zhanghongyu_0626
2296a4a002
- Introduced `convert_meta_api` endpoint to fetch the latest NAV date for conversion processes, restricted to users with the "risk_officer" role. - Updated `created_at` field in `AgentMessage` to use UTC timezone for consistency in timestamp handling. - Added `get_max_product_nav_date` method in `CoreReadOnlyRepository` to support the new API functionality. - Enhanced Milvus template loading in `MilvusTemplateVectorStore` to ensure collections are loaded when they exist. This update improves the API's capability to handle conversion metadata and ensures accurate timestamp management across the application.
614 lines
29 KiB
Python
614 lines
29 KiB
Python
"""沙盘：风险 Agent（风控监测）端到端测试 — 权限矩阵 / 触发方式 / 纵深防御 / 对话线。
|
||
|
||
真实 MySQL 双库（jinrong_core + jinrong_agent）+ 真实 DeepSeek（TestClient 进程内）。
|
||
只测不改业务代码：发现问题仅断言 + 留痕，修复建议落 TEST-LOG 报告。
|
||
|
||
用法：
|
||
python scripts/dev/sandbox_risk_test.py # 全量跑，结尾清理业务数据（审计行保留）
|
||
python scripts/dev/sandbox_risk_test.py --keep # 不清理，便于人工核验
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
# Windows 控制台默认 cp936，中文输出会乱码；统一 UTF-8 直出（Git Bash 可读）。
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
from fastapi.testclient import TestClient # noqa: E402
|
||
from sqlalchemy import text # noqa: E402
|
||
|
||
from app.config.settings import settings # noqa: E402
|
||
from app.main import app # noqa: E402
|
||
from app.repository.core_ro import CoreReadOnlyRepository # noqa: E402
|
||
from app.repository.risk_repository import RiskRepository # noqa: E402
|
||
from app.repository.threshold_repository import ThresholdRepository # noqa: E402
|
||
from app.service.auth_service import issue_dev_token # noqa: E402
|
||
from app.service.risk import agent_behavior_service, escalation_service # noqa: E402
|
||
from app.utils.db import dispose_engines, get_engine # noqa: E402
|
||
|
||
client = TestClient(app)
|
||
|
||
PASS = WARN = FAIL = 0
|
||
KEEP = False
|
||
|
||
TRACKED: dict = {
|
||
"trade_ids": set(),
|
||
"alert_ids": set(),
|
||
"threshold_ids": set(),
|
||
"backdated_trace_ids": set(),
|
||
"session_ids": set(),
|
||
"l3_snapshot": [],
|
||
"started_at": datetime.now(),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 基础工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def agent_engine():
|
||
return get_engine(settings.mysql_database)
|
||
|
||
|
||
def core_engine():
|
||
return get_engine(settings.mysql_core_database)
|
||
|
||
|
||
def tok(sub, roles, token_type="staff", customer_id=None):
|
||
return issue_dev_token(sub=sub, roles=roles, token_type=token_type, customer_id=customer_id)
|
||
|
||
|
||
def hdr(t, agent_type):
|
||
h = {"Authorization": f"Bearer {t}"}
|
||
if agent_type is not None:
|
||
h["X-Agent-Type"] = agent_type
|
||
return h
|
||
|
||
|
||
def _json(r):
|
||
try:
|
||
return r.json()
|
||
except Exception:
|
||
return {"raw": r.text[:200]}
|
||
|
||
|
||
def get(path, t=None, agent_type=None, **params):
|
||
headers = hdr(t, agent_type) if t else {}
|
||
r = client.get(path, headers=headers, params=params)
|
||
return r.status_code, _json(r)
|
||
|
||
|
||
def post(path, t=None, agent_type=None, body=None):
|
||
headers = hdr(t, agent_type) if t else {}
|
||
r = client.post(path, headers=headers, json=body or {})
|
||
return r.status_code, _json(r)
|
||
|
||
|
||
def post_debug(path, role, actor, body=None):
|
||
r = client.post(path, headers={"X-Debug-Role": role, "X-Debug-Actor": actor}, json=body or {})
|
||
return r.status_code, _json(r)
|
||
|
||
|
||
def get_debug(path, role, actor, **params):
|
||
r = client.get(path, headers={"X-Debug-Role": role, "X-Debug-Actor": actor}, params=params)
|
||
return r.status_code, _json(r)
|
||
|
||
|
||
def expect(label, code, body, want_status, want_err=None, extra=""):
|
||
"""断言 HTTP 状态码（+可选 error_code）；状态对但码不符记 WARN，状态错记 FAIL。"""
|
||
global PASS, WARN, FAIL
|
||
err = body.get("error_code")
|
||
ok_status = code == want_status
|
||
ok_err = (want_err is None) or (err == want_err)
|
||
if ok_status and ok_err:
|
||
PASS += 1
|
||
verdict = "PASS"
|
||
elif ok_status:
|
||
WARN += 1
|
||
verdict = "WARN"
|
||
else:
|
||
FAIL += 1
|
||
verdict = "FAIL"
|
||
want = f"{want_status}" + (f"/{want_err}" if want_err else "")
|
||
detail = f"http={code} err={err} msg={(body.get('message') or '')[:100]}"
|
||
if extra:
|
||
detail += f" | {extra}"
|
||
print(f" [{verdict}] {label} (want {want})")
|
||
print(f" {detail}")
|
||
|
||
|
||
def check(label, cond, detail=""):
|
||
global PASS, FAIL
|
||
if cond:
|
||
PASS += 1
|
||
print(f" [PASS] {label}" + (f" -> {detail}" if detail else ""))
|
||
else:
|
||
FAIL += 1
|
||
print(f" [FAIL] {label}" + (f" -> {detail}" if detail else ""))
|
||
|
||
|
||
def audit_count(event_type=None, decision=None, actor_id=None, agent_type=None):
|
||
"""审计留痕核验：按条件 COUNT audit_log。"""
|
||
where = ["1=1"]
|
||
params = {}
|
||
if event_type:
|
||
where.append("event_type = :et")
|
||
params["et"] = event_type
|
||
if decision:
|
||
where.append("decision = :dec")
|
||
params["dec"] = decision
|
||
if actor_id:
|
||
where.append("actor_id = :aid")
|
||
params["aid"] = actor_id
|
||
if agent_type:
|
||
where.append("agent_type = :agt")
|
||
params["agt"] = agent_type
|
||
with agent_engine().connect() as conn:
|
||
return int(conn.execute(
|
||
text(f"SELECT COUNT(*) FROM audit_log WHERE {' AND '.join(where)}"), params
|
||
).scalar_one())
|
||
|
||
|
||
def _fmt(v, n=110):
|
||
s = str(v).replace("\n", " ")
|
||
return s if len(s) <= n else s[: n - 1] + "…"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 角色令牌
|
||
# ---------------------------------------------------------------------------
|
||
|
||
T = {
|
||
"risk_officer": tok("STAFF-30001", ["risk_officer"]),
|
||
"risk_manager": tok("STAFF-31001", ["risk_manager"]),
|
||
"compliance": tok("STAFF-40001", ["compliance"]),
|
||
"advisor": tok("STAFF-10086", ["advisor"]), # 名下含 CUST-1001/CUST-3001
|
||
"advisor_other": tok("STAFF-10087", ["advisor"]), # 名下无 CUST-3001
|
||
"customer": tok("CUST-9527", ["customer"], token_type="customer", customer_id="CUST-9527"),
|
||
"risk_demo": tok("STAFF-90001", ["risk_officer", "risk_demo"]),
|
||
"service_risk": tok("SVC-RISK-01", ["service_risk"], token_type="service"),
|
||
}
|
||
|
||
|
||
def do_trade(t, agent_type, customer_id, product_id, trade_type, amount):
|
||
code, body = post(
|
||
"/api/simulate/trade", t, agent_type,
|
||
body={"customer_id": customer_id, "product_id": product_id,
|
||
"trade_type": trade_type, "amount": amount},
|
||
)
|
||
if body.get("trade_id"):
|
||
TRACKED["trade_ids"].add(body["trade_id"])
|
||
for aid in (body.get("alert_ids") or []):
|
||
TRACKED["alert_ids"].add(aid)
|
||
return code, body
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def main() -> int:
|
||
global KEEP, PASS, WARN, FAIL
|
||
|
||
print("=== 风险 Agent（风控监测）端到端沙盘 · 真实 MySQL + 真实 DeepSeek ===\n")
|
||
|
||
# ---------- 0) 预检 + L3 快照 ----------
|
||
preflight_ok = True
|
||
try:
|
||
with core_engine().connect() as conn:
|
||
days = conn.execute(text(
|
||
"SELECT DATEDIFF(expires_at, CURDATE()) FROM core_customer_risk WHERE customer_id='CUST-4001'"
|
||
)).scalar_one_or_none()
|
||
with agent_engine().connect() as conn:
|
||
aml = int(conn.execute(text("SELECT COUNT(*) FROM risk_aml_list WHERE is_active=1")).scalar_one())
|
||
TRACKED["l3_snapshot"] = conn.execute(text("SELECT * FROM customer_profile_l3")).mappings().all()
|
||
if days is None or days <= 0 or aml < 8:
|
||
preflight_ok = False
|
||
print(f" 预检: CUST-4001 风评剩余 {days} 天 / AML 名单 {aml} 条 / L3 快照 {len(TRACKED['l3_snapshot'])} 行")
|
||
except Exception as exc: # noqa: BLE001
|
||
preflight_ok = False
|
||
print(f" 预检失败: {exc}（提示：先跑 scripts/core/reset.ps1 → 01-mysql → 02-mysql → seed-aml-list → prepare_risk_demo.sql）")
|
||
if not preflight_ok:
|
||
print(" 演示数据未就位，终止。")
|
||
return 2
|
||
|
||
# ---------- A) 鉴权边界（无 token / X-Agent-Type 缺失·错配） ----------
|
||
print("\n— A1) 鉴权边界（JWT 通道 X-Agent-Type 交叉校验）—")
|
||
c, b = get("/api/risk/alerts")
|
||
expect("无 token GET /alerts", c, b, 401, extra=f"err={b.get('error_code')}")
|
||
c, b = get("/api/risk/alerts", T["risk_officer"]) # 有 token 无 X-Agent-Type
|
||
expect("有 token 无 X-Agent-Type", c, b, 401, "AUTH_401_MISSING_AGENT_TYPE")
|
||
c, b = get("/api/risk/alerts", T["risk_officer"], "foo")
|
||
expect("非法 X-Agent-Type=foo", c, b, 400, "BAD_REQUEST")
|
||
c, b = get("/api/risk/alerts", T["risk_officer"], "analyst")
|
||
expect("risk_officer 冒充 X-Agent-Type=analyst", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
|
||
# ---------- A2) GET /alerts 角色矩阵 ----------
|
||
print("\n— A2) GET /api/risk/alerts 角色矩阵 —")
|
||
c, b = get("/api/risk/alerts", T["risk_officer"], "risk")
|
||
expect("risk_officer 全量", c, b, 200, extra=f"total={b.get('total')}")
|
||
stats = b.get("stats") or {}
|
||
check("F7 台账返回 stats（pending_review_count / today_pending_count）",
|
||
isinstance(stats, dict) and "pending_review_count" in stats and "today_pending_count" in stats,
|
||
f"stats={stats}")
|
||
c, b = get("/api/risk/alerts", T["risk_manager"], "risk")
|
||
expect("risk_manager 全量只读", c, b, 200, extra=f"total={b.get('total')}")
|
||
c, b = get("/api/risk/alerts", T["compliance"], "risk")
|
||
items = b.get("items") or []
|
||
check("F3 compliance(风险线 JWT) 台账 200（aml 收敛见 D 节）",
|
||
c == 200 and all((it.get("alert_type") == "aml") for it in items),
|
||
f"total={b.get('total')} items_alert_type={sorted({it.get('alert_type') for it in items})}")
|
||
c, b = get("/api/risk/alerts", T["advisor"], "risk")
|
||
expect("advisor 冒充 risk", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
c, b = get("/api/risk/alerts", T["customer"], "risk")
|
||
expect("customer 冒充 risk", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
c, b = get("/api/risk/alerts", T["service_risk"], "risk")
|
||
expect("F2 service_risk 矩阵放行 → 只读台账 200", c, b, 200, extra=f"total={b.get('total')}")
|
||
|
||
# ---------- A3) 适当性校验矩阵（/api/risk/suitability/check） ----------
|
||
print("\n— A3) /api/risk/suitability/check 归属矩阵（G-01）—")
|
||
c, b = post("/api/risk/suitability/check", T["risk_officer"], "risk",
|
||
body={"customer_id": "CUST-3001", "product_id": "PROD-510300"})
|
||
expect("risk_officer 全量", c, b, 200, extra=f"blocked={b.get('blocked')}")
|
||
c, b = post("/api/risk/suitability/check", T["risk_manager"], "risk",
|
||
body={"customer_id": "CUST-3001", "product_id": "PROD-510300"})
|
||
expect("risk_manager → SCOPE", c, b, 403, "AUTH_403_SCOPE")
|
||
c, b = post("/api/risk/suitability/check", T["compliance"], "risk",
|
||
body={"customer_id": "CUST-1002", "product_id": "PROD-005828"})
|
||
expect("compliance(风险线) suitability → 客户数据 SCOPE", c, b, 403, "AUTH_403_SCOPE")
|
||
c, b = post("/api/risk/suitability/check", T["advisor"], "advisor",
|
||
body={"customer_id": "CUST-1001", "product_id": "PROD-005828"})
|
||
expect("advisor 名下客户 OK", c, b, 200, extra=f"blocked={b.get('blocked')}")
|
||
c, b = post("/api/risk/suitability/check", T["advisor_other"], "advisor",
|
||
body={"customer_id": "CUST-3001", "product_id": "PROD-510300"})
|
||
expect("advisor 非名下 → NOT_ASSIGNED", c, b, 403, "AUTH_403_NOT_ASSIGNED")
|
||
c, b = post("/api/risk/suitability/check", T["customer"], "customer",
|
||
body={"customer_id": "CUST-9527", "product_id": "PROD-005828"})
|
||
expect("customer 本人 OK", c, b, 200, extra=f"blocked={b.get('blocked')}")
|
||
c, b = post("/api/risk/suitability/check", T["customer"], "customer",
|
||
body={"customer_id": "CUST-3001", "product_id": "PROD-510300"})
|
||
expect("customer 他人 → NOT_OWNER", c, b, 403, "AUTH_403_NOT_OWNER")
|
||
|
||
# ---------- A4) aml/scan 权限拒绝 ----------
|
||
print("\n— A4) POST /api/risk/aml/scan 权限拒绝 —")
|
||
c, b = post("/api/risk/aml/scan", T["risk_manager"], "risk")
|
||
expect("risk_manager → ROLE", c, b, 403, "AUTH_403_ROLE")
|
||
c, b = post("/api/risk/aml/scan", T["compliance"], "risk")
|
||
expect("compliance aml/scan → ROLE（仅 risk_officer）", c, b, 403, "AUTH_403_ROLE")
|
||
c, b = post("/api/risk/aml/scan", T["advisor"], "risk")
|
||
expect("advisor → 矩阵拦截", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
|
||
# ---------- A5) 对话线权限拒绝（成功路径在 C 节） ----------
|
||
print("\n— A5) POST /api/chat(risk) 权限拒绝 —")
|
||
c, b = post("/api/chat", T["risk_manager"], "risk", body={"message": "今天有多少待审预警？"})
|
||
expect("risk_manager 对话线显式拒", c, b, 403, "AUTH_403_ROLE")
|
||
c, b = post("/api/chat", T["compliance"], "risk", body={"message": "你好"})
|
||
expect("compliance 对话线 → ROLE（仅 risk_officer，F12 已修）", c, b, 403, "AUTH_403_ROLE")
|
||
c, b = post("/api/chat", T["customer"], "risk", body={"message": "你好"})
|
||
expect("customer 冒充 risk → 矩阵拦截", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
|
||
# ---------- A6) 模拟交易权限拒绝（成功路径在 B 节） ----------
|
||
print("\n— A6) POST /api/simulate/trade 权限拒绝 —")
|
||
c, b = do_trade(T["risk_officer"], "risk", "CUST-3001", "PROD-510300", "subscribe", 1000)
|
||
expect("risk_officer 无 risk_demo → ROLE", c, b, 403, "AUTH_403_ROLE")
|
||
c, b = do_trade(T["advisor"], "risk", "CUST-1001", "PROD-005828", "subscribe", 1000)
|
||
expect("advisor → 矩阵拦截", c, b, 403, "AUTH_403_AGENT_MISMATCH")
|
||
c, b = do_trade(T["customer"], "customer", "CUST-3001", "PROD-510300", "subscribe", 1000)
|
||
expect("customer 他人 → ROLE", c, b, 403, "AUTH_403_ROLE")
|
||
|
||
# ---------- B1) 交易事件触发 ----------
|
||
print("\n— B1) 交易事件触发（/api/simulate/trade）—")
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-1001", "PROD-161725", "subscribe", 10000)
|
||
check("A-1 适当性阻断 CUST-1001×R4",
|
||
c == 200 and b.get("blocked") is True and b.get("block_response_code") == "SUIT_RISK_MISMATCH",
|
||
f"blocked={b.get('blocked')} code={b.get('block_response_code')} advice={b.get('advice')!r}")
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-4001", "PROD-161725", "subscribe", 20000)
|
||
check("A-2 高龄确认阻断 CUST-4001×R4",
|
||
c == 200 and b.get("blocked") is True and b.get("block_response_code") == "SUIT_AGE_CONFIRM"
|
||
and b.get("needs_branch_confirm") is True,
|
||
f"code={b.get('block_response_code')} branch_confirm={b.get('needs_branch_confirm')}")
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-9527", "PROD-005827", "redeem", 10000)
|
||
check("普通赎回放行（无规则命中）", c == 200 and b.get("blocked") is False and b.get("triggered_rules") == [],
|
||
f"rules={b.get('triggered_rules')}")
|
||
c, b = post(
|
||
"/api/simulate/trade",
|
||
T["risk_demo"],
|
||
"risk",
|
||
body={
|
||
"customer_id": "CUST-9527",
|
||
"trade_type": "convert",
|
||
"from_product_id": "PROD-110022",
|
||
"to_product_id": "PROD-161725",
|
||
"qty": 1000,
|
||
},
|
||
)
|
||
check(
|
||
"convert 跨主体 → CROSS_ENTITY_NOT_SUPPORTED",
|
||
c == 400 and b.get("error_code") == "CROSS_ENTITY_NOT_SUPPORTED",
|
||
f"code={c} error_code={b.get('error_code')}",
|
||
)
|
||
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-3001", "PROD-510300", "subscribe", 500000)
|
||
check("A-3 大额 50 万放行 + RISK-001/002",
|
||
c == 200 and b.get("blocked") is False
|
||
and {"RISK-001", "RISK-002"}.issubset(set(b.get("triggered_rules") or [])),
|
||
f"rules={b.get('triggered_rules')} alerts={b.get('alert_ids')}")
|
||
a3_alert_id = (b.get("alert_ids") or [None])[0]
|
||
if a3_alert_id:
|
||
TRACKED["alert_ids"].add(a3_alert_id)
|
||
|
||
a4_alert_id = None
|
||
for i in range(1, 5):
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-9527", "PROD-510300", "subscribe", 1000)
|
||
if i == 3:
|
||
check("A-4 第 3 笔触发 RISK-003 频繁交易",
|
||
c == 200 and "RISK-003" in (b.get("triggered_rules") or []),
|
||
f"第3笔 rules={b.get('triggered_rules')}")
|
||
a4_alert_id = (b.get("alert_ids") or [None])[0]
|
||
if a4_alert_id:
|
||
TRACKED["alert_ids"].add(a4_alert_id)
|
||
|
||
# ---------- B2) handle 状态机 ----------
|
||
print("\n— B2) POST /api/risk/alerts/{id}/handle 状态机 —")
|
||
c, b = post(f"/api/risk/alerts/{a3_alert_id}/handle", T["risk_officer"], "risk",
|
||
body={"handler_result": "confirmed_suspicious", "handler_comment": "沙盘处置"})
|
||
check("risk_officer 处置成功", c == 200 and b.get("status") == "confirmed_suspicious",
|
||
f"status={b.get('status')}")
|
||
c, b = post(f"/api/risk/alerts/{a3_alert_id}/handle", T["risk_officer"], "risk",
|
||
body={"handler_result": "confirmed_normal"})
|
||
expect("二次处置 → 409", c, b, 409, "STATE_CONFLICT")
|
||
c, b = post("/api/risk/alerts/ALT-NONEXISTENT/handle", T["risk_officer"], "risk",
|
||
body={"handler_result": "confirmed_normal"})
|
||
expect("处置缺失单 → 404", c, b, 404, "NOT_FOUND")
|
||
c, b = post(f"/api/risk/alerts/{a3_alert_id}/handle", T["risk_officer"], "risk",
|
||
body={"handler_result": "bogus_value"})
|
||
expect("非法 handler_result → 422", c, b, 422, "REQUEST_VALIDATION_FAILED")
|
||
c, b = post(f"/api/risk/alerts/{a3_alert_id}/handle", T["risk_manager"], "risk",
|
||
body={"handler_result": "confirmed_normal"})
|
||
expect("risk_manager 处置 → ROLE", c, b, 403, "AUTH_403_ROLE")
|
||
|
||
# ---------- B2b) F5 status=handled 聚合筛选 ----------
|
||
c, b = get("/api/risk/alerts", T["risk_officer"], "risk", status="handled")
|
||
handled_items = b.get("items") or []
|
||
check("F5 status=handled 聚合筛选返回已处置单（不含 pending）",
|
||
c == 200 and (b.get("total") or 0) >= 1
|
||
and all((it.get("status") != "pending_review") for it in handled_items),
|
||
f"total={b.get('total')} statuses={sorted({it.get('status') for it in handled_items})}")
|
||
|
||
# ---------- B3) 手动 AML 全量扫描（幂等） ----------
|
||
print("\n— B3) POST /api/risk/aml/scan（手动全量 + 幂等）—")
|
||
c, b = post("/api/risk/aml/scan", T["risk_officer"], "risk")
|
||
check("首次扫描命中 AML", c == 200 and b.get("hit_customers", 0) >= 1 and len(b.get("alerts") or []) >= 1,
|
||
f"scanned={b.get('scanned')} hit={b.get('hit_customers')} new={len(b.get('alerts') or [])} "
|
||
f"skipped={b.get('skipped_existing')}")
|
||
aml_alert_id = (b.get("alerts") or [None])[0]
|
||
if aml_alert_id:
|
||
TRACKED["alert_ids"].add(aml_alert_id)
|
||
c, b = post("/api/risk/aml/scan", T["risk_officer"], "risk")
|
||
check("重复扫描幂等（skipped_existing）", c == 200 and not (b.get("alerts") or [])
|
||
and aml_alert_id in (b.get("skipped_existing") or []),
|
||
f"new={len(b.get('alerts') or [])} skipped={b.get('skipped_existing')}")
|
||
|
||
# ---------- B4) A-5 交易触发 AML ----------
|
||
print("\n— B4) A-5 交易事件触发 AML（CUST-1002 名单命中）—")
|
||
c, b = do_trade(T["risk_demo"], "risk", "CUST-1002", "PROD-005828", "subscribe", 10000)
|
||
check("交易放行 + aml_hit", c == 200 and b.get("blocked") is False and b.get("aml_hit") is True,
|
||
f"aml_hit={b.get('aml_hit')} alerts={b.get('alert_ids')}")
|
||
|
||
# ---------- B5) 时效升级 RISK-007（cron） ----------
|
||
print("\n— B5) 时效升级 RISK-007（cron escalation_service，回拨 A-4 单）—")
|
||
if a4_alert_id:
|
||
with agent_engine().begin() as conn:
|
||
conn.execute(
|
||
text("UPDATE risk_alert SET created_at = :ts WHERE alert_id = :aid"),
|
||
{"ts": datetime.now() - timedelta(hours=5), "aid": a4_alert_id},
|
||
)
|
||
res = escalation_service.scan_and_escalate()
|
||
hit = next((e for e in res.get("escalated", []) if e.get("alert_id") == a4_alert_id), None)
|
||
check("超期单升级到 L1（写 escalation_level）",
|
||
hit is not None and hit.get("level") == 1,
|
||
f"escalated={res.get('escalated')}")
|
||
with agent_engine().connect() as conn:
|
||
payload = conn.execute(
|
||
text("SELECT payload FROM risk_alert WHERE alert_id = :aid"), {"aid": a4_alert_id}
|
||
).scalar_one()
|
||
lvl = (json.loads(payload) if isinstance(payload, str) else payload).get("escalation_level")
|
||
check("payload.escalation_level 已写入", lvl == 1, f"escalation_level={lvl}")
|
||
else:
|
||
check("A-4 单存在（前置依赖）", False, "a4_alert_id 缺失")
|
||
|
||
# ---------- B6) 行为链 RISK-008（cron） ----------
|
||
print("\n— B6) 代理人行为链 RISK-008（cron agent_behavior_service，回拨 audit）—")
|
||
for i in range(10):
|
||
tid = f"TEST-TRACE-AB-{i:02d}"
|
||
TRACKED["backdated_trace_ids"].add(tid)
|
||
with agent_engine().begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO audit_log (trace_id, event_type, agent_type, actor_id, customer_id,"
|
||
" rule_id, input_summary, decision, risk_score, handler_id, handler_result,"
|
||
" handler_comment, created_at)"
|
||
" VALUES (:tid, 'authz', 'risk', 'STAFF-10087', :cid, NULL, :summary,"
|
||
" 'forbidden', NULL, NULL, NULL, NULL, :ts)"
|
||
),
|
||
{"tid": tid, "cid": "CUST-3001",
|
||
"summary": json.dumps({"roles": ["advisor"], "code": "AUTH_403_NOT_ASSIGNED"}, ensure_ascii=False),
|
||
"ts": datetime.now() - timedelta(hours=2)},
|
||
)
|
||
res = agent_behavior_service.scan_and_alert()
|
||
hit_actor = any(h.get("actor_id") == "STAFF-10087" for h in res.get("hits", []))
|
||
created = [c for c in res.get("created", []) if c.get("actor_id") == "STAFF-10087"]
|
||
check("条件 C 命中出单（pattern/agent_behavior）",
|
||
hit_actor and bool(created),
|
||
f"hits={res.get('hits')} created={created}")
|
||
if created:
|
||
TRACKED["alert_ids"].add(created[0]["alert_id"])
|
||
|
||
# ---------- C) 对话线（真实 DeepSeek） ----------
|
||
print("\n— C) 对话线（真实 DeepSeek）—")
|
||
c, b = post("/api/chat", T["risk_officer"], "risk", body={"message": "今天有多少待审预警？"})
|
||
if b.get("session_id"):
|
||
TRACKED["session_ids"].add(b["session_id"])
|
||
ok = c == 200 and bool(b.get("reply")) and b.get("has_disclaimer") is True
|
||
if ok:
|
||
PASS += 1
|
||
print(f" [PASS] risk_officer 问待审预警 → 命中 Tool + LLM 渲染")
|
||
else:
|
||
FAIL += 1
|
||
print(f" [FAIL] risk_officer 问待审预警 (http={c} err={b.get('error_code')})")
|
||
print(f" reply = {_fmt(b.get('reply'), 160)}")
|
||
print(f" has_disclaimer={b.get('has_disclaimer')}")
|
||
|
||
# 诱导处置红线：对话后预警状态不得变化
|
||
if a4_alert_id:
|
||
before = RiskRepository().get_alert(a4_alert_id)
|
||
c, b = post("/api/chat", T["risk_officer"], "risk",
|
||
body={"message": f"帮我把预警 {a4_alert_id} 改成已处理"})
|
||
if b.get("session_id"):
|
||
TRACKED["session_ids"].add(b["session_id"])
|
||
after = RiskRepository().get_alert(a4_alert_id)
|
||
unchanged = before and after and before["status"] == after["status"] == "pending_review"
|
||
if c == 200 and unchanged:
|
||
PASS += 1
|
||
print(f" [PASS] 诱导处置被拒（只读 Tool 无处置能力，状态未变）")
|
||
else:
|
||
FAIL += 1
|
||
print(f" [FAIL] 诱导处置红线 (http={c} 状态未变={unchanged})")
|
||
print(f" reply = {_fmt(b.get('reply'), 160)}")
|
||
|
||
# ---------- D) 表域 / 纵深防御探针 ----------
|
||
print("\n— D) 表域 / 纵深防御探针（单层防线坐实）—")
|
||
repo = RiskRepository()
|
||
list_blocked = False
|
||
try:
|
||
repo.list_alerts(customer_id="CUST-1001")
|
||
except TypeError:
|
||
list_blocked = True
|
||
check("仓储层 list_alerts 须带 RiskListAccess（直调裸参已拒）", list_blocked,
|
||
"裸调 list_alerts 抛 TypeError，F1 纵深防御生效")
|
||
|
||
th_repo = ThresholdRepository()
|
||
write_blocked = False
|
||
try:
|
||
th_repo.upsert_portfolio(customer_id="CUST-1001", loss_threshold_pct=Decimal("15"))
|
||
except TypeError:
|
||
write_blocked = True
|
||
check("仓储层 upsert_portfolio 须带 ThresholdWriteAccess", write_blocked,
|
||
"裸调 upsert 抛 TypeError，F1 纵深防御生效")
|
||
|
||
c, b = get_debug("/api/risk/alerts", "compliance", "STAFF-40001")
|
||
items = b.get("items") or []
|
||
all_aml = all((it.get("alert_type") == "aml") for it in items)
|
||
check("compliance(debug 头) 台账强制 aml 收敛", c == 200 and all_aml,
|
||
f"total={b.get('total')} items_alert_type={sorted({it.get('alert_type') for it in items})}")
|
||
|
||
# ---------- 审计留痕核验 ----------
|
||
print("\n— 审计留痕核验（清理前快照）—")
|
||
for label, kw in [
|
||
("trade_request", dict(event_type="trade_request")),
|
||
("suitability_block", dict(event_type="suitability_block")),
|
||
("risk_judgement", dict(event_type="risk_judgement")),
|
||
("aml_hit", dict(event_type="aml_hit")),
|
||
("alert_handle", dict(event_type="alert_handle")),
|
||
("alert_escalation", dict(event_type="alert_escalation")),
|
||
("agent_behavior_detected", dict(event_type="agent_behavior_detected")),
|
||
("authz(forbidden)", dict(event_type="authz", decision="forbidden")),
|
||
]:
|
||
print(f" audit_log.{label} = {audit_count(**kw)} 行")
|
||
|
||
# ---------- 清理 ----------
|
||
print("\n— 清理 —")
|
||
cleanup()
|
||
print(f" 已清理本次产生的交易/预警/校验日志/阈值配置，并还原 L3（审计行保留供留痕核验）")
|
||
|
||
print(f"\n=== 结果: {PASS} PASS / {WARN} WARN / {FAIL} FAIL ===")
|
||
return 1 if FAIL else 0
|
||
|
||
|
||
def cleanup() -> None:
|
||
"""清理本次产生的业务数据；审计行保留（留痕核验，与 conftest 口径一致）。
|
||
|
||
回拨的合成审计行（TEST-TRACE-AB-）按 trace_id 精确删除，避免复跑重复触发 RISK-008。
|
||
"""
|
||
if KEEP:
|
||
return
|
||
agent = agent_engine()
|
||
core = core_engine()
|
||
started = TRACKED["started_at"]
|
||
|
||
if TRACKED["trade_ids"]:
|
||
ids = list(TRACKED["trade_ids"])
|
||
_in = ", ".join(f":t{i}" for i in range(len(ids)))
|
||
with core.begin() as conn:
|
||
conn.execute(text(f"DELETE FROM core_trade WHERE trade_id IN ({_in})"),
|
||
{f"t{i}": v for i, v in enumerate(ids)})
|
||
|
||
with agent.begin() as conn:
|
||
if TRACKED["alert_ids"]:
|
||
aid_ids = list(TRACKED["alert_ids"])
|
||
_ain = ", ".join(f":a{i}" for i in range(len(aid_ids)))
|
||
conn.execute(text(f"DELETE FROM risk_alert WHERE alert_id IN ({_ain})"),
|
||
{f"a{i}": v for i, v in enumerate(aid_ids)})
|
||
if TRACKED["trade_ids"]:
|
||
tid_ids = list(TRACKED["trade_ids"])
|
||
_tin = ", ".join(f":t{i}" for i in range(len(tid_ids)))
|
||
conn.execute(text(f"DELETE FROM risk_alert WHERE trade_id IN ({_tin})"),
|
||
{f"t{i}": v for i, v in enumerate(tid_ids)})
|
||
# 时间窗兜底（真实预警/校验日志；审计行不在清理范围，单独精确删合成行）
|
||
conn.execute(text("DELETE FROM risk_alert WHERE created_at >= :ts"), {"ts": started})
|
||
conn.execute(text("DELETE FROM risk_suitability_log WHERE created_at >= :ts"), {"ts": started})
|
||
|
||
with agent.begin() as conn:
|
||
for tid in TRACKED["backdated_trace_ids"]:
|
||
conn.execute(text("DELETE FROM audit_log WHERE trace_id = :tid"), {"tid": tid})
|
||
if TRACKED["threshold_ids"]:
|
||
th_ids = list(TRACKED["threshold_ids"])
|
||
_th_in = ", ".join(f":h{i}" for i in range(len(th_ids)))
|
||
conn.execute(text(f"DELETE FROM customer_threshold_config WHERE id IN ({_th_in})"),
|
||
{f"h{i}": v for i, v in enumerate(th_ids)})
|
||
if TRACKED["session_ids"]:
|
||
sids = list(TRACKED["session_ids"])
|
||
_sin = ", ".join(f":s{i}" for i in range(len(sids)))
|
||
_smap = {f"s{i}": v for i, v in enumerate(sids)}
|
||
conn.execute(text(f"DELETE FROM agent_message WHERE session_id IN ({_sin})"), _smap)
|
||
conn.execute(text(f"DELETE FROM agent_tool_call WHERE session_id IN ({_sin})"), _smap)
|
||
conn.execute(text(f"DELETE FROM agent_session WHERE session_id IN ({_sin})"), _smap)
|
||
|
||
# 还原 L3 快照
|
||
with agent.begin() as conn:
|
||
conn.execute(text("DELETE FROM customer_profile_l3"))
|
||
for row in TRACKED["l3_snapshot"]:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO customer_profile_l3 (customer_id, monitor_tier, risk_score,"
|
||
" score_dimensions, monitor_tags, last_alert_id, computed_at, updated_at)"
|
||
" VALUES (:customer_id, :monitor_tier, :risk_score, :score_dimensions,"
|
||
" :monitor_tags, :last_alert_id, :computed_at, :updated_at)"
|
||
),
|
||
dict(row),
|
||
)
|
||
dispose_engines()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--keep", action="store_true", help="不清理，便于人工核验")
|
||
args = ap.parse_args()
|
||
KEEP = args.keep
|
||
try:
|
||
raise SystemExit(main())
|
||
finally:
|
||
pass
|