From 2fcc4939e70f5c4b512d76792badadda69ec4440 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Sun, 13 Sep 2026 21:39:37 +0800 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFfeat(tools):=20=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E5=85=A8=E6=B5=81=E7=A8=8B=E7=AB=AF=E5=88=B0=E7=AB=AF=E5=86=92?= =?UTF-8?q?=E7=83=9F=E8=84=9A=E6=9C=AC=20e2e=5Fsmoke=5Ftest.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把验收清单里"能用接口判定"的部分自动化：6 条线 40 项，打真实 HTTP、逐条打印 PASS/FAIL。 不是 pytest 的替代 —— 单测看代码正确性，它看"部署后底座是不是活的"。 覆盖： A 访客 令牌 → 提问 → 收到回答 → 未转人工 B 客户 登录 → 看板/持仓/流水 → 真实下单成交 → 委托详情 → 会话与转人工 C 风控 概览/队列/详情/八类证据/通知/扫描 → 处置闭环（确认→调查→结案） D 投顾 已发布方案 E 运营 场外邮件与邮箱状态 F 管理员 角色/权限/身份/配置发布/模型端点/审计/转人工工单/画像候选 两个实现细节值得留意： - **各端点 data 形状不统一**（角色列表裸数组、持仓在 holdings、成交明细在 transactions、 资金流水在 entries、知识列表不套 data），所以这里用 listed(payload, *keys) 依次尝试， 并在注释里写明"别猜字段名"—— 我自己写联调脚本时连续猜错三次。 - 库里没有"待处理"预警时**自动造一条**再验证处置闭环，否则该环节只能跳过 （实测库里的预警常常都已被处置过，第一次跑就跳过了）。 新增 --read-only：跳过所有写操作，用于生产或不想动数据时。 实测：40/40 通过（含真实下单 201 已成交、风控处置闭环三段全 PASS）。 docs/40 补上"一键冒烟"一节。 --- docs/40-前端验收清单.md | 13 ++ tools/e2e_smoke_test.py | 436 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 449 insertions(+) create mode 100644 tools/e2e_smoke_test.py diff --git a/docs/40-前端验收清单.md b/docs/40-前端验收清单.md index 17da9a3..a93a571 100644 --- a/docs/40-前端验收清单.md +++ b/docs/40-前端验收清单.md @@ -65,6 +65,19 @@ python -m uvicorn app.main:app --host 127.0.0.1 --port 8000 python -m app.worker ``` +### 一键冒烟（推荐先跑这个） + +前置跑完后，用一条命令确认 6 条线都是活的： + +```powershell +python tools/e2e_smoke_test.py # 含写操作（下单、风控处置） +python tools/e2e_smoke_test.py --read-only # 只看读链路，不动数据 +``` + +它打真实 HTTP、逐条打印 PASS/FAIL（**40 项**，覆盖访客/客户/风控/投顾/运营/管理员六条线）， +并在库里没有"待处理"预警时**自动造一条**来验证风控处置闭环。 +它**不是** pytest 的替代 —— 单测看代码正确性，它看"部署后底座是不是活的"。 + > 模块级变量是 **`app`**，不是 `application`。 --- diff --git a/tools/e2e_smoke_test.py b/tools/e2e_smoke_test.py new file mode 100644 index 0000000..9eee92f --- /dev/null +++ b/tools/e2e_smoke_test.py @@ -0,0 +1,436 @@ +"""平台端到端冒烟：按角色分线跑一遍真实请求，报告哪一环坏了。 + +## 为什么要有它 + +`docs/40` 的验收清单是**人工步骤 + 预期值**；本脚本把这些步骤里"能用接口判定"的部分 +自动化，用于**部署后、演示前**快速确认底座是活的。它不做断言式测试（那是 pytest 的活）， +而是**打真实 HTTP、打印每条的结果**，让人一眼看到哪条线断了。 + +## 覆盖范围（6 条线） + + A 访客 令牌 → 提问 → 收到回答 + B 客户 登录 → 看板/持仓/流水 → **真实下单成交** → 会话与转人工 + C 风控 概览/队列/详情/**八类证据**/通知/扫描/日报 → **处置闭环**（确认→调查→结案） + D 投顾 已发布方案 + E 运营 场外邮件与邮箱状态 + F 管理员 角色/权限/身份/配置发布/模型端点/审计/转人工工单/画像候选 + +## 前置（缺一项就有整条线是红的） + + python tools/seed_test_rbac.py # 账号与权限 + python tools/set_user_password.py # 演示口令（非幂等） + python -m tools.seed_sim_account_demo # 虚拟资金账户 + python tools/publish_risk_agent_config.py # 风控工具白名单 + python tools/sync_market_prices.py # **下单前置**：没有行情会全线 503 + python -m app.worker # **客服对话前置**：没有它 run 永远 queued + +## 用法 + + python tools/e2e_smoke_test.py # 全跑（含写操作：下单、风控处置） + python tools/e2e_smoke_test.py --read-only # 跳过所有写操作，只看读链路 + +写操作都会带唯一 `Idempotency-Key`；风控处置会在需要时**自动造一条待处理预警** +（否则该环节只能跳过 —— 库里的预警常常都已被处置过）。 +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +import urllib.error +import urllib.request +import uuid +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from sqlalchemy import func, select # noqa: E402 + +from app.infrastructure.db import SessionFactory # noqa: E402 +from app.model.fund import FundRiskAlert # noqa: E402 + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + +BASE = "http://127.0.0.1:8000" +DEMO_CUSTOMER = ("cust_t", "123456") +DEMO_RISK = ("risk_t", "666666") +DEMO_ADVISOR = ("advisor_t", "abc12345") +DEMO_OPERATOR = ("offsite_t", "offsite123") +DEMO_ADMIN = ("admin_t", "88888888") + +RESULTS: list[tuple[str, bool, str]] = [] +READ_ONLY = False + + +def call( + method: str, + path: str, + token: str | None = None, + body: dict[str, Any] | None = None, + *, + idem: bool = False, + timeout: int = 90, +) -> tuple[int, dict[str, Any]]: + data = json.dumps(body).encode("utf-8") if body is not None else None + request = urllib.request.Request(BASE + path, data=data, method=method) + if data is not None: + request.add_header("Content-Type", "application/json") + if token: + request.add_header("Authorization", "Bearer " + token) + if idem: + 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") + return response.status, (json.loads(raw) if raw else {}) + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", errors="replace") + try: + return exc.code, json.loads(raw) + except ValueError: + return exc.code, {"_raw": raw[:250]} + except Exception as exc: # noqa: BLE001 - 网络层失败也要报出来，而不是崩掉整轮 + return 0, {"_err": repr(exc)} + + +def check(label: str, ok: bool, detail: str = "") -> bool: + RESULTS.append((label, ok, detail)) + print(f"{'PASS' if ok else 'FAIL'} {label}" + (f" —— {detail}" if detail else "")) + return ok + + +def info(label: str, detail: str) -> None: + print(f"INFO {label} —— {detail}") + + +def section(title: str) -> None: + print() + print("=" * 96) + print(title) + print("=" * 96) + + +def body_of(payload: dict[str, Any]) -> dict[str, Any]: + data = payload.get("data") + return data if isinstance(data, dict) else {} + + +def listed(payload: dict[str, Any], *keys: str) -> list[Any]: + """从响应里取列表。 + + ⚠️ 本平台各端点的 `data` 形状**不统一**：角色列表是裸数组、持仓在 `holdings`、 + 成交明细在 `transactions`、**资金流水在 `entries`**。所以这里按候选键依次尝试 —— + 写联调脚本时别猜字段名，会连续猜错（实测踩过三次）。 + """ + data = payload.get("data") + if isinstance(data, list): + return data + if isinstance(data, dict): + for key in keys: + value = data.get(key) + if isinstance(value, list): + return value + for key in keys: + value = payload.get(key) + if isinstance(value, list): + return value + return [] + + +def ok_code(payload: dict[str, Any]) -> bool: + return payload.get("code") in (0, None) + + +def login(credentials: tuple[str, str]) -> str | None: + status, payload = call( + "POST", "/api/v1/auth/tokens", + body={"username": credentials[0], "password": credentials[1]}, + ) + return body_of(payload).get("access_token") if status == 200 else None + + +def create_pending_alert() -> str | None: + """造一条「待处理」预警，供处置闭环验证（库里的常常都已被处置过）。""" + + async def run() -> str: + alert_no = f"ALDEMO{datetime.now(UTC).strftime('%H%M%S')}" + now = datetime.now(UTC).replace(tzinfo=None) + async with SessionFactory() as session: + async with session.begin(): + max_id = await session.scalar( + select(func.coalesce(func.max(FundRiskAlert.id), 0)) + ) + session.add(FundRiskAlert( + id=int(max_id or 0) + 1, alert_no=alert_no, customer_id=9001, + alert_type="频繁交易", alert_level="高", + trigger_rule_codes=["RW-DEMO-01"], + evidence_summary="端到端冒烟脚本创建的演示预警，可安全处置。", + evidence_snapshot={"demo": True}, + priority_score=90, event_status="待处理", status="待处理", + ack_status="未确认", is_escalated=0, + due_at=now + timedelta(hours=24), created_at=now, updated_at=now, + )) + return alert_no + + try: + import asyncio + + return asyncio.run(run()) + except Exception as exc: # noqa: BLE001 + info("造预警失败", f"{type(exc).__name__}: {exc}") + return None + + +# ---------------------------------------------------------------- A 访客 + +def section_guest() -> None: + section("A. 访客线") + status, payload = call("POST", "/api/v1/visitor-tokens") + data = payload if isinstance(payload.get("access_token"), str) else payload.get("data") + token = (data or {}).get("access_token") if isinstance(data, dict) else None + check("A1 访客令牌 V001", bool(token), f"HTTP={status}") + if not token: + return + started = time.perf_counter() + status, accepted = call("POST", "/api/v1/agent-runs", token, { + "agent_type": "customer_service", "session_id": str(uuid.uuid4()), + "message": "场内基金的管理费率是多少", "idempotency_key": uuid.uuid4().hex, + }) + run_id = body_of(accepted).get("run_id") + check("A2 提问受理 R001", status == 202 and bool(run_id), f"HTTP={status}") + if not run_id: + return + state, snapshot = None, {} + for attempt in range(40): + time.sleep(0.5 if attempt else 0.3) + _, snapshot = call("GET", f"/api/v1/agent-runs/{run_id}", token) + state = body_of(snapshot).get("status") + if state in {"succeeded", "failed", "cancelled"}: + break + elapsed = time.perf_counter() - started + check("A3 问答完成", state == "succeeded", f"status={state} 用时 {elapsed:.2f}s") + if state != "succeeded": + info("A3 提示", "若 status=queued：Agent Worker 没在跑（python -m app.worker）") + return + result = body_of(snapshot).get("result") or {} + check("A4 知识库覆盖（未转人工）", result.get("transfer_required") is False, + f"transfer_required={result.get('transfer_required')}") + + +# ---------------------------------------------------------------- B 客户 + +def section_customer() -> None: + section("B. 客户线") + token = login(DEMO_CUSTOMER) + check("B1 客户登录 A034", bool(token)) + if not token: + return + + status, payload = call("GET", "/api/v1/onboarding/risk-questionnaire", token) + check("B2 风险测评问卷 ONB001", status == 200 and ok_code(payload), f"HTTP={status}") + + status, payload = call("GET", "/api/v1/users/me/account/dashboard", token) + data = body_of(payload) + account, summary = data.get("account") or {}, data.get("summary") or {} + check("B3 账户看板 T001", status == 200 and bool(account), f"HTTP={status}") + if account: + info("B3 账户", json.dumps({ + "可用资金": account.get("available_cash"), "总资产": summary.get("total_asset"), + "持仓市值": summary.get("total_market_value")}, ensure_ascii=False)) + + status, payload = call("GET", "/api/v1/users/me/holdings", token) + holdings = listed(payload, "holdings") + check("B4 持仓 T006", status == 200 and ok_code(payload), f"条数={len(holdings)}") + + status, payload = call("GET", "/api/v1/users/me/cash-ledger", token) + ledger = listed(payload, "entries", "cash_ledger") + check("B5 资金流水 T009", status == 200 and ok_code(payload), f"条数={len(ledger)}") + + print() + print("--- 下单链路 ---") + if READ_ONLY: + info("B6 下单", "跳过（--read-only）") + else: + status, payload = call("POST", "/api/v1/users/me/orders", token, + {"product_code": "510300", "order_side": "buy", "quantity": 100}, + idem=True) + order = body_of(payload) + check("B6 买入下单 T002", status in (200, 201) and ok_code(payload), + f"HTTP={status} {payload.get('message') or ''}") + if not order: + info("B6 提示", "503 通常是行情过期：跑 python tools/sync_market_prices.py") + else: + info("B6 成交", json.dumps({ + "委托号": order.get("order_no"), "状态": order.get("status"), + "成交价": order.get("executed_price"), "金额": order.get("gross_amount")}, + ensure_ascii=False)) + status, _ = call("GET", f"/api/v1/users/me/orders/{order.get('order_no')}", token) + check("B7 委托详情 T004", status == 200, f"HTTP={status}") + + status, payload = call("GET", "/api/v1/users/me/transactions", token) + check("B8 成交明细 T007", status == 200 and ok_code(payload), + f"条数={len(listed(payload, 'transactions'))}") + + print() + print("--- 客服会话与转人工 ---") + status, payload = call("POST", "/api/v1/conversations", token, + {"agent_type": "customer_service"}, idem=True) + session_id = body_of(payload).get("session_id") + check("B9 创建会话 C001", status in (200, 201) and bool(session_id), + f"HTTP={status} session_id={session_id}") + if not session_id: + return + status, _ = call("GET", f"/api/v1/conversations/{session_id}", token) + check("B10 会话详情 C002", status == 200, f"HTTP={status}") + status, _ = call("GET", f"/api/v1/conversations/{session_id}/messages", token) + check("B11 会话消息 C003", status == 200, f"HTTP={status}") + if READ_ONLY: + info("B12 转人工", "跳过（--read-only）") + else: + status, payload = call( + "POST", f"/api/v1/conversations/{session_id}/handover-requests", token, + {"reason_code": "user_requested", "reason_detail": "端到端冒烟：客户主动请求人工"}, + idem=True) + check("B12 转人工 C005", status in (200, 201, 202) and ok_code(payload), + f"HTTP={status}") + + +# ---------------------------------------------------------------- C 风控 + +def section_risk() -> None: + section("C. 风控线") + token = login(DEMO_RISK) + check("C1 风控登录", bool(token)) + if not token: + return + + status, payload = call("GET", "/api/v1/risk/overview", token) + overview = body_of(payload) + check("C2 风险概览 RK001", status == 200 and ok_code(payload), + f"total={overview.get('total')} pending={overview.get('pending')} " + f"levels={overview.get('levels')}") + + status, payload = call("GET", "/api/v1/risk/alerts", token) + alerts = listed(payload, "items", "alerts") + check("C3 预警队列 RK002", status == 200 and ok_code(payload), f"条数={len(alerts)}") + + sources = ("customers", "products", "transactions", "capital_flows", + "holdings", "login_records", "alerts", "notifications") + passed = sum( + 1 for source in sources + if ok_code(call("GET", f"/api/v1/risk/evidence/{source}", token)[1]) + ) + check("C4 八类证据 RK004", passed == len(sources), f"{passed}/{len(sources)} 通过") + + status, payload = call("GET", "/api/v1/risk/notifications", token) + check("C5 通知记录 RK005", status == 200 and ok_code(payload), f"HTTP={status}") + + if READ_ONLY: + info("C6 手动扫描 RK006", "跳过（--read-only）") + info("C7 处置闭环", "跳过（--read-only）") + return + + status, payload = call("POST", "/api/v1/risk/alerts/scan", token, {}, idem=True) + check("C6 手动扫描 RK006", status == 200 and ok_code(payload), + f"新建 {body_of(payload).get('created_count')} 条") + + print() + print("--- 风控处置闭环 ---") + target = next( + (item.get("alert_no") for item in alerts if item.get("status") == "待处理"), None + ) + if target is None: + target = create_pending_alert() + info("造预警", f"{target}（库里没有待处理预警，现造一条）") + if target is None: + info("C7 处置闭环", "无法获得待处理预警，跳过") + return + for label, path, body in ( + ("C7 确认接收 RK007", f"/api/v1/risk/alerts/{target}/acknowledgements", {}), + ("C8 进入调查 RK008", f"/api/v1/risk/alerts/{target}/investigations", {}), + ("C9 完成结案 RK010", f"/api/v1/risk/alerts/{target}/resolutions", + {"resolution": "端到端冒烟：已核实为演示数据并完成处置"}), + ): + status, payload = call("POST", path, token, body, idem=True) + check(label, status == 200 and ok_code(payload), + f"{target} HTTP={status} {payload.get('message') or ''}") + + +# ---------------------------------------------------------------- D/E/F + +def section_advisor_operator_admin() -> None: + section("D. 投顾线") + advisor = login(DEMO_ADVISOR) + check("D1 投顾登录", bool(advisor)) + if advisor: + status, payload = call("GET", "/api/v1/advisor/recommendations/published", advisor) + check("D2 已发布方案", status == 200 and ok_code(payload), + f"条数={len(listed(payload))}") + + section("E. 运营线") + operator = login(DEMO_OPERATOR) + check("E1 运营登录", bool(operator)) + if operator: + status, payload = call("GET", "/api/v1/offsite-fund/mails?page=1&page_size=8", operator) + check("E2 场外邮件", status == 200 and ok_code(payload), + f"total={body_of(payload).get('total')}") + status, payload = call("GET", "/api/v1/offsite-fund/mailbox-status", operator) + check("E3 邮箱状态", status == 200 and ok_code(payload), + f"status={body_of(payload).get('status')}") + + section("F. 管理员线") + admin = login(DEMO_ADMIN) + check("F1 管理员登录", bool(admin)) + if not admin: + return + for label, path in ( + ("F2 角色列表 A035", "/api/v1/admin/roles"), + ("F3 角色详情 A036", "/api/v1/admin/roles/customer"), + ("F4 角色权限 A037", "/api/v1/admin/roles/customer/permissions"), + ("F5 用户身份 A038", "/api/v1/admin/users/9001/roles"), + ("F6 配置发布 A002", "/api/v1/admin/config-releases"), + ("F7 模型端点 A012", "/api/v1/admin/model-endpoints"), + ("F8 审计记录 A033", "/api/v1/admin/audit-records"), + ("F9 转人工工单", "/api/v1/admin/customer-service/handover-tickets"), + ("F10 画像候选 A039", "/api/v1/admin/customer-profile-candidates"), + ): + status, payload = call("GET", path, admin) + data = payload.get("data") + shape = (f"list[{len(data)}]" if isinstance(data, list) + else f"dict[{len(data)}]" if isinstance(data, dict) else "?") + check(label, status == 200 and ok_code(payload), f"HTTP={status} {shape}") + + +def main() -> int: + global READ_ONLY # noqa: PLW0603 - 单进程脚本，命令行开关就设一次 + parser = argparse.ArgumentParser(description="平台端到端冒烟") + parser.add_argument("--read-only", action="store_true", + help="跳过所有写操作（下单、转人工、风控扫描与处置）") + args = parser.parse_args() + READ_ONLY = args.read_only + + print(f"目标 {BASE}" + ("（只读模式）" if READ_ONLY else "（含写操作）")) + section_guest() + section_customer() + section_risk() + section_advisor_operator_admin() + + passed = sum(1 for _, ok, _ in RESULTS if ok) + section(f"汇总：{passed}/{len(RESULTS)} 通过") + for label, ok, detail in RESULTS: + if not ok: + print(f" FAIL {label} {detail}") + if passed == len(RESULTS): + print("全部通过。") + return 0 if passed == len(RESULTS) else 1 + + +if __name__ == "__main__": + sys.exit(main())