from __future__ import annotations from datetime import date from pathlib import Path import pytest from checkin_cli import nutrition_onboarding as onboarding_domain from checkin_cli.nutrition_onboarding_clarification_policy import ( compile_clarification_policy, ) from checkin_cli.nutrition_onboarding_contract import ( QUESTION_FIELDS, MessageEvidence, OnboardingAuthority, canonical_digest, example_answer, ) from checkin_cli.nutrition_onboarding_reconciliation import ( build_reconciliation, recompute_reconciliation, ) REFERENCE_DATE = date(2026, 8, 18) def _answers(**overrides: object) -> dict[str, object]: answers = {field: example_answer(field) for field in QUESTION_FIELDS} answers.update(overrides) return answers def test_mixed_ambiguity_and_invalid_value_compile_complete_ordered_issue_set() -> None: result = compile_clarification_policy( _answers(height_cm="180 또는 185", meal_count=9), reference_date=REFERENCE_DATE, ) assert [(issue.field, issue.reason_code) for issue in result.issues] == [ ("height_cm", "multiple_measurements"), ("meal_count", "invalid_value"), ] assert result.visible_issue_ids == tuple( issue.issue_id for issue in result.issues ) def test_plain_invalid_date_exception_maps_to_date_of_birth() -> None: result = compile_clarification_policy( _answers(date_of_birth="태어난 날은 잘 모르겠습니다"), reference_date=REFERENCE_DATE, ) assert [(issue.field, issue.reason_code) for issue in result.issues] == [ ("date_of_birth", "invalid_date_of_birth") ] @pytest.mark.parametrize( ("overrides", "expected"), ( ( { "goal_type": "maintain", "target_weight_kg": "75", "target_date": "2026-12-01", }, [ ("target_weight_kg", "target_not_allowed"), ("target_date", "target_not_allowed"), ], ), ( {"goal_type": "loss", "target_weight_kg": None, "target_date": None}, [ ("target_weight_kg", "target_required"), ("target_date", "target_required"), ], ), ( { "goal_type": "gain", "target_weight_kg": "70", "target_date": "2026-12-01", }, [("target_weight_kg", "target_direction_invalid")], ), ), ) def test_cross_field_invariants_map_only_involved_fields( overrides: dict[str, object], expected: list[tuple[str, str]], ) -> None: result = compile_clarification_policy( _answers(**overrides), reference_date=REFERENCE_DATE, ) assert [(issue.field, issue.reason_code) for issue in result.issues] == expected assert all(issue.field != "goal_type" for issue in result.issues) def test_recompute_reuses_persisted_reference_date_and_compiler_digest() -> None: answers = _answers(date_of_birth="2008-08-19") live = build_reconciliation( answers=answers, answers_digest=canonical_digest(answers), advisory={"summary_ko": "advisory"}, reference_date=REFERENCE_DATE, ) replay = recompute_reconciliation( live, answers=answers, answers_digest=canonical_digest(answers), ) next_day = compile_clarification_policy( answers, reference_date=date(2026, 8, 19), ) assert live == replay assert live["reference_date"] == REFERENCE_DATE.isoformat() assert live["compiler_digest"] != next_day.result_digest assert [issue["field"] for issue in live["issues"]] == ["date_of_birth"] assert next_day.issues == () @pytest.mark.parametrize( ("field", "value"), ( ("conditions", {"status": "provided", "items": []}), ("medications", {"status": "provided", "items": "철분"}), ("pregnancy_breastfeeding", "maybe"), ("pregnancy_breastfeeding", 1), ("eating_disorder_risk", "maybe"), ("eating_disorder_risk", 1), ), ) def test_malformed_sensitive_values_are_issues_never_holds( field: str, value: object, ) -> None: result = compile_clarification_policy( _answers(**{field: value}), reference_date=REFERENCE_DATE, ) assert [(issue.field, issue.reason_code) for issue in result.issues] == [ (field, "invalid_value") ] assert result.holds == () def test_unknown_safety_answers_are_holds_not_clarification_issues() -> None: result = compile_clarification_policy( _answers( allergies={"status": "unknown", "items": []}, intolerances={"status": "unknown", "items": []}, conditions={"status": "unknown", "items": []}, medications={"status": "unknown", "items": []}, pregnancy_breastfeeding=None, eating_disorder_risk=None, ), reference_date=REFERENCE_DATE, ) assert result.issues == () assert tuple(hold.field for hold in result.holds) == ( "allergies", "intolerances", "conditions", "medications", "pregnancy_breastfeeding", "eating_disorder_risk", ) def test_nested_structured_item_ambiguity_is_compiler_owned() -> None: result = compile_clarification_policy( _answers( allergies={ "status": "provided", "items": ["땅콩 또는 호두"], } ), reference_date=REFERENCE_DATE, ) assert [(issue.field, issue.reason_code) for issue in result.issues] == [ ("allergies", "multiple_values") ] def test_canonical_types_and_field_local_issue_identity_are_stable() -> None: first = compile_clarification_policy( _answers(height_cm=180.0, meal_count=True), reference_date=REFERENCE_DATE, ) second = compile_clarification_policy( _answers(height_cm=180, meal_count=True, budget_band="높음"), reference_date=REFERENCE_DATE, ) assert first.answers["height_cm"] == "180" assert type(first.answers["meal_count"]) is bool assert [(issue.field, issue.reason_code) for issue in first.issues] == [ ("meal_count", "invalid_value") ] assert first.issues[0].issue_id == second.issues[0].issue_id def _authority() -> OnboardingAuthority: return OnboardingAuthority( customer_key="client_001", customer_user_id=10, customer_chat_id=-100, customer_topic_id=20, owner_user_id=12, owner_chat_id=-100, owner_topic_id=22, consent_notice_version="privacy-v1", consent_granted=True, customer_enabled=False, ) def _evidence(index: int, *, owner: bool = False) -> MessageEvidence: return MessageEvidence( actor_user_id=12 if owner else 10, chat_id=-100, topic_id=22 if owner else 20, message_id=100 + index, update_id=1000 + index, ) def test_owner_arbitrary_issue_binding_cannot_mutate_lifecycle(tmp_path: Path) -> None: service = onboarding_domain.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, enforce_reconciliation=True, ) authority = _authority() service.start_or_resume( authority=authority, evidence=_evidence(0), reference_date=REFERENCE_DATE, ) answers = _answers() for index, field in enumerate(QUESTION_FIELDS, start=1): service.submit_answer( field=field, value=answers[field], authority=authority, evidence=_evidence(index), ) record = service.record_reconciliation( answers_digest=canonical_digest(answers), advisory={}, clarifications=[], authority=authority, reference_date=REFERENCE_DATE, ) service.attest_baseline(authority=authority, evidence=_evidence(30)) with pytest.raises(ValueError, match="binding is stale"): service.review_as_owner( decision="revise", revision_issue_id="a" * 64, revision_field="meal_count", expected_answers_digest=canonical_digest(answers), expected_reconciliation_digest=record["digest"], authority=authority, evidence=_evidence(31, owner=True), ) assert service.status().state.value == "owner_review" assert service.baseline_path.exists()