138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
"""Tests for preserving local avatar ownership when Huihui IDs change."""
|
|
|
|
from datetime import datetime
|
|
|
|
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
|
|
|
|
|
|
@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")
|