132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
from database import init_db, SessionLocal
|
|
from models import (
|
|
Authorization,
|
|
Avatar,
|
|
ChatAttachment,
|
|
TakeoverCursor,
|
|
TakeoverMessage,
|
|
TakeoverReplyTask,
|
|
TokenAccount,
|
|
TokenPaymentOrder,
|
|
TokenUsage,
|
|
User,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def setup_database():
|
|
"""Initialize DB tables and seed an Authorization row."""
|
|
init_db()
|
|
db = SessionLocal()
|
|
try:
|
|
existing = db.query(Authorization).first()
|
|
if existing is None:
|
|
auth = Authorization(
|
|
id="test-auth-1",
|
|
avatar_id="test-avatar-1",
|
|
target_type="user",
|
|
target_id="test-user-1",
|
|
target_name="Test User",
|
|
permissions=["read", "write"],
|
|
status="active",
|
|
)
|
|
db.add(auth)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def authorization_context():
|
|
"""Create isolated users, avatars, and one authorization for API tests."""
|
|
suffix = uuid.uuid4().hex
|
|
owner = User(
|
|
id=f"owner-{suffix}",
|
|
huihui_user_id=f"huihui-owner-{suffix}",
|
|
nickname="授权测试用户",
|
|
app_token=f"owner-token-{suffix}",
|
|
)
|
|
other = User(
|
|
id=f"other-{suffix}",
|
|
huihui_user_id=f"huihui-other-{suffix}",
|
|
nickname="其他用户",
|
|
app_token=f"other-token-{suffix}",
|
|
)
|
|
avatar = Avatar(
|
|
id=f"avatar-{suffix}",
|
|
owner_id=owner.huihui_user_id,
|
|
name="授权测试分身",
|
|
status="active",
|
|
config={},
|
|
)
|
|
other_avatar = Avatar(
|
|
id=f"other-avatar-{suffix}",
|
|
owner_id=other.huihui_user_id,
|
|
name="其他分身",
|
|
status="active",
|
|
config={},
|
|
)
|
|
authorization = Authorization(
|
|
id=f"authorization-{suffix}",
|
|
avatar_id=avatar.id,
|
|
target_type="user",
|
|
target_id=f"contact-{suffix}",
|
|
target_name="测试联系人",
|
|
permissions=["chat", "browse"],
|
|
status="active",
|
|
)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
db.add_all([owner, other, avatar, other_avatar, authorization])
|
|
db.commit()
|
|
yield {
|
|
"owner": owner,
|
|
"other": other,
|
|
"avatar": avatar,
|
|
"other_avatar": other_avatar,
|
|
"authorization": authorization,
|
|
"owner_headers": {"Authorization": f"Bearer {owner.app_token}"},
|
|
"other_headers": {"Authorization": f"Bearer {other.app_token}"},
|
|
"suffix": suffix,
|
|
}
|
|
finally:
|
|
db.rollback()
|
|
avatar_ids = [avatar.id, other_avatar.id]
|
|
db.query(ChatAttachment).filter(
|
|
ChatAttachment.avatar_id.in_(avatar_ids)
|
|
).delete(synchronize_session=False)
|
|
db.query(TakeoverReplyTask).filter(
|
|
TakeoverReplyTask.avatar_id.in_(avatar_ids)
|
|
).delete(synchronize_session=False)
|
|
db.query(TakeoverMessage).filter(
|
|
TakeoverMessage.avatar_id.in_(avatar_ids)
|
|
).delete(synchronize_session=False)
|
|
db.query(TakeoverCursor).filter(
|
|
TakeoverCursor.avatar_id.in_(avatar_ids)
|
|
).delete(synchronize_session=False)
|
|
db.query(Authorization).filter(
|
|
Authorization.avatar_id.in_(avatar_ids)
|
|
).delete(synchronize_session=False)
|
|
db.query(Avatar).filter(Avatar.id.in_(avatar_ids)).delete(
|
|
synchronize_session=False
|
|
)
|
|
user_ids = [owner.id, other.id]
|
|
db.query(TokenPaymentOrder).filter(TokenPaymentOrder.user_id.in_(user_ids)).delete(
|
|
synchronize_session=False
|
|
)
|
|
db.query(TokenUsage).filter(TokenUsage.user_id.in_(user_ids)).delete(
|
|
synchronize_session=False
|
|
)
|
|
db.query(TokenAccount).filter(TokenAccount.user_id.in_(user_ids)).delete(
|
|
synchronize_session=False
|
|
)
|
|
db.query(User).filter(User.id.in_([owner.id, other.id])).delete(
|
|
synchronize_session=False
|
|
)
|
|
db.commit()
|
|
db.close()
|