220 lines
8.8 KiB
Python
220 lines
8.8 KiB
Python
|
"""
|
|||
|
|
幂等契约 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]}"
|