danyang-match/tests/e2e/idempotency_test.py

"""
幂等契约 E2E（规范第八章「Idempotency Key」+ 第一百四十六章「接口幂等验收」）。

背景：本轮检修发现后端**零处**读取 `Idempotency-Key` 头 —— 小程序有 26 处
生成并发送该头，全部石沉大海；唯一真正实现幂等的是 counseling 的 body 字段。
规范第 146 章的验收口径是「**最终不能生成重复核心数据**」。

本文件验证的是**机制本身**，不是某个端点的业务规则：
- 同一把键的重复投递 → 重放首次响应，业务代码只执行一次；
- 不同键 / 不同主体 → 互不干扰；
- 并发同键 → 业务只落地一次，且**绝不出现 500**。

设计纪律与 conftest 一致：栈不可达时退出（码 4），不退化为 skip。
"""
from __future__ import annotations

import uuid
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Dict

import pytest

pytestmark = pytest.mark.e2e

IDEMPOTENCY_HEADER = "Idempotency-Key"
SERVICE_TYPE = "relationship_guidance"


def _key(scope: str) -> str:
    """每次调用生成一把全新的键 —— 模拟「一次新的逻辑操作」。"""
    return f"e2e-{scope}-{uuid.uuid4()}"


def _headers(key: str) -> Dict[str, str]:
    return {IDEMPOTENCY_HEADER: key}


def _counseling_body() -> Dict[str, Any]:
    return {
        "serviceType": SERVICE_TYPE,
        "summary": "e2e 幂等验证",
        "requestSource": "user_self",
    }


def _my_services(api: Any, token: str) -> list:
    return api.ok("GET", "/counseling/services", token=token) or []


# ------------------------------------------------------------------ 重放语义


def test_same_key_replays_identical_response(api, new_user):
    """同一把键的重复投递必须**重放**首次响应，而不是重新执行。

    这是幂等键最核心的语义，也是改造前完全失效的地方。

    刻意用「咨询申请」而不是「喜欢」来验证：新注册用户的账号状态尚未通过
    审核，Like 会被业务规则拒绝（`账号状态或资料未通过，无法 Like`）。
    那是**业务前置条件**，与幂等机制无关，混进来会让用例红得没有信息量。
    """
    me = new_user()
    key = _key("replay")
    before = len(_my_services(api, me["token"]))

    first = api.call("POST", "/counseling/services", token=me["token"],
                     json=_counseling_body(), headers=_headers(key))
    second = api.call("POST", "/counseling/services", token=me["token"],
                      json=_counseling_body(), headers=_headers(key))

    assert first.status_code == 200, (
        f"首次提交失败：HTTP {first.status_code} {first.text[:300]}"
    )
    assert second.status_code == 200, (
        f"重复请求未被重放（HTTP {second.status_code}）：{second.text[:300]}"
    )
    assert first.json() == second.json(), (
        "同一幂等键的两次请求必须返回完全一致的响应，"
        f"实际不一致：\nfirst={first.json()}\nsecond={second.json()}"
    )
    assert len(_my_services(api, me["token"])) == before + 1, (
        "重放不应额外落地记录"
    )


def test_repeated_submission_with_same_key_creates_single_record(api, new_user):
    """弱网重发场景：同一把键重复提交咨询申请，只允许落地一条记录。"""
    me = new_user()
    key = _key("counseling")
    before = len(_my_services(api, me["token"]))

    first = api.call("POST", "/counseling/services", token=me["token"],
                     json=_counseling_body(), headers=_headers(key))
    second = api.call("POST", "/counseling/services", token=me["token"],
                      json=_counseling_body(), headers=_headers(key))

    assert first.status_code == 200, f"首次提交失败：{first.text[:300]}"
    assert second.status_code == 200, f"重复提交失败：{second.text[:300]}"

    after = _my_services(api, me["token"])
    assert len(after) == before + 1, (
        f"同一幂等键重复提交产生了 {len(after) - before} 条申请，期望 1 条"
    )


def test_body_idempotency_key_still_prevents_duplicates(api, new_user):
    """counseling 原有的 body 字段幂等（V039 `uk_counseling_idem`）不得因新机制回归。"""
    me = new_user()
    body = _counseling_body()
    body["idempotencyKey"] = _key("counseling-body")
    before = len(_my_services(api, me["token"]))

    api.ok("POST", "/counseling/services", token=me["token"], json=body)
    api.ok("POST", "/counseling/services", token=me["token"], json=body)

    after = _my_services(api, me["token"])
    assert len(after) == before + 1, (
        f"body 幂等键失效：重复提交产生了 {len(after) - before} 条申请"
    )


# ------------------------------------------------------------------ 作用域隔离


def test_different_keys_are_independent_operations(api, new_user):
    """不同键 = 两次独立操作，都应被受理（幂等不能变成"同端点只准提交一次"）。"""
    me = new_user()
    before = len(_my_services(api, me["token"]))

    api.ok("POST", "/counseling/services", token=me["token"],
           json=_counseling_body(), headers=_headers(_key("scope-a")))
    api.ok("POST", "/counseling/services", token=me["token"],
           json=_counseling_body(), headers=_headers(_key("scope-b")))

    after = _my_services(api, me["token"])
    assert len(after) == before + 2, (
        f"不同幂等键的两次提交应各成一条，实际新增 {len(after) - before} 条"
    )


def test_same_key_is_scoped_per_actor(api, new_user):
    """同一把键被两个用户使用：各算各的操作，绝不能把 A 的响应重放给 B。

    键的作用域是「主体 + 端点 + 键」。若只按键去重，B 会拿到 A 的申请结果，
    这既是数据泄露也是功能错误。
    """
    user_a = new_user()
    user_b = new_user()
    key = _key("shared-key")
    before_a = len(_my_services(api, user_a["token"]))
    before_b = len(_my_services(api, user_b["token"]))

    resp_a = api.call("POST", "/counseling/services", token=user_a["token"],
                      json=_counseling_body(), headers=_headers(key))
    resp_b = api.call("POST", "/counseling/services", token=user_b["token"],
                      json=_counseling_body(), headers=_headers(key))

    assert resp_a.status_code == 200, f"A 提交失败：{resp_a.text[:300]}"
    assert resp_b.status_code == 200, (
        f"B 使用与 A 相同的键却被拒绝/重放（HTTP {resp_b.status_code}）：{resp_b.text[:300]}"
    )
    assert resp_a.json()["data"]["id"] != resp_b.json()["data"]["id"], (
        "两个用户使用同一把键时拿到了同一条记录 —— 幂等作用域未按主体隔离"
    )
    assert len(_my_services(api, user_a["token"])) == before_a + 1
    assert len(_my_services(api, user_b["token"])) == before_b + 1


# ------------------------------------------------------------------ 并发兜底


def test_concurrent_same_key_lands_business_exactly_once(api, new_user):
    """并发同键（连点两次的真实形态）：业务只落地一次，且不得出现 500。

    这是占位机制存在的理由 —— 只在结束时才记录的话，两个并发请求会
    **都跑完业务**再发现重复，副作用已经发生。
    """
    me = new_user()
    key = _key("concurrent")
    before = len(_my_services(api, me["token"]))

    def send(_: int):
        return api.call("POST", "/counseling/services", token=me["token"],
                        json=_counseling_body(), headers=_headers(key))

    with ThreadPoolExecutor(max_workers=5) as pool:
        responses = list(pool.map(send, range(5)))

    codes = [r.status_code for r in responses]
    assert 500 not in codes, (
        f"并发重复提交出现 500（应为 200 重放或 409 处理中）：{codes}"
        f"\n{responses[codes.index(500)].text[:400]}"
    )
    assert all(code in (200, 409) for code in codes), f"出现非预期状态码：{codes}"

    after = _my_services(api, me["token"])
    assert len(after) == before + 1, (
        f"并发同键落地了 {len(after) - before} 条记录，期望恰好 1 条"
    )


# ------------------------------------------------------------------ 无键路径不退化


def test_duplicate_like_without_key_is_not_500(api, new_user):
    """不带幂等键的重复提交仍不得 500。

    这条守的是另一条防线：业务层的 SELECT-then-INSERT 与 DB 唯一约束
    （`uk_user_relation`）。并发撞约束时应是 409，不是服务器错误。
    """
    me = new_user()
    target = new_user()
    payload = {"targetUserId": target["user_id"]}

    first = api.call("POST", "/likes", token=me["token"], json=payload)
    second = api.call("POST", "/likes", token=me["token"], json=payload)

    assert first.status_code != 500, f"首次喜欢 500：{first.text[:300]}"
    assert second.status_code != 500, f"重复喜欢 500：{second.text[:300]}"