"""Characterization tests for the default-off weekly-operations boundary.""" from __future__ import annotations from dataclasses import replace import sys from datetime import datetime from pathlib import Path from subprocess import run from typing import TypeAlias import pytest from tests.gateway._weekly_operations_authority_support import ( test_registry_identity_binding_digest as registry_identity_binding_digest, test_registry_identity_document as registry_identity_document, ) from gateway.platforms.nutrition_weekly_operations_authority import ( WeeklyOperationsRuntimeContext, parse_weekly_operations_authority, weekly_operations_is_authorized, ) from gateway.platforms.nutrition_weekly_operations_config import ( JsonValue, WeeklyOperationsAuthorityError, WeeklyOperationsConfig, WeeklyOperationsConfigError, parse_weekly_operations_config, ) PROFILE_PACKAGE = Path(__file__).resolve().parents[2] / "dualcoach" / "profile" WeeklyField: TypeAlias = bool | int | str NutritionCoachingExtra: TypeAlias = dict[str, dict[str, dict[str, WeeklyField]]] def test_baseline_schedule_is_unchanged_when_weekly_operations_is_absent() -> None: """Given no v1.4 config, preserve existing daily and weekly scheduling exactly.""" # Given script = ( "import sys; from datetime import time; " f"sys.path.insert(0, {str(PROFILE_PACKAGE)!r}); " "from checkin_cli.customer_coaching import CustomerSchedule; " "schedule = CustomerSchedule(daily_time=time(8), weekly_weekday=0, monthly_day=1); " "print(f'{schedule.daily_time}|{schedule.weekly_weekday}')" ) extra = { "nutrition_coaching": { "enabled": True, "registry_path": "customers/registry.json", } } # When completed = run([sys.executable, "-B", "-c", script], check=True, capture_output=True, text=True) from gateway.platforms.nutrition_coaching_config import NutritionCoachingConfig config = NutritionCoachingConfig.from_extra(extra) # Then assert completed.stdout == "08:00:00|0\n" assert config is not None assert config.registry_path == Path("customers/registry.json") assert not hasattr(config, "weekly_operations") assert not hasattr(config, "weekly_operations_authority") def _enabled_extra() -> NutritionCoachingExtra: return { "nutrition_coaching": { "operator_review": {"user_id": "owner", "chat_id": "review", "topic_id": 59}, "weekly_operations": { "enabled": True, "reminder_time": "20:00:00", "missed_cutoff_time": "23:00:00", "weekly_weekday": 0, "feature_epoch": "weekly-operations-v1", "registry_identity_binding_digest": registry_identity_binding_digest(), }, } } def _authority(config: WeeklyOperationsConfig) -> dict[str, JsonValue]: return { "schema": "nutricoach-weekly-operations-authority-v2", "candidate_digest": "a" * 64, "config_digest": config.digest, "enabled_customer_keys": ["client-001"], "owner": {"user_id": "owner", "chat_id": "owner-dm", "version": 7}, "consent_digest": "b" * 64, "registry_identity": registry_identity_document(), "issued_at": "2026-08-24T00:00:00+09:00", "expires_at": "2026-08-24T01:00:00+09:00", "feature_epoch": config.feature_epoch, } def _context() -> WeeklyOperationsRuntimeContext: return WeeklyOperationsRuntimeContext( candidate_digest="a" * 64, customer_key="client-001", owner_user_id="owner", owner_chat_id="owner-dm", owner_version=7, consent_digest="b" * 64, consent_granted=True, feature_epoch="weekly-operations-v1", now=datetime.fromisoformat("2026-08-24T00:30:00+09:00"), ) def test_config_defaults_off_with_stable_kst_schedule_when_absent() -> None: # Given extra = {"nutrition_coaching": {"enabled": True, "registry_path": "customers/registry.json"}} # When first = parse_weekly_operations_config(extra) second = parse_weekly_operations_config(extra) # Then assert (first.enabled, first.reminder_time, first.missed_cutoff_time, first.weekly_weekday) == (False, "20:00:00", "23:00:00", 0) assert first.digest == second.digest assert first.review_route is None def test_config_binds_only_the_existing_topic_59_review_route() -> None: # Given extra = _enabled_extra() # When config = parse_weekly_operations_config(extra) # Then assert config.enabled is True assert config.review_route is not None assert config.review_route.key == ("owner", "review", "59") @pytest.mark.parametrize( ("field", "value"), (("reminder_time", None), ("reminder_time", "20:00"), ("reminder_time", "20:00+00:00"), ("missed_cutoff_time", "2026-08-24T23:00:00+09:00")), ) def test_config_rejects_missing_naive_and_absolute_recurring_times(field: str, value: str | None) -> None: # Given extra = _enabled_extra() weekly = extra["nutrition_coaching"]["weekly_operations"] if value is None: del weekly[field] else: weekly[field] = value # When / Then with pytest.raises(WeeklyOperationsConfigError): _ = parse_weekly_operations_config(extra) def test_config_rejects_reminder_at_or_after_cutoff_and_wrong_route() -> None: # Given reversed_times = _enabled_extra() reversed_times["nutrition_coaching"]["weekly_operations"].update( reminder_time="23:00:00", missed_cutoff_time="20:00:00" ) wrong_route = _enabled_extra() wrong_route["nutrition_coaching"]["operator_review"]["topic_id"] = 58 # When / Then with pytest.raises(WeeklyOperationsConfigError): _ = parse_weekly_operations_config(reversed_times) with pytest.raises(WeeklyOperationsConfigError): _ = parse_weekly_operations_config(wrong_route) def test_config_authority_requires_compiled_configured_authorized_and_consented() -> None: # Given enabled = parse_weekly_operations_config(_enabled_extra()) receipt = parse_weekly_operations_authority(_authority(enabled)) context = _context() # When authorized = weekly_operations_is_authorized(enabled, receipt, context) disabled = weekly_operations_is_authorized(parse_weekly_operations_config({}), receipt, context) revoked_consent = weekly_operations_is_authorized(enabled, receipt, replace(context, consent_granted=False)) # Then assert enabled.is_compiled is True assert authorized is True assert disabled is False assert revoked_consent is False @pytest.mark.parametrize( ("field", "value"), (("candidate_digest", "c" * 64), ("config_digest", "d" * 64), ("enabled_customer_keys", ["client-002"]), ("owner", {"user_id": "other", "chat_id": "owner-dm", "version": 7}), ("owner", {"user_id": "owner", "chat_id": "owner-dm", "version": 8}), ("consent_digest", "e" * 64), ("feature_epoch", "weekly-operations-v2")), ) def test_config_authority_fails_closed_for_every_binding(field: str, value: JsonValue) -> None: # Given config = parse_weekly_operations_config(_enabled_extra()) raw = _authority(config) raw[field] = value # When authorized = weekly_operations_is_authorized(config, parse_weekly_operations_authority(raw), _context()) # Then assert authorized is False @pytest.mark.parametrize( ("issued_at", "expires_at"), (("2026-08-24T00:45:00+09:00", "2026-08-24T01:45:00+09:00"), ("2026-08-23T23:00:00+09:00", "2026-08-24T00:00:00+09:00")), ) def test_config_authority_rejects_stale_and_expired_receipts(issued_at: str, expires_at: str) -> None: # Given config = parse_weekly_operations_config(_enabled_extra()) raw = _authority(config) raw.update(issued_at=issued_at, expires_at=expires_at) # When authorized = weekly_operations_is_authorized(config, parse_weekly_operations_authority(raw), _context()) # Then assert authorized is False def test_config_authority_rejects_missing_or_naive_issue_expiry() -> None: # Given config = parse_weekly_operations_config(_enabled_extra()) missing = _authority(config) _ = missing.pop("issued_at") naive = _authority(config) naive["expires_at"] = "2026-08-24T01:00:00" # When / Then with pytest.raises(WeeklyOperationsAuthorityError): _ = parse_weekly_operations_authority(missing) with pytest.raises(WeeklyOperationsAuthorityError): _ = parse_weekly_operations_authority(naive) def test_enabled_authority_requires_sealed_registry_identity() -> None: config = parse_weekly_operations_config(_enabled_extra()) missing = _authority(config) _ = missing.pop("registry_identity") drifted = _authority(config) drifted["registry_identity"] = { **registry_identity_document(), "inode": 2 } with pytest.raises(WeeklyOperationsAuthorityError): _ = parse_weekly_operations_authority(missing) with pytest.raises(WeeklyOperationsAuthorityError): _ = parse_weekly_operations_authority(drifted) def test_authority_integrity_digest_binds_registry_identity_stably() -> None: config = parse_weekly_operations_config(_enabled_extra()) first = parse_weekly_operations_authority(_authority(config)) second = parse_weekly_operations_authority(_authority(config)) changed_raw = _authority(config) changed_raw["registry_identity"] = registry_identity_document(inode=2) changed = parse_weekly_operations_authority(changed_raw) assert first.integrity_digest == second.integrity_digest assert first.registry_identity == second.registry_identity assert changed.integrity_digest != first.integrity_digest def test_enabled_config_requires_registry_identity_binding_digest() -> None: missing = _enabled_extra() _ = missing["nutrition_coaching"]["weekly_operations"].pop( "registry_identity_binding_digest" ) changed = _enabled_extra() changed["nutrition_coaching"]["weekly_operations"][ "registry_identity_binding_digest" ] = registry_identity_binding_digest(inode=2) with pytest.raises(WeeklyOperationsConfigError): _ = parse_weekly_operations_config(missing) assert ( parse_weekly_operations_config(changed).digest != parse_weekly_operations_config(_enabled_extra()).digest )