group_fqcd_jr/tests/unit/service/test_suitability_service.py

"""适当性校验（B2）单元测试：权威风险等级、测评有效期、专业投资者规则、越权与审计。"""

from datetime import UTC, datetime, timedelta
from typing import Any

import pytest
from pydantic import ValidationError

from app.core.contracts import RequestContext
from app.core.errors import ForbiddenAgentError
from app.service.suitability_service import (
    SuitabilityService,
    SuitabilityToolInput,
    suitability_tool_handler,
)

NOW = datetime(2026, 9, 9, 10, 0, tzinfo=UTC)
VALID_UNTIL = NOW + timedelta(days=30)


class FakeResult:
    def __init__(self, row: dict[str, Any] | None) -> None:
        self._row = row

    def mappings(self) -> "FakeResult":
        return self

    def first(self) -> dict[str, Any] | None:
        return self._row


class FakeSession:
    """同时支持只读查询与审计写入的最小替身。"""

    def __init__(self, row: dict[str, Any] | None, added: list[Any]) -> None:
        self._row = row
        self.added = added

    async def __aenter__(self) -> "FakeSession":
        return self

    async def __aexit__(self, *args: object) -> None:
        return None

    def begin(self) -> "FakeSession":
        return self

    async def execute(self, *args: object, **kwargs: object) -> FakeResult:
        return FakeResult(self._row)

    def add(self, item: Any) -> None:
        self.added.append(item)


def authority_row(**overrides: Any) -> dict[str, Any]:
    row: dict[str, Any] = {
        "is_professional_investor": 0,
        "professional_investor_status": "未申请",
        "investor_type": "C3",
        "assessed_at": NOW - timedelta(days=5),
        "valid_until": VALID_UNTIL,
    }
    row.update(overrides)
    return row


def query(**overrides: Any) -> SuitabilityToolInput:
    values: dict[str, Any] = {"customer_id": "7", "product_risk_level": 3}
    values.update(overrides)
    return SuitabilityToolInput.model_validate(values)


def context(**overrides: Any) -> RequestContext:
    values: dict[str, Any] = {"user_id": "7", "trace_id": "trace-suitability"}
    values.update(overrides)
    return RequestContext(**values)


def service_with_row(
    row: dict[str, Any] | None, added: list[Any] | None = None
) -> SuitabilityService:
    sink = added if added is not None else []
    return SuitabilityService(session_factory=lambda: FakeSession(row, sink))


async def test_authority_risk_level_replaces_caller_supplied_level() -> None:
    decision = await service_with_row(authority_row(investor_type="C5")).evaluate(
        query(product_risk_level=5), context(), now=NOW
    )
    assert decision.allowed is True
    assert decision.reason_code == "SUITABLE"
    assert decision.customer_risk_level == 5
    assert decision.risk_level_source == "fin_risk_assessment"


@pytest.mark.parametrize("forged", [{"customer_risk_level": 5}, {"professional_investor": True},
                                    {"assessment_expires_at": "2099-01-01T00:00:00Z"}])
def test_caller_cannot_declare_risk_facts(forged: dict[str, Any]) -> None:
    with pytest.raises(ValidationError):
        query(**forged)


async def test_insufficient_authority_level_is_denied() -> None:
    """低两个等级及以上仍必须拒绝（第十四条第 2、3 款，矩阵里的"❌ 禁止"）。"""
    decision = await service_with_row(authority_row(investor_type="C1")).evaluate(
        query(product_risk_level=3), context(), now=NOW
    )
    assert decision.allowed is False
    assert decision.reason_code == "RISK_LEVEL_MISMATCH"
    assert decision.requires_recording is True


@pytest.mark.parametrize(
    ("investor_type", "product_level", "allowed", "reason_code"),
    [
        # 逐格抄自 knowledge/policy/个人投资者适当性管理指南.md 第十二条矩阵。
        # 这张表的价值在于：任何一格被改动，都必须是有意为之并在此处说明理由。
        ("C1", 1, True, "SUITABLE"), ("C1", 2, True, "SUITABLE"),
        ("C1", 3, False, "RISK_LEVEL_MISMATCH"), ("C1", 4, False, "RISK_LEVEL_MISMATCH"),
        ("C1", 5, False, "RISK_LEVEL_MISMATCH"),
        ("C2", 1, True, "SUITABLE"), ("C2", 2, True, "SUITABLE"), ("C2", 3, True, "SUITABLE"),
        ("C2", 4, False, "RISK_LEVEL_MISMATCH"), ("C2", 5, False, "RISK_LEVEL_MISMATCH"),
        ("C3", 1, True, "SUITABLE"), ("C3", 2, True, "SUITABLE"), ("C3", 3, True, "SUITABLE"),
        ("C3", 4, True, "SUITABLE_WITH_DISCLOSURE"), ("C3", 5, False, "RISK_LEVEL_MISMATCH"),
        ("C4", 1, True, "SUITABLE"), ("C4", 2, True, "SUITABLE"), ("C4", 3, True, "SUITABLE"),
        ("C4", 4, True, "SUITABLE"), ("C4", 5, True, "SUITABLE_WITH_DISCLOSURE"),
        ("C5", 1, True, "SUITABLE"), ("C5", 2, True, "SUITABLE"), ("C5", 3, True, "SUITABLE"),
        ("C5", 4, True, "SUITABLE"), ("C5", 5, True, "SUITABLE"),
    ],
)
async def test_full_matrix_matches_policy_document(
    investor_type: str, product_level: int, allowed: bool, reason_code: str
) -> None:
    """客服回答必须与知识库里的矩阵一致 —— 这是同一个 Agent 的两条出口。"""
    decision = await service_with_row(authority_row(investor_type=investor_type)).evaluate(
        query(product_risk_level=product_level), context(), now=NOW
    )
    assert decision.allowed is allowed
    assert decision.reason_code == reason_code


async def test_disclosure_tier_always_requires_disclosure_and_recording() -> None:
    """C3→R4、C4→R5 是第十五条豁免档：可买，但必须揭示、确认、录音。"""
    for investor_type, product_level in (("C3", 4), ("C4", 5)):
        decision = await service_with_row(authority_row(investor_type=investor_type)).evaluate(
            query(product_risk_level=product_level), context(), now=NOW
        )
        assert decision.allowed is True
        assert decision.reason_code == "SUITABLE_WITH_DISCLOSURE"
        assert decision.required_disclosure is True
        assert decision.requires_confirmation is True
        assert decision.requires_recording is True


async def test_expired_assessment_is_denied_even_for_eligible_level() -> None:
    row = authority_row(investor_type="C5", valid_until=NOW - timedelta(seconds=1))
    decision = await service_with_row(row).evaluate(query(product_risk_level=1), context(), now=NOW)
    assert decision.allowed is False
    assert decision.reason_code == "ASSESSMENT_EXPIRED"
    assert decision.customer_risk_level == 5


async def test_missing_valid_until_is_treated_as_unusable() -> None:
    decision = await service_with_row(authority_row(valid_until=None)).evaluate(
        query(), context(), now=NOW
    )
    assert decision.allowed is False
    assert decision.reason_code == "ASSESSMENT_EXPIRED"


async def test_no_assessment_is_denied() -> None:
    decision = await service_with_row(authority_row(investor_type=None)).evaluate(
        query(), context(), now=NOW
    )
    assert decision.allowed is False
    assert decision.reason_code == "ASSESSMENT_MISSING"
    assert decision.customer_risk_level is None


async def test_unknown_customer_is_denied() -> None:
    decision = await service_with_row(None).evaluate(query(), context(), now=NOW)
    assert decision.allowed is False
    assert decision.reason_code == "CUSTOMER_NOT_FOUND"


@pytest.mark.parametrize("investor_type", ["X9", "C6", "C0", ""])
async def test_invalid_investor_type_fails_closed(investor_type: str) -> None:
    decision = await service_with_row(authority_row(investor_type=investor_type)).evaluate(
        query(), context(), now=NOW
    )
    assert decision.allowed is False
    assert decision.reason_code == "RISK_LEVEL_INVALID"


async def test_certified_professional_investor_gets_level_exemption_with_disclosure() -> None:
    row = authority_row(
        investor_type="C1",
        is_professional_investor=1,
        professional_investor_status="已认定",
    )
    decision = await service_with_row(row).evaluate(
        query(product_risk_level=5, product_requires_disclosure=True),
        context(),
        now=NOW,
    )
    assert decision.allowed is True
    assert decision.reason_code == "SUITABLE_PROFESSIONAL_INVESTOR"
    assert decision.professional_investor is True
    assert decision.required_disclosure is True
    assert decision.requires_confirmation is True
    assert decision.requires_recording is True


@pytest.mark.parametrize("status", ["未申请", "审核中", "已拒绝"])
async def test_uncertified_professional_status_does_not_exempt_level(status: str) -> None:
    row = authority_row(
        investor_type="C1", is_professional_investor=1, professional_investor_status=status
    )
    decision = await service_with_row(row).evaluate(
        query(product_risk_level=5), context(), now=NOW
    )
    assert decision.allowed is False
    assert decision.reason_code == "RISK_LEVEL_MISMATCH"


async def test_professional_investor_cannot_bypass_expired_assessment() -> None:
    row = authority_row(
        investor_type="C5",
        is_professional_investor=1,
        professional_investor_status="已认定",
        valid_until=NOW - timedelta(days=1),
    )
    decision = await service_with_row(row).evaluate(query(product_risk_level=1), context(), now=NOW)
    assert decision.allowed is False
    assert decision.reason_code == "ASSESSMENT_EXPIRED"


async def test_assessment_of_other_customer_is_forbidden() -> None:
    service = service_with_row(authority_row())
    with pytest.raises(ForbiddenAgentError):
        await service.evaluate(query(customer_id="9"), context(customer_ids=("8",)), now=NOW)


async def test_assigned_customer_is_allowed_and_admin_is_exempt() -> None:
    service = service_with_row(authority_row())
    assigned = await service.evaluate(query(customer_id="9"), context(customer_ids=("9",)), now=NOW)
    assert assigned.customer_risk_level == 3
    admin = await service.evaluate(
        query(customer_id="9"), context(roles=("admin",), permissions=()), now=NOW
    )
    assert admin.customer_risk_level == 3


async def test_decision_is_audited_with_authority_source() -> None:
    added: list[Any] = []
    service = service_with_row(authority_row(investor_type="C2"), added)
    decision = await service.evaluate_and_audit(query(product_risk_level=4), context(), now=NOW)
    assert decision.allowed is False
    assert len(added) == 1
    detail = added[0].detail
    assert detail["reason_code"] == "RISK_LEVEL_MISMATCH"
    assert detail["customer_risk_level"] == 2
    assert detail["risk_level_source"] == "fin_risk_assessment"
    assert detail["professional_investor"] is False
    assert "answers" not in detail


async def test_tool_handler_uses_same_service(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    added: list[Any] = []
    sink = added
    monkeypatch.setattr(
        "app.service.suitability_service.SessionFactory",
        lambda: FakeSession(authority_row(investor_type="C1"), sink),
    )
    result = await suitability_tool_handler(query(product_risk_level=5), context())
    assert result["allowed"] is False
    assert result["reason_code"] == "RISK_LEVEL_MISMATCH"
    assert result["customer_risk_level"] == 1
    assert len(added) == 1