GaoYiYuan_0626
66ec7eba08
代理人异常行为链识别（RISK-008）落地： - 新增 agent_behavior_service：三条件证据聚合（A 诱导调仓/B AUTH_403_SCOPE 越权试探/C AUTH_403_NOT_OWNER|NOT_ASSIGNED 越权查询），按代理人维度独立出 pattern 单，payload.actor_id 指向代理人，审计仅 INSERT event_type=agent_behavior_detected。 - risk_repository 新增 list_audit_events / find_agent_behavior_alert / merge_agent_behavior_payload（同日同代理人一张单，证据并集）。 - trade_gateway.submit_trade 补 actor_id 透传（代理人发起交易归属发起人，缺省 SYSTEM）；simulate 路由传入 auth.actor_id。 - chat_tools 新增 query_agent_behavior 只读 Tool（agent_id 过滤 + 客户脱敏），tool_service 补意图词与摘要。 - scripts/cron/agent_behavior_scan.py 定时扫描脚本。 - 修复 append_alert_event 序列化缺 default=str（C6 evidence 含 datetime 字段）。 - 单测 12 例（_count_induce 边界 / 三条件 / 出单去重 / payload 归属 / Tool 过滤脱敏）。 全量 pytest 482 passed 0 failed（470 基线 + 12 C6）。
261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""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, timedelta
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import create_sqlite_engine, seed_suitability_matrix
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
DEMO_SQL = ROOT / "scripts" / "demo" / "prepare_risk_demo.sql"
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _disable_concentration_rule(monkeypatch):
|
||
"""RISK-006 回归隔离（C4 起）：默认把集中度阈值推到不可达，现有用例断言零改动。
|
||
|
||
RISK-006 在引擎里对**每一笔**交易都会查持仓画像，若沿用默认 0.80，
|
||
现有 436 条用例中断言「命中规则集合 / 单日一张单」的会被新规则打穿。
|
||
故全局默认禁用；RISK-006 专属测试内再显式传 `RiskThresholds(...)` 或
|
||
再次 monkeypatch 回真实阈值 0.80（实现方案 §6 全量回归口径，评审 P1-2）。
|
||
|
||
前置条件：`RiskThresholds.from_settings()` 必须补读 concentration_threshold，
|
||
否则此处 monkeypatch 不生效（rules.py 已补读并注明）。
|
||
"""
|
||
from app.config.settings import settings
|
||
|
||
monkeypatch.setattr(settings, "risk_concentration_threshold", 1.01)
|
||
|
||
|
||
@pytest.fixture()
|
||
def sqlite_engine():
|
||
"""内存 sqlite 全表引擎（B8 前 DDL 散落各测试文件，收敛后统一走这里）。
|
||
|
||
AL-05 起统一灌 C×R 矩阵种子（check_suitability 以矩阵表为 L0 权威，
|
||
缺失即全量 forbidden，见 _ddl.seed_suitability_matrix）。
|
||
"""
|
||
engine = create_sqlite_engine()
|
||
seed_suitability_matrix(engine)
|
||
yield engine
|
||
engine.dispose()
|
||
|
||
|
||
@pytest.fixture()
|
||
def backdated_alert(sqlite_engine):
|
||
"""注入 created_at 回拨的 pending 单（C5/C6 免等待基建）；teardown 按 alert_id 精确删除。
|
||
|
||
用法：
|
||
aid = backdated_alert(alert_id="ALT-TEST-X", customer_id="C1", hours_ago=5,
|
||
payload={...}, alert_type="pattern", risk_score=60)
|
||
注意：注入与读取必须共用同一个 sqlite_engine（StaticPool 单连接共享）——测试内
|
||
一律 `RiskRepository(engine=sqlite_engine)`；回拨行 teardown 精确按 alert_id 删，
|
||
不污染正常 trace 空间（评审必查点，_cleanup_test_rows 的时间窗清不掉回拨行）。
|
||
"""
|
||
from sqlalchemy import text
|
||
|
||
import json
|
||
|
||
created_ids: list[str] = []
|
||
|
||
def _make(
|
||
alert_id: str,
|
||
customer_id: str,
|
||
hours_ago: float,
|
||
payload: dict | None = None,
|
||
alert_type: str = "pattern",
|
||
risk_score: int = 60,
|
||
) -> str:
|
||
created_at = datetime.now() - timedelta(hours=hours_ago)
|
||
with sqlite_engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO risk_alert (alert_id, trace_id, customer_id, trade_id,"
|
||
" alert_type, triggered_rules, risk_score, status, payload, created_at)"
|
||
" VALUES (:aid, 'TRACE-TEST', :cid, 'TRD-TEST-0', :atype, :rules,"
|
||
" :score, 'pending_review', :payload, :created_at)"
|
||
),
|
||
{
|
||
"aid": alert_id,
|
||
"cid": customer_id,
|
||
"atype": alert_type,
|
||
"rules": json.dumps(["RISK-007"]),
|
||
"score": risk_score,
|
||
"payload": json.dumps(payload or {}, ensure_ascii=False),
|
||
"created_at": created_at,
|
||
},
|
||
)
|
||
created_ids.append(alert_id)
|
||
return alert_id
|
||
|
||
yield _make
|
||
with sqlite_engine.begin() as conn:
|
||
for aid in created_ids:
|
||
conn.execute(
|
||
text("DELETE FROM risk_alert WHERE alert_id = :aid"), {"aid": aid}
|
||
)
|
||
|
||
|
||
@pytest.fixture()
|
||
def backdated_audit_event(sqlite_engine):
|
||
"""注入 created_at 回拨的 audit_log 行（C6）；teardown 按 trace_id（TEST-TRACE- 前缀）精确删除。
|
||
|
||
用法：
|
||
tid = backdated_audit_event(event_type="authz", actor_id="STAFF-X",
|
||
customer_id="CUST-Y",
|
||
input_summary={"code": "AUTH_403_SCOPE"},
|
||
decision="forbidden", hours_ago=5)
|
||
注意：注入与读取必须共用同一个 sqlite_engine；回拨行 teardown 按 trace_id 前缀
|
||
精确删除，不污染正常 trace 空间（评审必查点，_cleanup_test_rows 时间窗清不掉回拨行）。
|
||
"""
|
||
import json
|
||
from uuid import uuid4
|
||
|
||
created_trace_ids: list[str] = []
|
||
|
||
def _make(
|
||
event_type: str,
|
||
actor_id: str,
|
||
customer_id: str | None = None,
|
||
input_summary: dict | None = None,
|
||
decision: str = "forbidden",
|
||
hours_ago: float = 5.0,
|
||
agent_type: str = "risk",
|
||
trace_id: str | None = None,
|
||
) -> str:
|
||
tid = trace_id or f"TEST-TRACE-{uuid4().hex[:12].upper()}"
|
||
created_at = datetime.now() - timedelta(hours=hours_ago)
|
||
with sqlite_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, :et, :agent_type, :aid, :cid, NULL, :summary,"
|
||
" :decision, NULL, NULL, NULL, NULL, :created_at)"
|
||
),
|
||
{
|
||
"tid": tid,
|
||
"et": event_type,
|
||
"agent_type": agent_type,
|
||
"aid": actor_id,
|
||
"cid": customer_id,
|
||
"summary": json.dumps(input_summary or {}, ensure_ascii=False),
|
||
"decision": decision,
|
||
"created_at": created_at,
|
||
},
|
||
)
|
||
created_trace_ids.append(tid)
|
||
return tid
|
||
|
||
yield _make
|
||
with sqlite_engine.begin() as conn:
|
||
for tid in created_trace_ids:
|
||
conn.execute(text("DELETE FROM audit_log WHERE trace_id = :tid"), {"tid": tid})
|
||
|
||
|
||
# ---------- 真库集成环境（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:
|
||
# FM-03 口径（AL-05 换核）：expires_at 未过期即就位（原 evaluated_at+365 天口径退役）
|
||
days = conn.execute(
|
||
text(
|
||
"SELECT DATEDIFF(expires_at, CURDATE()) FROM core_customer_risk"
|
||
" WHERE customer_id = 'CUST-4001'"
|
||
)
|
||
).scalar_one_or_none()
|
||
if days is None or days <= 0:
|
||
pytest.skip(f"演示数据未就位（CUST-4001 风评 expires_at 剩余 {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 Exception as exc:
|
||
# pytest.skip 抛出的 Skipped 继承 BaseException，不会被此处吞掉，直接向上传播
|
||
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()
|
||
l3_snapshot: list = [] # 哨兵：setup 失败时 teardown 不因未绑定变量掩盖原始异常
|
||
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()
|