""" V1.1 活动互选二阶段同意 E2E —— 场景真实实现（Phase 23 Task 2.1，2026-09-19） 原文件 8 个测试函数**全部**为 `pass` 空壳（基线见 git b46b48e），依赖不存在的 `activity_factory` / `user_a` / `user_b` 等 fixture，从未执行过， 却在 `docs/STATUS.md` 中被计为 P23-T11「活动互选 E2E ✅」。 本文件按文件头保留的场景清单逐条实现为**真实 HTTP 用例**。 实现纪律（与 conftest 一致，不得回退） ------------------------------------ 1. 不 mock 业务行为：走真实 HTTP + 真实数据库； 2. 每个用例使用**独立新用户**，避免用例互相污染； 3. 栈不可达时由 conftest `pytest.exit(returncode=4)`，**不退化为 skip**。 被验证的调用链（实测自 ActivityMutualConsentController / ActivityMutualConsentService） -------------------------------------------------------------------------------------- POST /activities/{aid}/mutual/consent {partnerId} → 建一对 pending 记录 POST /activities/{aid}/mutual/confirm {partnerId} → 本方 accepted；双 accepted 才建 Match POST /activities/{aid}/mutual/decline {partnerId} → 本方 declined GET /activities/{aid}/mutual/status/{partnerId} → 本方/对方状态（INV-006 保密） GET /matches → 本人 active Match（data.items[]） POST /matches/{id}/unmatch → 单方解除 前置数据（为什么必须 seed） -------------------------- `activity_mutual_connection_consents.activity_id` 有 **FK 到 activities(id)**， `activities.organizer_org_id` 又有 FK 到 organizations(id) 且 NOT NULL， 而全新库里 organizations / activities 均为**空表**。 不走运营 API 建的原因：实测 `POST /operator/organizations` 当前**必然 500** （`OrganizationService.createOrganization` 从不赋 id，而 BaseEntity 用 `@TableId(type = IdType.INPUT)` 且 organizations.id 无 DB 默认值）—— 该缺陷另行记录。此处按同目录 `seed_e2e_operators.sql` 的既有约定直接落库， seed 由 `_ensure_activity()` 在首次使用时幂等执行（本机与 CI 通用）。 关于超时场景（第 7 条） ---------------------- `confirmSideIfPending` 的 SQL 带 `AND (expires_at IS NULL OR expires_at > NOW())`， 即「确认时」的过期闸门**已实现**。但 `createMutualConsent` **从不写入 expires_at** （列无默认值，服务也不设置），所以记录实际上永不过期 —— 这属计划批 3 的 P16-T13「互选超时接线（expires_at 已由 V041 提供）」，尚未接线。 本文件的超时用例因此**直接回拨库内 expires_at**，验证的是已实现的确认闸门本身； 不变量本身（过期不得推动成双确认）与「谁负责写入 expires_at」是两件事。 """ from __future__ import annotations import os import re import subprocess import uuid from typing import Any, Dict, List import pytest pytestmark = pytest.mark.e2e # 容器名取 compose 的 container_name，本地与 CI 一致（infrastructure/docker-compose.yml） PG_CONTAINER = os.environ.get("E2E_PG_CONTAINER", "danyang-match-postgres") PG_DB = os.environ.get("E2E_PG_DB", "danyang_match") # 与 tests/e2e/sql/seed_e2e_activity.sql 中的固定 UUID 一一对应 ACTIVITY_A = "e2e00000-0000-0000-0000-0000000000a2" ACTIVITY_B = "e2e00000-0000-0000-0000-0000000000a3" _UUID_RE = re.compile(r"^[0-9a-fA-F-]{36}$") _activity_ready = False # ------------------------------------------------------------------ 库内辅助 def _psql(sql: str) -> str: """在容器内执行一段 SQL 并返回输出。 失败时明确报错而不是跳过 —— 「没跑成」必须与「通过」区分开。 与仓库既有做法一致（tests/e2e/run_v11_tests.py 亦用 docker exec 取库内数据）。 """ proc = subprocess.run( ["docker", "exec", PG_CONTAINER, "psql", "-U", "postgres", "-d", PG_DB, "-v", "ON_ERROR_STOP=1", "-t", "-c", sql], capture_output=True, text=True, ) assert proc.returncode == 0, ( f"库操作失败（需要容器 {PG_CONTAINER} 可访问）：\n" f"stdout={proc.stdout}\nstderr={proc.stderr}" ) return proc.stdout def _ensure_activity(which: str = "A") -> str: """幂等确保互选所需的活动存在，返回活动 id。""" global _activity_ready if not _activity_ready: seed_path = os.path.join(os.path.dirname(__file__), "sql", "seed_e2e_activity.sql") with open(seed_path, encoding="utf-8") as fh: seed = fh.read() proc = subprocess.run( ["docker", "exec", "-i", PG_CONTAINER, "psql", "-U", "postgres", "-d", PG_DB, "-v", "ON_ERROR_STOP=1", "-q"], input=seed, capture_output=True, text=True, encoding="utf-8", ) assert proc.returncode == 0, ( f"互选前置数据 seed 失败（需要容器 {PG_CONTAINER} 可访问）：\n" f"stdout={proc.stdout}\nstderr={proc.stderr}" ) _activity_ready = True return ACTIVITY_A if which == "A" else ACTIVITY_B def _expire_consent(consent_id: str) -> None: """把互选记录的 expires_at 回拨到过去，触发确认闸门。 这是**唯一**能触发超时路径的手段：服务端从不写 expires_at（见模块 docstring）。 """ assert _UUID_RE.match(consent_id or ""), f"非法的互选记录 id: {consent_id!r}" out = _psql( "UPDATE activity_mutual_connection_consents " f"SET expires_at = NOW() - INTERVAL '1 hour' WHERE id = '{consent_id}';" ) assert "UPDATE 1" in out, f"预期恰好更新 1 行，实际输出：{out!r}" # ------------------------------------------------------------------ HTTP 辅助 def _consent(api: Any, token: str, activity_id: str, partner_id: str) -> Dict[str, Any]: return api.ok( "POST", f"/activities/{activity_id}/mutual/consent", token=token, json={"partnerId": partner_id}, ) def _confirm(api: Any, token: str, activity_id: str, partner_id: str) -> Dict[str, Any]: return api.ok( "POST", f"/activities/{activity_id}/mutual/confirm", token=token, json={"partnerId": partner_id}, ) def _decline(api: Any, token: str, activity_id: str, partner_id: str) -> Dict[str, Any]: return api.ok( "POST", f"/activities/{activity_id}/mutual/decline", token=token, json={"partnerId": partner_id}, ) def _status(api: Any, token: str, activity_id: str, partner_id: str) -> Dict[str, Any]: return api.ok("GET", f"/activities/{activity_id}/mutual/status/{partner_id}", token=token) def _active_matches(api: Any, token: str) -> List[Dict[str, Any]]: data = api.ok("GET", "/matches", token=token) or {} return data.get("items") or [] def _match_ids(api: Any, token: str) -> set: return {m.get("id") for m in _active_matches(api, token)} def _pair(new_user: Any) -> tuple: """两个互相独立的真实用户。""" return new_user(), new_user() def _double_confirm(api: Any, a: Dict[str, Any], b: Dict[str, Any], activity_id: str) -> Dict[str, Any]: """走完「建记录 → A 确认 → B 确认」，返回 B 侧确认响应。""" _consent(api, a["token"], activity_id, b["user_id"]) _confirm(api, a["token"], activity_id, b["user_id"]) return _confirm(api, b["token"], activity_id, a["user_id"]) # ------------------------------------------------------------------ 场景 1（INV-014） def test_mutual_interest_does_not_create_match(api, new_user): """INV-014：建立互选记录（表达意向）**不得**直接创建 Match。""" a, b = _pair(new_user) aid = _ensure_activity() _consent(api, a["token"], aid, b["user_id"]) assert _match_ids(api, a["token"]) == set(), "INV-014: 仅建立互选记录不得创建 Match（A 侧）" assert _match_ids(api, b["token"]) == set(), "INV-014: 仅建立互选记录不得创建 Match（B 侧）" # ------------------------------------------------------------------ 场景 2（INV-015） def test_one_side_confirm_no_match(api, new_user): """INV-015：只有一方确认时**不得**创建 Match。""" a, b = _pair(new_user) aid = _ensure_activity() _consent(api, a["token"], aid, b["user_id"]) result = _confirm(api, a["token"], aid, b["user_id"]) assert result["bothConfirmed"] is False, f"单方确认不得判定为双确认: {result}" assert result["matchCreated"] is False, f"INV-015: 单方确认不得创建 Match: {result}" assert _match_ids(api, a["token"]) == set(), "INV-015: 单方确认后 A 侧不得出现 Match" assert _match_ids(api, b["token"]) == set(), "INV-015: 单方确认后 B 侧不得出现 Match" # ------------------------------------------------------------------ 场景 3 def test_both_confirm_creates_match_episode(api, new_user): """双方分别确认后创建 Match Episode，且对双方立即可见。""" a, b = _pair(new_user) aid = _ensure_activity() _consent(api, a["token"], aid, b["user_id"]) first = _confirm(api, a["token"], aid, b["user_id"]) assert first["bothConfirmed"] is False, f"第一方确认后不应双确认: {first}" second = _confirm(api, b["token"], aid, a["user_id"]) assert second["bothConfirmed"] is True, f"双方确认后必须 bothConfirmed=true: {second}" assert second["matchCreated"] is True, f"双确认必须创建 Match: {second}" ids_a = _match_ids(api, a["token"]) ids_b = _match_ids(api, b["token"]) assert len(ids_a) == 1, f"双确认后应恰好 1 个 active Match（A 侧），实得 {ids_a}" assert ids_a == ids_b, f"Match 对双方必须一致可见（A={ids_a}, B={ids_b}）" # ------------------------------------------------------------------ 场景 4 def test_existing_active_match_not_duplicated(api, new_user): """AGENTS.md #19：已有 active Match 时再次双确认必须**复用**，不得重复创建。""" a, b = _pair(new_user) _double_confirm(api, a, b, _ensure_activity("A")) before = _match_ids(api, a["token"]) assert len(before) == 1, f"前置条件失败：第一次双确认应产生 1 个 Match，实得 {before}" _double_confirm(api, a, b, _ensure_activity("B")) after = _match_ids(api, a["token"]) assert after == before, ( f"AGENTS.md #19: 已有 active Match 不得重复创建（before={before}, after={after}）" ) # ------------------------------------------------------------------ 场景 5 def test_prior_unmatch_creates_new_episode(api, new_user): """AGENTS.md #20：曾 Unmatch 后重新双确认必须开启新 Episode（episode +1）。""" a, b = _pair(new_user) _double_confirm(api, a, b, _ensure_activity("A")) first = _active_matches(api, a["token"]) assert len(first) == 1, f"前置条件失败：应恰好 1 个 active Match，实得 {first}" match_id = first[0]["id"] episode_before = first[0].get("episode") api.ok("POST", f"/matches/{match_id}/unmatch", token=a["token"]) assert _active_matches(api, a["token"]) == [], "Unmatch 后不得再有 active Match" _double_confirm(api, a, b, _ensure_activity("B")) second = _active_matches(api, a["token"]) assert len(second) == 1, f"重新确认后应恰好 1 个 active Match，实得 {second}" episode_after = second[0].get("episode") assert episode_after == (episode_before or 1) + 1, ( f"AGENTS.md #20: 历史 Unmatch 后 episode 必须 +1" f"（before={episode_before}, after={episode_after}）" ) # ------------------------------------------------------------------ 场景 6 def test_one_side_decline_no_match(api, new_user): """一方拒绝则不创建 Match；对方随后确认也不得创建。""" a, b = _pair(new_user) aid = _ensure_activity() _consent(api, a["token"], aid, b["user_id"]) declined = _decline(api, a["token"], aid, b["user_id"]) assert declined["myStatus"] == "declined", f"拒绝后本方状态应为 declined: {declined}" assert _match_ids(api, a["token"]) == set(), "拒绝后不得创建 Match" after = _confirm(api, b["token"], aid, a["user_id"]) assert after["bothConfirmed"] is False, f"对方已拒绝，本方确认不得判定双确认: {after}" assert _match_ids(api, b["token"]) == set(), "对方拒绝后，本方确认也不得创建 Match" # ------------------------------------------------------------------ 场景 7（超时） def test_timeout_handled(api, new_user): """超时（expires_at）处理：已过期的互选记录**不得**再被推动成双确认。 注意本用例先做一次正常确认作为**对照**，证明「确认闸门本身可用」， 否则「过期后确认不生效」可能只是因为确认功能整体坏掉（假绿）。 """ a, b = _pair(new_user) # 对照：未过期的记录，确认必须生效 aid_ok = _ensure_activity() _consent(api, a["token"], aid_ok, b["user_id"]) control = _confirm(api, a["token"], aid_ok, b["user_id"]) assert control["myStatus"] == "accepted", f"对照用例失败：未过期记录确认应生效: {control}" # 目标：已过期（expires_at 回拨）的记录，确认不得生效 aid_expired = _ensure_activity("B") consent = _consent(api, a["token"], aid_expired, b["user_id"]) _expire_consent(consent["id"]) expired_a = _confirm(api, a["token"], aid_expired, b["user_id"]) assert expired_a["myStatus"] == "pending", f"过期后确认不得写入 accepted: {expired_a}" expired_b = _confirm(api, b["token"], aid_expired, a["user_id"]) assert expired_b["bothConfirmed"] is False, f"过期记录不得判定双确认: {expired_b}" assert expired_b["matchCreated"] is False, f"过期记录不得创建 Match: {expired_b}" assert _match_ids(api, a["token"]) == set(), "过期互选不得创建 Match" # ------------------------------------------------------------------ 场景 8（INV-006） def test_single_interest_not_leaked(api, new_user): """INV-006：单向意向不泄露 —— 本方未确认时不得回显对方是否已确认。""" a, b = _pair(new_user) aid = _ensure_activity() _consent(api, a["token"], aid, b["user_id"]) _confirm(api, a["token"], aid, b["user_id"]) # A 已确认，B 尚未 status_b = _status(api, b["token"], aid, a["user_id"]) assert status_b["exists"] is True, f"记录应存在: {status_b}" assert status_b["partnerStatus"] == "pending", ( f"INV-006: 本方(B)未确认时不得回显对方(A)意向（实得 {status_b}）" ) assert status_b["bothConfirmed"] is False, f"未双确认不得报 bothConfirmed: {status_b}" status_a = _status(api, a["token"], aid, b["user_id"]) assert status_a["myStatus"] == "accepted", f"A 侧本方状态应为 accepted: {status_a}" assert status_a["partnerStatus"] == "pending", f"B 未确认，A 侧不应看到 accepted: {status_a}" # 反证：B 确认后双方都应看到 accepted —— 证明上面的 pending 不是「永远 pending」的假绿 _confirm(api, b["token"], aid, a["user_id"]) status_b_after = _status(api, b["token"], aid, a["user_id"]) assert status_b_after["partnerStatus"] == "accepted", ( f"B 确认后应看到对方 accepted（反证前面的 pending 是条件性保密而非恒定）: {status_b_after}" )