XingHuo/tests/test_suitability.py

"""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）