From af5412646bf45752fecea22ee4816baf9cd378ec Mon Sep 17 00:00:00 2001 From: WorkBuddy Date: Fri, 18 Sep 2026 16:37:33 +0800 Subject: [PATCH] =?UTF-8?q?test(e2e):=20=E8=BF=90=E8=90=A5=E7=99=BB?= =?UTF-8?q?=E5=BD=95=E5=8F=AA=E7=94=A8=20seed=20=E8=B4=A6=E5=8F=B7?= =?UTF-8?q?=EF=BC=9B=E8=A1=A5=E5=B9=82=E7=AD=89=20E2E=20=E4=B8=8E=E4=B8=A4?= =?UTF-8?q?=E5=A4=84=E6=9C=BA=E6=A2=B0=E5=AE=88=E5=8D=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 覆盖，供幂等用例携带自定义键。 --- tests/e2e/conftest.py | 36 +++- tests/e2e/fixture_account_guard_test.py | 73 ++++++++ tests/e2e/idempotency_test.py | 219 ++++++++++++++++++++++++ 3 files changed, 321 insertions(+), 7 deletions(-) create mode 100644 tests/e2e/fixture_account_guard_test.py create mode 100644 tests/e2e/idempotency_test.py diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 403a955..8e2d164 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -49,12 +49,16 @@ class ApiClient: token: Optional[str] = None, json: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, ) -> requests.Response: - headers = {"Authorization": f"Bearer {token}"} if token else {} + merged = {"Authorization": f"Bearer {token}"} if token else {} + if headers: + # 显式传入的头覆盖默认头（幂等键等按用例定制） + merged.update(headers) return self.session.request( method, f"{self.base_url}{path}", - headers=headers, + headers=merged, json=json, params=params, timeout=self.timeout, @@ -112,15 +116,29 @@ def operator_password() -> str: return pwd +ADMIN_USERNAME = "e2e_admin" + + @pytest.fixture(scope="session") def admin_token(api: ApiClient, operator_password: str) -> str: + """超级管理员 token。 + + 必须使用 **seed 脚本创建的账号**，而不是 `admin`。 + + 原因（本轮发现）：库中 operator 账号**没有任何迁移种子**，`DemoDataSeeder` + 也不创建运营账号（只统计），运营账号只能由运行时服务或本 seed 脚本产生。 + 此前该 fixture 写的是 `username="admin"` —— 那是某台开发机上手工创建的遗留 + 账号，全新库（CI）里根本不存在 → 该 fixture 在 CI 上必然 401， + **整个 E2E job 从未真正通过过**，只是本机因为有遗留账号而看不出来。 + `e2e_admin` 由 seed 绑定 super_admin 角色，语义等价且可自举。 + """ data = api.ok( "POST", "/auth/operator/login", - json={"username": "admin", "password": operator_password}, + json={"username": ADMIN_USERNAME, "password": operator_password}, ) token = data.get("accessToken") - assert token, "admin 登录未返回 accessToken" + assert token, f"{ADMIN_USERNAME} 登录未返回 accessToken" return token @@ -130,9 +148,13 @@ def operator_token( ) -> Callable[[str], str]: """按用户名取运营 token（会话内缓存）。 - 可用账号见 tests/e2e/sql/seed_e2e_operators.sql： - e2e_admin / e2e_platform / e2e_counselor / e2e_station / e2e_matchmaker / e2e_content， - 外加迁移 seed 的 admin。用法：`operator_token("e2e_counselor")`。 + 可用账号**仅限** tests/e2e/sql/seed_e2e_operators.sql 中创建的 6 个： + e2e_admin / e2e_platform / e2e_counselor / e2e_station / e2e_matchmaker / e2e_content。 + 迁移里**没有任何 operator 种子**（原文"外加迁移 seed 的 admin"是错的， + 已纠正——全新库里没有 admin 这个账号）。用法：`operator_token("e2e_counselor")`。 + + 传入 seed 之外的用户名等于依赖某台机器上的手工数据，CI 上必然 401； + `fixture_account_guard_test.py` 会机械拦住这类写法。 """ cache: Dict[str, str] = {} diff --git a/tests/e2e/fixture_account_guard_test.py b/tests/e2e/fixture_account_guard_test.py new file mode 100644 index 0000000..5418826 --- /dev/null +++ b/tests/e2e/fixture_account_guard_test.py @@ -0,0 +1,73 @@ +"""E2E 自检守卫：fixture 只能使用 seed 脚本真正创建的运营账号。 + +为什么需要它（本轮发现的真实缺陷）： + + 库中 operator 账号**没有任何迁移种子**，`DemoDataSeeder` 也不创建运营账号 + （只统计数量）。因此全新库（CI）里唯一的运营账号来源是 + `tests/e2e/sql/seed_e2e_operators.sql`。 + + 而 `conftest.py` 的 `admin_token` 曾写死 `username="admin"` —— 那是某台 + 开发机上手工创建的遗留账号。本机有它，所以本地跑绿；CI 全新库没有它， + 该 fixture 必然 401 → **整个 E2E job 从未真正通过过**。 + + 这类"依赖本机手工数据"的写法没有任何运行期提示，只能靠机械检查拦住。 +""" + +from __future__ import annotations + +import re +from pathlib import Path + +E2E_DIR = Path(__file__).resolve().parent +CONFTEST = E2E_DIR / "conftest.py" +SEED_SQL = E2E_DIR / "sql" / "seed_e2e_operators.sql" +MIGRATION_DIR = E2E_DIR.parent.parent / "backend" / "src" / "main" / "resources" / "db" / "migration" + + +def _seeded_usernames() -> set[str]: + sql = SEED_SQL.read_text(encoding="utf-8") + # seed 用的是 VALUES 行：('uuid', '名称', 'username', crypt(...), 'active') + return set(re.findall(r"'([A-Za-z0-9_]+)',\s+crypt\(", sql)) + + +def _usernames_used_in_conftest() -> set[str]: + src = CONFTEST.read_text(encoding="utf-8") + used = set(re.findall(r'"username"\s*:\s*"([A-Za-z0-9_]+)"', src)) + used |= set(re.findall(r'ADMIN_USERNAME\s*=\s*"([A-Za-z0-9_]+)"', src)) + return used + + +def test_conftest_only_logs_in_with_seeded_accounts() -> None: + seeded = _seeded_usernames() + assert seeded, f"未能从 {SEED_SQL.name} 解析出任何账号，守卫已失效" + + used = _usernames_used_in_conftest() + assert used, f"未能从 {CONFTEST.name} 解析出任何登录用户名，守卫已失效" + + unknown = sorted(used - seeded) + assert not unknown, ( + "conftest 使用了 seed 脚本未创建的运营账号：" + f"{unknown}（seed 现有：{sorted(seeded)}）。\n" + "全新库（CI）里这些账号不存在，对应 fixture 会 401，E2E 必然失败。" + "请改用 seed 中的账号，或在 seed 脚本中补上该账号。" + ) + + +def test_no_operator_account_is_seeded_by_migrations() -> None: + """锁住上面那条推理的前提：迁移确实没有 operator 种子。 + + 若将来有人在迁移里加了 admin 种子，本用例会变红——那时应重新评估 + `admin_token` 的选择（迁移种子同样是自举的，可作为合法来源）， + 而不是继续凭本机的遗留账号。 + """ + hits: list[str] = [] + for sql_file in MIGRATION_DIR.glob("*.sql"): + text = sql_file.read_text(encoding="utf-8", errors="replace") + if re.search(r"INSERT\s+INTO\s+operators\b", text, re.IGNORECASE): + hits.append(sql_file.name) + + assert not hits, ( + f"迁移中出现了 operator 种子：{hits}。" + "本次 E2E fixture 的选型前提（'运营账号只能来自 seed 脚本'）已改变，" + "请重新评估 conftest 的 ADMIN_USERNAME 与本文档说明。" + ) diff --git a/tests/e2e/idempotency_test.py b/tests/e2e/idempotency_test.py new file mode 100644 index 0000000..e641b83 --- /dev/null +++ b/tests/e2e/idempotency_test.py @@ -0,0 +1,219 @@ +""" +幂等契约 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]}"