"""门户接口体检：按前端的方式调用每个端点，核对状态、信封与字段。 ## 为什么需要它 2026-09-13 做过一轮"照着接口内容把前端测一遍"，靠它查出 6 个真缺陷： - `K002`/`K003` 是**裸信封**，前端按 `payload.data` 取 ⇒ 知识库页显示"已入库 0 块 / 为空"； - 委托与成交详情页照着**建表字段**写，而接口返回视图没带那些列 ⇒ 5 行永远显示 `--`； - 配置项/路由规则的 `PUT` 要求 `If-Match`，却没有端点能返回该 digest ⇒ **首次编辑必然 409**； - 回复模板 `scene` 只校验长度、不校验枚举 ⇒ 非法值撞数据库 CHECK、**冒成 500**； - 路由规则表单固定 `max_attempts=2` 且 `fallbacks` 为空 ⇒ 后端要求"重试次数不超过端点数量"、**必然 422**； - 模型端点手填 ID ⇒ 未激活的 ID 直接 422。 **那些验证原本只存在于一次会话里** —— 下次改动没人会重跑，同样的坑会再踩一遍。 本脚本把它们固化下来。 ## 与 `e2e_smoke_test.py` 的分工 | 脚本 | 回答的问题 | |---|---| | `e2e_smoke_test.py` | **业务链路通不通**：登录→下单→成交、风控扫描→处置闭环、客服问答 | | `portal_api_check.py`（本脚本） | **接口契约对不对**：字段名是否与前端一致、`data` 是 list 还是 dict、错误码是 422 还是 500、参数边界 | 两者互补，都跑一遍最稳。**前置与冒烟相同**（见 `e2e_smoke_test.py` 的"前置"一节， 其中 `sync_market_prices.py` 与常驻 Worker 尤其关键）。 ## 已作废用例（2026-09-20） 投顾模块整体清除后，原 `AD003` / `AD005` / `AD011` / `A047` 四条用例与 `advisor_t` 登录 **一并作废**（端点与账号都已不存在）。本脚本改为显式 `skip`，不再报假红。 **别把这几条的消失当成"少测了"** —— 它们对应的产品表面已经不在了。 ## 三档（默认只跑第一档） python tools/portal_api_check.py # 只读：GET 与只读分析，不动数据 python tools/portal_api_check.py --write # 加测写操作（会产生数据） python tools/portal_api_check.py --dangerous # 再加**会改生效配置**的操作 ⚠️ `--dangerous` 那一档包括**激活配置版本**。2026-09-13 有一次就是激活了空版本， `config_release` 是整版本替换语义 ⇒ 所有配置项（客服/风控的工具白名单）失效 ⇒ **Agent 全线失败关闭**。跑之前请确认你知道怎么恢复（用 `tools/publish_*.py` 重新发布， 或把上一个版本的配置项复制进新版本 —— 注意 `rollbacks` **不复制配置项**，救不回来）。 ## 退出码 0 = 全部通过；1 = 有 FAIL（`NOTE` 是已知问题，不影响退出码）。 """ from __future__ import annotations import argparse import base64 import json import sys import urllib.error import urllib.request import uuid from pathlib import Path from typing import Any PROJECT_ROOT = Path(__file__).resolve().parents[1] if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] BASE = "http://127.0.0.1:8000" DEMO_ACCOUNTS = { "customer": ("cust_t", "123456"), "risk": ("risk_t", "666666"), "operator": ("offsite_t", "offsite123"), "admin": ("admin_t", "88888888"), } #: **已作废角色（2026-09-20）**：投顾模块已于 2026-09-17 整体清除，`advisor_t` 账号 #: **已不存在**（`D4.4` / `D4.5` 清除报告；`客服agent\D2.5` §2.1 亦标注"不要念它"）。 #: 本脚本写于 2026-09-13，早于该清除 ⇒ 此前它会稳定报 3 条 FAIL #: （`advisor_t` 登录 / `AD011` 已发布方案 / `A047` 投顾待审）。 #: 那 3 条不是缺陷，是**脚本过期**：对应端点已随模块删除，前端 `api-client.js` #: 的端点表也早已不含它们（由 `tests/unit/api/test_frontend_boundaries.py` 的 #: 端点表 ↔ OpenAPI 对照守住）。这里按"作废"处理并显式 skip，避免制造假红。 PURGED_ROLES: dict[str, str] = { "advisor": "投顾模块已整体清除（D4.4/D4.5），账号与端点均不存在", } RESULTS: list[tuple[str, str, str]] = [] TOKENS: dict[str, str] = {} CONTEXT: dict[str, Any] = {} DO_WRITE = False DO_DANGEROUS = False # ---------------------------------------------------------------- 基础设施 def call( method: str, path: str, token: str | None = None, body: Any = None, *, headers: dict[str, str] | None = None, expect_json: bool = True, timeout: int = 60, ) -> tuple[int, Any, dict[str, str]]: """发一个与前端行为一致的请求（写操作自动带 Idempotency-Key）。""" data = json.dumps(body).encode() if body is not None else None request = urllib.request.Request(BASE + path, data=data, method=method) request.add_header("Accept", "application/json") if data is not None: request.add_header("Content-Type", "application/json") if token: request.add_header("Authorization", f"Bearer {token}") for key, value in (headers or {}).items(): request.add_header(key, value) if method != "GET": request.add_header("Idempotency-Key", uuid.uuid4().hex) try: with urllib.request.urlopen(request, timeout=timeout) as response: raw = response.read().decode("utf-8", errors="replace") payload = json.loads(raw) if (expect_json and raw.strip()) else raw return response.status, payload, dict(response.headers) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: return exc.code, json.loads(raw) if raw.strip() else {}, dict(exc.headers) except Exception: return exc.code, {"_raw": raw[:200]}, dict(exc.headers) except Exception as exc: # noqa: BLE001 return 0, {"error": {"code": type(exc).__name__, "message": str(exc)}}, {} def login(username: str, password: str) -> str | None: status, payload, _ = call( "POST", "/api/v1/auth/tokens", body={"username": username, "password": password} ) return (payload.get("data") or {}).get("access_token") if status == 200 else None def record(name: str, ok: bool, detail: str) -> None: RESULTS.append(("PASS" if ok else "FAIL", name, detail)) def note(name: str, detail: str) -> None: """已知问题或设计如此：既不算通过也不算失败。""" RESULTS.append(("NOTE", name, detail)) def skip(name: str, detail: str) -> None: RESULTS.append(("SKIP", name, detail)) # ---------------------------------------------------------------- data 形状 def unwrap(payload: Any) -> Any: """前端 `request()` 取的是 `payload.data`（端点标了 `raw` 的除外）。""" return payload["data"] if isinstance(payload, dict) and "data" in payload else payload def resolve_row( payload: Any, *, nested: str | None = None, list_key: str | None = None ) -> dict[str, Any]: """定位"被渲染的那一行"。 `data` 在本平台有三种形状，前端三种都要处理：**裸 list**、**dict[list_key]**、 **dict（单对象）**。把它们区分开是本脚本的第一价值 —— 曾经就是这里判断错， 把三个正常接口报成了"缺字段"。 """ data = unwrap(payload) if nested: return (data or {}).get(nested) or {} if isinstance(data, list): return data[0] if data else {} if isinstance(data, dict): if list_key: rows = data.get(list_key) if isinstance(rows, list): return rows[0] if rows else {} return data if list_key and isinstance(payload, dict): rows = payload.get(list_key) if isinstance(rows, list): return rows[0] if rows else {} return {} def check( name: str, method: str, path: str, role: str | None = None, *, expect: int | tuple[int, ...] = 200, fields: tuple[str, ...] = (), nested: str | None = None, list_key: str | None = None, ) -> Any: """调用并核对：状态码 +（可选）前端 render 真正读取的字段。""" token = TOKENS.get(role) if role else TOKENS.get("guest") status, payload, _ = call(method, path, token) expected = (expect,) if isinstance(expect, int) else expect ok = status in expected detail = f"HTTP={status}" if not ok: code = (payload.get("error") or {}).get("code") if isinstance(payload, dict) else None detail += f" {code or ''} 期望 {expected}" elif fields: row = resolve_row(payload, nested=nested, list_key=list_key) missing = [f for f in fields if f not in (row or {})] if missing: ok = False detail += f" 缺字段={missing}" else: detail += f" 字段齐全({len(fields)})" data = unwrap(payload) if isinstance(data, list): detail += f" 共{len(data)}条" elif list_key and isinstance(data, dict) and isinstance(data.get(list_key), list): detail += f" 共{len(data[list_key])}条" record(name, ok, detail) return payload # ---------------------------------------------------------------- 只读组 def check_guest_readonly() -> None: check("访客：公开产品 P001", "GET", "/api/v1/products", list_key="products", fields=("product_code", "product_name", "exchange_code", "product_category", "risk_level", "current_nav", "change_pct")) # 列表接口的 data 是 list（前端 `state.products = data`），不是 {products: [...]} listing = unwrap(call("GET", "/api/v1/products?limit=20")[1]) rows = listing.get("products") if isinstance(listing, dict) else listing product_code = (rows or [{}])[0].get("product_code") or "510300" check("访客：净值走势 P002", "GET", f"/api/v1/products/{product_code}/nav-history?days=120", list_key="points", fields=("nav_date", "nav")) def check_customer_readonly() -> None: check("客户：资产看板 T001·account", "GET", "/api/v1/users/me/account/dashboard", "customer", nested="account", fields=("available_cash",)) check("客户：资产看板 T001·summary", "GET", "/api/v1/users/me/account/dashboard", "customer", nested="summary", fields=("total_asset", "total_market_value")) check("客户：持仓 T006", "GET", "/api/v1/users/me/holdings", "customer", list_key="holdings", fields=("product_code", "product_name", "total_quantity", "average_cost", "market_value")) check("客户：委托列表 T003", "GET", "/api/v1/users/me/orders?limit=20", "customer", fields=("order_no", "product_name", "product_code", "order_side", "quantity", "average_executed_price", "quote_price", "status", "submitted_at")) check("客户：成交列表 T007", "GET", "/api/v1/users/me/transactions?limit=20", "customer", list_key="transactions", fields=("transaction_no", "order_no", "product_name", "product_code", "order_side", "executed_quantity", "executed_price", "gross_amount", "fee_amount", "net_amount", "executed_at")) check("客户：资金流水 T009", "GET", "/api/v1/users/me/cash-ledger?limit=20", "customer", list_key="entries") check("客户：风险测评 ONB001", "GET", "/api/v1/onboarding/risk-questionnaire", "customer") # 详情类：需要真实单据号，从列表里取 orders = resolve_row( call("GET", "/api/v1/users/me/orders?limit=1", TOKENS.get("customer"))[1] ) order_no = orders.get("order_no") if order_no: check("客户：委托详情 T004", "GET", f"/api/v1/users/me/orders/{order_no}", "customer", fields=("order_no", "product_name", "product_code", "order_side", "price_type", "quantity", "filled_quantity", "quote_price", "average_executed_price", "status", "submitted_at", "quote_at")) else: skip("客户：委托详情 T004", "库里没有委托，先跑一次下单或 e2e_smoke_test") txns = unwrap(call("GET", "/api/v1/users/me/transactions?limit=1", TOKENS.get("customer"))[1]) txn_no = (txns.get("transactions") or [{}])[0].get("transaction_no") if isinstance(txns, dict) else None if txn_no: check("客户：成交详情 T008", "GET", f"/api/v1/users/me/transactions/{txn_no}", "customer", fields=("transaction_no", "order_no", "product_name", "product_code", "order_side", "executed_price", "executed_quantity", "gross_amount", "fee_amount", "net_amount", "quote_at", "executed_at")) else: skip("客户：成交详情 T008", "库里没有成交记录") def check_risk_readonly() -> None: check("风控：概览 RK001", "GET", "/api/v1/risk/overview", "risk", fields=("total", "pending", "levels")) # ⚠️ limit 上限是 5（RiskAlertPageQuery），传大了会 422 —— 前端传的正是 5 check("风控：预警列表 RK002（limit=5）", "GET", "/api/v1/risk/alerts?limit=5", "risk", fields=("alert_no", "risk_level", "customer_no", "product_name", "alert_type", "status")) alerts = call("GET", "/api/v1/risk/alerts?limit=1", TOKENS.get("risk"))[1] rows = unwrap(alerts) if isinstance(unwrap(alerts), list) else [] alert_no = rows[0].get("alert_no") if rows else None if alert_no: check("风控：预警详情 RK003", "GET", f"/api/v1/risk/alerts/{alert_no}", "risk") # ⚠️ source 是**枚举**：customers/products/transactions/capital_flows/holdings/ # login_records/alerts/notifications —— 传别的会 422 check("风控：证据 RK004（source=customers）", "GET", f"/api/v1/risk/evidence/customers?alert_no={alert_no}&limit=10", "risk") else: skip("风控：预警详情 RK003 / 证据 RK004", "库里没有预警") check("风控：通知 RK005（limit=10）", "GET", "/api/v1/risk/notifications?limit=10", "risk") def check_admin_readonly() -> None: check("管理：配置发布 A002", "GET", "/api/v1/admin/config-releases?limit=20", "admin", fields=("id", "release_no", "title", "status", "created_by", "updated_at")) check("管理：模型端点 A012", "GET", "/api/v1/admin/model-endpoints?limit=20", "admin") check("管理：审计 A033", "GET", "/api/v1/admin/audit-records?limit=20", "admin") check("管理：角色列表 A035", "GET", "/api/v1/admin/roles", "admin") check("管理：角色权限 A037", "GET", "/api/v1/admin/roles/customer/permissions", "admin") check("管理：用户身份 A038", "GET", "/api/v1/admin/users/9001/roles", "admin") check("管理：画像候选 A039", "GET", "/api/v1/admin/customer-profile-candidates?limit=20", "admin") check("管理：转人工工单", "GET", "/api/v1/admin/customer-service/handover-tickets?limit=20", "admin") # ⚠️ K003 是**裸信封**（{items, count}），这是前端最容易搞错的一处 status, payload, _ = call("GET", "/api/v1/knowledge/list?limit=50", TOKENS.get("admin")) bare = isinstance(payload, dict) and "items" in payload and "data" not in payload record("管理：知识库 K003（裸信封）", status == 200 and bare, f"HTTP={status} {'顶层 items：符合裸信封' if bare else '信封形状变了，前端按裸体取会拿不到数据'}") # 配置项/路由规则列表：顺带取一个 release_id 给写入档用 releases = unwrap(call("GET", "/api/v1/admin/config-releases?limit=20", TOKENS.get("admin"))[1]) if isinstance(releases, list) and releases: CONTEXT["release_id"] = releases[0].get("id") check("管理：配置项列表 A009", "GET", f"/api/v1/admin/config-releases/{CONTEXT['release_id']}/platform-config-items", "admin") check("管理：路由规则列表 A019", "GET", f"/api/v1/admin/config-releases/{CONTEXT['release_id']}/model-routing-rules", "admin") def check_operator_readonly() -> None: check("运营：场外邮件列表", "GET", "/api/v1/offsite-fund/mails?limit=20", "operator") check("运营：邮箱状态", "GET", "/api/v1/offsite-fund/mailbox-status", "operator") # ---------------------------------------------------------------- 写入档 def check_write() -> None: admin = TOKENS.get("admin") # ---- 配置版本 + 配置项（含 If-Match 链路）---- status, payload, _ = call("POST", "/api/v1/admin/config-releases", admin, { "release_no": f"CHECK-{uuid.uuid4().hex[:8]}", "title": "接口体检版本", "change_summary": "由 portal_api_check.py --write 创建，用于验证配置项与路由规则的增改链路。"}) release_id = (payload.get("data") or {}).get("id") record("写入：新建配置版本 A001", status == 201 and bool(release_id), f"HTTP={status} id={release_id}") if not release_id: return CONTEXT["check_release_id"] = release_id # 配置项的 key 必须是「已注册 agent_type:声明的意图」，而且白名单**不能超出 # 该 Agent 的代码上限** —— `admin_service` 会拿 `AgentFactory.definition()` 校验 # `allowed_tools`，手编一个工具名必然 422「配置超出 Agent 工具上限」。 # （我第一版就写死了 `search_knowledge`，结果选中的是风控 Agent，直接被打回。） # 正确做法：从接口声明的意图配置里取工具，保证落在上限内 —— 这同时也验证了 # 「意图配置的 allowed_tools 能被配置项接口接受」这条跨接口一致性。 raw_intents = unwrap(call("GET", "/api/v1/admin/agent-intent-configs?limit=100", admin)[1]) or [] if isinstance(raw_intents, dict): raw_intents = (raw_intents.get("items") or raw_intents.get("intent_configs") or raw_intents.get("records") or []) usable = [ row for row in raw_intents if isinstance(row, dict) and row.get("agent_type") and row.get("intent_code") and row.get("allowed_tools") ] if usable: seed = usable[0] agent = str(seed["agent_type"]) intent = str(seed["intent_code"]) tools = [str(t) for t in seed["allowed_tools"]][:1] else: agent, intent, tools = "customer_service", "faq", [] item_key = f"{agent}:{intent}" if not tools: skip("写入：新增配置项 A008", "没有声明了工具的白名单意图，无法构造合法配置项") else: status, payload, _ = call( "POST", f"/api/v1/admin/config-releases/{release_id}/platform-config-items", admin, {"namespace": "agent_tools", "item_key": item_key, "value_json": {"allowed_tools": tools}, "schema_version": "1"}) item_id = (payload.get("data") or {}).get("id") error = (payload.get("error") or {}) record("写入：新增配置项 A008", status == 201 and bool(item_id), f"HTTP={status} id={item_id} key={item_key} tools={tools} " f"{error.get('code') or ''} {error.get('message') or ''}") if item_id: # 更新必须带 If-Match；etag 只能从**详情端点**取（列表的 meta 没有它） detail = call("GET", f"/api/v1/admin/config-releases/{release_id}/platform-config-items/{item_id}", admin)[1] etag = ((detail.get("meta") or {}).get("etag")) if isinstance(detail, dict) else None record("写入：配置项详情取 ETag A048", bool(etag), f"etag={'有' if etag else '没有'}（没有就说明 PUT 链路是死的）") if etag: status, payload, _ = call( "PUT", f"/api/v1/admin/config-releases/{release_id}/platform-config-items/{item_id}", admin, {"namespace": "agent_tools", "item_key": item_key, "value_json": {"allowed_tools": tools}, "schema_version": "1"}, ) # 注意：这里**故意不带 If-Match** —— 期望 409，证明乐观并发没被削弱 record("写入：不带 If-Match 应 409", status == 409, f"HTTP={status}") # ---- 路由规则：max_attempts 不能超过端点数量 ---- endpoints = unwrap(call("GET", "/api/v1/admin/model-endpoints?limit=20", admin)[1]) or [] active = [e for e in endpoints if e.get("status") == "active"] if active: endpoint_id = active[0]["id"] status, payload, _ = call( "POST", f"/api/v1/admin/config-releases/{release_id}/model-routing-rules", admin, {"rule_code": f"check-{uuid.uuid4().hex[:8]}", "agent_type": agent, "task_type": "intent_classify", "model_policy": "primary_only", "primary_endpoint_id": endpoint_id, "fallbacks": [], "max_attempts": 1, "latency_budget_ms": 15000, "priority": 100}) record("写入：新增路由规则 A018", status == 201, f"HTTP={status} endpoint={endpoint_id} {(payload.get('error') or {}).get('message') or ''}") status, payload, _ = call( "POST", f"/api/v1/admin/config-releases/{release_id}/model-routing-rules", admin, {"rule_code": f"check-bad-{uuid.uuid4().hex[:8]}", "agent_type": agent, "task_type": "intent_classify", "model_policy": "primary_only", "primary_endpoint_id": endpoint_id, "fallbacks": [], "max_attempts": 2, "latency_budget_ms": 15000, "priority": 100}) # fallbacks 为空时 max_attempts=2 必须被拒 —— 前端曾固定写 2，导致表单必然失败 record("写入：max_attempts 超端点数量应 422", status == 422, f"HTTP={status}") else: skip("写入：路由规则 A018", "没有已激活的模型端点") # ---- 知识库（裸信封）---- text = "# 接口体检文档\n\n## 一、目的\n\n" + ("这用于确认上传能被切分并写入知识表。" * 12) + "\n" status, payload, _ = call("POST", "/api/v1/knowledge/upload", admin, { "filename": f"check-{uuid.uuid4().hex[:8]}.md", "content_base64": base64.b64encode(text.encode()).decode(), "knowledge_type": "faq"}) ids = (payload.get("knowledge_ids") or []) if isinstance(payload, dict) else [] record("写入：知识库上传 K002（裸信封）", status == 201 and bool(ids), f"HTTP={status} 切分 {len(ids)} 块") for knowledge_id in ids: status, _, _ = call("DELETE", f"/api/v1/knowledge/{knowledge_id}", admin) record(f"写入：知识库失效 K004（id={knowledge_id}）", status == 200, f"HTTP={status}") # ---- 回复模板：scene 是枚举，非法值必须是 422 而不是 500 ---- status, payload, _ = call("POST", "/api/v1/admin/reply-templates", admin, { "template_code": f"ZZCHECK-{uuid.uuid4().hex[:6]}", "scene": "not_a_scene", "title": "体检用", "content_text": "不应被创建"}) record("写入：回复模板非法 scene 应 422（非 500）", status == 422, f"HTTP={status} ← 500 表示 schema 漏了枚举校验") # ---- 敏感词（用绝不会被说出的词）---- status, payload, _ = call("POST", "/api/v1/admin/negative-word-rules", admin, { "rule_code": f"ZZCHECK-{uuid.uuid4().hex[:6]}", "word_pattern": f"ZZCHECKTOKEN{uuid.uuid4().hex[:6]}", "match_type": "contains", "category": "体检", "severity": "warn"}) record("写入：创建敏感词规则", status == 201, f"HTTP={status} {(payload.get('error') or {}).get('message') or ''}") # ---- 推广物料：完整链路（合规会拦不合规输入，这里用合规的最小输入）---- status, payload, _ = call("POST", "/api/v1/fund-promotion-materials", admin, { "product_name": "红利低波50ETF南方", "product_code": "515450", "material_title": f"接口体检材料 {uuid.uuid4().hex[:6]}", "style_code": "steady_professional", "output_formats": ["pptx"]}) task_no = (payload.get("data") or {}).get("task_no") record("写入：创建推广任务", bool(task_no), f"HTTP={status} task_no={task_no}") if task_no: na = "不适用" status, payload, _ = call( "PUT", f"/api/v1/fund-promotion-materials/{task_no}/inputs", admin, { "product_info": {"fund_type": "股票型-指数", "operation_mode": "契约型开放式", "product_status": "new_product", "investment_objective": "紧密跟踪标的指数。", "benchmark": "红利低波50指数", "risk_level": "R3"}, "manager_info": {"manager_name": "张三", "management_company": "南方基金管理股份有限公司", "registration_code": "F0000000000000"}, "team_info": {"team_description": "指数投资团队。"}, "strategy_info": {"investment_scope": "标的指数成份股。", "strategy": "完全复制法。", "restrictions": "遵守合同约定。"}, "fee_structure": {"subscription_fee": na, "purchase_fee": na, "redemption_fee": na, "sales_service_fee": na, "management_fee": "0.50%/年", "custody_fee": "0.10%/年", "client_maintenance_fee": na}, "performance_info": {"show_product_performance": False, "show_manager_performance": False}, "risk_disclosure": {"special_risks": ["指数跟踪偏离风险"]}, "source_notes": {}}) record("写入：更新材料输入", status == 200, f"HTTP={status} {(payload.get('error') or {}).get('message') or ''}") status, payload, _ = call( "POST", f"/api/v1/fund-promotion-materials/{task_no}/generations", admin, {"output_formats": ["pptx"]}) version_id = (payload.get("data") or {}).get("material_version_id") or \ (payload.get("data") or {}).get("id") record("写入：生成推介材料", status in (200, 201) and bool(version_id), f"HTTP={status} version_id={version_id}") status, payload, _ = call( "GET", f"/api/v1/fund-promotion-materials/{task_no}/compliance-checks", admin) findings = ((unwrap(payload) or {}).get("findings") or []) if isinstance(unwrap(payload), dict) else [] record("写入：合规检查", status == 200, f"HTTP={status} findings={len(findings)}" + (f"（{findings[0].get('rule_code')}）" if findings else "")) # ---------------------------------------------------------------- 危险档 def check_dangerous() -> None: """会改**生效配置**的操作。默认不跑。""" admin = TOKENS.get("admin") release_id = CONTEXT.get("check_release_id") if not release_id: skip("危险：配置发布四态", "没有可用的测试版本（先跑 --write）") return base = f"/api/v1/admin/config-releases/{release_id}" def etag_headers() -> dict[str, str]: _, _, headers = call("GET", base, admin) value = headers.get("ETag") or headers.get("etag") or "" return {"If-Match": value} if value else {} # ⚠️ 这一档会把该版本**激活**。若该版本内容不全，生效配置会被替换掉， # 客服/风控的工具白名单可能随之失效（config_release 是整版本替换语义）。 status, payload, _ = call("POST", f"{base}/validations", admin, {}) record("危险：提交校验 A004", status in (200, 422), f"HTTP={status} {(payload.get('error') or {}).get('message') or ''}") if status != 200: note("危险：后续审核/激活", "校验未通过，按设计不再继续") return status, payload, _ = call("POST", f"{base}/reviews", admin, {"decision": "approved", "comment": "接口体检"}, headers=etag_headers()) record("危险：审核 A005", status == 200, f"HTTP={status} {(payload.get('error') or {}).get('message') or ''}") if status != 200: return note("危险：激活 A006", "**故意不执行** —— 激活会替换生效配置。" "需要时手工调，或先确认该版本已继承全部既有配置项。") # ---------------------------------------------------------------- 主流程 def main() -> int: global DO_WRITE, DO_DANGEROUS parser = argparse.ArgumentParser(description="门户接口体检") parser.add_argument("--write", action="store_true", help="加测写操作（会产生数据）") parser.add_argument("--dangerous", action="store_true", help="再加测会改生效配置的操作") args = parser.parse_args() DO_WRITE = args.write or args.dangerous DO_DANGEROUS = args.dangerous print("=" * 100) print("门户接口体检：按前端的方式调用端点，核对状态、信封与字段") print(f"档位：{'危险（含写）' if DO_DANGEROUS else '写入（含只读）' if DO_WRITE else '只读'}") print("=" * 100) # 前置：健康检查 + 访客令牌 + 五个角色登录 status, _, _ = call("GET", "/internal/health/ready") if status != 200: print(f"\n[失败] API 未就绪（GET /internal/health/ready -> {status}）。请先起服务。") return 1 status, payload, _ = call("POST", "/api/v1/visitor-tokens") # 访客令牌端点是 `raw` 形状：令牌在顶层，没有 data 信封 TOKENS["guest"] = (payload.get("data") or payload).get("access_token") if status == 201 else None record("前置：访客令牌 V001", bool(TOKENS["guest"]), f"HTTP={status}") for role, (username, password) in DEMO_ACCOUNTS.items(): token = login(username, password) TOKENS[role] = token record(f"前置：{role} 登录（{username}）", bool(token), "" if token else "登录失败，该角色全部用例会红") # 作废角色：显式 skip（见 PURGED_ROLES 的说明），不计入失败 for role, reason in PURGED_ROLES.items(): skip(f"前置：{role} 登录（{DEMO_ACCOUNTS.get(role, ('—',))[0]}）", reason) print() print("--- 只读：访客 / 客户 ---") check_guest_readonly() check_customer_readonly() print("--- 只读：风控 ---") check_risk_readonly() skip("投顾线（AD003 / AD005 / AD011 / A047）", "投顾模块已整体清除，用例随端点一并作废") print("--- 只读：管理员 / 运营 ---") check_admin_readonly() check_operator_readonly() if DO_WRITE: print("--- 写入档（会产生数据）---") check_write() else: skip("写入档", "未加 --write，跳过") if DO_DANGEROUS: print("--- 危险档（会改生效配置）---") check_dangerous() else: skip("危险档", "未加 --dangerous，跳过") # 汇总 print() print("=" * 100) for verdict, name, detail in RESULTS: print(f"{verdict:<6}{name:<40}{detail}") print("=" * 100) failed = [r for r in RESULTS if r[0] == "FAIL"] passed = [r for r in RESULTS if r[0] == "PASS"] noted = [r for r in RESULTS if r[0] == "NOTE"] skipped = [r for r in RESULTS if r[0] == "SKIP"] print(f"合计 {len(RESULTS)} 项：通过 {len(passed)}，失败 {len(failed)}，" f"已知问题 {len(noted)}，跳过 {len(skipped)}") if failed: print("\n失败项：") for _, name, detail in failed: print(f" · {name} {detail}") print("\n提示：字段缺失通常是「前端按建表字段写、接口返回视图没带」；") print(" 500 通常是「表上有 CHECK 而 schema 只做长度校验」；") print(" 422 通常是「前端传的参数超出接口约束（如 limit 上限）」。") return 1 print("\n全部通过。") if DO_WRITE: print("\n注意：写入档会留下测试数据（配置版本、敏感词等**没有删除端点**）。" "\n 命名前缀 CHECK-/ZZCHECK- 便于识别；知识库条目已在脚本内自动失效。") return 0 if __name__ == "__main__": sys.exit(main())