355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""Integration tests for global match service (match_service.py).
|
|
|
|
Tests cover: get_candidates, like_user, list_my_matches,
|
|
get_match_detail, get_public_user_info, ai_suggest_candidates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.match import Match, MatchLike
|
|
from app.models.user import User
|
|
from app.services.match_service import (
|
|
ai_suggest_candidates,
|
|
get_candidates,
|
|
get_match_detail,
|
|
get_public_user_info,
|
|
like_user,
|
|
list_my_matches,
|
|
)
|
|
|
|
from tests.conftest import register_user
|
|
|
|
|
|
# Patch sanitize_user_media_payload to be a no-op (skip filesystem checks)
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_sanitize():
|
|
with patch(
|
|
"app.services.match_service.sanitize_user_media_payload",
|
|
side_effect=lambda d: d,
|
|
):
|
|
yield
|
|
|
|
|
|
# ============================================================================
|
|
# get_candidates
|
|
# ============================================================================
|
|
|
|
class TestGetCandidates:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_opposite_gender(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User, user_female_d: User,
|
|
):
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id in user_ids
|
|
assert user_female_d.id in user_ids
|
|
assert user_male_a.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excludes_liked_users(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
like = MatchLike(
|
|
from_user_id=user_male_a.id,
|
|
to_user_id=user_female_b.id,
|
|
source="manual",
|
|
created_at=datetime.now(),
|
|
)
|
|
db_session.add(like)
|
|
await db_session.flush()
|
|
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excludes_matched_users(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
match = Match(
|
|
user_a_id=min(user_male_a.id, user_female_b.id),
|
|
user_b_id=max(user_male_a.id, user_female_b.id),
|
|
match_score=80.0,
|
|
match_type="manual",
|
|
matched_at=datetime.now(),
|
|
)
|
|
db_session.add(match)
|
|
await db_session.flush()
|
|
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_level1_info(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
assert len(items) >= 1
|
|
item = next(i for i in items if i["user_id"] == user_female_b.id)
|
|
# Level-1 fields present
|
|
assert "nickname" in item
|
|
assert "birth_year_range" in item
|
|
assert "city" in item
|
|
assert "match_score" in item
|
|
# Level-2/3 fields should NOT be present
|
|
assert "avatar_url" not in item
|
|
assert "job_industry" not in item
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_candidates_when_all_same_gender(
|
|
self, db_session: AsyncSession, user_male_a: User, user_male_c: User,
|
|
):
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_male_c.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excludes_inactive_users(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
user_female_b.is_active = 0
|
|
db_session.add(user_female_b)
|
|
await db_session.flush()
|
|
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id not in user_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_excludes_unaudited_users(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
user_female_b.audit_status = 0
|
|
db_session.add(user_female_b)
|
|
await db_session.flush()
|
|
|
|
items = await get_candidates(db_session, user_male_a, "manual")
|
|
user_ids = [item["user_id"] for item in items]
|
|
assert user_female_b.id not in user_ids
|
|
|
|
|
|
# ============================================================================
|
|
# like_user
|
|
# ============================================================================
|
|
|
|
class TestLikeUser:
|
|
@pytest.mark.asyncio
|
|
async def test_one_way_like(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
assert result["is_mutual"] is False
|
|
assert result["match_id"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_like(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
# B likes A first
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
# A likes B → mutual
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
assert result["is_mutual"] is True
|
|
assert result["match_id"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_like_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
with pytest.raises(ValueError, match="不能对自己"):
|
|
await like_user(db_session, user_male_a, user_male_a, "manual")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_gender_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User, user_male_c: User,
|
|
):
|
|
with pytest.raises(ValueError, match="异性"):
|
|
await like_user(db_session, user_male_a, user_male_c, "manual")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_like_idempotent(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
assert result["is_mutual"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_like_creates_match_like_record(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
from sqlalchemy import select
|
|
result = await db_session.execute(
|
|
select(MatchLike).where(
|
|
MatchLike.from_user_id == user_male_a.id,
|
|
MatchLike.to_user_id == user_female_b.id,
|
|
)
|
|
)
|
|
like = result.scalar_one_or_none()
|
|
assert like is not None
|
|
assert like.source == "manual"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_creates_match_record(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
match_id = result["match_id"]
|
|
from sqlalchemy import select
|
|
match = await db_session.get(Match, match_id)
|
|
assert match is not None
|
|
assert {match.user_a_id, match.user_b_id} == {user_male_a.id, user_female_b.id}
|
|
assert match.match_score is not None
|
|
|
|
|
|
# ============================================================================
|
|
# list_my_matches
|
|
# ============================================================================
|
|
|
|
class TestListMyMatches:
|
|
@pytest.mark.asyncio
|
|
async def test_empty_list(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
items = await list_my_matches(db_session, user_male_a)
|
|
assert items == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_like(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
items = await list_my_matches(db_session, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["match_status"] == "pending"
|
|
assert items[0]["other_user"]["user_id"] == user_female_b.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_match(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
items = await list_my_matches(db_session, user_male_a)
|
|
assert len(items) == 1
|
|
assert items[0]["match_status"] == "success"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sorted_by_time_desc(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User, user_female_d: User,
|
|
):
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
await like_user(db_session, user_male_a, user_female_d, "manual")
|
|
items = await list_my_matches(db_session, user_male_a)
|
|
assert len(items) == 2
|
|
# Most recent first
|
|
assert items[0]["matched_at"] >= items[1]["matched_at"]
|
|
|
|
|
|
# ============================================================================
|
|
# get_match_detail
|
|
# ============================================================================
|
|
|
|
class TestGetMatchDetail:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_level3_info(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
match_id = result["match_id"]
|
|
|
|
detail = await get_match_detail(db_session, user_male_a, match_id)
|
|
assert detail is not None
|
|
other = detail["other_user"]
|
|
# Level-3 fields present
|
|
assert "job_industry" in other
|
|
assert "self_intro" in other
|
|
assert "income_range" in other
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_participant_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User, user_male_c: User,
|
|
):
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
result = await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
match_id = result["match_id"]
|
|
|
|
with pytest.raises(PermissionError, match="无权限"):
|
|
await get_match_detail(db_session, user_male_c, match_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonexistent_match(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
detail = await get_match_detail(db_session, user_male_a, 99999)
|
|
assert detail is None
|
|
|
|
|
|
# ============================================================================
|
|
# get_public_user_info
|
|
# ============================================================================
|
|
|
|
class TestGetPublicUserInfo:
|
|
@pytest.mark.asyncio
|
|
async def test_stranger_gets_level1(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
info = await get_public_user_info(db_session, user_male_a, user_female_b)
|
|
assert "nickname" in info
|
|
assert "birth_year_range" in info
|
|
assert "job_industry" not in info
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_liked_user_gets_level2(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
info = await get_public_user_info(db_session, user_male_a, user_female_b)
|
|
assert "avatar_url" in info
|
|
assert "education" in info
|
|
assert "hobbies" in info
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutual_match_gets_level3(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User,
|
|
):
|
|
await like_user(db_session, user_female_b, user_male_a, "manual")
|
|
await like_user(db_session, user_male_a, user_female_b, "manual")
|
|
info = await get_public_user_info(db_session, user_male_a, user_female_b)
|
|
assert "job_industry" in info
|
|
assert "self_intro" in info
|
|
|
|
|
|
# ============================================================================
|
|
# ai_suggest_candidates
|
|
# ============================================================================
|
|
|
|
class TestAiSuggestCandidates:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_sorted_by_score(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User, user_female_d: User,
|
|
):
|
|
items = await ai_suggest_candidates(db_session, user_male_a, limit=5)
|
|
if len(items) >= 2:
|
|
scores = [item.get("match_score", 0) or 0 for item in items]
|
|
# First items should have higher or equal scores
|
|
assert scores[0] >= scores[-1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respects_limit(
|
|
self, db_session: AsyncSession, user_male_a: User, user_female_b: User, user_female_d: User,
|
|
):
|
|
items = await ai_suggest_candidates(db_session, user_male_a, limit=1)
|
|
assert len(items) <= 1
|