danyang-match/tests/e2e/activity_mutual_consent_test.py

"""
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,
        )
        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}"
    )