WorkBuddy
af5412646b
1) 修复一处从未暴露的 CI 缺陷 库中 operator 账号没有任何迁移种子，DemoDataSeeder 也不创建运营账号 （只统计数量），全新库唯一的账号来源是 seed_e2e_operators.sql。 而 conftest 的 admin_token 写死 username="admin" —— 那是某台开发机上 手工创建的遗留账号：本机有它所以本地一直跑绿，CI 全新库没有它 → 该 fixture 必然 401 → e2e job 从未真正通过过。 同一文件里「外加迁移 seed 的 admin」这句注释也是纯粹的虚构。 改用 seed 创建的 e2e_admin（已绑定 super_admin，语义等价）。 2) fixture_account_guard_test.py（2 用例） 机械拦住「fixture 依赖本机手工数据」：从 seed SQL 解析账号清单， 扫描 conftest 的所有登录用户名，任何一个不在 seed 内即变红； 另一条锁住推理前提（迁移里确实没有 operator 种子）。 两条都带「解析失败即失败」断言 —— 守卫自身静默失效是本项目反复踩过的坑。 3) idempotency_test.py（7 用例） 同键重放响应逐字节相同、同键重复提交只产生一条记录、body 字段幂等未回归、 不同键相互独立、同键作用域按 actor 隔离、5 线程并发同键业务恰好落库一次、 无键重复 Like 不得返回 500。 4) conftest.ApiClient.call 支持 headers 覆盖，供幂等用例携带自定义键。
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]}"
|