"""Explicit r71b maintenance preparation contracts.""" from __future__ import annotations import hashlib from datetime import date, datetime from pathlib import Path import pytest from pydantic import JsonValue, TypeAdapter from gateway.platforms.nutrition_weekly_operations_identifiers import ( TOPIC59_PROJECTION_SCHEMA, ) from gateway.platforms.nutrition_weekly_operations_ledger_history import ( Topic59LedgerEntry, ) from gateway.platforms.nutrition_weekly_operations_publication_contract import ( Topic59LedgerState, ) from scripts.nutricoach_v150_candidate_cli import CliNamespace from scripts.nutricoach_v150_r71b_task10_evidence import task10_evidence_spec from scripts.nutricoach_v150_r71b_preparation import ( HistoricalSendingRow, MaintenancePreparationError, MaintenanceWindow, NoSendFacts, SanitizedTopic59Preflight, collect_authoritative_r71b_preflight, prepare_r71b_maintenance, ) from scripts.seal_nutricoach_v150_candidate import copy_r71b_evidence _OBJECT = TypeAdapter(dict[str, JsonValue]) def _window() -> MaintenanceWindow: return MaintenanceWindow( kst_day=date(2026, 9, 1), not_before=datetime.fromisoformat("2026-09-01T16:00:00+09:00"), expires_at=datetime.fromisoformat("2026-09-01T16:10:00+09:00"), ) def _preflight() -> SanitizedTopic59Preflight: return SanitizedTopic59Preflight( candidate_digest="1" * 64, successor_config_digest="2" * 64, route_digest="3" * 64, customer_identity_digest="4" * 64, card_slot="5" * 64, publication_ledger_sha256="6" * 64, historical_sending_rows=( HistoricalSendingRow(card_slot="7" * 64, entry_digest="8" * 64), HistoricalSendingRow(card_slot="9" * 64, entry_digest="a" * 64), ), ) def test_preparer_derives_exact_scope_oracle_hold_and_two_digests() -> None: # Given facts = NoSendFacts( non_topic59_due_send_count=0, non_topic59_due_edit_count=0, activation_notice_due_count=0, owner_delivery_due_count=0, target_topic59_projection_count=1, ) seed: dict[str, JsonValue] = _OBJECT.validate_python({ "schema": "nutricoach-v150-r71b-package-binding-inputs-v1", "product_generation": "r71", "package_namespace": "r71b-maintenance", "authority_id": "nutricoach-v150-v15-runtime-authority-r71b-maintenance", "paths": {"observer_root": "/tmp/observer-r71"}, "candidate": {"candidate_digest": "1" * 64}, "wheels": [], "controller": {"files": {}}, "protected_inventory": {"explicit_maintenance_volatile_paths": []}, "dependency_snapshot": {"dependency_snapshot_sha256": "b" * 64}, "authority_baseline": {"candidate_digest": "c" * 64}, "target_binding_base": {"profile_root": "/tmp/profile"}, "service_preparation": {"ActiveState": "active"}, "diagnosed_blocker_evidence": [], }) # When prepared = prepare_r71b_maintenance(_preflight(), _window(), facts, seed) # Then assert prepared.scope.config_digest == "2" * 64 assert prepared.scope.route_digest == "3" * 64 assert prepared.oracle["target_topic59_projection_count"] == 1 assert prepared.oracle["non_topic59_due_send_count"] == 0 assert prepared.artifacts.authority.hold == prepared.artifacts.hold assert prepared.artifacts.package_binding_digest != prepared.artifacts.final_package_digest @pytest.mark.parametrize( "facts", ( NoSendFacts(1, 0, 0, 0, 1), NoSendFacts(0, 0, 0, 0, 0), ), ) def test_preparer_rejects_incomplete_no_send_oracle(facts: NoSendFacts) -> None: # Given / When / Then with pytest.raises(MaintenancePreparationError, match="no_send_oracle"): _ = prepare_r71b_maintenance(_preflight(), _window(), facts, {}) def test_explicit_kst_window_rejects_non_kst_or_misaligned_inputs() -> None: # Given from scripts.nutricoach_v150_r71b_preparation import parse_maintenance_window # Then assert parse_maintenance_window( "2026-09-01", "2026-09-01T16:00:00+09:00", "2026-09-01T16:10:00+09:00", ).kst_day == date(2026, 9, 1) with pytest.raises(MaintenancePreparationError, match="maintenance_window"): _ = parse_maintenance_window( "2026-09-01", "2026-09-01T16:00:00+00:00", "2026-09-01T16:10:00+00:00", ) with pytest.raises(MaintenancePreparationError, match="maintenance_window"): _ = parse_maintenance_window( "2026-09-01", "2026-09-02T00:00:00+09:00", "2026-09-02T00:10:00+09:00", ) def test_authoritative_collector_derives_profile_projection_and_full_oracle( tmp_path: Path, ) -> None: # Given candidate = tmp_path / "candidate" profile = tmp_path / "profile" candidate.mkdir() digest = "1" * 64 _ = (candidate / "manifest.json").write_text( '{"candidate_digest":"' + digest + '"}\n', encoding="utf-8" ) config = profile / "config.yaml" config.parent.mkdir() customer = "4" * 64 route = hashlib.sha256( "\0".join((TOPIC59_PROJECTION_SCHEMA, "route", "100", "200", "59")).encode() ).hexdigest() registry_digest = "a" * 64 _ = config.write_text("\n".join(( "platforms:", " telegram:", " extra:", " nutrition_coaching:", " operator_review:", " user_id: '100'", " chat_id: '200'", " 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", f" registry_identity_binding_digest: '{registry_digest}'", "", )), encoding="utf-8") data = profile / "data" data.mkdir() slot = "5" * 64 def row(sequence: int, day: date, state: str, entry: str, card: str) -> str: return Topic59LedgerEntry.model_validate({ "schema": "nutricoach-topic59-publication-ledger-v2", "sequence": sequence, "predecessor_digest": "0" * 64, "entry_digest": entry, "card_slot": card, "logical_key": "6" * 64, "customer_identity_digest": customer, "kst_day": day.isoformat(), "config_digest": "0" * 64, "candidate_digest": digest, "route_digest": route, "status_digest": "7" * 64, "payload_digest": "8" * 64, "operation": "send", "state": Topic59LedgerState(state), "payload": { "customer_identity_digest": customer, "kst_day": day.isoformat(), "state": "submitted", "completed_field_count": 0, "draft_state": "미작성", "approval_state": "요청 전", "delivery_state": "미발송", "event_id": "b" * 64, "status_row_id": "c" * 64, }, }).model_dump_json(by_alias=True) from gateway.platforms.nutrition_weekly_operations_config import parse_weekly_operations_config config_digest = parse_weekly_operations_config({ "nutrition_coaching": { "operator_review": {"user_id": "100", "chat_id": "200", "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_digest, }, }, }).digest target = row(3, date(2026, 9, 1), "prepared", "9" * 64, slot) target = target.replace( '"config_digest":"' + ("0" * 64) + '"', '"config_digest":"' + config_digest + '"', ) _ = (data / "weekly-operations-topic59.jsonl").write_text( "\n".join(( row(1, date(2026, 8, 30), "sending", "b" * 64, "c" * 64), row(2, date(2026, 8, 31), "sending", "d" * 64, "e" * 64), target, "", )), encoding="utf-8") _ = (data / "scheduled-deliveries.jsonl").write_bytes(b"") cron = profile / "cron" cron.mkdir() _ = (cron / "jobs.json").write_text( '{"jobs":[{"id":"6e042d5dff68","name":"NutriCoach schedule dispatcher",' + '"schedule_display":"* * * * *",' + '"last_error":"weekly operations failed: canonical authority identity drift",' + '"enabled":true}]}\n', encoding="utf-8", ) arguments = CliNamespace() arguments.r70_error_evidence = task10_evidence_spec("r70_error").source_path arguments.observer_evidence = task10_evidence_spec("observer").source_path arguments.fixed_collector_evidence = task10_evidence_spec( "health_recovery" ).source_path copies = copy_r71b_evidence(arguments, candidate) for path in copies: path.chmod(0o444) # When collected = collect_authoritative_r71b_preflight(candidate, profile, _window()) # Then assert collected.preflight.candidate_digest == digest assert collected.preflight.successor_config_digest == config_digest assert collected.preflight.route_digest == route assert collected.facts == NoSendFacts(0, 0, 0, 0, 1) assert len(collected.preflight.historical_sending_rows) == 2 assert collected.evidence["r70_error_evidence_sha256"] == task10_evidence_spec( "r70_error" ).source_sha256