415 lines
15 KiB
Python
415 lines
15 KiB
Python
"""Tests for activity registration flow and the registration → matching transition.
|
|
|
|
Covers: register_activity, cancel_registration, can_user_register,
|
|
and the end-to-end flow from registration through entering the match phase.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.activity import Activity
|
|
from app.models.registration import Registration
|
|
from app.models.user import User
|
|
from app.services.activity_match_service import (
|
|
create_activity_choice,
|
|
get_activity_match_state,
|
|
list_activity_candidates,
|
|
)
|
|
from app.services.activity_service import (
|
|
cancel_registration,
|
|
can_user_register,
|
|
register_activity,
|
|
)
|
|
|
|
from tests.conftest import register_user
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_media():
|
|
with patch("app.services.activity_service.sanitize_media_url", side_effect=lambda d: d), \
|
|
patch("app.services.activity_match_service.sanitize_user_media_payload", side_effect=lambda d: d), \
|
|
patch("app.services.activity_service.ensure_activity_share_token", side_effect=lambda s, a: a):
|
|
yield
|
|
|
|
|
|
# ============================================================================
|
|
# register_activity
|
|
# ============================================================================
|
|
|
|
class TestRegisterActivity:
|
|
@pytest.mark.asyncio
|
|
async def test_basic_registration(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
reg = await register_activity(db_session, activity_open, user_male_a)
|
|
assert reg.user_id == user_male_a.id
|
|
assert reg.activity_id == activity_open.id
|
|
assert reg.status == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotent_registration(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
reg1 = await register_activity(db_session, activity_open, user_male_a)
|
|
reg2 = await register_activity(db_session, activity_open, user_male_a)
|
|
assert reg1.id == reg2.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unpublished_activity_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
activity = Activity(
|
|
title="未发布活动",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
is_published=0,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ValueError, match="未发布"):
|
|
await register_activity(db_session, activity, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signup_deadline_passed_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
activity = Activity(
|
|
title="已截止活动",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
signup_deadline=datetime.now() - timedelta(hours=1),
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ValueError, match="报名已截止"):
|
|
await register_activity(db_session, activity, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_required_not_audited_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
user_male_a.audit_status = 0
|
|
db_session.add(user_male_a)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ValueError, match="审核"):
|
|
await register_activity(db_session, activity_open, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_gender_rejected(
|
|
self, db_session: AsyncSession, activity_open: Activity,
|
|
):
|
|
user = User(
|
|
openid="no_gender_user",
|
|
nickname="无性别",
|
|
gender=None,
|
|
audit_status=2,
|
|
is_active=1,
|
|
updated_at=datetime.now(),
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
|
|
with pytest.raises(ValueError, match="性别"):
|
|
await register_activity(db_session, activity_open, user)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_male_capacity_full_rejected(
|
|
self, db_session: AsyncSession, user_male_a: User, user_male_c: User,
|
|
):
|
|
activity = Activity(
|
|
title="男性满员活动",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
capacity_male=1,
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
await register_activity(db_session, activity, user_male_a)
|
|
with pytest.raises(ValueError, match="男性名额已满"):
|
|
await register_activity(db_session, activity, user_male_c)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_female_capacity_full_rejected(
|
|
self, db_session: AsyncSession, user_female_b: User, user_female_d: User,
|
|
):
|
|
activity = Activity(
|
|
title="女性满员活动",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
capacity_female=1,
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
await register_activity(db_session, activity, user_female_b)
|
|
with pytest.raises(ValueError, match="女性名额已满"):
|
|
await register_activity(db_session, activity, user_female_d)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_user_can_re_register(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
"""Cancelled (status=3) users can re-register — this is the expected behavior."""
|
|
# Register, then cancel via direct status manipulation
|
|
reg = await register_activity(db_session, activity_open, user_male_a)
|
|
reg.status = 3
|
|
db_session.add(reg)
|
|
await db_session.flush()
|
|
|
|
# Re-register
|
|
reg2 = await register_activity(db_session, activity_open, user_male_a)
|
|
assert reg2.status == 1
|
|
assert reg2.id == reg.id # same record reactivated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_user_re_register_respects_capacity(
|
|
self, db_session: AsyncSession, user_male_a: User, user_male_c: User,
|
|
):
|
|
"""Cancelled user re-registration is blocked when capacity is full.
|
|
The capacity check runs BEFORE the re-activation logic."""
|
|
activity = Activity(
|
|
title="容量测试",
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
capacity_male=1,
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
# A registers, then cancels
|
|
reg_a = await register_activity(db_session, activity, user_male_a)
|
|
reg_a.status = 3
|
|
db_session.add(reg_a)
|
|
await db_session.flush()
|
|
|
|
# C takes the slot
|
|
await register_activity(db_session, activity, user_male_c)
|
|
|
|
# A tries to re-register — capacity is full → rejected
|
|
with pytest.raises(ValueError, match="男性名额已满"):
|
|
await register_activity(db_session, activity, user_male_a)
|
|
|
|
|
|
# ============================================================================
|
|
# cancel_registration
|
|
# ============================================================================
|
|
|
|
class TestCancelRegistration:
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_success(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
activity = Activity(
|
|
title="可取消活动",
|
|
start_time=datetime.now() + timedelta(days=3),
|
|
end_time=datetime.now() + timedelta(days=4),
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
await register_activity(db_session, activity, user_male_a)
|
|
result = await cancel_registration(db_session, activity, user_male_a)
|
|
assert result.status == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_within_24h_blocked(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
activity = Activity(
|
|
title="即将开始",
|
|
start_time=datetime.now() + timedelta(hours=12),
|
|
end_time=datetime.now() + timedelta(days=1),
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
await register_activity(db_session, activity, user_male_a)
|
|
with pytest.raises(ValueError, match="24小时"):
|
|
await cancel_registration(db_session, activity, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_no_registration(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
with pytest.raises(ValueError, match="未找到"):
|
|
await cancel_registration(db_session, activity_open, user_male_a)
|
|
|
|
|
|
# ============================================================================
|
|
# can_user_register
|
|
# ============================================================================
|
|
|
|
class TestCanUserRegister:
|
|
@pytest.mark.asyncio
|
|
async def test_can_register_normal(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
assert await can_user_register(db_session, activity_open, user_male_a) is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_already_registered_returns_true(
|
|
self, db_session: AsyncSession, user_male_a: User, activity_open: Activity,
|
|
):
|
|
await register_activity(db_session, activity_open, user_male_a)
|
|
assert await can_user_register(db_session, activity_open, user_male_a) is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_none_user_returns_false(
|
|
self, db_session: AsyncSession, activity_open: Activity,
|
|
):
|
|
assert await can_user_register(db_session, activity_open, None) is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unpublished_returns_false(
|
|
self, db_session: AsyncSession, user_male_a: User,
|
|
):
|
|
activity = Activity(
|
|
title="未发布", is_published=0,
|
|
start_time=datetime.now() + timedelta(days=1),
|
|
end_time=datetime.now() + timedelta(days=2),
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
assert await can_user_register(db_session, activity, user_male_a) is False
|
|
|
|
|
|
# ============================================================================
|
|
# Registration → Matching transition (end-to-end)
|
|
# ============================================================================
|
|
|
|
class TestRegistrationToMatchingFlow:
|
|
@pytest.mark.asyncio
|
|
async def test_register_then_enter_match(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_open: Activity,
|
|
):
|
|
"""Full flow: register → check match state → view candidates → select."""
|
|
# Step 1: Both register
|
|
await register_activity(db_session, activity_open, user_male_a)
|
|
await register_activity(db_session, activity_open, user_female_b)
|
|
|
|
# Step 2: Check match state
|
|
state = await get_activity_match_state(db_session, activity_open, user_male_a)
|
|
assert state["is_bound"] is True
|
|
assert state["can_choose"] is True
|
|
assert state["stage"] == "matching_open"
|
|
assert state["selected_count"] == 0
|
|
assert state["remaining_count"] == 3
|
|
|
|
# Step 3: 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
|
|
|
|
# Step 4: Select
|
|
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
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unregistered_user_cannot_enter_match(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
activity_open: Activity,
|
|
):
|
|
"""Without registration, match state returns PermissionError."""
|
|
with pytest.raises(PermissionError, match="请先绑定"):
|
|
await get_activity_match_state(db_session, activity_open, user_male_a)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_before_start_cannot_view_candidates(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
activity_before_start: Activity,
|
|
):
|
|
"""Registered but activity hasn't started → cannot view candidates."""
|
|
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_matching_open_can_choose(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
activity_open: Activity,
|
|
):
|
|
"""During matching_open, can_choose is True."""
|
|
await register_user(db_session, user_male_a, activity_open)
|
|
state = await get_activity_match_state(db_session, activity_open, user_male_a)
|
|
assert state["can_choose"] is True
|
|
assert state["can_modify_choice"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_matching_closed_cannot_choose(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_closed: Activity,
|
|
):
|
|
"""After match_deadline, can_choose is False but can still view."""
|
|
await register_user(db_session, user_male_a, activity_closed)
|
|
state = await get_activity_match_state(db_session, activity_closed, user_male_a)
|
|
assert state["can_choose"] is False
|
|
assert state["can_modify_choice"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_matching_closed_rejects_selection(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
user_female_b: User,
|
|
activity_closed: Activity,
|
|
):
|
|
"""Selection during matching_closed is rejected."""
|
|
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)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_then_cannot_enter_match(
|
|
self,
|
|
db_session: AsyncSession,
|
|
user_male_a: User,
|
|
):
|
|
"""After cancelling registration, cannot access match features."""
|
|
activity = Activity(
|
|
title="可取消活动",
|
|
start_time=datetime.now() + timedelta(days=3),
|
|
end_time=datetime.now() + timedelta(days=4),
|
|
match_deadline=datetime.now() + timedelta(days=4),
|
|
is_published=1,
|
|
)
|
|
db_session.add(activity)
|
|
await db_session.flush()
|
|
|
|
await register_activity(db_session, activity, user_male_a)
|
|
await cancel_registration(db_session, activity, user_male_a)
|
|
|
|
with pytest.raises(PermissionError, match="请先绑定"):
|
|
await get_activity_match_state(db_session, activity, user_male_a)
|