85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.service.suitability_service import (
|
|
SuitabilityRequest,
|
|
SuitabilityService,
|
|
SuitabilityToolInput,
|
|
suitability_tool_handler,
|
|
)
|
|
|
|
|
|
def request(**overrides: object) -> SuitabilityRequest:
|
|
values: dict[str, object] = {
|
|
"customer_risk_level": 3,
|
|
"product_risk_level": 3,
|
|
"product_requires_disclosure": True,
|
|
}
|
|
values.update(overrides)
|
|
return SuitabilityRequest.model_validate(values)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("customer", "product", "allowed"),
|
|
[(1, 1, True), (1, 2, False), (3, 2, True), (5, 5, True)],
|
|
)
|
|
def test_risk_level_boundary(customer: int, product: int, allowed: bool) -> None:
|
|
decision = SuitabilityService().evaluate(
|
|
request(customer_risk_level=customer, product_risk_level=product)
|
|
)
|
|
assert decision.allowed is allowed
|
|
assert decision.reason_code == ("SUITABLE" if allowed else "RISK_LEVEL_MISMATCH")
|
|
|
|
|
|
def test_expired_assessment_is_denied() -> None:
|
|
now = datetime(2026, 9, 9, tzinfo=UTC)
|
|
decision = SuitabilityService().evaluate(
|
|
request(assessment_expires_at=now - timedelta(seconds=1)), now=now
|
|
)
|
|
assert decision.allowed is False
|
|
assert decision.reason_code == "ASSESSMENT_EXPIRED"
|
|
assert decision.requires_recording is True
|
|
|
|
|
|
def test_disclosure_requires_confirmation_and_recording() -> None:
|
|
decision = SuitabilityService().evaluate(
|
|
request(product_requires_disclosure=True, requires_confirmation=False)
|
|
)
|
|
assert decision.allowed is True
|
|
assert decision.required_disclosure is True
|
|
assert decision.requires_confirmation is True
|
|
assert decision.requires_recording is True
|
|
|
|
|
|
def test_missing_timezone_is_rejected() -> None:
|
|
with pytest.raises(ValueError, match="timezone"):
|
|
request(assessment_expires_at=datetime(2026, 9, 9))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_input_uses_same_rules(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class DummySession:
|
|
def add(self, _item: object) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self) -> "DummySession":
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
pass
|
|
|
|
def begin(self) -> "DummySession":
|
|
return self
|
|
|
|
monkeypatch.setattr(
|
|
"app.service.suitability_service.SessionFactory", lambda: DummySession()
|
|
)
|
|
context = RequestContext(user_id="7", trace_id="trace-suitability")
|
|
result = await suitability_tool_handler(
|
|
SuitabilityToolInput(customer_risk_level=1, product_risk_level=2), context
|
|
)
|
|
assert result["allowed"] is False
|
|
assert result["reason_code"] == "RISK_LEVEL_MISMATCH"
|