188 lines
6.5 KiB
Python
188 lines
6.5 KiB
Python
from decimal import Decimal
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.core.risk_questionnaire_contracts import RiskQuestionnaireSubmission
|
|
from app.model.fund import FundRiskAssessment as RiskAssessment
|
|
from app.model.memory import MemorySyncOutbox
|
|
from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag
|
|
from app.model.risk_questionnaire import ProfileSnapshot
|
|
from app.service.api_transaction_service import ApiTransactionService
|
|
from app.service.risk_questionnaire_service import RiskQuestionnaireService
|
|
|
|
|
|
def context() -> RequestContext:
|
|
return RequestContext(
|
|
user_id="7", trace_id="questionnaire-test", roles=("customer",)
|
|
)
|
|
|
|
|
|
def lowest_risk_answers() -> dict[str, int]:
|
|
return {
|
|
"q1": 12, "q2": 4, "q3": 5, "q4": 4, "q5": 4, "q6": 5, "q7": 5,
|
|
"q8": 4, "q9": 4, "q10": 1, "q11": 5, "q12": 4, "q13": 4,
|
|
}
|
|
|
|
|
|
def highest_risk_answers() -> dict[str, int]:
|
|
return {
|
|
"q1": 1, "q2": 1, "q3": 4, "q4": 1, "q5": 1, "q6": 1, "q7": 1,
|
|
"q8": 1, "q9": 1, "q10": 4, "q11": 1, "q12": 1, "q13": 1,
|
|
}
|
|
|
|
|
|
def submission() -> RiskQuestionnaireSubmission:
|
|
return RiskQuestionnaireSubmission(
|
|
answers=lowest_risk_answers(), declaration_accepted=True
|
|
)
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self) -> None:
|
|
self.items: list[object] = []
|
|
|
|
def add(self, item: object) -> None:
|
|
self.items.append(item)
|
|
|
|
async def flush(self) -> None:
|
|
for item in self.items:
|
|
if isinstance(item, AdvisorProfileDriftReview) and item.id is None:
|
|
item.id = 99
|
|
|
|
|
|
class FakeRepository:
|
|
def __init__(self, session: FakeSession) -> None:
|
|
self.session = session
|
|
self.deactivated = False
|
|
|
|
async def assessments_since(self, _customer_id: int, _since: object) -> int:
|
|
return 0
|
|
|
|
async def next_profile_version(self, _customer_id: int) -> int:
|
|
return 1
|
|
|
|
async def pending_drift_review(self, _customer_id: int, *, lock: bool = False) -> None:
|
|
del lock
|
|
return None
|
|
|
|
async def active_tags(
|
|
self, _customer_id: int, *, lock: bool = False
|
|
) -> list[AdvisorProfileTag]:
|
|
del lock
|
|
return []
|
|
|
|
async def deactivate_current_profile(self, _customer_id: int, _now: object) -> None:
|
|
self.deactivated = True
|
|
|
|
def add_assessment(self, assessment: RiskAssessment) -> None:
|
|
self.session.add(assessment)
|
|
|
|
def add_profile(self, profile: ProfileSnapshot) -> None:
|
|
self.session.add(profile)
|
|
|
|
def add_tag(self, tag: AdvisorProfileTag) -> None:
|
|
self.session.add(tag)
|
|
|
|
def add_drift_review(self, review: AdvisorProfileDriftReview) -> None:
|
|
self.session.add(review)
|
|
|
|
async def supersede_active_tags(
|
|
self, _customer_id: int, _tag_keys: tuple[str, ...], _now: object
|
|
) -> None:
|
|
return None
|
|
|
|
def add_sync_event(self, event: MemorySyncOutbox) -> None:
|
|
self.session.add(event)
|
|
|
|
|
|
def test_submission_requires_all_questions_and_declaration() -> None:
|
|
with pytest.raises(ValidationError, match="every questionnaire question"):
|
|
RiskQuestionnaireSubmission(answers={"q1": 1}, declaration_accepted=True)
|
|
with pytest.raises(ValidationError):
|
|
RiskQuestionnaireSubmission.model_validate({
|
|
"answers": lowest_risk_answers(), "declaration_accepted": False,
|
|
})
|
|
|
|
|
|
def test_score_bands_are_deterministic_and_server_only() -> None:
|
|
low = RiskQuestionnaireService.score(lowest_risk_answers())
|
|
high = RiskQuestionnaireService.score(highest_risk_answers())
|
|
assert (low.total, low.risk_level) == (13, "C1")
|
|
assert (high.total, high.risk_level) == (57, "C5")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_submission_persists_assessment_and_profile_without_exposing_them(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session = FakeSession()
|
|
|
|
async def execute(
|
|
_self: ApiTransactionService,
|
|
_context: RequestContext,
|
|
_scope: str,
|
|
_key: str | None,
|
|
_body: object,
|
|
operation: Any,
|
|
) -> dict[str, Any]:
|
|
return cast(dict[str, Any], await operation(session))
|
|
|
|
monkeypatch.setattr(ApiTransactionService, "execute", execute)
|
|
monkeypatch.setattr(
|
|
"app.service.risk_questionnaire_service.RiskQuestionnaireRepository", FakeRepository
|
|
)
|
|
response = await RiskQuestionnaireService().submit(
|
|
submission(), context(), "questionnaire-key-0001"
|
|
)
|
|
|
|
data = cast(dict[str, object], response["data"])
|
|
assert set(data) == {"completed", "questionnaire_version", "valid_until"}
|
|
assert data["completed"] is True
|
|
assert not {"total_score", "risk_level", "risk_profile", "profile"}.intersection(data)
|
|
assert any(isinstance(item, RiskAssessment) for item in session.items)
|
|
assert any(isinstance(item, ProfileSnapshot) for item in session.items)
|
|
tags = [item for item in session.items if isinstance(item, AdvisorProfileTag)]
|
|
assert {tag.tag_key for tag in tags} == {
|
|
"risk_level", "risk_profile", "investment_horizon", "preferred_asset_classes",
|
|
}
|
|
assert all(tag.source_type == "formal_risk_assessment" for tag in tags)
|
|
assert all(tag.status == "active" for tag in tags)
|
|
assert len([item for item in session.items if isinstance(item, MemorySyncOutbox)]) == 2
|
|
|
|
|
|
def test_tag_confidence_is_lower_near_a_risk_band_boundary() -> None:
|
|
assert RiskQuestionnaireService._risk_tag_confidence(22) == Decimal("0.7000")
|
|
assert RiskQuestionnaireService._risk_tag_confidence(17) == Decimal("0.9000")
|
|
|
|
|
|
def test_tag_value_change_requires_drift_review() -> None:
|
|
candidates = RiskQuestionnaireService._profile_tag_candidates(
|
|
RiskQuestionnaireService.score(highest_risk_answers()), highest_risk_answers(), 8
|
|
)
|
|
previous = AdvisorProfileTag(
|
|
id=1,
|
|
tag_uuid="previous-risk-level",
|
|
customer_id=7,
|
|
tag_key="risk_level",
|
|
tag_value="C1",
|
|
tag_value_hash="different-value-hash",
|
|
confidence=0.95,
|
|
source_type="formal_risk_assessment",
|
|
source_reference="fin_risk_assessment:7",
|
|
source_confidence=1,
|
|
profile_version=1,
|
|
drift_review_id=None,
|
|
previous_tag_id=None,
|
|
drift_reason=None,
|
|
status="active",
|
|
active_customer_tag="7:risk_level",
|
|
created_at=object(),
|
|
updated_at=object(),
|
|
)
|
|
changes = RiskQuestionnaireService._drift_changes([previous], candidates)
|
|
assert changes[0]["tag_key"] == "risk_level"
|
|
assert changes[0]["reason"] == "value_changed"
|