lzf_0626
f37fd922a1
## 结论：risk_action_service 没问题，我先前是误判
esolve（完成结案）写全了 status="已结案" + closed_at + handle_result（risk_action_service.py:105-107），
exclude 同样（:67-71）。我报的"结案没写 closed_at"看到的那条 ALDEMO0003
其实是**种子预置样本**，不是我以为的自己创建的那条 —— 我的冒烟脚本用
ALDEMO{时分秒} 命名，**撞上了种子的 ALDEMO0001-0003**，"已存在则跳过"把创建挡掉，
随后处置全 409，串起来就像"结案没写时间"。
真正的问题在种子数据，两处：
1) **状态口径与代码不一致**：ALDEMO0003 写成 status="已关闭"，而
isk_action_service 里**没有任何动作会产生这个状态**（状态机只有
待处理 / 调查中 / 已排除 / 已结案）。演示时会被当成系统行为，排查时又查不到出处。
按它的 close_reason 语义（低风险 + 已核验为正常交易），它就是**关闭误报** → 改为"已排除"。
2) **终态缺 closed_at**：那条有 close_reason 却没有 closed_at，自相矛盾；
而且
isk_repository 按 closed_at 统计"当日结案"，缺了它会被漏掉。
INSERT 补上该列，样本用 closed_at_offset_minutes 给值。
## 顺带修掉冒烟脚本的撞号
预警编号前缀 ALDEMO → E2E，注释里写明原因 —— 免得下次又操作到种子样本、
还把现象当成新 bug。
## 验证
- ALDEMO0003 现为 status=已排除 + closed_at 有值
- 库里所有终态预警（已排除/已结案）都带 closed_at；进行中的（调查中）不带 —— 口径一致
- 自己创建的两条走 resolve 后 status=已结案 且 closed_at/handle_result 都有值
- 冒烟重跑：40/40 通过，新造预警编号 E2E134351，处置闭环三段全 PASS
门禁：ruff 通过 / mypy 250 文件 0 错 / 单元+契约 1381 passed 2 skipped。
440 lines
18 KiB
Python
440 lines
18 KiB
Python
"""平台端到端冒烟：按角色分线跑一遍真实请求，报告哪一环坏了。
|
||
|
||
## 为什么要有它
|
||
|
||
`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:
|
||
# 前缀**刻意避开 `ALDEMO`**：那是 `seed_risk_alert_demo_data.py` 的演示样本编号
|
||
# （ALDEMO0001-0003）。撞号过一次 —— 本脚本"已存在则跳过"的检查会把创建挡掉，
|
||
# 随后处置全 409，看起来像"结案没写 closed_at"，其实是操作了别人的样本。
|
||
alert_no = f"E2E{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())
|