feat(avatar): support production H5 token SSO
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
"""Tests for preserving local avatar ownership when Huihui IDs change."""
|
||||
|
||||
from datetime import datetime
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
@@ -9,7 +10,8 @@ from sqlalchemy.pool import StaticPool
|
||||
|
||||
from database import Base
|
||||
from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
|
||||
from routers.huihui_auth import _issue_session
|
||||
from routers.huihui_auth import _issue_session, token_login
|
||||
from services.boxim_client import BoxIMError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -135,3 +137,55 @@ def test_ambiguous_phone_matches_do_not_move_existing_avatars(db):
|
||||
assert db.query(User).count() == 3
|
||||
_assert_avatar_data_owner(db, first_avatar.id, "fat-1")
|
||||
_assert_avatar_data_owner(db, second_avatar.id, "fat-2")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_login_uses_huihui_user_id_and_keeps_upstream_token_server_side(db):
|
||||
existing = User(
|
||||
id="existing-local",
|
||||
huihui_user_id="huihui-user-88",
|
||||
app_token="existing-app-session",
|
||||
)
|
||||
db.add(existing)
|
||||
db.commit()
|
||||
|
||||
client = AsyncMock()
|
||||
client.exchange_access_token.return_value = {"accessToken": "boxim-token"}
|
||||
client.get_self.return_value = {
|
||||
"id": 998877,
|
||||
"huihuiUserId": "huihui-user-88",
|
||||
"nickName": "会会用户",
|
||||
"headImage": "https://cdn.example/avatar.jpg",
|
||||
}
|
||||
with patch("routers.huihui_auth._cfg_ready", return_value=True), patch(
|
||||
"routers.huihui_auth._create_boxim_client", return_value=client
|
||||
):
|
||||
response = await token_login({"token": "production-huihui-token"}, db)
|
||||
|
||||
assert response["code"] == 200
|
||||
assert response["data"]["token"] == "existing-app-session"
|
||||
assert "token" not in response["data"]["huihui"]
|
||||
user = db.query(User).one()
|
||||
assert user.huihui_user_id == "huihui-user-88"
|
||||
assert user.huihui_user_id != "998877"
|
||||
assert user.huihui_token == "production-huihui-token"
|
||||
assert user.nickname == "会会用户"
|
||||
assert user.avatar_url == "https://cdn.example/avatar.jpg"
|
||||
client.exchange_access_token.assert_awaited_once_with("production-huihui-token")
|
||||
client.get_self.assert_awaited_once_with("boxim-token")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_login_rejects_expired_huihui_token_without_creating_user(db):
|
||||
client = AsyncMock()
|
||||
client.exchange_access_token.side_effect = BoxIMError(
|
||||
"expired", auth_error=True
|
||||
)
|
||||
with patch("routers.huihui_auth._cfg_ready", return_value=True), patch(
|
||||
"routers.huihui_auth._create_boxim_client", return_value=client
|
||||
):
|
||||
response = await token_login({"token": "expired-token"}, db)
|
||||
|
||||
assert response["code"] == 401
|
||||
assert response["message"] == "会会登录凭证无效或已过期"
|
||||
assert db.query(User).count() == 0
|
||||
|
||||
Reference in New Issue
Block a user