202 lines
8.4 KiB
Python
202 lines
8.4 KiB
Python
"""suitability 单测（A4 · 验收 A-8）：矩阵 25 组合 + SUIT-006/008 边界 + 服务集成。"""
|
||
|
||
from datetime import date, datetime, timedelta
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import create_sqlite_engine
|
||
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.risk_repository import RiskRepository
|
||
from app.service.suitability import (
|
||
cap_by_age,
|
||
check_core,
|
||
grade_number,
|
||
is_assessment_valid,
|
||
match_by_matrix,
|
||
suitability_check,
|
||
)
|
||
|
||
TODAY = date(2026, 9, 6)
|
||
|
||
|
||
def _customer(level, age=40, evaluated=TODAY - timedelta(days=90)):
|
||
return {
|
||
"customer_id": "C1",
|
||
"risk_code": level,
|
||
"age": age,
|
||
"risk_evaluated_at": evaluated,
|
||
}
|
||
|
||
|
||
def _product(level):
|
||
return {"product_id": "P1", "min_risk_code": level}
|
||
|
||
|
||
class TestMatrix25:
|
||
"""SUIT-001~005：C1~C5 × R1~R5 全组合（期望值为硬编码字面量，直接对照附表矩阵审查）。"""
|
||
|
||
# 行=客户 C1~C5，列=产品 R1~R5（✓=可购），与附-风控规则表 §1 矩阵逐格对照
|
||
EXPECTED = [
|
||
# R1 R2 R3 R4 R5
|
||
[True, False, False, False, False], # C1
|
||
[True, True, False, False, False], # C2
|
||
[True, True, True, False, False], # C3
|
||
[True, True, True, True, False], # C4
|
||
[True, True, True, True, True], # C5
|
||
]
|
||
CUSTOMERS = ["C1", "C2", "C3", "C4", "C5"]
|
||
PRODUCTS = ["R1", "R2", "R3", "R4", "R5"]
|
||
|
||
@pytest.mark.parametrize("c", CUSTOMERS)
|
||
@pytest.mark.parametrize("p", PRODUCTS)
|
||
def test_matrix(self, c, p):
|
||
expect = self.EXPECTED[self.CUSTOMERS.index(c)][self.PRODUCTS.index(p)]
|
||
result = check_core(_customer(c), _product(p), today=TODAY)
|
||
assert result.is_matched is expect
|
||
assert result.blocked is (not expect) # 测评有效期内 blocked ≡ not matched
|
||
assert result.customer_level == c and result.effective_level == c
|
||
|
||
def test_a1_c1_buys_r4(self):
|
||
"""验收 A-1 断言口径：C1+R4 → rule_id=SUIT-001（客户维度，PRD 字面）。"""
|
||
result = check_core(_customer("C1"), _product("R4"), today=TODAY)
|
||
assert result.is_matched is False and result.blocked is True
|
||
assert result.rule_id == "SUIT-001"
|
||
assert "SUIT-001" in result.block_reason
|
||
|
||
|
||
class TestSUIT006AgeCap:
|
||
def test_70yo_c5_capped_to_c3(self):
|
||
level, reason = cap_by_age("C5", 70)
|
||
assert level == "C3" and "SUIT-006" in reason
|
||
|
||
def test_69yo_c5_not_capped(self):
|
||
assert cap_by_age("C5", 69) == ("C5", None)
|
||
|
||
def test_70yo_c3_unchanged(self):
|
||
assert cap_by_age("C3", 72) == ("C3", None)
|
||
|
||
def test_null_age_skipped_with_review_hint(self):
|
||
level, reason = cap_by_age("C5", None)
|
||
assert level == "C5" and reason == "年龄缺失，建议人工复核"
|
||
|
||
def test_a2_70yo_c5_buys_r4_blocked(self):
|
||
"""验收 A-2 断言口径：70 岁 C5 买 R4 → 封顶说明 + SUIT-003 不匹配，无过期干扰。"""
|
||
result = check_core(_customer("C5", age=70), _product("R4"), today=TODAY)
|
||
assert result.blocked is True and result.is_matched is False
|
||
assert result.effective_level == "C3"
|
||
assert result.rule_id == "SUIT-003" # 客户判定等级 C3 → SUIT-003（PRD A-2 字面）
|
||
assert any("SUIT-006" in r for r in result.reasons)
|
||
assert "SUIT-003" in result.block_reason
|
||
assert not any("SUIT-008" in r for r in result.reasons)
|
||
|
||
|
||
class TestSUIT008Validity:
|
||
def test_364_days_valid(self):
|
||
assert is_assessment_valid(TODAY - timedelta(days=364), TODAY, 365) is True
|
||
|
||
def test_365_days_expired(self):
|
||
assert is_assessment_valid(TODAY - timedelta(days=365), TODAY, 365) is False
|
||
|
||
def test_none_evaluated_expired(self):
|
||
assert is_assessment_valid(None, TODAY, 365) is False
|
||
|
||
def test_expired_but_matched_still_blocked(self):
|
||
"""C3 买 R3 等级匹配，但测评过期 → 仍阻断（双字段语义）+ rule_id=SUIT-008。"""
|
||
customer = _customer("C3", evaluated=TODAY - timedelta(days=400))
|
||
result = check_core(customer, _product("R3"), today=TODAY)
|
||
assert result.is_matched is True and result.blocked is True
|
||
assert result.rule_id == "SUIT-008"
|
||
assert "SUIT-008" in result.reasons[-1]
|
||
assert "重新测评" in result.block_reason
|
||
|
||
def test_custom_valid_days(self):
|
||
customer = _customer("C3", evaluated=TODAY - timedelta(days=100))
|
||
assert is_assessment_valid(customer["risk_evaluated_at"], TODAY, 90) is False
|
||
|
||
def test_str_and_datetime_input_compat(self):
|
||
"""str/datetime 输入兼容（驱动差异防御，评审 P2-4②）。"""
|
||
assert is_assessment_valid("2026-06-01", TODAY, 365) is True
|
||
assert is_assessment_valid("2025-06-01", TODAY, 365) is False
|
||
assert is_assessment_valid(datetime(2026, 6, 1, 12, 0), TODAY, 365) is True
|
||
|
||
|
||
class TestCheckService:
|
||
"""suitability_check 服务集成（sqlite 内存库）：落日志 + 原等级落库。"""
|
||
|
||
@pytest.fixture()
|
||
def repos(self):
|
||
engine = create_sqlite_engine() # DDL 单一事实源（B4 评审 P3-12）
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, age, open_date)"
|
||
" VALUES ('CUST-1', '客户·测**', 70, '2020-01-01')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at)"
|
||
" VALUES ('CUST-1', 'C5', :evaluated)"
|
||
),
|
||
{"evaluated": TODAY - timedelta(days=90)},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product (product_id, product_name, product_type, min_risk_code)"
|
||
" VALUES ('PROD-R4', '测试R4产品', 'stock', 'R4')"
|
||
)
|
||
)
|
||
core_ro = CoreReadOnlyRepository(engine=engine)
|
||
risk_repo = RiskRepository(engine=engine)
|
||
yield core_ro, risk_repo, engine
|
||
engine.dispose()
|
||
|
||
def test_service_blocked_and_log_written(self, repos):
|
||
core_ro, risk_repo, engine = repos
|
||
result = suitability_check(
|
||
"CUST-1", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo, today=TODAY
|
||
)
|
||
assert result.blocked is True and result.effective_level == "C3"
|
||
# 落库 customer_risk_level = 原测评等级 C5（封顶只进 reasons，PRD FR-2）
|
||
row = engine.connect().execute(
|
||
text(
|
||
"SELECT customer_risk_level, product_risk_level, is_matched, is_blocked"
|
||
" FROM risk_suitability_log"
|
||
)
|
||
).first()
|
||
assert row[0] == "C5" and row[1] == "R4" and row[2] == 0 and row[3] == 1
|
||
|
||
def test_service_unknown_customer_raises(self, repos):
|
||
core_ro, risk_repo, _ = repos
|
||
with pytest.raises(LookupError):
|
||
suitability_check("NOPE", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo, today=TODAY)
|
||
|
||
def test_service_default_valid_days_from_settings(self, repos, monkeypatch):
|
||
"""P1-1 回归：不传 valid_days 时取 settings（.env 可配性）。
|
||
|
||
CUST-1 为 70 岁 C5（封顶 C3 买 R4 本就不匹配，主因 SUIT-003 优先）；
|
||
配置阈值降到 90 后 90 天前的测评命中过期 → reasons 出现 SUIT-008 行即证明配置生效。
|
||
"""
|
||
from app.config.settings import settings as app_settings
|
||
|
||
core_ro, risk_repo, _ = repos
|
||
monkeypatch.setattr(app_settings, "risk_assessment_valid_days", 90)
|
||
result = suitability_check("CUST-1", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo, today=TODAY)
|
||
assert result.blocked is True and result.rule_id == "SUIT-003"
|
||
assert any("SUIT-008" in r for r in result.reasons)
|
||
|
||
def test_service_request_ref_passthrough(self, repos):
|
||
"""P2-2：request_ref 透传落库（B5 网关传 trade_id 用）。"""
|
||
core_ro, risk_repo, engine = repos
|
||
suitability_check(
|
||
"CUST-1", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo,
|
||
today=TODAY, request_ref="TRD-TEST-1",
|
||
)
|
||
row = engine.connect().execute(
|
||
text("SELECT request_ref, trace_id FROM risk_suitability_log")
|
||
).first()
|
||
assert row[0] == "TRD-TEST-1"
|
||
assert row[1] and row[1].startswith("trc-") # 无上游 trace 时兜底生成（P2-1）
|