from __future__ import annotations import hashlib import json import os import subprocess import sys from datetime import date, datetime, timedelta, timezone from pathlib import Path import pytest from checkin_cli import customer_admin from checkin_cli.customer_admin import ( CustomerAdminError, validate_committed_activation, ) from checkin_cli.nutrition_readiness import audit_nutrition_start_readiness from checkin_cli.nutrition_readiness_io import canonical_digest from tests.test_customer_admin import _activation_fixture, activate_customer def _write_document(path: Path, document: dict[str, object]) -> dict[str, object]: payload = dict(document) payload.pop("digest", None) encoded = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") payload["digest"] = hashlib.sha256(encoded).hexdigest() path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") os.chmod(path, 0o600) return payload def _write_complete_readiness(profile_root: Path, customer_key: str) -> None: kb = _write_document( profile_root / "data/global/nutrition-safety/restriction-kb-v1.json", { "schema_version": "1.0", "knowledge_version": "2026-08-01", "effective_at_kst": "2026-08-01T00:00:00+09:00", "reviewed_at_kst": "2026-08-01T00:00:00+09:00", "reviewed_by": "owner-receipt", "approved": True, "sources": [{"source_id": "source-1", "title": "approved source"}], "allergens": [{"rule_id": "allergen-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]}], "intolerances": [{"rule_id": "intolerance-1", "action": "require_human_review", "severity": "medium", "applicability": "all", "source_ids": ["source-1"]}], "religious_ethical_exclusions": [ {"rule_id": "ethical-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]} ], "medication_condition_rules": [ {"rule_id": "medical-1", "action": "require_human_review", "severity": "high", "applicability": "all", "source_ids": ["source-1"]} ], "hard_contraindications": [ {"rule_id": "contraindication-1", "action": "require_human_review", "severity": "critical", "applicability": "all", "source_ids": ["source-1"]} ], "cross_contact_rules": [{"rule_id": "contact-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]}], "substitution_rules": [{"rule_id": "substitution-1", "action": "inform", "severity": "low", "applicability": "all", "source_ids": ["source-1"]}], }, ) baseline = _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/baseline-v1.json", { "schema_version": "2.0", "customer_key": customer_key, "adult_age": 30, "equation_sex_basis": "female", "height_cm": 165.0, "weight_kg": 65.0, "activity_category": "moderate", "activity_rationale": "reviewed", "goal_type": "maintain", "target_weight_kg": 65.0, "target_date": "2026-10-01", "maintenance_intent": True, "dietary_preferences": [], "disliked_foods": [], "allergies": [], "intolerances": [], "religious_ethical_exclusions": [], "conditions": [], "medications": [], "pregnancy_breastfeeding": False, "eating_disorder_risk": False, "cooking_access": "home", "budget_band": "standard", "meal_count": 3, "schedule_constraints": [], "customer_attested_at_kst": "2026-08-01T01:00:00+09:00", "owner_review_receipt": "owner-review", "input_reconciliation_digest": "d" * 64, }, ) reconciliation = _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/restriction-reconciliation-v1.json", { "schema_version": "1.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "resolution_status": "resolved", "unresolved_hard_contraindications": [], "reviewed_by": "owner-review", "reviewed_at_kst": "2026-08-01T01:30:00+09:00", }, ) calculation = _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/initial-plan-v1.json", { "schema_version": "1.0", "method_id": "mifflin_st_jeor_1990", "method_version": "1.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "normalized_inputs_digest": "a" * 64, "bmr_kcal": 1400, "tdee_kcal": 2100, "minimum_calories_kcal": 1200, "maximum_calories_kcal": 3000, "activity_assumption": "moderate", "goal_trajectory": "maintain", "goal_trajectory_safe": True, "safe_rate_guardrail": "reviewed", "requested_target_weight_kg": 65.0, "requested_target_date": "2026-10-01", "requested_trajectory_within_guardrail": True, "recommended_target_date": None, "weeks": [ { "week": week, "calories_kcal": 2100, "protein_g": 120, "carbohydrate_g": 250, "fat_g": 70, } for week in range(1, 13) ], "reviewed_by": "owner-review", "reviewed_at_kst": "2026-08-01T02:00:00+09:00", "review_decision": "approved", }, ) policy = _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/adjustment-policy-v1.json", { "schema_version": "1.0", "policy_version": "1.0", "effective_at_kst": "2026-08-01T00:00:00+09:00", "effective_from": "2026-08-01", "goal_mode": "maintain", "approved": True, "minimum_observation_days": 7, "minimum_current_samples": 4, "minimum_total_samples": 10, "minimum_adherent_days": 5, "weight_trend_method": "two_non_overlapping_seven_day_means", "adherence_inputs": ["calories", "meal_plan"], "actual_intake_inputs": ["calories"], "activity_change_inputs": ["training"], "adjustment_thresholds": { "minimum_current_samples": 4, "minimum_total_samples": 10, "minimum_adherent_days": 5, "goal_mode": "maintain", }, "maximum_step_kcal": 100, "calorie_step": 100, "calorie_floor": 1500, "calorie_ceiling": 4500, "cooldown_days": 7, "desired_weekly_change_min": "-0.25", "desired_weekly_change_max": "0.25", "safety_hold": False, "escalation_rules": ["contradictory_adherence", "safety_hold"], "safety_stop_rules": ["medical_review", "eating_disorder_risk"], "manual_override_receipt_required": True, "reviewed_by": "owner-review", }, ) receipt = _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/readiness-receipt-v1.json", { "schema_version": "2.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "restriction_reconciliation_digest": reconciliation["digest"], "calculation_digest": calculation["digest"], "adjustment_policy_digest": policy["digest"], "input_reconciliation_digest": "d" * 64, "owner_review_receipt": "owner-review", "privacy_consent_version": "privacy-v1", "privacy_consent_digest": "b" * 64, "feature_epoch_digest": "c" * 64, "delivery_enabled": False, "activation_enabled": False, "issued_at_kst": "2026-08-01T03:00:00+09:00", "expires_at_kst": "2099-08-01T03:00:00+09:00", }, ) readiness_digests = { "restriction_kb": kb["digest"], "baseline": baseline["digest"], "restriction_reconciliation": reconciliation["digest"], "calculation": calculation["digest"], "adjustment_policy": policy["digest"], "receipt": receipt["digest"], "input_reconciliation": "d" * 64, } _write_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/readiness-current.json", { "schema_version": "nutrition_readiness_pointer_v1", "revision": 1, "bundle_digest": canonical_digest(readiness_digests), "readiness_receipt_digest": receipt["digest"], "input_reconciliation_digest": "d" * 64, }, ) def test_existing_audit_contract_exposes_readiness_reason_codes() -> None: assert "baseline_missing" in customer_admin._GATE_PREFLIGHT_REASON_CODES assert "authority_receipt_stale" in customer_admin._GATE_PREFLIGHT_REASON_CODES def test_activation_fails_closed_without_nutrition_readiness_evidence(tmp_path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture( tmp_path, nutrition_readiness=False, ) registry_before = registry_path.read_bytes() with pytest.raises(CustomerAdminError, match="nutrition readiness"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) assert registry_path.read_bytes() == registry_before assert not (profile_root / "data" / "customer-activation-journal.json").exists() assert not (profile_root / "data" / "customer-activation-audit.jsonl").exists() def test_complete_readiness_allows_canonical_activation(tmp_path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") result = activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) assert result.customer.enabled is True assert b'"enabled": true' in registry_path.read_bytes() def test_readiness_projection_is_redacted(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") audit = audit_nutrition_start_readiness(profile_root, "client_001") encoded = json.dumps(audit.to_dict(), ensure_ascii=False) assert audit.ready is True assert "female" not in encoded assert "165" not in encoded assert "65.0" not in encoded assert "owner-review" not in encoded def test_readiness_cli_blocks_without_personal_values(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path, nutrition_readiness=False) completed = subprocess.run( [ sys.executable, "-m", "checkin_cli.readiness_cli", "--profile-root", str(profile_root), "--customer-key", "client_001", ], check=False, capture_output=True, text=True, ) payload = json.loads(completed.stdout) assert completed.returncode == 2 assert payload["ready"] is False assert "baseline_missing" in payload["reason_codes"] assert "client_001" not in completed.stdout def test_readiness_cli_passes_complete_evidence(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") completed = subprocess.run( [ sys.executable, "-m", "checkin_cli.readiness_cli", "--profile-root", str(profile_root), "--customer-key", "client_001", ], check=False, capture_output=True, text=True, ) payload = json.loads(completed.stdout) assert completed.returncode == 0 assert payload["ready"] is True assert payload["reason_codes"] == [] assert "female" not in completed.stdout assert "165" not in completed.stdout assert "65.0" not in completed.stdout def test_missing_restriction_reconciliation_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") ( profile_root / "data/customers/client_001/nutrition-onboarding/restriction-reconciliation-v1.json" ).unlink() audit = audit_nutrition_start_readiness(profile_root, "client_001") assert audit.ready is False assert "customer_restrictions_unresolved" in audit.reason_codes def test_unsafe_goal_trajectory_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") root = profile_root / "data/customers/client_001/nutrition-onboarding" calculation_path = root / "initial-plan-v1.json" calculation = json.loads(calculation_path.read_text(encoding="utf-8")) calculation["goal_trajectory_safe"] = False calculation = _write_document(calculation_path, calculation) receipt_path = root / "readiness-receipt-v1.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) receipt["calculation_digest"] = calculation["digest"] _write_document(receipt_path, receipt) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert audit.ready is False assert "goal_trajectory_unsafe" in audit.reason_codes def test_empty_global_restriction_coverage_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") kb_path = profile_root / "data/global/nutrition-safety/restriction-kb-v1.json" kb = json.loads(kb_path.read_text(encoding="utf-8")) for field in ( "allergens", "intolerances", "religious_ethical_exclusions", "medication_condition_rules", "hard_contraindications", "cross_contact_rules", "substitution_rules", ): kb[field] = [] _write_document(kb_path, kb) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert audit.ready is False assert "restriction_sources_incomplete" in audit.reason_codes def test_uncited_restriction_rule_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") kb_path = profile_root / "data/global/nutrition-safety/restriction-kb-v1.json" kb = json.loads(kb_path.read_text(encoding="utf-8")) kb["allergens"][0].pop("source_ids", None) _write_document(kb_path, kb) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert "restriction_sources_incomplete" in audit.reason_codes def test_implausible_baseline_measurement_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") baseline_path = ( profile_root / "data/customers/client_001/nutrition-onboarding/baseline-v1.json" ) baseline = json.loads(baseline_path.read_text(encoding="utf-8")) baseline["height_cm"] = 1 _write_document(baseline_path, baseline) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert "measurement_implausible" in audit.reason_codes def test_week_without_calorie_macro_targets_blocks_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") calculation_path = ( profile_root / "data/customers/client_001/nutrition-onboarding/initial-plan-v1.json" ) calculation = json.loads(calculation_path.read_text(encoding="utf-8")) calculation["weeks"][0].pop("calories_kcal") _write_document(calculation_path, calculation) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert "weekly_targets_incomplete" in audit.reason_codes def test_invalid_adjustment_bounds_block_readiness(tmp_path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") policy_path = ( profile_root / "data/customers/client_001/nutrition-onboarding/adjustment-policy-v1.json" ) policy = json.loads(policy_path.read_text(encoding="utf-8")) policy["maximum_step_kcal"] = 0 _write_document(policy_path, policy) audit = audit_nutrition_start_readiness(profile_root, "client_001") assert "adjustment_policy_unapproved" in audit.reason_codes def test_committed_activation_rejects_stale_readiness(tmp_path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) baseline_path = ( profile_root / "data/customers/client_001/nutrition-onboarding/baseline-v1.json" ) baseline = json.loads(baseline_path.read_text(encoding="utf-8")) baseline["weight_kg"] = 66.0 _write_document(baseline_path, baseline) with pytest.raises(CustomerAdminError, match="nutrition readiness"): validate_committed_activation(profile_root, registry_path, "client_001") def test_readiness_rejects_hardlinked_receipt(tmp_path: Path) -> None: profile_root, _, _, _ = _activation_fixture(tmp_path) _write_complete_readiness(profile_root, "client_001") receipt_path = ( profile_root / "data/customers/client_001/nutrition-onboarding/readiness-receipt-v1.json" ) alias = tmp_path / "receipt-alias.json" os.link(receipt_path, alias) 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 False assert "readiness_receipt_missing" in audit.reason_codes