"""不属任何单线的**孤立路由**端到端补测（真实 HTTP · GX1–GX5）。 盘点（计划 §1.1）查出四条路由**零 E2E 覆盖**，且其中两条**连单测都没有**： | 路由 | 盘点结论 | 本脚本分组 | | --- | --- | --- | | `POST /api/customers/{id}/threshold-check` | ★ 全仓 e2e 零命中；前端**有按钮**（`CustomerHoldingsPage.tsx:21-27`）；`tests/test_main.py:104` 仅路径字符串断言 | GX1 | | `GET /api/staff/me` | 仅单测；前端零调用 | GX2 | | `GET /api/products/{product_id}` | 仅单测；前端零调用 | GX3 | | `POST /api/chat/sessions/close-all` · `/sessions/{id}/close` | 仅单测；前端有按钮（`ChatPanel.tsx:72,176`）但**没人点过** | GX4 | | （文档类）`DEMO-功能测试流程.md` / `功能演示版验收标准.md` | 门槛陈旧 + 编号一号两义 | GX5（**只记录不修**） | **为什么单独成脚本：** 这四条路由横跨客户/平台/对话三条线，塞进任一业务线包都会 污染那个包的结论边界（用户拍板口径 7「一线一包」）。故归入横切（crosscut）包。 **关于 GX4 的数据副作用（本脚本最需要小心的一处）：** `close-all` / `close` 会**改变 `agent_session.status`**，而 `jinrong_agent` 按纪律 **不还原**（追加型），且会话列表是演示界面的一部分。故： - 状态变更类用例一律用**备用账号 `STAFF-20003`**（答辩记号本标注为"问数第三账号"）， **不用**主演示账号 —— 避免把 CUST-9527 / STAFF-10086 的对话历史关掉。 - 该账号若无任何会话，则 `close-all` 实测 `closed_count=0`，**零副作用**； 有会话时才走状态机分支，并把实际走的分支记入 INFO。 用法： python -m uvicorn app.main:app --host 127.0.0.1 --port 8000 python scripts/dev/gap_routes_e2e_smoke.py python scripts/dev/gap_routes_e2e_smoke.py --only GX1 --only GX4 python scripts/dev/gap_routes_e2e_smoke.py --report docs/.../_raw/xc-gap.md 退出码：0 全过 / 1 有 FAIL / 2 前置自检失败。 """ from __future__ import annotations import argparse import sys import traceback from pathlib import Path import httpx sys.path.insert(0, str(Path(__file__).resolve().parents[2])) sys.path.insert(0, str(Path(__file__).resolve().parent)) import customer_trade_e2e_smoke as e2e # noqa: E402 from customer_trade_e2e_smoke import ( # noqa: E402 FAIL, INFO, PASS, SKIP, Ctx, bad, body, data_of, err_code, expect_status, expect_true, info, login, ok, skip, ) # ── 演示账号 ── CUSTOMER = "CUST-9527" # 本人 OTHER_CUSTOMER = "CUST-1002" # 非本人（但同属 STAFF-10086） ADVISOR = "STAFF-10086" OWNED_CUSTOMER = "CUST-1001" # STAFF-10086 名下 UNOWNED_CUSTOMER = "CUST-1004" # 存在，但归属 STAFF-10087（实测自 core_customer_advisor） RISK = "STAFF-30001" # risk_officer（threshold-check / staff-me 的正例角色） #: GX4 的状态变更专用账号：答辩记号本标注为"问数第三账号（dev 映射）"， #: 非主演示账号 —— 用它关会话不会动到演示叙事。 SPARE = "STAFF-20003" UNKNOWN_ID = "CUST-000000" #: 已知存在的产品（用于 GX3 正例；取自 Core 模拟库事实）。 PRODUCT = "PROD-110022" REQUIRED_PATHS = { "/api/customers/{customer_id}/threshold-check", "/api/staff/me", "/api/products/{product_id}", "/api/chat/sessions", "/api/chat/sessions/close-all", "/api/chat/sessions/{session_id}/close", "/api/risk/alerts", } def expect_code_in(group: str, name: str, resp: httpx.Response, want_status: int, want_codes: set[str]) -> bool: got = err_code(resp) if resp.status_code == want_status and got in want_codes: ok(group, name, f"HTTP {want_status} code∈{sorted(want_codes)}", f"HTTP {resp.status_code} {got}") return True bad(group, name, f"HTTP {want_status} code∈{sorted(want_codes)}", f"HTTP {resp.status_code} {got}", resp.text[:400]) return False # -------------------------------------------------------------------------- # 前置自检 # -------------------------------------------------------------------------- def preflight(ctx: Ctx, base_url: str) -> bool: print(f"== 孤立路由端到端补测 @ {base_url} ==\n") failures: list[str] = [] try: r = ctx.client.get("/api/ready") print(f" /api/ready -> HTTP {r.status_code}") if r.status_code != 200: failures.append(f"/api/ready HTTP {r.status_code}") except Exception as exc: # noqa: BLE001 failures.append(f"后端不可达: {exc}") try: paths = set(ctx.client.get("/openapi.json").json().get("paths", {})) missing = sorted(REQUIRED_PATHS - paths) print(f" openapi 路径总数 {len(paths)}（基线 66）· 必需路径缺 {len(missing)}") failures.extend(f"openapi 缺路径 {p}（后端跑的是旧进程？）" for p in missing) except Exception as exc: # noqa: BLE001 failures.append(f"取 openapi 失败: {exc}") # ⚠️ 陷阱（2026-09-13 实测）：`login()` 默认 `token_type="staff"`，而客户账号用默认值 # 登录时签发的 JWT **`customer_id` 为 None**（实测 claims：`token_type='staff', # roles=['customer'], customer_id=None`）。而 `assert_customer_access` 的 self 域 # **靠 `customer_id` 判定** ⇒ 客户查自己也会 403 AUTH_403_NOT_OWNER（假 FAIL）。 # 客户账号必须显式 `token_type="customer"`。 for key, actor, tt in (("customer", CUSTOMER, "customer"), ("other", OTHER_CUSTOMER, "customer"), ("advisor", ADVISOR, "staff"), ("risk", RISK, "staff"), ("spare", SPARE, "staff")): try: ctx.h[key] = login(ctx, actor, tt) print(f" 登录 {actor} -> ok") except Exception as exc: # noqa: BLE001 if key == "spare": # 备用账号不可用不该阻断整轮：GX4 会据此 SKIP 而不是假 FAIL。 print(f" ⚠️ 备用账号 {actor} 登录失败：{exc}") continue failures.append(f"登录 {actor} 失败: {exc}") if failures: print("\n前置自检未通过：") for f in failures: print(f" - {f}") return not failures # -------------------------------------------------------------------------- # GX1 · POST /api/customers/{id}/threshold-check ★零覆盖 # -------------------------------------------------------------------------- def gx1(ctx: Ctx) -> None: G = "GX1" # 该路由走 `get_platform_auth_context`（`customers.py:67`）—— **不要求** X-Agent-Type。 # 前端调用方：`CustomerHoldingsPage.tsx:21-27` 的「检查提醒阈值」按钮。 for key, cid, label in ( ("risk", OWNED_CUSTOMER, "risk_officer"), ("customer", CUSTOMER, "customer 本人"), ("advisor", OWNED_CUSTOMER, "归属 advisor"), ): r = ctx.client.post(f"/api/customers/{cid}/threshold-check", headers=ctx.h[key]) if not expect_status(G, f"1-01 POST {cid}/threshold-check（{label}）", r, 200): continue d = data_of(r) expect_true(G, f"1-01a 返回体含 alert/pushed 两键（{label}）", "alert" in d and "pushed" in d, "data.alert + data.pushed", f"keys={sorted(d)} alert={d.get('alert')!r} pushed={d.get('pushed')!r}") # 不带 `push` 参数时**不得**推送（`threshold_service.py:122-124` 只在 alert 真值时推） expect_true(G, f"1-01b 未带 push 参数 ⇒ pushed=false（{label}）", d.get("pushed") is False, "pushed=False", f"pushed={d.get('pushed')!r}") info(G, f"1-01c alert 实测值（{label}）", f"{str(d.get('alert'))[:120]}") # `push=true` 正例：只在有 alert 时才真的发布（无 alert 时是 no-op，不构成副作用） r = ctx.client.post(f"/api/customers/{OWNED_CUSTOMER}/threshold-check?push=true", headers=ctx.h["risk"]) expect_status(G, "1-02 push=true 正常返回", r, 200) if r.status_code == 200: d = data_of(r) info(G, "1-02a push=true 实测（alert 为空时按设计为 no-op）", f"alert={str(d.get('alert'))[:80]!r} pushed={d.get('pushed')!r}") # ── 负例：越权 ── r = ctx.client.post(f"/api/customers/{OWNED_CUSTOMER}/threshold-check", headers=ctx.h["other"]) expect_code_in(G, f"1-03 非本人 customer 查 {OWNED_CUSTOMER} → 403（AUTH_403_NOT_OWNER）", r, 403, {"AUTH_403_NOT_OWNER"}) r = ctx.client.post("/api/customers/CUST-1004/threshold-check", headers=ctx.h["advisor"]) expect_code_in(G, "1-04 非归属 advisor 查 CUST-1004 → 403（AUTH_403_NOT_ASSIGNED）", r, 403, {"AUTH_403_NOT_ASSIGNED"}) r = ctx.client.post(f"/api/customers/{UNKNOWN_ID}/threshold-check", headers=ctx.h["risk"]) # 事实（已读源码核对）：**不 404**。持仓查空 → alert=None → 200。 # 这条是本组最容易被"想当然"写错的地方 —— 断言按**实测事实**写，并把 # "未知客户不报错"这一行为本身记为 INFO（是否算缺陷取决于产品口径，本轮不下判）。 if r.status_code == 200: ok(G, f"1-05 未知客户 {UNKNOWN_ID} → 200 且 alert=null（实测口径，不 404）", f"HTTP 200 alert=None", f"HTTP {r.status_code} alert={data_of(r).get('alert')!r}") info(G, "1-05a 未知客户的实际行为", "持仓查空 ⇒ alert=None ⇒ 200 静默返回（不区分" "「客户不存在」与「无提醒」，见 `threshold_service.py:64-70`）") else: bad(G, f"1-05 未知客户 {UNKNOWN_ID} → 期望与实测口径一致", "HTTP 200（源码事实）", f"HTTP {r.status_code} {err_code(r)}", r.text[:200]) # -------------------------------------------------------------------------- # GX2 · GET /api/staff/me（仅单测 · 前端零调用） # -------------------------------------------------------------------------- def gx2(ctx: Ctx) -> None: G = "GX2" r = ctx.client.get("/api/staff/me", headers=ctx.h["advisor"]) expect_status(G, "2-01 GET /api/staff/me（staff token）", r, 200) if r.status_code == 200: d = data_of(r) expect_true(G, "2-01a 返回体含 actor_id 且等于登录账号", d.get("actor_id") == ADVISOR or d.get("staff_id") == ADVISOR, f"actor_id={ADVISOR}", f"keys={sorted(d)} actor_id={d.get('actor_id')!r} staff_id={d.get('staff_id')!r}") info(G, "2-01b /staff/me 返回字段", f"{sorted(d)}") expect_true(G, "2-01c roles 为 list（非 JSON 字符串）", isinstance(d.get("roles"), list), "list", f"{type(d.get('roles')).__name__}") # customer token → `require_staff_token`（`deps.py:353-356`）拒 r = ctx.client.get("/api/staff/me", headers=ctx.h["customer"]) expect_code_in(G, "2-02 customer token → 403（AUTH_403_SCOPE）", r, 403, {"AUTH_403_SCOPE"}) # 无 token 的 401 **不是**由 token 校验发出的，而是先撞上开发态的调试头网关 # （`deps.py:268-270`：dev 下无 Authorization 时改用 `X-Debug-Role`/`X-Debug-Actor`， # 两样都没有 → `AUTH_401_MISSING_DEBUG_HEADERS`）。实测首跑冻结，按集合接受。 r = ctx.client.get("/api/staff/me") expect_code_in(G, "2-03 无 token → 401", r, 401, {"AUTH_401_MISSING_DEBUG_HEADERS", "AUTH_401_MISSING_TOKEN", "AUTH_401"}) # -------------------------------------------------------------------------- # GX3 · GET /api/products/{product_id}（仅单测 · 前端零调用） # -------------------------------------------------------------------------- def gx3(ctx: Ctx) -> None: G = "GX3" r = ctx.client.get(f"/api/products/{PRODUCT}", headers=ctx.h["risk"]) expect_status(G, f"3-01 GET /api/products/{PRODUCT}", r, 200) if r.status_code == 200: d = data_of(r) expect_true(G, "3-01a 返回体含 product_id 且与请求一致", d.get("product_id") == PRODUCT, f"product_id={PRODUCT}", f"keys={sorted(d)[:12]}") info(G, "3-01b 产品字段清单", f"{sorted(d)}") # customer 角色也在 `PLATFORM_FULL_READ_ROLES` 白名单（`deps.py:40`）内 r = ctx.client.get(f"/api/products/{PRODUCT}", headers=ctx.h["customer"]) expect_status(G, "3-02 customer 角色可读产品目录", r, 200) r = ctx.client.get("/api/products/PROD-DOES-NOT-EXIST", headers=ctx.h["risk"]) expect_code_in(G, "3-03 未知产品 → 404（NOT_FOUND）", r, 404, {"NOT_FOUND"}) r = ctx.client.get(f"/api/products/{PRODUCT}") # 同 2-03：dev 态下先撞调试头网关。 expect_code_in(G, "3-04 无 token → 401", r, 401, {"AUTH_401_MISSING_DEBUG_HEADERS", "AUTH_401_MISSING_TOKEN", "AUTH_401"}) # -------------------------------------------------------------------------- # GX2b · 登录接口的 token_type × actor 组合面（**只记录**，供后续拍板） # -------------------------------------------------------------------------- def gx2b(ctx: Ctx) -> None: """`/api/auth/login` 允许给**客户账号**签发 `token_type="staff"` 的 token。 实测三条事实（本组不判缺陷，只取证）： 1. `token_type="staff"` + actor=`CUST-9527` → JWT `customer_id=None`， 于是 `assert_customer_access` 的 self 域判定失败 ⇒ **客户查自己 403 NOT_OWNER**； 2. 同一 token 打 `GET /api/staff/me`：`require_staff_token`（`deps.py:353`）**只看 `token_type` 不看角色** ⇒ 放行 → 再查 staff 表查无此人 → **404 `staff not found`**； 3. 换 `token_type="customer"` → `customer_id` 正常签发，self 域通过（200）， 而 `/api/staff/me` 正确拒为 403 AUTH_403_SCOPE。 ⇒ 结论：客户身份**必须**用 `token_type="customer"` 登录；用错类型不会越权 （staff/me 只回自己且 404），但会**制造 403/404 的语义混淆**。 """ G = "GX2b" wrong = login(ctx, CUSTOMER) # 故意用默认的 token_type="staff" right = login(ctx, CUSTOMER, "customer") r = ctx.client.get("/api/staff/me", headers=wrong) expect_code_in(G, "2b-01 staff 型 token 载客户号 → /staff/me 实测为 404（非 403）", r, 404, {"NOT_FOUND"}) info(G, "2b-01a 该行为的原因", "`require_staff_token` 只比 `auth.token_type`（deps.py:353-356），staff 型即放行，" "随后 `staff_service.get_staff_me('CUST-9527')` 查无此人 → 404") r = ctx.client.get("/api/staff/me", headers=right) expect_code_in(G, "2b-02 customer 型 token → /staff/me 正确拒 403（AUTH_403_SCOPE）", r, 403, {"AUTH_403_SCOPE"}) # -------------------------------------------------------------------------- # GX4 · 对话会话侧栏（close-all / close）—— 仅单测，前端有按钮但没人点过 # -------------------------------------------------------------------------- def gx4(ctx: Ctx) -> None: G = "GX4" spare = ctx.h.get("spare") if not spare: skip(G, "4-01..4-08", f"备用账号 {SPARE} 登录失败，按纪律不用主演示账号做状态变更") return # `/api/chat/*` 走 `get_auth_context` ⇒ **要求** X-Agent-Type（`deps.py:248-251`）。 h = {**spare, "X-Agent-Type": "analyst"} # (a) 缺头负例 r = ctx.client.post("/api/chat/sessions/close-all", headers=spare) expect_code_in(G, "4-01 缺 X-Agent-Type → 401（AUTH_401_MISSING_AGENT_TYPE）", r, 401, {"AUTH_401_MISSING_AGENT_TYPE"}) # (b) 非法头负例 r = ctx.client.post("/api/chat/sessions/close-all", headers={**spare, "X-Agent-Type": "nope"}) expect_code_in(G, "4-02 非法 X-Agent-Type → 400（BAD_REQUEST）", r, 400, {"BAD_REQUEST"}) # (c) 正常路径 r = ctx.client.post("/api/chat/sessions/close-all", headers=h) expect_status(G, f"4-03 POST /sessions/close-all（{SPARE} · 非主演示账号）", r, 200) if r.status_code != 200: return d = body(r) expect_true(G, "4-03a 返回体为裸 dict 且含 closed_count/agent_type", "closed_count" in d and d.get("agent_type") == "analyst", "closed_count:int + agent_type='analyst'", f"keys={sorted(d)} closed_count={d.get('closed_count')!r} agent_type={d.get('agent_type')!r}") info(G, "4-03b close-all 实际关闭条数（0 = 该账号本次零副作用）", f"{d.get('closed_count')}") # (d) 幂等：紧接着再调一次，两次都不该报错 r2 = ctx.client.post("/api/chat/sessions/close-all", headers=h) expect_status(G, "4-04 close-all 幂等（紧接着再调仍 200）", r2, 200) if r2.status_code == 200: expect_true(G, "4-04a 第二次 close-all 的 closed_count=0", body(r2).get("closed_count") == 0, "closed_count=0", f"closed_count={body(r2).get('closed_count')!r}") # (e) 列表 r = ctx.client.get("/api/chat/sessions", headers=h) expect_status(G, "4-05 GET /sessions（会话侧栏数据源）", r, 200) sessions: list = [] if r.status_code == 200: dd = data_of(r) sessions = dd if isinstance(dd, list) else (dd.get("items") or []) info(G, "4-05a 该账号会话条数与状态分布", f"n={len(sessions)} statuses={sorted({s.get('status') for s in sessions if isinstance(s, dict)})}") # (f) 不存在的会话 → 404 r = ctx.client.post("/api/chat/sessions/does-not-exist/close", headers=h) expect_code_in(G, "4-06 关闭不存在的会话 → 404（NOT_FOUND）", r, 404, {"NOT_FOUND"}) # (g) 状态机：已 closed 的会话再关闭 → 409（**并发双点的兜底码**，`chat.py:400-403`） target = next((s for s in sessions if isinstance(s, dict) and s.get("status") == "closed"), None) if target is None: info(G, "4-07 已关闭会话再关闭 → 409", "该账号无 closed 会话，状态机分支未覆盖（记录）") else: r = ctx.client.post(f"/api/chat/sessions/{target['session_id']}/close", headers=h) expect_code_in(G, "4-07 已关闭会话再关闭 → 409（STATE_CONFLICT）", r, 409, {"STATE_CONFLICT"}) # (h) 越权：会话归属他人 → AUTH_403_SESSION_AGENT other = ctx.h.get("advisor") if other and target is not None: h2 = {**other, "X-Agent-Type": "analyst"} r = ctx.client.post(f"/api/chat/sessions/{target['session_id']}/close", headers=h2) expect_code_in(G, "4-08 他人会话（actor 不一致）→ 403（AUTH_403_SESSION_AGENT）", r, 403, {"AUTH_403_SESSION_AGENT"}) else: skip(G, "4-08 他人会话越权 → 403", "无可用对照会话") # -------------------------------------------------------------------------- # GX5 · 文档类发现（**只记录不修**，静态取证） # -------------------------------------------------------------------------- def gx5(ctx: Ctx) -> None: G = "GX5" root = Path(__file__).resolve().parents[2] # ── 5-01 `D2`/`D3` 一号两义（答辩稿缺陷）── doc = root / "docs" / "答辩" / "DEMO-功能测试流程.md" if not doc.exists(): skip(G, "5-01 D2/D3 编号歧义取证", f"文档不存在：{doc}") else: lines = doc.read_text(encoding="utf-8", errors="ignore").splitlines() # 行号按 ADV-001 v1.0 记录（第 64-65 行 = 问数线用例；第 79/89 行 = 缺陷编号） hits = [(i + 1, ln.strip()[:90]) for i, ln in enumerate(lines) if ("D2" in ln or "D3" in ln)] use_lines = [h for h in hits if "D2" in h[1] and "D3" in h[1]] info(G, "5-01 D2/D3 在答辩稿中的出现位置", f"共 {len(hits)} 行；同时含 D2 与 D3 的行={[h[0] for h in use_lines]}") info(G, "5-01a 前 6 行原文", " ｜ ".join(f"L{n}:{t}" for n, t in hits[:6])) info(G, "5-01b 结论", "同一符号既作用例编号又作缺陷编号 ⇒ 本包报告一律不用裸 D2/D3，" "改写为「问数线 D2」/「ADV 缺陷-2」") # ── 5-02 验收标准陈旧门槛 ── std = root / "docs" / "演示文档" / "功能演示版验收标准.md" if std.exists(): txt = std.read_text(encoding="utf-8", errors="ignore") for needle, why in (("/api/v1/ping", "该路由在当前 openapi 中不存在"), ("run_demo_smoke.ps1", "该脚本仍打 /api/v1/*，实测已失效")): info(G, f"5-02 验收标准引用 {needle}", "命中" if needle in txt else "未命中", why) try: paths = set(ctx.client.get("/openapi.json").json().get("paths", {})) info(G, "5-02a 运行时 openapi 是否含 /api/v1/ping", "含" if any(p.startswith("/api/v1") for p in paths) else "不含（文档门槛已陈旧）") except Exception as exc: # noqa: BLE001 info(G, "5-02a 运行时 openapi 探测", f"失败：{type(exc).__name__}: {exc}") else: skip(G, "5-02 验收标准陈旧门槛取证", f"文档不存在：{std}") # ── 5-03 失效脚本仍在仓库 ── for rel, note in (("scripts/demo/run_demo_smoke.ps1", "唯一 demo 冒烟脚本"), ("scripts/eval/evaluate_agent.py", "第 49/59 行同打 /api/v1/*")): p = root / rel if not p.exists(): info(G, f"5-03 {rel}", "不存在", note) continue t = p.read_text(encoding="utf-8", errors="ignore") n = t.count("/api/v1/") info(G, f"5-03 {rel} 引用 /api/v1/ 次数", f"{n} 次", f"{note} ⇒ {'已失效（会 404）' if n else '无 /api/v1 引用'}") GROUPS = {"GX1": gx1, "GX2": gx2, "GX2b": gx2b, "GX3": gx3, "GX4": gx4, "GX5": gx5} ORDER = ["GX1", "GX2", "GX2b", "GX3", "GX4", "GX5"] def main() -> int: parser = argparse.ArgumentParser(description="孤立路由端到端补测（GX1–GX5）") parser.add_argument("--base-url", default="http://127.0.0.1:8000") parser.add_argument("--report", default=None) parser.add_argument("--only", action="append", default=None) parser.add_argument("--timeout", type=float, default=120.0) args = parser.parse_args() try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass ctx = Ctx(client=httpx.Client(base_url=args.base_url.rstrip("/"), timeout=args.timeout)) try: if not preflight(ctx, args.base_url): print("\n前置自检未通过 —— 未产生任何用例判定。") return 2 selected = [g for g in ORDER if not args.only or g in args.only] print(f"\n-- 执行分组: {', '.join(selected)} --\n") for name in selected: try: GROUPS[name](ctx) except Exception as exc: # noqa: BLE001 bad(name, f"{name}-00 分组执行异常", "分组正常跑完", f"{type(exc).__name__}: {exc}", traceback.format_exc()[-800:]) finally: ctx.client.close() counts = {s: sum(1 for r in e2e.RESULTS if r.status == s) for s in (PASS, FAIL, SKIP, INFO)} print(f"\n== 结果: {counts[PASS]} PASS / {counts[FAIL]} FAIL / " f"{counts[SKIP]} SKIP / {counts[INFO]} INFO ==") for r in e2e.RESULTS: if r.status == FAIL: print(f" - {r.group} · {r.name}: 期望 {r.expected} / 实际 {r.actual[:160]}") if args.report: e2e.write_report(Path(args.report)) print(f"\n报告已写入 {args.report}") return 1 if counts[FAIL] else 0 if __name__ == "__main__": sys.exit(main())