from __future__ import annotations import importlib.util import json from datetime import date, datetime, timedelta, timezone from pathlib import Path import pytest import checkin_cli.nutrition_onboarding_finalization as finalization_module from checkin_cli.nutrition_onboarding_finalization import validate_clinical_clearance from checkin_cli.nutrition_readiness import audit_nutrition_start_readiness from checkin_cli.nutrition_readiness_io import canonical_digest from checkin_cli.nutrition_restriction_kb import ( load_restriction_kb_template, seed_restriction_kb, ) MODULE = "checkin_cli.nutrition_onboarding" def _load(): if importlib.util.find_spec(MODULE) is None: pytest.fail("nutrition onboarding service contract missing") return __import__(MODULE, fromlist=["*"]) def _authority(mod, **overrides: object): values: dict[str, object] = { "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, } values.update(overrides) return mod.OnboardingAuthority(**values) def _evidence(mod, *, actor: int = 10, topic: int = 20, message: int = 1): return mod.MessageEvidence( actor_user_id=actor, chat_id=-100, topic_id=topic, message_id=message, update_id=message + 100, ) def test_service_contract_is_present() -> None: mod = _load() assert mod.NUTRITION_ONBOARDING_API_VERSION == "2.0" assert len(mod.QUESTION_FIELDS) >= 20 def test_start_rejects_symlinked_profile_data_ancestor(tmp_path: Path) -> None: mod = _load() profile_root = tmp_path / "profile" outside = tmp_path / "outside" profile_root.mkdir() outside.mkdir() (profile_root / "data").symlink_to(outside, target_is_directory=True) with pytest.raises(ValueError, match="symlink"): mod.NutritionOnboardingService( profile_root=profile_root, customer_key="client_001", enforce_current_authority=False, ) assert not (outside / "customers" / "client_001").exists() def test_start_rejects_symlinked_profile_root(tmp_path: Path) -> None: mod = _load() outside = tmp_path / "outside" outside.mkdir() profile_root = tmp_path / "profile" profile_root.symlink_to(outside, target_is_directory=True) with pytest.raises(ValueError, match="profile root"): mod.NutritionOnboardingService( profile_root=profile_root, customer_key="client_001", enforce_current_authority=False, ) assert not (outside / "data").exists() def test_start_rejects_profile_data_swapped_to_symlink_after_construction( tmp_path: Path, ) -> None: mod = _load() profile_root = tmp_path / "profile" outside = tmp_path / "outside" (profile_root / "data").mkdir(parents=True) outside.mkdir() service = mod.NutritionOnboardingService( profile_root=profile_root, customer_key="client_001", enforce_current_authority=False, ) data_path = profile_root / "data" data_path.rename(profile_root / "data-original") data_path.symlink_to(outside, target_is_directory=True) with pytest.raises(ValueError, match="symlink"): service.store.initialize() with pytest.raises(ValueError, match="symlink"): service.start_or_resume( authority=_authority(mod), evidence=_evidence(mod), ) assert not (outside / "customers" / "client_001").exists() def test_start_requires_current_privacy_consent_and_disabled_target(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) with pytest.raises(ValueError, match="privacy-v1"): service.start_or_resume( authority=_authority(mod, consent_granted=False), evidence=_evidence(mod), ) with pytest.raises(ValueError, match="disabled"): service.start_or_resume( authority=_authority(mod, customer_enabled=True), evidence=_evidence(mod), ) def test_service_rejects_mutation_without_current_registry_authority( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", ) with pytest.raises(ValueError, match="authority"): service.start_or_resume( authority=_authority(mod), evidence=_evidence(mod), ) assert not service.session_path.exists() def test_customer_answers_require_exact_route_and_question_order(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) with pytest.raises(ValueError, match="route"): service.submit_answer( field=mod.QUESTION_FIELDS[0], value="1996-08-01", authority=authority, evidence=_evidence(mod, actor=999), ) with pytest.raises(ValueError, match="expected question"): service.submit_answer( field=mod.QUESTION_FIELDS[1], value="male", authority=authority, evidence=_evidence(mod, message=2), ) def test_restart_preserves_answer_cursor_without_exposing_values(tmp_path: Path) -> None: mod = _load() authority = _authority(mod) first = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) first.start_or_resume(authority=authority, evidence=_evidence(mod)) first.submit_answer( field=mod.QUESTION_FIELDS[0], value="1996-08-01", authority=authority, evidence=_evidence(mod, message=2), ) restarted = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) status = restarted.status() assert status.answer_count == 1 assert status.next_field == mod.QUESTION_FIELDS[1] assert "1996-08-01" not in status.model_dump_json() def test_attestation_goes_directly_to_owner_review(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) for index, field in enumerate(mod.QUESTION_FIELDS): service.submit_answer( field=field, value=mod.example_answer(field), authority=authority, evidence=_evidence(mod, message=index + 2), ) status = service.status() assert status.state == mod.OnboardingState.CUSTOMER_ATTESTATION attested = service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) assert attested.state == mod.OnboardingState.OWNER_REVIEW owned = service.review_as_owner( decision="approved", authority=authority, evidence=_evidence(mod, actor=12, topic=22, message=100), ) assert owned.state == mod.OnboardingState.FINALIZING def test_customer_v2_maintain_flow_uses_fifteen_screens_and_age_only( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) status = service.start_or_resume( authority=authority, evidence=_evidence(mod), questionnaire_version="customer_v2", reference_date=date(2026, 9, 1), ) assert status.next_field == "goal_type" assert status.question_number == 1 assert status.question_total == 17 answers: tuple[tuple[str, object], ...] = ( ("goal_type", "maintain"), ("meal_count", 3), ( "activity_profile", { "activity_category": "moderate", "activity_rationale": "주 3회 헬스", }, ), ( "food_preferences_profile", { "religious_ethical_exclusions": {"status": "none", "items": []}, "disliked_foods": {"status": "provided", "items": ["고수"]}, "dietary_preferences": {"status": "provided", "items": ["한식"]}, }, ), ( "lifestyle_profile", { "cooking_access": "전자레인지와 가스레인지 사용 가능", "budget_band": "한 끼 1만 원 안팎", "schedule_constraints": "평일 점심은 외식", }, ), ("allergies", {"status": "none", "items": []}), ("intolerances", {"status": "none", "items": []}), ("date_of_birth", 30), ("equation_sex_basis", "male"), ("height_cm", "175"), ("weight_kg", "70"), ("conditions", {"status": "none", "items": []}), ("medications", {"status": "none", "items": []}), ("pregnancy_breastfeeding", False), ("eating_disorder_risk", False), ) for index, (field, value) in enumerate(answers, start=2): status = service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index), ) assert status.state == mod.OnboardingState.CUSTOMER_ATTESTATION assert status.answer_count == 15 candidate = service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) assert candidate.state == mod.OnboardingState.OWNER_REVIEW baseline = json.loads(service.baseline_path.read_text()) assert baseline["baseline"]["adult_age"] == 30 assert baseline["source_answers"]["date_of_birth"] == 30 assert baseline["source_answers"]["target_weight_kg"] is None assert baseline["source_answers"]["target_date"] is None def test_customer_v2_loss_flow_requires_both_target_screens( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), questionnaire_version="customer_v2", reference_date=date(2026, 9, 1), ) status = service.submit_answer( field="goal_type", value="loss", authority=authority, evidence=_evidence(mod, message=2), ) assert status.next_field == "target_weight_kg" assert status.question_number == 2 assert status.question_total == 17 def test_answer_failures_persist_handoff_and_owner_resume( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), questionnaire_version="customer_v2", reference_date=date(2026, 9, 1), ) first = service.record_answer_failure( field="goal_type", authority=authority, evidence=_evidence(mod, message=2), ) restarted = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) handed_off = restarted.record_answer_failure( field="goal_type", authority=authority, evidence=_evidence(mod, message=3), ) assert first.state == mod.OnboardingState.COLLECTING assert first.answer_failure_count == 1 assert handed_off.state == mod.OnboardingState.INPUT_HELP assert handed_off.help_field == "goal_type" assert handed_off.answer_failure_count == 2 resumed = restarted.resume_input_help( authority=authority, evidence=_evidence(mod, actor=12, topic=22, message=4), ) assert resumed.state == mod.OnboardingState.COLLECTING assert resumed.next_field == "goal_type" assert resumed.answer_failure_count == 0 def test_formal_required_unknown_hands_off_on_first_attempt( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), questionnaire_version="customer_v2", ) status = service.record_answer_failure( field="goal_type", authority=authority, evidence=_evidence(mod, message=2), force_handoff=True, ) assert status.state == mod.OnboardingState.INPUT_HELP assert status.answer_failure_count == 1 def test_reconciliation_uses_session_reference_date_across_restart( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), reference_date=date(2026, 8, 18), ) answers: dict[str, object] = {} for index, field in enumerate(mod.QUESTION_FIELDS, start=2): value = mod.example_answer(field) answers[field] = value _ = service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index), ) restarted = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) record = restarted.record_reconciliation( answers_digest=canonical_digest(answers), advisory={}, clarifications=[], authority=authority, ) assert record["reference_date"] == "2026-08-18" with pytest.raises(ValueError, match="reference date"): _ = restarted.replace_stale_reconciliation( expected_stale_answers_digest="0" * 64, answers_digest=canonical_digest(answers), advisory={}, clarifications=[], authority=authority, reference_date=date(2026, 8, 19), ) def test_customer_v2_group_submission_is_atomic_restartable_and_rewindable( tmp_path: Path, ) -> None: mod = _load() authority = _authority(mod) service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), questionnaire_version="customer_v2", reference_date=date(2026, 9, 1), ) _ = service.submit_answer( field="goal_type", value="maintain", authority=authority, evidence=_evidence(mod, message=2), ) _ = service.submit_answer( field="meal_count", value=3, authority=authority, evidence=_evidence(mod, message=3), ) with pytest.raises(ValueError, match="grouped"): service.submit_answer( field="activity_profile", value={"activity_category": "moderate"}, authority=authority, evidence=_evidence(mod, message=4), ) persisted = json.loads(service.session_path.read_text()) assert "activity_category" not in persisted["answers"] assert service.status().next_field == "activity_profile" status = service.submit_answer( field="activity_profile", value={ "activity_category": "moderate", "activity_rationale": "주 3회 헬스", }, authority=authority, evidence=_evidence(mod, message=5), ) assert status.next_field == "food_preferences_profile" restarted = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) assert restarted.status().next_field == "food_preferences_profile" _ = restarted.submit_answer( field="food_preferences_profile", value={ "religious_ethical_exclusions": {"status": "none", "items": []}, "disliked_foods": {"status": "provided", "items": ["고수"]}, "dietary_preferences": {"status": "provided", "items": ["한식"]}, }, authority=authority, evidence=_evidence(mod, message=6), ) rewound = restarted.rewind_collection( authority=authority, evidence=_evidence(mod, message=7), target_cursor=3, ) assert rewound.next_field == "food_preferences_profile" persisted = json.loads(service.session_path.read_text()) for field in ( "religious_ethical_exclusions", "disliked_foods", "dietary_preferences", ): assert field not in persisted["answers"] @pytest.mark.parametrize("meal_count", (1, 3, 7, 8)) def test_onboarding_baseline_accepts_supported_meal_counts( meal_count: int, ) -> None: mod = _load() answers = { field: mod.example_answer(field) for field in mod.QUESTION_FIELDS } answers["meal_count"] = meal_count baseline = mod.build_onboarding_baseline( { **answers, "schema_version": "1.0", "customer_key": "client_001", "session_status": "completed", "review_decision": "pending", "publication_status": "unpublished", }, as_of=date(2026, 9, 1), ) assert baseline.meal_count == meal_count def test_reconciliation_answers_are_authorized_and_copy_only( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) expected = {} for index, field in enumerate(mod.QUESTION_FIELDS): value = mod.example_answer(field) expected[field] = value service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index + 2), ) answers = service.reconciliation_answers(authority=authority) assert answers == expected answers["meal_count"] = 99 assert service.reconciliation_answers(authority=authority) == expected with pytest.raises(ValueError, match="customer attestation"): service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) service.reconciliation_answers(authority=authority) def test_reconciliation_revision_is_digest_bound_and_customer_authorized( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) original = {} for index, field in enumerate(mod.QUESTION_FIELDS): value = mod.example_answer(field) original[field] = value service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index + 2), ) answers_digest = canonical_digest(original) revised = service.revise_reconciliation_answer( field="meal_count", value=5, expected_answers_digest=answers_digest, authority=authority, evidence=_evidence(mod, message=99), ) assert revised.state == mod.OnboardingState.CUSTOMER_ATTESTATION expected = {**original, "meal_count": 5} assert service.reconciliation_answers(authority=authority) == expected with pytest.raises(ValueError, match="digest"): service.revise_reconciliation_answer( field="meal_count", value=6, expected_answers_digest=answers_digest, authority=authority, evidence=_evidence(mod, message=100), ) with pytest.raises(ValueError, match="canonical"): service.revise_reconciliation_answer( field="not_a_field", value="unsafe", expected_answers_digest=canonical_digest(expected), authority=authority, evidence=_evidence(mod, message=101), ) assert service.reconciliation_answers(authority=authority) == expected def test_unresolved_clarification_hands_off_after_two_revisions( tmp_path: Path, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, enforce_reconciliation=True, ) authority = _authority(mod) _ = service.start_or_resume( authority=authority, evidence=_evidence(mod), ) for index, field in enumerate(mod.QUESTION_FIELDS, start=2): value = ( "170 또는 180" if field == "height_cm" else mod.example_answer(field) ) _ = service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index), ) answers = service.reconciliation_answers(authority=authority) _ = service.record_reconciliation( answers_digest=canonical_digest(answers), advisory={"status": "unavailable"}, clarifications=[], authority=authority, ) first = service.revise_reconciliation_answer( field="height_cm", value="대략 170", expected_answers_digest=canonical_digest(answers), authority=authority, evidence=_evidence(mod, message=100), ) revised_answers = service.reconciliation_answers(authority=authority) second = service.revise_reconciliation_answer( field="height_cm", value="대략 170", expected_answers_digest=canonical_digest(revised_answers), authority=authority, evidence=_evidence(mod, message=101), ) assert first.state == mod.OnboardingState.CUSTOMER_ATTESTATION assert first.answer_failure_count == 1 assert second.state == mod.OnboardingState.INPUT_HELP assert second.help_field == "height_cm" assert second.answer_failure_count == 2 def test_redacted_status_document_contains_no_answer_values(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) service.submit_answer( field=mod.QUESTION_FIELDS[0], value="1996-08-01", authority=authority, evidence=_evidence(mod, message=2), ) rendered = json.dumps(service.status().model_dump(mode="json"), sort_keys=True) assert "1996-08-01" not in rendered assert "answers" not in rendered def _collect_complete_answers( mod: object, service: object, authority: object, *, equation_sex_basis: str, ) -> dict[str, object]: answers: dict[str, object] = {} for index, field in enumerate(mod.QUESTION_FIELDS): value = ( equation_sex_basis if field == "equation_sex_basis" else mod.example_answer(field) ) answers[field] = value service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index + 2), ) return answers @pytest.mark.parametrize( ("equation_answer", "canonical_answer"), ( ("넵 남성입니다", "male"), ("네 여성입니다", "female"), ("선택 안 합니다", "decline"), ), ) def test_unambiguous_equation_basis_reaches_owner_review_without_redundant_clarification( tmp_path: Path, equation_answer: str, canonical_answer: str, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, enforce_reconciliation=True, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) _collect_complete_answers( mod, service, authority, equation_sex_basis=equation_answer, ) status = service.status() assert status.answer_count == 22 assert status.state == mod.OnboardingState.CUSTOMER_ATTESTATION answers = service.reconciliation_answers(authority=authority) assert answers["equation_sex_basis"] == canonical_answer reconciliation = service.record_reconciliation( answers_digest=canonical_digest(answers), advisory={"summary_ko": "남성 계산 기준으로 확인했습니다."}, clarifications=[ { "field": "equation_sex_basis", "kind": "ambiguity", "question_ko": "계산 기준을 남성으로 적용할까요?", } ], authority=authority, ) assert reconciliation["state"] == "resolved" assert reconciliation["clarifications"] == [] attested = service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) assert attested.state == mod.OnboardingState.OWNER_REVIEW candidate = json.loads(service.baseline_path.read_text(encoding="utf-8")) assert candidate["source_answers"]["equation_sex_basis"] == canonical_answer assert candidate["baseline"]["equation_sex_basis"] == canonical_answer @pytest.mark.parametrize("kind", ["ambiguity", "contradiction"]) def test_noncanonical_equation_basis_keeps_clarification_and_requires_enum_resolution( tmp_path: Path, kind: str, ) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, enforce_reconciliation=True, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) _collect_complete_answers( mod, service, authority, equation_sex_basis="남성 또는 여성 중 하나", ) answers = service.reconciliation_answers(authority=authority) reconciliation = service.record_reconciliation( answers_digest=canonical_digest(answers), advisory={"summary_ko": "계산 기준 확인이 필요합니다."}, clarifications=[ { "field": "equation_sex_basis", "kind": kind, "question_ko": "계산 기준 하나를 선택해 주세요.", } ], authority=authority, ) assert reconciliation["state"] == "clarifying" assert reconciliation["clarifications"] == [] assert len(reconciliation["issues"]) == 1 issue = reconciliation["issues"][0] assert issue["field"] == "equation_sex_basis" assert issue["reason_code"] == "multiple_values" assert issue["question_ko"] == ( "계산 기준을 남성, 여성, 선택 안 함 중 하나로 입력해 주세요." ) assert "계산 기준 하나를 선택해 주세요." not in str(reconciliation) binding = { "expected_issue_id": issue["issue_id"], "expected_answers_digest": reconciliation["answers_digest"], "expected_reconciliation_digest": reconciliation["digest"], } with pytest.raises(ValueError, match="male, female, or decline"): service.revise_reconciliation_answer( field="equation_sex_basis", value="남성이라는 단어가 포함된 모호한 답변", authority=authority, evidence=_evidence(mod, message=99), **binding, ) assert service.reconciliation_answers(authority=authority) == answers revised = service.revise_reconciliation_answer( field="equation_sex_basis", value="Yes, male", authority=authority, evidence=_evidence(mod, message=100), **binding, ) assert revised.state == mod.OnboardingState.CUSTOMER_ATTESTATION assert service.reconciliation_answers(authority=authority)[ "equation_sex_basis" ] == "male" assert service.reconciliation_record(authority=authority)["state"] == "resolved" def test_safety_hold_requires_scoped_owner_clinical_clearance(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) for index, field in enumerate(mod.QUESTION_FIELDS): value = mod.example_answer(field) if field == "conditions": value = {"status": "provided", "items": ["외부 검토 필요"]} if field == "pregnancy_breastfeeding": value = True service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index + 2), ) held = service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) assert held.state == mod.OnboardingState.SAFETY_HOLD cleared = service.record_clinical_review( cleared=True, external_reference="opaque-review-001", authority=authority, evidence=_evidence(mod, actor=12, topic=22, message=100), ) assert cleared.state == mod.OnboardingState.OWNER_REVIEW clinical_path = service.root / "clinical-review.json" clinical = json.loads(clinical_path.read_text(encoding="utf-8")) assert validate_clinical_clearance( clinical_path, baseline_digest=str(clinical["baseline_digest"]), ) == clinical for field, value in ( ("cleared_for_nonmedical_coaching", False), ("baseline_digest", "f" * 64), ): tampered = dict(clinical) tampered[field] = value tampered["digest"] = canonical_digest(tampered) clinical_path.write_text(json.dumps(tampered), encoding="utf-8") with pytest.raises(ValueError, match="clinical clearance"): validate_clinical_clearance( clinical_path, baseline_digest=str(clinical["baseline_digest"]), ) def test_cancel_purges_answers_and_leaves_redacted_tombstone(tmp_path: Path) -> None: mod = _load() service = mod.NutritionOnboardingService( profile_root=tmp_path, customer_key="client_001", enforce_current_authority=False, ) authority = _authority(mod) service.start_or_resume(authority=authority, evidence=_evidence(mod)) service.submit_answer( field=mod.QUESTION_FIELDS[0], value="1996-08-01", authority=authority, evidence=_evidence(mod, message=2), ) cancelled = service.cancel( authority=authority, evidence=_evidence(mod, message=3), ) assert cancelled.state == mod.OnboardingState.CANCELLED assert cancelled.answer_count == 0 assert cancelled.next_field is None assert "1996-08-01" not in service.session_path.read_text(encoding="utf-8") def test_owner_finalization_writes_ready_artifacts_and_disabled_projection( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: mod = _load() profile_root = tmp_path / "profile" profile_root.mkdir(mode=0o700) source = tmp_path / "kb-template.json" source.write_text(json.dumps(load_restriction_kb_template()), encoding="utf-8") seed_restriction_kb( profile_root=profile_root, source=source, owner_digest="a" * 64, commit=True, as_of=date(2026, 8, 1), ) (profile_root / "data").chmod(0o700) registry_path = profile_root / "customers" / "registry.json" registry_path.parent.mkdir(parents=True) registry_path.write_text( json.dumps( { "customers": [ { "customer_key": "client_001", "display_name": "Client", "enabled": False, "routes": {"customer": "opaque", "owner": "opaque"}, "consent": {"privacy-v1": "granted"}, "nutrition_profile": {"status": "draft"}, "plan": {"weeks": []}, } ] } ), encoding="utf-8", ) authority = _authority(mod) service = mod.NutritionOnboardingService( profile_root=profile_root, customer_key="client_001", enforce_current_authority=False, enforce_reconciliation=True, ) service.start_or_resume(authority=authority, evidence=_evidence(mod)) answers = {} for index, field in enumerate(mod.QUESTION_FIELDS): value = mod.example_answer(field) if field == "weight_kg": value = "82.3" if field == "goal_type": value = "loss" if field == "target_weight_kg": value = "75" if field == "target_date": value = "2026-10-22" if field == "conditions": value = {"status": "provided", "items": ["외부 검토 필요"]} if field == "pregnancy_breastfeeding": value = True answers[field] = value service.submit_answer( field=field, value=value, authority=authority, evidence=_evidence(mod, message=index + 2), ) reconciliation = service.record_reconciliation( answers_digest=canonical_digest(answers), advisory={"summary_ko": "입력 내용을 확인했습니다."}, clarifications=[], authority=authority, ) reconciliation_digest = str(reconciliation["digest"]) held = service.attest_baseline( authority=authority, evidence=_evidence(mod, message=99), ) assert held.state == mod.OnboardingState.SAFETY_HOLD binding = mod.OwnerRiskAcceptanceBinding( candidate_epoch="c" * 64, bootstrap_session_id="session-1", customer_key="client_001", baseline_digest=str(held.baseline_digest), reconciliation_digest=reconciliation_digest, ) service.record_owner_risk_acceptance( request=mod.OwnerRiskAcceptanceRequest( binding=binding, acceptance_identity=mod.owner_risk_acceptance_identity( binding, ), authority=authority, evidence=_evidence( mod, actor=12, topic=22, message=100, ), ), ) service.review_as_owner( decision="approved", authority=authority, evidence=_evidence(mod, actor=12, topic=22, message=101), ) candidate_text = service.baseline_path.read_text(encoding="utf-8") tampered_candidate = json.loads(candidate_text) tampered_candidate["baseline"]["weight_kg"] = 999 service.baseline_path.write_text( json.dumps(tampered_candidate), encoding="utf-8", ) with pytest.raises(ValueError, match="baseline candidate"): service.finalize( starts_on=date(2026, 8, 21), issued_at_kst=datetime( 2026, 8, 1, 3, tzinfo=timezone(timedelta(hours=9)), ), privacy_consent_digest="b" * 64, feature_epoch_digest="c" * 64, ) service.baseline_path.write_text( candidate_text, encoding="utf-8", ) original_projection = finalization_module.apply_nutrition_onboarding_projection def crash_after_projection(*args: object, **kwargs: object) -> None: original_projection(*args, **kwargs) raise RuntimeError("simulated crash after registry projection") monkeypatch.setattr( finalization_module, "apply_nutrition_onboarding_projection", crash_after_projection, ) with pytest.raises(RuntimeError, match="simulated crash"): _ = service.finalize( starts_on=date(2026, 8, 21), issued_at_kst=datetime( 2026, 8, 1, 3, tzinfo=timezone(timedelta(hours=9)), ), privacy_consent_digest="b" * 64, feature_epoch_digest="c" * 64, ) monkeypatch.setattr( finalization_module, "apply_nutrition_onboarding_projection", original_projection, ) ready = service.finalize( starts_on=date(2026, 8, 21), issued_at_kst=datetime( 2026, 8, 1, 4, tzinfo=timezone(timedelta(hours=9)), ), privacy_consent_digest="b" * 64, feature_epoch_digest="c" * 64, ) assert ready.state == mod.OnboardingState.READY audit = audit_nutrition_start_readiness( profile_root, "client_001", now_kst=datetime(2026, 8, 1, 4, tzinfo=timezone(timedelta(hours=9))), ) assert audit.ready is True assert audit.checks["pointer"] is True baseline = json.loads( (service.root / "baseline-v1.json").read_text(encoding="utf-8") ) receipt = json.loads( (service.root / "readiness-receipt-v1.json").read_text(encoding="utf-8") ) calculation = json.loads( (service.root / "initial-plan-v1.json").read_text( encoding="utf-8", ) ) pointer = json.loads( (service.root / "readiness-current.json").read_text(encoding="utf-8") ) ready_document = json.loads(service.ready_path.read_text(encoding="utf-8")) scrubbed_candidate = json.loads( service.baseline_path.read_text(encoding="utf-8") ) assert "source_answers" not in scrubbed_candidate assert len(scrubbed_candidate["source_answers_digest"]) == 64 for document in (baseline, receipt, pointer, ready_document): assert document["input_reconciliation_digest"] == reconciliation_digest assert baseline["owner_risk_acceptance_receipt"][ "clinical_approval" ] is False assert baseline["eating_disorder_risk"] is False assert baseline["schema_version"] == "2.1" assert len(ready_document["owner_risk_acceptance_receipts"]) == 1 assert calculation["requested_target_weight_kg"] == 75.0 assert calculation["requested_target_date"] == "2026-10-22" assert calculation["requested_trajectory_within_guardrail"] is False assert calculation["recommended_target_date"] == "2026-10-23" baseline_path = service.root / "baseline-v1.json" baseline_text = baseline_path.read_text(encoding="utf-8") tampered_baseline = json.loads(baseline_text) tampered_acceptance = dict( tampered_baseline["owner_risk_acceptance_receipt"] ) tampered_acceptance["clinical_approval"] = True tampered_acceptance["digest"] = canonical_digest( { key: value for key, value in tampered_acceptance.items() if key != "digest" } ) tampered_baseline["owner_risk_acceptance_receipt"] = ( tampered_acceptance ) tampered_baseline["digest"] = canonical_digest( { key: value for key, value in tampered_baseline.items() if key != "digest" } ) baseline_path.write_text( json.dumps(tampered_baseline), encoding="utf-8", ) invalid_acceptance = audit_nutrition_start_readiness( profile_root, "client_001", now_kst=datetime(2026, 8, 1, 4, tzinfo=timezone(timedelta(hours=9))), ) assert invalid_acceptance.ready is False assert "baseline_invalid" in invalid_acceptance.reason_codes baseline_path.write_text(baseline_text, encoding="utf-8") baseline_path.chmod(0o600) pointer_path = service.root / "readiness-current.json" pointer_text = pointer_path.read_text(encoding="utf-8") pointer = json.loads(pointer_text) pointer_path.unlink() missing_pointer = audit_nutrition_start_readiness( profile_root, "client_001", now_kst=datetime(2026, 8, 1, 4, tzinfo=timezone(timedelta(hours=9))), ) assert missing_pointer.ready is False assert "readiness_receipt_invalid" in missing_pointer.reason_codes pointer_path.write_text(pointer_text, encoding="utf-8") pointer["bundle_digest"] = "f" * 64 pointer["digest"] = canonical_digest(pointer) pointer_path.write_text(json.dumps(pointer), encoding="utf-8") tampered = audit_nutrition_start_readiness( profile_root, "client_001", now_kst=datetime(2026, 8, 1, 4, tzinfo=timezone(timedelta(hours=9))), ) assert tampered.ready is False assert "readiness_receipt_invalid" in tampered.reason_codes registry = json.loads(registry_path.read_text(encoding="utf-8")) assert registry["customers"][0]["enabled"] is False assert len(registry["customers"][0]["plan"]["weeks"]) == 12 assert not service.session_path.exists()