261 lines
7.8 KiB
Python
261 lines
7.8 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import event
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.database import Base, get_db
|
|
from app.main import create_app
|
|
|
|
# Import all models so Base.metadata is populated
|
|
from app.models.activity import Activity
|
|
from app.models.activity_guest_choice import ActivityGuestChoice
|
|
from app.models.match import Match, MatchLike
|
|
from app.models.registration import Registration
|
|
from app.models.user import User
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Database fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_engine():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(db_engine):
|
|
"""Yield a session wrapped in a transaction that rolls back after each test."""
|
|
async with db_engine.connect() as conn:
|
|
txn = await conn.begin()
|
|
session = AsyncSession(bind=conn, expire_on_commit=False)
|
|
try:
|
|
yield session
|
|
finally:
|
|
await session.close()
|
|
await txn.rollback()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(db_session: AsyncSession):
|
|
"""Async test client with DB and auth dependencies overridden."""
|
|
app = create_app()
|
|
|
|
async def _override_get_db():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = _override_get_db
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def auth_headers(user_id: int) -> dict[str, str]:
|
|
"""Return a dummy Authorization header. The real dependency is overridden in tests."""
|
|
return {"Authorization": f"Bearer test-token-{user_id}"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# User fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest_asyncio.fixture
|
|
async def user_male_a(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=1,
|
|
openid="openid_male_a",
|
|
nickname="小王",
|
|
gender=1,
|
|
birth_year=1995,
|
|
city="广州市",
|
|
height=178,
|
|
education=4,
|
|
job_industry="IT",
|
|
job_company="科技公司",
|
|
income_range=3,
|
|
personality_tags=["阳光", "幽默"],
|
|
hobbies=["篮球", "电影", "旅行"],
|
|
prefer_age_min=1993,
|
|
prefer_age_max=2000,
|
|
prefer_education=3,
|
|
self_intro="热爱生活",
|
|
avatar_url="/static/uploads/avatar1.jpg",
|
|
avatar_blur_url="/static/uploads/avatar1_blur.jpg",
|
|
profile_images=["/static/uploads/p1.jpg"],
|
|
audit_status=2,
|
|
is_active=1,
|
|
subscribe_match=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
# value_answers is not a mapped column; set as plain attribute for scoring
|
|
user.value_answers = {"q1": "A", "q2": "B", "q3": "A", "q4": "C"}
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def user_female_b(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=2,
|
|
openid="openid_female_b",
|
|
nickname="小李",
|
|
gender=2,
|
|
birth_year=1997,
|
|
city="深圳市",
|
|
height=165,
|
|
education=4,
|
|
job_industry="金融",
|
|
job_company="银行",
|
|
income_range=2,
|
|
personality_tags=["温柔", "独立"],
|
|
hobbies=["阅读", "电影", "瑜伽"],
|
|
prefer_age_min=1990,
|
|
prefer_age_max=1998,
|
|
prefer_education=3,
|
|
self_intro="乐观开朗",
|
|
avatar_url="/static/uploads/avatar2.jpg",
|
|
avatar_blur_url="/static/uploads/avatar2_blur.jpg",
|
|
profile_images=["/static/uploads/p2.jpg"],
|
|
audit_status=2,
|
|
is_active=1,
|
|
subscribe_match=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
user.value_answers = {"q1": "A", "q2": "B", "q3": "B", "q4": "C"}
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def user_male_c(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=3,
|
|
openid="openid_male_c",
|
|
nickname="小张",
|
|
gender=1,
|
|
birth_year=1993,
|
|
city="广州市",
|
|
height=170,
|
|
education=3,
|
|
job_industry="教育",
|
|
job_company="学校",
|
|
income_range=2,
|
|
personality_tags=["稳重"],
|
|
hobbies=["篮球", "游戏"],
|
|
avatar_url="/static/uploads/avatar3.jpg",
|
|
audit_status=2,
|
|
is_active=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
user.value_answers = {"q1": "C", "q2": "A", "q3": "A", "q4": "D"}
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def user_female_d(db_session: AsyncSession) -> User:
|
|
user = User(
|
|
id=4,
|
|
openid="openid_female_d",
|
|
nickname="小陈",
|
|
gender=2,
|
|
birth_year=2000,
|
|
city="北京市",
|
|
height=160,
|
|
education=2,
|
|
avatar_url="/static/uploads/avatar4.jpg",
|
|
audit_status=2,
|
|
is_active=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
return user
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Activity fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest_asyncio.fixture
|
|
async def activity_open(db_session: AsyncSession) -> Activity:
|
|
activity = Activity(
|
|
id=1,
|
|
title="周末相亲活动",
|
|
description="轻松交友",
|
|
start_time=datetime.now() - timedelta(hours=1),
|
|
end_time=datetime.now() + timedelta(hours=5),
|
|
match_deadline=datetime.now() + timedelta(hours=48),
|
|
selection_limit=3,
|
|
capacity_male=10,
|
|
capacity_female=10,
|
|
is_published=1,
|
|
require_audit=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
return activity
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def activity_closed(db_session: AsyncSession) -> Activity:
|
|
activity = Activity(
|
|
id=2,
|
|
title="已结束活动",
|
|
start_time=datetime.now() - timedelta(days=3),
|
|
end_time=datetime.now() - timedelta(days=2),
|
|
match_deadline=datetime.now() - timedelta(days=1),
|
|
selection_limit=3,
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
return activity
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def activity_before_start(db_session: AsyncSession) -> Activity:
|
|
activity = Activity(
|
|
id=3,
|
|
title="未开始活动",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
match_deadline=datetime.now() + timedelta(days=3),
|
|
selection_limit=5,
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
return activity
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registration helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def register_user(db_session: AsyncSession, user: User, activity: Activity, status: int = 1) -> Registration:
|
|
reg = Registration(user_id=user.id, activity_id=activity.id, status=status)
|
|
db_session.add(reg)
|
|
await db_session.flush()
|
|
return reg
|