617 lines
22 KiB
Python
617 lines
22 KiB
Python
"""Integration tests for activity match service (activity_match_service.py).
|
|
|
|
Tests cover: resolve_activity_stage, create_activity_choice, cancel_activity_choice,
|
|
list_activity_candidates, list_my_activity_choices, list_activity_liked_me,
|
|
list_activity_mutual_matches, get_activity_match_detail.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.activity import Activity
|
|
from app.models.activity_guest_choice import ActivityGuestChoice
|
|
from app.models.registration import Registration
|
|
from app.models.user import User
|
|
from app.services.activity_match_service import (
|
|
cancel_activity_choice,
|
|
create_activity_choice,
|
|
get_activity_match_detail,
|
|
list_activity_candidates,
|
|
list_activity_liked_me,
|
|
list_activity_mutual_matches,
|
|
list_my_activity_choices,
|
|
resolve_activity_stage,
|
|
)
|
|
|
|
from tests.conftest import register_user
|
|
|
|
|
|
# Patch sanitize_user_media_payload to skip filesystem checks
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_sanitize():
|
|
with patch(
|
|
"app.services.activity_match_service.sanitize_user_media_payload",
|
|
side_effect=lambda d: d,
|
|
):
|
|
yield
|
|
|
|
|
|
# ============================================================================
|
|
# resolve_activity_stage
|
|
# ============================================================================
|
|
|
|
class TestResolveActivityStage:
|
|
def test_before_start(self):
|
|
activity = Activity(
|
|
start_time=datetime.now() + timedelta(hours=1),
|
|
end_time=datetime.now() + timedelta(hours=5),
|
|
match_deadline=datetime.now() + timedelta(hours=48),
|
|
)
|
|
assert resolve_activity_stage(activity) == "before_start"
|
|
|
|
def test_matching_open(self):
|
|
activity = Activity(
|
|
start_time=datetime.now() - timedelta(hours=1),
|
|
end_time=datetime.now() + timedelta(hours=5),
|
|
match_deadline=datetime.now() + timedelta(hours=48),
|
|
)
|
|
assert resolve_activity_stage(activity) == "matching_open"
|
|
|
|
def test_matching_closed(self):
|
|
activity = Activity(
|
|
start_time=datetime.now() - timedelta(hours=1),
|
|
end_time=datetime.now() + timedelta(hours=5),
|
|
match_deadline=datetime.now() - timedelta(hours=1),
|
|
)
|
|
assert resolve_activity_stage(activity) == "matching_closed"
|
|
|
|
def test_ended(self):
|
|
activity = Activity(
|
|
start_time=datetime.now() - timedelta(days=2),
|
|
end_time=datetime.now() - timedelta(days=1),
|
|
match_deadline=datetime.now() - timedelta(hours=12),
|
|
)
|
|
assert resolve_activity_stage(activity) == "ended"
|
|
|
|
def test_no_deadline_still_open(self):
|
|
activity = Activity(
|
|
start_time=datetime.now() - timedelta(hours=1),
|
|
end_time=datetime.now() + timedelta(hours=5),
|
|
match_deadline=None,
|
|
)
|
|
assert resolve_activity_stage(activity) == "matching_open"
|
|
|
|
|
|
# ============================================================================
|
|
# list_activity_candidates
|
|
# ============================================================================
|
|
|
|
class TestListActivityCandidates:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_opposite_gender(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
items = await list_activity_candidates(db_session, activity_open, user_male_a)
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id in user_ids
|
|
assert user_male_a.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excludes_unregistered_users(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
# B is NOT registered
|
|
items = await list_activity_candidates(db_session, activity_open, user_male_a)
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_before_start_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
activity_before_start: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_before_start)
|
|
with pytest.raises(PermissionError, match="活动开始前"):
|
|
await list_activity_candidates(db_session, activity_before_start, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unregistered_user_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
# A is NOT registered
|
|
with pytest.raises(PermissionError, match="请先绑定"):
|
|
await list_activity_candidates(db_session, activity_open, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_selected_flag(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
# A selects B
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
|
|
items = await list_activity_candidates(db_session, activity_open, user_male_a)
|
|
b_item = next(i for i in items if i["user_id"] == user_female_b.id)
|
|
assert b_item["selected"] is True
|
|
|
|
|
|
# ============================================================================
|
|
# create_activity_choice
|
|
# ============================================================================
|
|
|
|
class TestCreateActivityChoice:
|
|
@pytest.mark.asyncio
|
|
async def test_basic_selection(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
result = await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
assert result["selected_count"] == 1
|
|
assert result["remaining_count"] == 2
|
|
assert result["is_mutual"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_selection(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
result = await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
assert result["is_mutual"] is True
|
|
assert result["match_user_id"] == user_male_a.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_selection_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
with pytest.raises(ValueError, match="不能选择自己"):
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_gender_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_male_c: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_male_c, activity_open)
|
|
with pytest.raises(ValueError, match="异性"):
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_male_c)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unregistered_target_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
# B is NOT registered
|
|
with pytest.raises(PermissionError, match="请先绑定"):
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_selection_limit_enforced(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
user_female_d: User,
|
|
activity_open: Activity,
|
|
):
|
|
# activity_open has selection_limit=3
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
await register_user(db_session, user_female_d, activity_open)
|
|
|
|
# Create a third female user for the limit test
|
|
user_female_e = User(
|
|
id=5,
|
|
openid="openid_female_e",
|
|
nickname="小周",
|
|
gender=2,
|
|
birth_year=1996,
|
|
audit_status=2,
|
|
is_active=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
db_session.add(user_female_e)
|
|
await db_session.flush()
|
|
await register_user(db_session, user_female_e, activity_open)
|
|
|
|
user_female_f = User(
|
|
id=6,
|
|
openid="openid_female_f",
|
|
nickname="小吴",
|
|
gender=2,
|
|
birth_year=1998,
|
|
audit_status=2,
|
|
is_active=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
db_session.add(user_female_f)
|
|
await db_session.flush()
|
|
await register_user(db_session, user_female_f, activity_open)
|
|
|
|
# Select 3 users (the limit)
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_d)
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_e)
|
|
|
|
# 4th should fail
|
|
with pytest.raises(ValueError, match="可选人数已达上限"):
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_f)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_selection_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
with pytest.raises(ValueError, match="已选择该嘉宾"):
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closed_stage_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_closed: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_closed)
|
|
await register_user(db_session, user_female_b, activity_closed)
|
|
|
|
with pytest.raises(ValueError, match="已锁定"):
|
|
await create_activity_choice(db_session, activity_closed, user_male_a, user_female_b)
|
|
|
|
|
|
# ============================================================================
|
|
# cancel_activity_choice
|
|
# ============================================================================
|
|
|
|
class TestCancelActivityChoice:
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_choice(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
result = await cancel_activity_choice(db_session, activity_open, user_male_a, user_female_b.id)
|
|
assert result["selected_count"] == 0
|
|
assert result["remaining_count"] == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_nonexistent_choice(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
with pytest.raises(ValueError, match="未找到"):
|
|
await cancel_activity_choice(db_session, activity_open, user_male_a, user_female_b.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_and_reselect(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
await cancel_activity_choice(db_session, activity_open, user_male_a, user_female_b.id)
|
|
|
|
# Re-select should work
|
|
result = await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
assert result["selected_count"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_on_closed_stage_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_closed: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_closed)
|
|
await register_user(db_session, user_female_b, activity_closed)
|
|
|
|
# Add a choice directly to DB for testing cancel on closed stage
|
|
choice = ActivityGuestChoice(
|
|
activity_id=activity_closed.id,
|
|
from_user_id=user_male_a.id,
|
|
to_user_id=user_female_b.id,
|
|
status=1,
|
|
)
|
|
db_session.add(choice)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ValueError, match="已锁定"):
|
|
await cancel_activity_choice(db_session, activity_closed, user_male_a, user_female_b.id)
|
|
|
|
|
|
# ============================================================================
|
|
# list_my_activity_choices
|
|
# ============================================================================
|
|
|
|
class TestListMyActivityChoices:
|
|
@pytest.mark.asyncio
|
|
async def test_lists_selected_users(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
items = await list_my_activity_choices(db_session, activity_open, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["user_id"] == user_female_b.id
|
|
assert items[0]["match_status"] == "pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_mutual_status(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
|
|
items = await list_my_activity_choices(db_session, activity_open, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["match_status"] == "success"
|
|
assert items[0]["can_view_detail"] is True
|
|
|
|
|
|
# ============================================================================
|
|
# list_activity_liked_me
|
|
# ============================================================================
|
|
|
|
class TestListActivityLikedMe:
|
|
@pytest.mark.asyncio
|
|
async def test_lists_who_chose_me(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
items = await list_activity_liked_me(db_session, activity_open, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["user_id"] == user_female_b.id
|
|
assert items[0]["match_status"] == "pending"
|
|
|
|
|
|
# ============================================================================
|
|
# list_activity_mutual_matches
|
|
# ============================================================================
|
|
|
|
class TestListActivityMutualMatches:
|
|
@pytest.mark.asyncio
|
|
async def test_only_mutual(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
user_female_d: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
await register_user(db_session, user_female_d, activity_open)
|
|
|
|
# Mutual with B
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
|
|
# One-way with D
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_d)
|
|
|
|
items = await list_activity_mutual_matches(db_session, activity_open, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["user_id"] == user_female_b.id
|
|
assert items[0]["match_status"] == "success"
|
|
|
|
|
|
# ============================================================================
|
|
# get_activity_match_detail
|
|
# ============================================================================
|
|
|
|
class TestGetActivityMatchDetail:
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_can_view_detail(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
|
|
detail = await get_activity_match_detail(
|
|
db_session, activity_open, user_male_a, user_female_b.id
|
|
)
|
|
assert detail is not None
|
|
assert detail["target_user_id"] == user_female_b.id
|
|
other = detail["other_user"]
|
|
# Level-3 fields
|
|
assert "job_industry" in other
|
|
assert "self_intro" in other
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_mutual_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
# Only A selects B, no mutual
|
|
await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
|
|
with pytest.raises(PermissionError, match="互选成功"):
|
|
await get_activity_match_detail(
|
|
db_session, activity_open, user_male_a, user_female_b.id
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unregistered_user_rejected(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
# A is NOT registered
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
with pytest.raises(PermissionError, match="请先绑定"):
|
|
await get_activity_match_detail(
|
|
db_session, activity_open, user_male_a, user_female_b.id
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Full flow: register → select → mutual → view detail
|
|
# ============================================================================
|
|
|
|
class TestFullActivityMatchFlow:
|
|
@pytest.mark.asyncio
|
|
async def test_end_to_end(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
# Step 1: Both register
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
await register_user(db_session, user_female_b, activity_open)
|
|
|
|
# Step 2: View candidates
|
|
candidates = await list_activity_candidates(db_session, activity_open, user_male_a)
|
|
assert len(candidates) == 1
|
|
assert candidates[0]["user_id"] == user_female_b.id
|
|
assert candidates[0]["selected"] is False
|
|
|
|
# Step 3: A selects B
|
|
result = await create_activity_choice(db_session, activity_open, user_male_a, user_female_b)
|
|
assert result["is_mutual"] is False
|
|
assert result["selected_count"] == 1
|
|
|
|
# Step 4: Check my-choices
|
|
my_choices = await list_my_activity_choices(db_session, activity_open, user_male_a)
|
|
assert len(my_choices) == 1
|
|
assert my_choices[0]["match_status"] == "pending"
|
|
|
|
# Step 5: Check liked-me for B
|
|
liked_me = await list_activity_liked_me(db_session, activity_open, user_female_b)
|
|
assert len(liked_me) == 1
|
|
assert liked_me[0]["user_id"] == user_male_a.id
|
|
|
|
# Step 6: B selects A → mutual
|
|
result = await create_activity_choice(db_session, activity_open, user_female_b, user_male_a)
|
|
assert result["is_mutual"] is True
|
|
|
|
# Step 7: Mutual matches list
|
|
mutual = await list_activity_mutual_matches(db_session, activity_open, user_male_a)
|
|
assert len(mutual) == 1
|
|
assert mutual[0]["match_status"] == "success"
|
|
|
|
# Step 8: View match detail
|
|
detail = await get_activity_match_detail(
|
|
db_session, activity_open, user_male_a, user_female_b.id
|
|
)
|
|
assert detail is not None
|
|
assert detail["other_user"]["nickname"] == "小李"
|