From e7bd8880e22867efd3587e0e6d365f46754ddf68 Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 23:27:37 +0800 Subject: [PATCH] =?UTF-8?q?test:=20B8=20=E4=B8=BB=E4=BD=93=E2=80=94?= =?UTF-8?q?=E2=80=94conftest(=E6=BC=94=E7=A4=BA=E6=95=B0=E6=8D=AE=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C/=E5=B9=82=E7=AD=89=E4=BB=A3=E8=B7=91/TRD-TEST-=20tear?= =?UTF-8?q?down)+=E9=9B=86=E6=88=90=E6=B5=8B=E8=AF=95=20A-1~A-5/A-7/A-9+tr?= =?UTF-8?q?ace=E4=B8=80=E8=87=B4=E6=80=A7+=E5=AE=A1=E8=AE=A1JSON(11?= =?UTF-8?q?=E4=BE=8B)+=5Fddl.py=20=E5=8D=95=E4=B8=80=E4=BA=8B=E5=AE=9E?= =?UTF-8?q?=E6=BA=90+L3=20DEL=20=E8=A1=8C=E4=B8=BA=E6=96=AD=E8=A8=80(B7?= =?UTF-8?q?=E5=A4=8D=E5=AE=A1P2-1)+locks=20=E6=96=87=E6=A1=88=E4=B8=AD?= =?UTF-8?q?=E6=80=A7=E5=8C=96(P3-2),=20196=20=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/service/risk/locks.py | 2 +- tests/_ddl.py | 91 ++++++++ tests/conftest.py | 124 +++++++++++ tests/test_integration_risk.py | 383 +++++++++++++++++++++++++++++++++ tests/test_profile_l3.py | 42 +++- 5 files changed, 637 insertions(+), 5 deletions(-) create mode 100644 tests/_ddl.py create mode 100644 tests/conftest.py create mode 100644 tests/test_integration_risk.py diff --git a/app/service/risk/locks.py b/app/service/risk/locks.py index 116583d..24fd4c4 100644 --- a/app/service/risk/locks.py +++ b/app/service/risk/locks.py @@ -32,7 +32,7 @@ def run_locked(key: str, fn: Callable[[bool], Any]) -> Any: """锁内执行 fn(locked=True)；获取超时降级 fn(locked=False)。""" lock = lock_for(key) if not lock.acquire(timeout=LOCK_TIMEOUT_SECONDS): - logger.warning("agg lock timeout, degrade to standalone alert: %s", key) + logger.warning("agg lock timeout, run without lock (conflicts bounded by caller): %s", key) return fn(locked=False) try: return fn(locked=True) diff --git a/tests/_ddl.py b/tests/_ddl.py new file mode 100644 index 0000000..b59adaa --- /dev/null +++ b/tests/_ddl.py @@ -0,0 +1,91 @@ +"""sqlite 测试 DDL 单一事实源（B8 · B4 评审 P3-12 收敛 / B5 评审 P3-4 时区）。 + +以 VARCHAR/TEXT/INTEGER/REAL 近似 MySQL 的 ENUM/SMALLINT/JSON/DECIMAL， +枚举防线不在测试 DB 层（由 service 白名单保证，见 test_profile_l3 口径注释）。 +时间默认值统一 `(datetime('now','localtime'))`：sqlite CURRENT_TIMESTAMP 为 UTC， +本地 23:00~24:00 跑测试时与 MySQL localtime 口径错位一个日界（B5 评审 P3-4）。 +列集为各测试所需列的并集；新表/新列只改此处，禁止再回测试文件手写。 +""" + +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +_TS = "TIMESTAMP DEFAULT (datetime('now','localtime'))" + +SQLITE_TABLES: dict[str, str] = { + # ---- Core 模拟库（只读侧） ---- + "core_customer": """ + CREATE TABLE core_customer ( + customer_id VARCHAR(64) PRIMARY KEY, display_name VARCHAR(128), age INTEGER, + occupation VARCHAR(64), open_date DATE, is_active TINYINT DEFAULT 1) + """, + "core_customer_risk": """ + CREATE TABLE core_customer_risk ( + customer_id VARCHAR(64), risk_code VARCHAR(8), evaluated_at TIMESTAMP) + """, + "core_customer_advisor": """ + CREATE TABLE core_customer_advisor ( + advisor_id VARCHAR(64), customer_id VARCHAR(64), rel_status VARCHAR(16)) + """, + "core_product": """ + CREATE TABLE core_product ( + product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128), + min_risk_code VARCHAR(8), product_type VARCHAR(32)) + """, + "core_trade": """ + CREATE TABLE core_trade ( + trade_id VARCHAR(64) PRIMARY KEY, customer_id VARCHAR(64), product_id VARCHAR(64), + trade_type VARCHAR(16), amount DECIMAL, trade_status VARCHAR(16), + traded_at TIMESTAMP) + """, + # ---- agent 库 ---- + "audit_log": f""" + CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id VARCHAR(64), event_type VARCHAR(64), + agent_type VARCHAR(16), actor_id VARCHAR(64), customer_id VARCHAR(64), + rule_id VARCHAR(64), input_summary TEXT, decision VARCHAR(64), risk_score INTEGER, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at {_TS}) + """, + "risk_alert": f""" + CREATE TABLE risk_alert ( + alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64), + trade_id VARCHAR(64), alert_type VARCHAR(16), triggered_rules TEXT, + risk_score INTEGER, status VARCHAR(24) DEFAULT 'pending_review', payload TEXT, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at {_TS}, handled_at TIMESTAMP) + """, + "risk_suitability_log": f""" + CREATE TABLE risk_suitability_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id VARCHAR(64), customer_id VARCHAR(64), + product_id VARCHAR(64), customer_risk_level VARCHAR(8), product_risk_level VARCHAR(8), + is_matched TINYINT, is_blocked TINYINT, block_reason VARCHAR(512), + request_ref VARCHAR(64), profile_l1_version VARCHAR(32), + created_at {_TS}) + """, + "customer_profile_l3": f""" + CREATE TABLE customer_profile_l3 ( + customer_id VARCHAR(64) PRIMARY KEY, monitor_tier VARCHAR(16) NOT NULL, + risk_score INTEGER, score_dimensions TEXT, monitor_tags TEXT, + last_alert_id VARCHAR(64), computed_at TIMESTAMP NOT NULL, + updated_at {_TS}) + """, + "risk_aml_list": f""" + CREATE TABLE risk_aml_list ( + id INTEGER PRIMARY KEY AUTOINCREMENT, list_id VARCHAR(64), list_type VARCHAR(16), + full_name VARCHAR(128), id_no VARCHAR(32), bank_card_no VARCHAR(32), + match_threshold REAL, source VARCHAR(64), list_version VARCHAR(16), + effective_date DATE, is_active TINYINT DEFAULT 1, created_at {_TS}) + """, +} + + +def create_sqlite_engine(): + """内存 sqlite + 全部表（StaticPool 单连接共享，同现有测试模式）。""" + engine = create_engine( + "sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False} + ) + with engine.begin() as conn: + for ddl in SQLITE_TABLES.values(): + conn.execute(text(ddl)) + return engine diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..6f0733a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,124 @@ +"""pytest 全局 fixture（B8 · 开发计划 B8 行）。 + +- sqlite_engine：内存库 + tests/_ddl.py 单一事实源建表（单测共用）。 +- 集成测试（test_integration_risk.py）收集期先调 ensure_risk_demo_ready()： + 本机演示数据未就位（未跑 FLOW §0 ③④）时整模块 skip 并给提示，单测不受影响。 +- risk_demo_env（session）：真 MySQL（jinrong_agent/jinrong_core）集成环境， + setup 幂等代跑 prepare_risk_demo.sql（UPDATE 语义）+ L3 快照，teardown 按 + TRD-TEST- 前缀 + session 时间窗清理测试交易及其预警/校验/审计行，并还原 L3。 + 注意：时间窗清理假定测试期间无他人向演示库写入（学习项目单机约定）。 +""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path + +import pytest +from sqlalchemy import text + +from _ddl import create_sqlite_engine + +ROOT = Path(__file__).resolve().parent.parent +DEMO_SQL = ROOT / "scripts" / "demo" / "prepare_risk_demo.sql" + + +@pytest.fixture() +def sqlite_engine(): + """内存 sqlite 全表引擎（B8 前 DDL 散落各测试文件，收敛后统一走这里）。""" + engine = create_sqlite_engine() + yield engine + engine.dispose() + + +# ---------- 真库集成环境（B8 集成测试专用） ---------- + + +def ensure_risk_demo_ready() -> None: + """演示数据就位校验（开发计划 B8-a）：缺失则 skip 集成模块并提示 bootstrap。 + + 在集成测试模块收集期调用；失败 pytest.skip(allow_module_level=True) 只跳过 + 该模块，sqlite 单测不受影响。校验口径：CUST-4001 测评 <365 天、AML 名单 ≥8。 + """ + from app.config.settings import settings + from app.utils.db import get_engine + + hint = "先跑 FLOW §0 ③④：scripts/core/reset.ps1 → 01-mysql-共用底座.sql →" " 02-mysql-agent专用.sql → scripts/agent/seed-aml-list.sql → prepare_risk_demo.sql" + try: + core = get_engine(settings.mysql_core_database) + agent = get_engine(settings.mysql_database) + with core.connect() as conn: + days = conn.execute( + text( + "SELECT DATEDIFF(CURDATE(), evaluated_at) FROM core_customer_risk" + " WHERE customer_id = 'CUST-4001'" + ) + ).scalar_one_or_none() + if days is None or days >= settings.risk_assessment_valid_days: + pytest.skip(f"演示数据未就位（CUST-4001 测评 {days} 天）：{hint}", allow_module_level=True) + with agent.connect() as conn: + aml_count = conn.execute( + text("SELECT COUNT(*) FROM risk_aml_list WHERE is_active = 1") + ).scalar_one() + if aml_count < 8: + pytest.skip(f"演示数据未就位（AML 名单仅 {aml_count} 条）：{hint}", allow_module_level=True) + except pytest.SkipRequested: + raise + except Exception as exc: + pytest.skip(f"本机 MySQL 不可用（{exc}）：{hint}", allow_module_level=True) + + +def _run_demo_sql(core_engine) -> None: + """幂等代跑 prepare_risk_demo.sql（开发计划 B8-b）：文件须保持无存储过程/DELIMITER。""" + statements = [s.strip() for s in DEMO_SQL.read_text(encoding="utf-8").split(";") if s.strip()] + with core_engine.begin() as conn: + for stmt in statements: + if stmt.upper().startswith("USE "): + continue # engine 默认库已由 URL 指定 + conn.execute(text(stmt)) + + +def _cleanup_test_rows(agent_engine, core_engine, started_at: datetime) -> None: + """TRD-TEST- teardown（开发计划 B8-c）：测试交易及其关联行 + 时间窗兜底 + L3 还原。""" + with core_engine.begin() as conn: + conn.execute(text("DELETE FROM core_trade WHERE trade_id LIKE 'TRD-TEST-%'")) + with agent_engine.begin() as conn: + conn.execute(text("DELETE FROM risk_alert WHERE trade_id LIKE 'TRD-TEST-%' OR created_at >= :ts"), {"ts": started_at}) + conn.execute(text("DELETE FROM risk_suitability_log WHERE request_ref LIKE 'TRD-TEST-%' OR created_at >= :ts"), {"ts": started_at}) + conn.execute(text("DELETE FROM audit_log WHERE created_at >= :ts"), {"ts": started_at}) + + +@pytest.fixture(scope="session") +def risk_demo_env(): + """真 MySQL 集成环境：setup 快照 L3 → yield engines → teardown 清理 + 引擎 dispose。""" + from app.config.settings import settings + from app.utils.db import dispose_engines, get_engine + + core = get_engine(settings.mysql_core_database) + agent = get_engine(settings.mysql_database) + started_at = datetime.now() + try: + with agent.begin() as conn: + l3_snapshot = conn.execute( + text("SELECT * FROM customer_profile_l3") + ).mappings().all() + _run_demo_sql(core) + yield {"agent": agent, "core": core, "started_at": started_at} + finally: + try: + _cleanup_test_rows(agent, core, started_at) + with agent.begin() as conn: + conn.execute(text("DELETE FROM customer_profile_l3")) + for row in 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), + ) + finally: + dispose_engines() diff --git a/tests/test_integration_risk.py b/tests/test_integration_risk.py new file mode 100644 index 0000000..c014f81 --- /dev/null +++ b/tests/test_integration_risk.py @@ -0,0 +1,383 @@ +"""风控集成测试（B8 · PRD §8 验收 A-1~A-5/A-7/A-9 + trace 一致性 + 审计 JSON）。 + +真链路：TestClient(main app)（路由 + trace 中间件 + lifespan）→ 真本机 MySQL +（jinrong_core/jinrong_agent，演示数据就位校验见 conftest.ensure_risk_demo_ready）。 +仅注入两点：trade_gateway._new_trade_id 统一 TRD-TEST- 前缀（teardown 按前缀 +清理，B5 评审 P3-2）；redis_gateway fake（断言推送，不依赖本机 Redis）。 +用例间按演示时间线顺序耦合（A-3 产生的预警单供 A-7 处置），模块内保序。 +""" + +from __future__ import annotations + +import json +from uuid import uuid4 + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import text + +from conftest import ensure_risk_demo_ready + +from app.config.settings import settings # noqa: E402 +from app.gateway import trade_gateway # noqa: E402 +from app.main import app # noqa: E402 +from app.service.risk import redis_gateway # noqa: E402 + +ensure_risk_demo_ready() + +OFFICER = {"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-90001"} +COMPLIANCE = {"X-Debug-Role": "compliance", "X-Debug-Actor": "STAFF-40001"} +CUSTOMER_1001 = {"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-1001"} +CUSTOMER_1002 = {"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-1002"} +# simulate 交易白名单：risk_demo 演示账号或客户本人（FR-1 鉴权） +DEMO = {"X-Debug-Role": "risk_demo", "X-Debug-Actor": "STAFF-DEMO"} +# STAFF-10087 真实存在但名下无 CUST-3001（归属表 28 行分布于 5 个 advisor） +ADVISOR_10087 = {"X-Debug-Role": "advisor", "X-Debug-Actor": "STAFF-10087"} + +# 演示时间线状态（A-1/A-3 产物供 A-7 审计与处置用例续用） +_state: dict[str, str] = {} + + +class FakePub: + def __init__(self): + self.messages = [] + self.deletes = [] + + def publish(self, channel, payload): + self.messages.append((channel, payload)) + + def delete(self, *keys): + self.deletes.append(keys) + + +def _test_trade_id(now): + """集成交易统一 TRD-TEST- 前缀（conftest teardown 定位清理）。""" + return f"TRD-TEST-{uuid4().hex[:8].upper()}" + + +@pytest.fixture() +def iclient(risk_demo_env, monkeypatch): + monkeypatch.setattr(trade_gateway, "_new_trade_id", _test_trade_id) + fake = FakePub() + with TestClient(app) as c: # 真 lifespan：dev 放行 + Redis 网关注册 + monkeypatch.setattr(redis_gateway, "_gateway", fake) # 覆盖 lifespan 注册的真网关 + yield c, fake + + +def _trade(customer_id, product_id, trade_type, amount): + return { + "customer_id": customer_id, + "product_id": product_id, + "trade_type": trade_type, + "amount": amount, + } + + +def _one(env, sql, **params): + with env["agent"].connect() as conn: + return conn.execute(text(sql), params).mappings().first() + + +def _core_one(env, sql, **params): + with env["core"].connect() as conn: + return conn.execute(text(sql), params).mappings().first() + + +def _audit_by_trade(env, trade_id, event_type=None): + sql = ( + "SELECT * FROM audit_log WHERE event_type = :et" + " AND JSON_UNQUOTE(JSON_EXTRACT(input_summary, '$.trade_id')) = :tid" + ) + return _one(env, sql, et=event_type or "trade_request", tid=trade_id) + + +# ---------- A-1：C1 申购 R4 → SUIT-001 阻断（含 trace 一致性） ---------- + + +def test_a1_suit001_blocked_with_trace_consistency(iclient, risk_demo_env): + c, fake = iclient + env = risk_demo_env + trace = f"trc-integ-{uuid4().hex[:12]}" + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-1001", "PROD-161725", "subscribe", 10000), + headers={**CUSTOMER_1001, "X-Trace-Id": trace}, + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is True + assert "SUIT-001" in body["block_reason"] # C1 仅可购 R1 + trade_id = body["trade_id"] + assert trade_id.startswith("TRD-TEST-") + + # 阻断不落交易（FR-1） + assert _core_one(env, "SELECT * FROM core_trade WHERE trade_id = :t", t=trade_id) is None + # 校验日志 blocked + slog = _one( + env, + "SELECT * FROM risk_suitability_log WHERE request_ref = :t", + t=trade_id, + ) + assert slog is not None and slog["is_blocked"] == 1 and slog["is_matched"] == 0 + # suitability 预警单生成（R-02 阻断单），trace 与请求一致 + alert = _one(env, "SELECT * FROM risk_alert WHERE trade_id = :t", t=trade_id) + assert alert["alert_type"] == "suitability" and alert["status"] == "pending_review" + assert alert["trace_id"] == trace + # 审计可查（platform 阻断留痕），trace 一致 + audit = _audit_by_trade(env, trade_id, "trade_request") + assert audit["decision"] == "suitability_blocked" and audit["trace_id"] == trace + assert audit["agent_type"] == "platform" + # Redis 推送 trace 一致 + pushed = [p for ch, p in fake.messages if ch == "risk:pub:alert"] + assert any(p["alert_id"] == alert["alert_id"] and p["trace_id"] == trace for p in pushed) + _state["a1_trade_id"] = trade_id + + +def test_a1_audit_input_summary_json(iclient, risk_demo_env): + """B5 复审 L1：platform 审计 input_summary 可 JSON 解析（含阻断 reasons）。""" + c, _ = iclient + env = risk_demo_env + audit = _audit_by_trade(env, _state["a1_trade_id"], "trade_request") + summary = json.loads(audit["input_summary"]) + assert summary["trade_id"] == _state["a1_trade_id"] + assert summary["product_id"] == "PROD-161725" + assert summary["trade_type"] == "subscribe" + assert summary["amount"] == "10000" + assert "SUIT-001" in summary["block_reason"] + assert isinstance(summary["reasons"], list) and summary["reasons"] + + +# ---------- A-2：70 岁 C5 → SUIT-006 封顶 + SUIT-003，无 SUIT-008 ---------- + + +def test_a2_age70_cap_suit006_no_suit008(iclient, risk_demo_env): + c, _ = iclient + env = risk_demo_env + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-4001", "PROD-161725", "subscribe", 20000), + headers={**DEMO, "X-Trace-Id": f"trc-integ-{uuid4().hex[:12]}"}, + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is True + reasons = body["reasons"] + assert any("SUIT-006" in x for x in reasons), reasons # ≥70 按 C3 封顶 + assert any("SUIT-003" in x for x in reasons), reasons # C3 < R4 不匹配 + assert not any("SUIT-008" in x for x in reasons), reasons # 测评已刷新，不得干扰 + trade_id = body["trade_id"] + assert _core_one(env, "SELECT * FROM core_trade WHERE trade_id = :t", t=trade_id) is None + _state["a2_trade_id"] = trade_id + + +# ---------- A-3：C3 单笔 50 万 R3 → 放行 + RISK-001/002 预警 + 推送 ---------- + + +def test_a3_large_amount_alert_and_publish(iclient, risk_demo_env): + c, fake = iclient + env = risk_demo_env + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-3001", "PROD-510300", "subscribe", 500000), + headers=DEMO, + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is False + assert body["triggered_rules"] == ["RISK-001", "RISK-002"] + assert body["aml_hit"] is False + trade_id = body["trade_id"] + alert_id = body["alert_ids"][0] + + assert _core_one(env, "SELECT * FROM core_trade WHERE trade_id = :t", t=trade_id) is not None + alert = _one(env, "SELECT * FROM risk_alert WHERE alert_id = :a", a=alert_id) + assert alert["status"] == "pending_review" and alert["risk_score"] == 70 + assert json.loads(alert["triggered_rules"]) == ["RISK-001", "RISK-002"] + pushed = [p for ch, p in fake.messages if ch == "risk:pub:alert"] + assert any(p["alert_id"] == alert_id for p in pushed) + # 放行审计全量引擎输出（B5 复审 L1） + audit = _audit_by_trade(env, trade_id, "trade_request") + assert audit["decision"] == "trade_accepted" + summary = json.loads(audit["input_summary"]) + assert summary["triggered_rules"] == ["RISK-001", "RISK-002"] + assert summary["alert_ids"] == [alert_id] + _state["a3_alert_id"] = alert_id + _state["a3_trade_id"] = trade_id + + +# ---------- A-4：同产品当日第 3 笔 → freq 并入当日预警单 ---------- + + +def test_a4_freq_trade_merged_into_same_alert(iclient, risk_demo_env): + c, _ = iclient + env = risk_demo_env + alert_id = None + for _ in range(3): + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-9527", "PROD-510300", "subscribe", 1000), + headers=DEMO, + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is False + if body["triggered_rules"]: + assert body["triggered_rules"] == ["RISK-003"] + alert_id = body["alert_ids"][0] + assert alert_id, "第 3 笔应触发 RISK-003" + + # 第 4 笔：并入既有单，不另开新单（PRD FR-4） + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-9527", "PROD-510300", "subscribe", 1000), + headers=DEMO, + ) + body = r.json() + assert body["blocked"] is False and body["triggered_rules"] == ["RISK-003"] + assert body["alert_ids"] == [alert_id] + alerts = _one( + env, + "SELECT COUNT(*) AS n FROM risk_alert WHERE customer_id = 'CUST-9527'" + " AND alert_type = 'freq_trade'", + ) + assert alerts["n"] == 1 + row = _one(env, "SELECT payload FROM risk_alert WHERE alert_id = :a", a=alert_id) + assert len(json.loads(row["payload"])["events"]) == 2 + + +# ---------- A-5：AML 命中 → 独立单 + L3 high + compliance 可见 + 不冻户 ---------- + + +def test_a5_aml_hit_independent_alert(iclient, risk_demo_env): + c, fake = iclient + env = risk_demo_env + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-1002", "PROD-005828", "subscribe", 100), # C2+R2 匹配，不阻断 + headers=DEMO, + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is False and body["aml_hit"] is True + trade_id = body["trade_id"] + + alert = _one( + env, + "SELECT * FROM risk_alert WHERE customer_id = 'CUST-1002' AND alert_type = 'aml'", + ) + assert alert is not None + assert alert["risk_score"] == 95 and alert["status"] == "pending_review" + assert json.loads(alert["triggered_rules"]) == ["AML-001"] + # L3 置 high（FR-7） + l3 = _one(env, "SELECT * FROM customer_profile_l3 WHERE customer_id = 'CUST-1002'") + assert l3["monitor_tier"] == "high" + # 紧急推送含 compliance + pushed = [p for ch, p in fake.messages if ch == "risk:pub:alert"] + aml_push = [p for p in pushed if p["alert_id"] == alert["alert_id"]] + assert aml_push and "compliance" in aml_push[-1]["notify_role"] + # compliance 账号台账可见该单（A-7 强制 aml 过滤的另一面） + r = c.get("/api/risk/alerts", headers=COMPLIANCE) + assert r.status_code == 200 + items = r.json()["items"] + assert items and all(i["alert_type"] == "aml" for i in items) + assert any(i["alert_id"] == alert["alert_id"] for i in items) + # 账户未被冻结（R-03 禁止自动冻户） + row = _core_one(env, "SELECT is_active FROM core_customer WHERE customer_id = 'CUST-1002'") + assert row["is_active"] == 1 + _state["a5_alert_id"] = alert["alert_id"] + + +# ---------- A-7：人工处置状态机 + compliance 403 + GET 仅 aml ---------- + + +def test_a7_handle_state_machine_compliance_forbidden(iclient, risk_demo_env): + c, _ = iclient + env = risk_demo_env + alert_id = _state["a3_alert_id"] + r = c.post( + f"/api/risk/alerts/{alert_id}/handle", + json={"handler_result": "confirmed_suspicious", "handler_comment": "确认可疑"}, + headers=OFFICER, + ) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "confirmed_suspicious" and body["handler_id"] == "STAFF-90001" + # 状态机：已处置单禁止跳改 + r2 = c.post( + f"/api/risk/alerts/{alert_id}/handle", + json={"handler_result": "confirmed_normal"}, + headers=OFFICER, + ) + assert r2.status_code == 409 + # compliance 无处置权 + r3 = c.post( + f"/api/risk/alerts/{alert_id}/handle", + json={"handler_result": "confirmed_normal"}, + headers=COMPLIANCE, + ) + assert r3.status_code == 403 + # 处置审计留痕（同事务，agent_type='risk'） + audit = _one( + env, + "SELECT * FROM audit_log WHERE event_type = 'alert_handle' AND actor_id = 'STAFF-90001'" + " AND decision = 'alert_handled' AND handler_result = 'confirmed_suspicious'", + ) + assert audit is not None + + +# ---------- A-9：越权 403 + 审计 ---------- + + +def test_a9_cross_customer_and_unassigned_advisor_403(iclient, risk_demo_env): + c, _ = iclient + env = risk_demo_env + # customer 查他人 + r = c.post( + "/api/risk/suitability/check", + json={"customer_id": "CUST-3001", "product_id": "PROD-510300"}, + headers=CUSTOMER_1002, + ) + assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_OWNER" + # advisor 查非名下（STAFF-10086 名下无 CUST-3001） + r = c.post( + "/api/risk/suitability/check", + json={"customer_id": "CUST-3001", "product_id": "PROD-510300"}, + headers=ADVISOR_10087, # 名下无 CUST-3001 + ) + assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_ASSIGNED" + denials = _one( + env, + "SELECT COUNT(*) AS n FROM audit_log WHERE event_type = 'authz' AND decision = 'forbidden'" + " AND actor_id IN ('CUST-1002', 'STAFF-10087')", + ) + assert denials["n"] == 2 + + +# ---------- 参数校验与无 trace 头兜底 ---------- + + +def test_convert_400_and_no_new_trade_audit(iclient, risk_demo_env): + c, _ = iclient + env = risk_demo_env + before = _one(env, "SELECT COUNT(*) AS n FROM audit_log WHERE event_type = 'trade_request'") + r = c.post( + "/api/simulate/trade", + json=_trade("CUST-3001", "PROD-510300", "convert", 1000), + headers=DEMO, + ) + assert r.status_code == 400 and r.json()["error_code"] == "BAD_REQUEST" + after = _one(env, "SELECT COUNT(*) AS n FROM audit_log WHERE event_type = 'trade_request'") + assert after["n"] == before["n"] # convert 属参数校验失败，不落审计 + + +def test_missing_trace_header_generates_one(iclient): + c, _ = iclient + r = c.get("/api/risk/alerts", headers=OFFICER) + assert r.status_code == 200 + assert r.headers["X-Trace-Id"].startswith("trc-") + + +def test_settings_thresholds_loaded(): + """冒烟：阈值配置与冻结规则一致（.env 未覆盖时）。""" + assert settings.risk_large_amount == 500000 + assert settings.risk_assessment_valid_days == 365 diff --git a/tests/test_profile_l3.py b/tests/test_profile_l3.py index b93fba0..712d0ef 100644 --- a/tests/test_profile_l3.py +++ b/tests/test_profile_l3.py @@ -26,16 +26,22 @@ from app.service.risk.profile_l3 import ( @pytest.fixture(autouse=True) def _fake_redis(monkeypatch): - """L3 写侧 DEL 钩子隔离（B7 挂账②）：upsert 后失效缓存不触真 Redis。""" + """L3 写侧 DEL 钩子隔离（B7 挂账②）+ 行为断言记录（B7 复审 P2-1）。""" class _FakeGateway: + def __init__(self): + self.messages = [] + self.deletes = [] + def publish(self, channel, payload): - pass + self.messages.append((channel, payload)) def delete(self, *keys): - pass + self.deletes.append(keys) - monkeypatch.setattr(redis_gateway, "_gateway", _FakeGateway()) + fake = _FakeGateway() + monkeypatch.setattr(redis_gateway, "_gateway", fake) + return fake @pytest.fixture() @@ -326,3 +332,31 @@ def test_persistent_conflict_raises_not_silent(env, monkeypatch): _upsert(repo, "C1", "large_amount", alert_id="ALT-2") row = _row(engine, "C1") assert row["monitor_tier"] == "watch" # 未被静默改写 + + +# ---------- 写侧缓存 DEL 钩子行为（B7 复审 P2-1） ---------- + + +def test_upsert_deletes_l3_cache_key(env, _fake_redis): + """upsert 成功后 DEL profile:l3:{customer_id}（PRD §5.1，B7 挂账②）。""" + repo, _ = env + _upsert(repo, "CUST-DEL-1", "pattern", alert_id="ALT-1") + assert ("profile:l3:CUST-DEL-1",) in _fake_redis.deletes + + +def test_cache_delete_failure_degrades_to_ttl(env, monkeypatch): + """降级路径：DEL 失败只告警不阻塞 upsert（TTL 兜底，redis_gateway 契约）。""" + repo, _ = env + + class _BrokenGateway: + def publish(self, channel, payload): + pass + + def delete(self, *keys): + raise ConnectionError("redis down") + + monkeypatch.setattr(redis_gateway, "_gateway", _BrokenGateway()) + merged = _upsert(repo, "C1", "pattern", alert_id="ALT-1") # 不抛即降级成功 + assert merged["monitor_tier"] == "watch" + row = _row(env[1], "C1") + assert row["monitor_tier"] == "watch" # DB 权威数据不受缓存失败影响