张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
626 lines
32 KiB
Python
626 lines
32 KiB
Python
"""门户接口体检：按前端的方式调用每个端点，核对状态、信封与字段。
|
||
|
||
## 为什么需要它
|
||
|
||
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 尤其关键）。
|
||
|
||
## 三档（默认只跑第一档）
|
||
|
||
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"),
|
||
"advisor": ("advisor_t", "abc12345"),
|
||
"operator": ("offsite_t", "offsite123"),
|
||
"admin": ("admin_t", "88888888"),
|
||
}
|
||
|
||
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_advisor_readonly() -> None:
|
||
check("投顾：已发布方案 AD011", "GET", "/api/v1/advisor/recommendations/published", "advisor",
|
||
fields=("content_id", "customer_id", "content_type", "plan", "published_at"))
|
||
goals = call("GET", "/api/v1/advisor/customers/9001/investment-goals/current",
|
||
TOKENS.get("advisor"))[1]
|
||
goal = unwrap(goals) or {}
|
||
if goal.get("goal_no"):
|
||
check("投顾：客户目标 AD003", "GET",
|
||
"/api/v1/advisor/customers/9001/investment-goals/current", "advisor",
|
||
fields=("goal_no", "customer_id", "status", "goal_book", "confirmed_at"))
|
||
check("投顾：目标方案书 AD005", "GET",
|
||
f"/api/v1/advisor/investment-goals/{goal['goal_no']}/goal-book", "advisor",
|
||
fields=("goal_no", "goal_status", "review_status", "content"))
|
||
else:
|
||
skip("投顾：客户目标 AD003 / 方案书 AD005", "客户 9001 还没有投资目标")
|
||
|
||
|
||
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")
|
||
check("管理：投顾待审 A047", "GET", "/api/v1/admin/advisor/pending-contents", "admin",
|
||
fields=("content_id", "customer_id", "content_type", "review_status", "goal_no"))
|
||
# ⚠️ 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 "登录失败，该角色全部用例会红")
|
||
|
||
print()
|
||
print("--- 只读：访客 / 客户 ---")
|
||
check_guest_readonly()
|
||
check_customer_readonly()
|
||
print("--- 只读：风控 / 投顾 ---")
|
||
check_risk_readonly()
|
||
check_advisor_readonly()
|
||
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())
|