huihuiSquare/digital-avatar-app/backend/tests/test_huihui_auth.py

"""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
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from database import Base
from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
from routers.huihui_auth import _issue_session, token_login
from services.boxim_client import BoxIMError


@pytest.fixture
def db():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()
    try:
        yield session
    finally:
        session.close()


def _add_avatar_data(db, owner_id: str, suffix: str = "1") -> Avatar:
    avatar = Avatar(id=f"avatar-{suffix}", owner_id=owner_id, name="冯医生")
    db.add_all(
        [
            avatar,
            TakeoverCursor(id=f"cursor-{suffix}", avatar_id=avatar.id, owner_id=owner_id),
            TakeoverMessage(
                id=f"message-{suffix}",
                avatar_id=avatar.id,
                owner_id=owner_id,
                boxim_message_id=f"box-{suffix}",
                peer_id="peer",
                direction="incoming",
                send_time=datetime(2026, 8, 20, 12, 0, 0),
            ),
            TakeoverReplyTask(
                id=f"task-{suffix}",
                avatar_id=avatar.id,
                owner_id=owner_id,
                peer_id="peer",
                trigger_message_id=f"trigger-{suffix}",
                scheduled_at=datetime(2026, 8, 20, 12, 0, 3),
                boxim_local_id=f"local-{suffix}",
            ),
        ]
    )
    db.commit()
    return avatar


def _assert_avatar_data_owner(db, avatar_id: str, owner_id: str):
    assert db.query(Avatar).filter_by(id=avatar_id).one().owner_id == owner_id
    assert db.query(TakeoverCursor).filter_by(avatar_id=avatar_id).one().owner_id == owner_id
    assert db.query(TakeoverMessage).filter_by(avatar_id=avatar_id).one().owner_id == owner_id
    assert db.query(TakeoverReplyTask).filter_by(avatar_id=avatar_id).one().owner_id == owner_id


def test_unique_phone_user_is_reused_when_huihui_id_changes(db):
    legacy = User(
        id="legacy-local",
        huihui_user_id="fat-user-id",
        phone="18500000000",
        app_token="old-session",
    )
    db.add(legacy)
    db.commit()
    avatar = _add_avatar_data(db, legacy.huihui_user_id)

    response = _issue_session(
        db,
        "18500000000",
        {"userId": "prod-user-id", "nickname": "用户", "token": "prod-token"},
    )

    users = db.query(User).all()
    assert len(users) == 1
    assert users[0].id == "legacy-local"
    assert users[0].huihui_user_id == "prod-user-id"
    assert response["data"]["token"] == users[0].app_token
    _assert_avatar_data_owner(db, avatar.id, "prod-user-id")


def test_existing_production_user_claims_one_legacy_phone_account(db):
    current = User(
        id="prod-local",
        huihui_user_id="prod-user-id",
        phone="18500000000",
    )
    legacy = User(
        id="legacy-local",
        huihui_user_id="fat-user-id",
        phone="18500000000",
        app_token="old-session",
        huihui_token="fat-token",
    )
    db.add_all([current, legacy])
    db.commit()
    avatar = _add_avatar_data(db, legacy.huihui_user_id)

    _issue_session(
        db,
        "18500000000",
        {"userId": "prod-user-id", "nickname": "用户", "token": "prod-token"},
    )

    db.refresh(legacy)
    assert legacy.app_token == ""
    assert legacy.huihui_token == ""
    _assert_avatar_data_owner(db, avatar.id, "prod-user-id")


def test_ambiguous_phone_matches_do_not_move_existing_avatars(db):
    first = User(id="first", huihui_user_id="fat-1", phone="18500000000")
    second = User(id="second", huihui_user_id="fat-2", phone="18500000000")
    db.add_all([first, second])
    db.commit()
    first_avatar = _add_avatar_data(db, first.huihui_user_id, "1")
    second_avatar = _add_avatar_data(db, second.huihui_user_id, "2")

    _issue_session(
        db,
        "18500000000",
        {"userId": "prod-user-id", "nickname": "用户", "token": "prod-token"},
    )

    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