"""Characterize nutrition-daily's known and explicit-missing value contract.""" from __future__ import annotations from datetime import date from pathlib import Path import pytest from pydantic import ValidationError from checkin_cli.customer_reporting import build_customer_period_report from checkin_cli.models import CurrentView, Event from checkin_cli.wizard import WizardService from checkin_cli.wizard_models import ( WizardBranch, WizardContext, WizardResult, WizardSession, WizardStatus, ) RESERVED_UNKNOWN_LITERALS = ("unknown", "모름") FREE_TEXT_NUTRITION_STEPS = ( "meals", "digestion", "appetite_stress", "training_summary", "optional_note", ) UNKNOWN_CAPABLE_STEPS = ( "bodyweight", "calories", "macros", "meals", "water", "sleep_duration", "sleep_quality", "digestion", "condition", "appetite_stress", "training_summary", ) def _session(home: Path, session_id: str) -> WizardSession: return WizardSession.model_validate_json( (home / "drafts" / f"{session_id}.json").read_text(encoding="utf-8") ) def _assert_invalid_session(payload: object, message: str) -> None: try: _ = WizardSession.model_validate(payload) except ValidationError as error: assert message in str(error) else: raise AssertionError(f"expected invalid session containing {message!r}") def _mark_all_nutrition_values_unknown( service: WizardService, context: WizardContext, result: WizardResult, ) -> WizardResult: for step in UNKNOWN_CAPABLE_STEPS: assert result.step == step result = service.answer(context, result.session_id, result.version, "unknown") assert result.status is WizardStatus.ADVANCED assert result.step == "optional_note" return result def test_complete_known_nutrition_flow_remains_canonical(tmp_path: Path) -> None: """A complete known flow remains the pre-feature canonical event shape.""" service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") result = service.start_nutrition(context, "2026-08-31") for action, value in ( ("value", "70.2"), ("value", "2300"), ("value", "280 150 65"), ("value", "three planned meals"), ("value", "2.5"), ("value", "7.5"), ("select", "4"), ("select", "normal"), ("select", "4"), ("value", "appetite steady"), ("value", "synthetic training"), ("select", "skip"), ): result = service.answer(context, result.session_id, result.version, action, value) assert result.status is WizardStatus.ADVANCED saved = service.answer(context, result.session_id, result.version, "save") event = Event.model_validate_json((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert saved.status is WizardStatus.SAVED assert saved.branch is WizardBranch.NORMAL assert event.event_type.value == "nutrition_checkin" assert event.check_in is not None assert event.check_in.calories_kcal == 2300 assert event.check_in.protein_g == 150 assert event.check_in.carbohydrate_g == 280 assert event.check_in.fat_g == 65 def test_nutrition_session_v2_models_known_unknown_and_unanswered_exclusively( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") started = service.start_nutrition(context, "2026-08-31") initial = _session(tmp_path, started.session_id) assert initial.state_schema_version == 2 assert initial.unknown_steps == () assert "bodyweight" not in initial.answers marked_unknown = service.answer( context, started.session_id, started.version, "unknown", ) assert marked_unknown.status is WizardStatus.ADVANCED persisted_unknown = _session(tmp_path, started.session_id) assert "bodyweight" not in persisted_unknown.answers assert persisted_unknown.unknown_steps == ("bodyweight",) rewound = service.answer( context, started.session_id, marked_unknown.version, "previous", ) assert rewound.status is WizardStatus.ADVANCED assert rewound.step == "bodyweight" known = service.answer( context, started.session_id, rewound.version, "value", "70.2", ) persisted_known = _session(tmp_path, started.session_id) assert known.status is WizardStatus.ADVANCED assert persisted_known.answers["bodyweight"] == "70.2" assert "bodyweight" not in persisted_known.unknown_steps assert "calories" not in persisted_known.answers assert "calories" not in persisted_known.unknown_steps contradictory = persisted_known.model_dump() contradictory["unknown_steps"] = ("bodyweight",) _assert_invalid_session(contradictory, "known and unknown") def test_unknown_action_is_supported_for_exactly_all_eleven_nutrition_fields( tmp_path: Path, ) -> None: context = WizardContext("synthetic-owner", "synthetic-topic") assert len(UNKNOWN_CAPABLE_STEPS) == 11 for target_index, target_step in enumerate(UNKNOWN_CAPABLE_STEPS): home = tmp_path / target_step service = WizardService.for_standalone(home) result = service.start_nutrition(context, "2026-08-31") for _ in range(target_index): result = service.answer(context, result.session_id, result.version, "unknown") assert result.status is WizardStatus.ADVANCED assert result.step == target_step result = service.answer(context, result.session_id, result.version, "unknown") session = _session(home, result.session_id) assert result.status is WizardStatus.ADVANCED assert target_step in session.unknown_steps assert target_step not in session.answers service = WizardService.for_standalone(tmp_path / "optional-note") result = service.start_nutrition(context, "2026-08-31") result = _mark_all_nutrition_values_unknown(service, context, result) invalid = service.answer(context, result.session_id, result.version, "unknown") assert invalid.status is WizardStatus.INVALID assert invalid.step == "optional_note" def test_unknown_macros_project_to_three_missing_canonical_fields(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") result = service.start_nutrition(context, "2026-08-31") result = _mark_all_nutrition_values_unknown(service, context, result) result = service.answer(context, result.session_id, result.version, "skip") saved = service.answer(context, result.session_id, result.version, "save") event = service.finalized_event(saved.session_id) assert saved.status is WizardStatus.SAVED assert event is not None and event.check_in is not None assert event.check_in.protein_g is None assert event.check_in.carbohydrate_g is None assert event.check_in.fat_g is None def test_unknown_survives_defer_resume_previous_and_correction_replacement( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") started = service.start_nutrition(context, "2026-08-31") first_unknown = service.answer(context, started.session_id, started.version, "unknown") deferred = service.answer( context, started.session_id, first_unknown.version, "defer", ) resumed = WizardService.for_standalone(tmp_path).start_nutrition(context, "2026-08-31") assert deferred.status is WizardStatus.DEFERRED assert resumed.step == "calories" resumed_session = _session(tmp_path, resumed.session_id) assert resumed_session.unknown_steps == ("bodyweight",) previous = service.answer(context, resumed.session_id, resumed.version, "previous") assert previous.step == "bodyweight" after_previous = _session(tmp_path, resumed.session_id) assert after_previous.unknown_steps == ("bodyweight",) result = _mark_all_nutrition_values_unknown(service, context, previous) result = service.answer(context, result.session_id, result.version, "skip") saved = service.answer(context, result.session_id, result.version, "save") correction = service.start_nutrition_correction(context, "2026-08-31") correction_session = _session(tmp_path, correction.session_id) assert saved.status is WizardStatus.SAVED assert correction.status is WizardStatus.ADVANCED assert correction_session.unknown_steps == UNKNOWN_CAPABLE_STEPS assert correction_session.state_schema_version == 2 editing = service.answer( context, correction.session_id, correction.version, "edit", "bodyweight", ) replaced = service.answer( context, correction.session_id, editing.version, "value", "70.2", ) replaced_session = _session(tmp_path, correction.session_id) assert replaced.status is WizardStatus.ADVANCED assert replaced_session.answers["bodyweight"] == "70.2" assert "bodyweight" not in replaced_session.unknown_steps def test_all_unknown_nutrition_is_accepted_without_fabricated_downstream_values( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") result = service.start_nutrition(context, "2026-08-31") result = _mark_all_nutrition_values_unknown(service, context, result) summary = service.answer(context, result.session_id, result.version, "skip") saved = service.answer(context, summary.session_id, summary.version, "save") event = service.finalized_event(saved.session_id) persisted = _session(tmp_path, saved.session_id) serialized_text = (tmp_path / "events.jsonl").read_text(encoding="utf-8") serialized = Event.model_validate_json(serialized_text) current = CurrentView.model_validate_json( (tmp_path / "views" / "current.json").read_text(encoding="utf-8") ) assert saved.status is WizardStatus.SAVED assert saved.branch is None assert event is not None and event.check_in is not None assert persisted.branch is None event_day = date.fromisoformat(event.occurred_at_kst[:10]) report = build_customer_period_report( tmp_path / "events.jsonl", event_day, event_day, ) assert event.check_in.body_weight_kg is None assert event.check_in.calories_kcal is None assert event.check_in.protein_g is None assert event.check_in.carbohydrate_g is None assert event.check_in.fat_g is None assert event.check_in.sleep_hours is None assert event.check_in.sleep_quality_1to5 is None assert event.check_in.readiness_1to5 is None assert event.check_in.training_summary is None assert event.check_in.digestion_summary is None assert event.check_in.meal_summary is None assert event.check_in.water_liters is None assert event.check_in.appetite_stress_summary is None assert event.check_in.notes is None assert persisted.unknown_steps == UNKNOWN_CAPABLE_STEPS assert serialized.status.value == "accepted" assert serialized.check_in is not None assert serialized.check_in.model_dump(exclude_none=True) == {} assert current.eligible_checkins == () assert report.average_calories_kcal is None assert report.average_protein_g is None assert report.calorie_target_adherence_percent is None assert report.protein_target_adherence_percent is None assert report.missing_calories_count == 1 assert '"unknown"' not in serialized_text assert "모름" not in serialized_text replay = service.answer(context, saved.session_id, saved.version, "save") assert replay.status is WizardStatus.REJECTED assert len(serialized_text.splitlines()) == 1 def test_legacy_nutrition_session_defaults_to_v1_and_upgrades_on_transition( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") started = service.start_nutrition(context, "2026-08-31") bodyweight = service.answer( context, started.session_id, started.version, "value", "70.2", ) draft_path = tmp_path / "drafts" / f"{started.session_id}.json" legacy = _session(tmp_path, started.session_id) _ = draft_path.write_text( legacy.model_dump_json(exclude={"state_schema_version", "unknown_steps"}), encoding="utf-8", ) resumed = WizardService.for_standalone(tmp_path).start_nutrition(context, "2026-08-31") legacy_session = _session(tmp_path, resumed.session_id) upgraded = service.answer( context, resumed.session_id, resumed.version, "unknown", ) upgraded_session = _session(tmp_path, resumed.session_id) assert bodyweight.step == "calories" assert legacy_session.state_schema_version == 1 assert legacy_session.unknown_steps == () assert upgraded.status is WizardStatus.ADVANCED assert upgraded_session.state_schema_version == 2 assert upgraded_session.answers["bodyweight"] == "70.2" assert upgraded_session.unknown_steps == ("calories",) def test_unknown_action_rejects_stale_version_and_malformed_state(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") started = service.start_nutrition(context, "2026-08-31") advanced = service.answer(context, started.session_id, started.version, "unknown") stale = service.answer(context, started.session_id, started.version, "unknown") session = _session(tmp_path, started.session_id) assert advanced.status is WizardStatus.ADVANCED assert stale.status is WizardStatus.REJECTED malformed = session.model_dump() malformed["unknown_steps"] = ("optional_note",) _assert_invalid_session(malformed, "unknown-capable") @pytest.mark.parametrize("literal", RESERVED_UNKNOWN_LITERALS) @pytest.mark.parametrize("step", FREE_TEXT_NUTRITION_STEPS) def test_reserved_unknown_literals_are_rejected_at_every_nutrition_text_boundary( tmp_path: Path, step: str, literal: str, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") result = service.start_nutrition(context, "2026-08-31") for expected_step in UNKNOWN_CAPABLE_STEPS: if expected_step == step: break result = service.answer(context, result.session_id, result.version, "unknown") assert result.status is WizardStatus.ADVANCED assert result.step == step rejected = service.answer( context, result.session_id, result.version, "value", literal, ) persisted = _session(tmp_path, result.session_id) assert rejected.status is WizardStatus.INVALID assert rejected.step == step assert rejected.version == result.version assert step not in persisted.answers assert step not in persisted.unknown_steps def test_v2_nutrition_cursor_rejects_future_markers_and_unrepresented_history( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) context = WizardContext("synthetic-owner", "synthetic-topic") started = service.start_nutrition(context, "2026-08-31") initial = _session(tmp_path, started.session_id) future_marker = initial.model_dump() future_marker["unknown_steps"] = ("calories",) _assert_invalid_session(future_marker, "future or unvisited") unrepresented_history = initial.model_dump() unrepresented_history["step_history"] = ("bodyweight",) _assert_invalid_session(unrepresented_history, "represented by known or unknown")