214 lines
8.6 KiB
Python
214 lines
8.6 KiB
Python
"""B9a 演示/运维脚本单测（开发计划 B9a · sqlite）。
|
||
|
||
rebuild_alerts：补偿重放出单并推送、幂等跳过（防重复 append/aml 重复出单）、
|
||
missing 不落库；subscribe_alerts：payload → 单行可读文本。
|
||
脚本目录非包，动态入 sys.path 后按模块名导入。
|
||
"""
|
||
|
||
import sys
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import create_sqlite_engine
|
||
|
||
DEMO_DIR = Path(__file__).resolve().parents[1] / "scripts" / "demo"
|
||
sys.path.insert(0, str(DEMO_DIR))
|
||
|
||
from rebuild_alerts import rebuild_trade # noqa: E402
|
||
from subscribe_alerts import format_alert # noqa: E402
|
||
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.risk_repository import RiskRepository
|
||
from app.service.risk import alert_service
|
||
from app.service.risk.alert_service import handle_alert
|
||
from app.service.risk.engine import process_trade_event
|
||
|
||
|
||
class FakePublisher:
|
||
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)
|
||
|
||
|
||
@pytest.fixture()
|
||
def env():
|
||
engine = create_sqlite_engine()
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, age, is_active) VALUES"
|
||
" ('C1', '张某某', 40, 1), ('C2', '李四', 35, 1)"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product (product_id, product_name, min_risk_code, product_type)"
|
||
" VALUES ('P1', '测试混合基金', 'R3', 'mixed')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO risk_aml_list (list_id, list_type, full_name, match_threshold,"
|
||
" source, list_version, is_active) VALUES ('PEP-1', 'pep', '李四', 0.85, 'mock', 'v1', 1)"
|
||
)
|
||
)
|
||
core = CoreReadOnlyRepository(engine=engine)
|
||
repo = RiskRepository(engine=engine)
|
||
pub = FakePublisher()
|
||
alert_service.set_publisher(pub)
|
||
yield core, repo, pub, engine
|
||
alert_service.set_publisher(None)
|
||
engine.dispose()
|
||
|
||
|
||
def _seed_trade(conn, trade_id, amount, customer="C1", at=datetime(2026, 9, 6, 14, 0, 0)):
|
||
"""直接落 core_trade 不调引擎（engine_error 补偿场景：交易已成立、预警缺失）。"""
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount,"
|
||
" trade_status, traded_at) VALUES (:tid, :cid, 'P1', 'subscribe', :amt, 'confirmed', :at)"
|
||
),
|
||
{"tid": trade_id, "cid": customer, "amt": amount, "at": at},
|
||
)
|
||
|
||
|
||
def _counts(engine, table, where="1=1"):
|
||
with engine.connect() as conn:
|
||
return conn.execute(text(f"SELECT COUNT(*) FROM {table} WHERE {where}")).scalar_one()
|
||
|
||
|
||
def test_rebuild_creates_alert_and_publishes(env):
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB1", "600000")
|
||
out = rebuild_trade("TRD-TEST-RB1", core, repo)
|
||
assert out["state"] == "rebuilt"
|
||
assert out["triggered_rules"] == ["RISK-001", "RISK-002"]
|
||
assert len(out["alert_ids"]) == 1 and out["aml_hit"] is False
|
||
alert = repo.get_alert(out["alert_ids"][0])
|
||
assert alert["status"] == "pending_review" and alert["risk_score"] == 70
|
||
(channel, _), = pub.messages
|
||
assert channel == "risk:pub:alert"
|
||
assert _counts(engine, "audit_log", "decision='alert_created'") == 1 # P-05 留痕链
|
||
|
||
|
||
def test_rebuild_idempotent_skips_second_run(env):
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB2", "600000")
|
||
first = rebuild_trade("TRD-TEST-RB2", core, repo)
|
||
second = rebuild_trade("TRD-TEST-RB2", core, repo)
|
||
assert first["state"] == "rebuilt" and second["state"] == "skipped"
|
||
assert second["alert_ids"] == first["alert_ids"]
|
||
assert _counts(engine, "risk_alert") == 1
|
||
assert len(repo.get_alert(first["alert_ids"][0])["payload"]["events"]) == 1
|
||
assert len(pub.messages) == 1 # 重放不重复推送
|
||
|
||
|
||
def test_rebuild_missing_trade_touches_nothing(env):
|
||
core, repo, pub, engine = env
|
||
out = rebuild_trade("TRD-NO-SUCH", core, repo)
|
||
assert out["state"] == "missing" and out["alert_ids"] == []
|
||
assert _counts(engine, "risk_alert") == 0
|
||
assert _counts(engine, "audit_log") == 0
|
||
assert pub.messages == []
|
||
|
||
|
||
def test_rebuild_aml_hit_then_idempotent(env):
|
||
"""aml 单幂等是 LIKE 检查的关键价值：record_aml_alert 本身无去重，重放防二次出单。"""
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB3", "1000", customer="C2")
|
||
first = rebuild_trade("TRD-TEST-RB3", core, repo)
|
||
assert first["state"] == "rebuilt" and first["aml_hit"] is True
|
||
aml_ids = [
|
||
aid
|
||
for aid in first["alert_ids"]
|
||
if repo.get_alert(aid)["alert_type"] == "aml"
|
||
]
|
||
assert aml_ids
|
||
second = rebuild_trade("TRD-TEST-RB3", core, repo)
|
||
assert second["state"] == "skipped" and second["alert_ids"] == first["alert_ids"]
|
||
assert _counts(engine, "risk_alert", "alert_type='aml'") == 1
|
||
|
||
|
||
def test_rebuild_non_first_trade_of_merged_alert_skips(env):
|
||
"""B8 复审口径（P2-2）：幂等保障依赖 LIKE 对 events[n]（n≥1）命中——
|
||
同日第二笔 append 进同单后 rebuild，必须 skip 且不重复 append。"""
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB4", "600000")
|
||
_seed_trade(conn, "TRD-TEST-RB5", "600000", at=datetime(2026, 9, 6, 15, 0, 0))
|
||
first = rebuild_trade("TRD-TEST-RB4", core, repo)
|
||
assert first["state"] == "rebuilt"
|
||
# 第二笔走正常引擎路径（等价网关同步调用）→ append 进同单
|
||
process_trade_event(core.get_trade_by_id("TRD-TEST-RB5"), core_ro=core, risk_repo=repo)
|
||
merged_id = first["alert_ids"][0]
|
||
assert len(repo.get_alert(merged_id)["payload"]["events"]) == 2
|
||
second = rebuild_trade("TRD-TEST-RB5", core, repo)
|
||
assert second["state"] == "skipped" and second["alert_ids"] == [merged_id]
|
||
assert len(repo.get_alert(merged_id)["payload"]["events"]) == 2 # 不重复 append
|
||
|
||
|
||
def test_rebuild_after_disposal_still_skips(env):
|
||
"""已处置单仍 skip（重放不得绕过处置结论，评审 P3-7）。"""
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB6", "600000")
|
||
out = rebuild_trade("TRD-TEST-RB6", core, repo)
|
||
aid = out["alert_ids"][0]
|
||
handle_alert(aid, "confirmed_normal", "STAFF-R1", risk_repo=repo)
|
||
again = rebuild_trade("TRD-TEST-RB6", core, repo)
|
||
assert again["state"] == "skipped" and again["alert_ids"] == [aid]
|
||
assert _counts(engine, "risk_alert") == 1
|
||
|
||
|
||
def test_rebuild_warns_on_engine_error_middle_state(env):
|
||
"""P2-1：首张单落库后 engine_error 中断（部分失败中间态）→ skip + warning 提示人工核对。"""
|
||
core, repo, pub, engine = env
|
||
with engine.begin() as conn:
|
||
_seed_trade(conn, "TRD-TEST-RB7", "600000")
|
||
out = rebuild_trade("TRD-TEST-RB7", core, repo)
|
||
assert "warning" not in out # 正常补偿无警示
|
||
# 模拟中断留痕：同一笔再走一遍"交易已成立但引擎异常"的审计（如 aml 单缺失场景）
|
||
repo.insert_audit_log(
|
||
{
|
||
"trace_id": "tr-x", "event_type": "trade_request", "agent_type": "platform",
|
||
"actor_id": "SYSTEM", "customer_id": "C1", "rule_id": None,
|
||
"input_summary": {"trade_id": "TRD-TEST-RB7", "error_stage": "process_trade_event"},
|
||
"decision": "risk_engine_error", "risk_score": None,
|
||
"handler_id": None, "handler_result": None, "handler_comment": None,
|
||
}
|
||
)
|
||
again = rebuild_trade("TRD-TEST-RB7", core, repo)
|
||
assert again["state"] == "skipped"
|
||
assert "人工核对" in again["warning"]
|
||
|
||
|
||
def test_format_alert_renders_payload_fields():
|
||
line = format_alert(
|
||
{
|
||
"alert_id": "ALT-20260906-ABC",
|
||
"alert_type": "aml",
|
||
"customer_id_mask": "CUST-9**",
|
||
"risk_score": 95,
|
||
"trace_id": "tr-1",
|
||
"notify_role": ["risk_officer", "compliance"],
|
||
}
|
||
)
|
||
for frag in ("aml", "ALT-20260906-ABC", "score=95", "CUST-9**", "tr-1",
|
||
"risk_officer,compliance"):
|
||
assert frag in line
|
||
assert "CUST-9527" not in line # payload 只有脱敏掩码，无原始 id
|
||
assert format_alert({"raw": "not-a-json-dict"}) == "[raw] not-a-json-dict"
|