"""Typed profile-YAML boundary regressions for r71b preparation.""" from __future__ import annotations import json from datetime import date, datetime from pathlib import Path import pytest from pydantic import JsonValue, TypeAdapter from scripts.nutricoach_v150_candidate_cli import CliNamespace from scripts.nutricoach_v150_r71b_maintenance_transaction import ( KNOWN_R70_CRON_ERROR, KNOWN_R70_CRON_JOB_ID, KNOWN_R70_CRON_JOB_NAME, KNOWN_R70_CRON_SCHEDULE, ) from scripts.nutricoach_v150_r71b_task10_evidence import task10_evidence_spec from scripts.nutricoach_v150_r71b_preparation import ( MAX_PROFILE_YAML_BYTES, MaintenancePreparationError, MaintenanceWindow, NoSendFacts, collect_authoritative_r71b_preflight, parse_profile_yaml, 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 test_profile_yaml_accepts_machine_consumed_sequence_entries() -> None: # Given payload = b"""toolsets: - hermes-cli - gateway platforms: telegram: extra: nutrition_coaching: weekly_operations_capacity: 5 """ # When document = parse_profile_yaml(payload) # Then assert document["toolsets"] == ["hermes-cli", "gateway"] platforms = document.get("platforms") assert isinstance(platforms, dict) telegram = platforms.get("telegram") assert isinstance(telegram, dict) extra = telegram.get("extra") assert isinstance(extra, dict) nutrition = extra.get("nutrition_coaching") assert isinstance(nutrition, dict) assert nutrition["weekly_operations_capacity"] == 5 @pytest.mark.parametrize( "payload", ( b"- not-a-profile-mapping\n", b"1: non-string-key\n", b"broken: [unterminated\n", b"unsafe: !!python/object/apply:os.system ['nope']\n", b"recursive: &self [*self]\n", ), ) def test_profile_yaml_rejects_noncanonical_documents(payload: bytes) -> None: # Given / When / Then with pytest.raises(MaintenancePreparationError, match="profile_config"): _ = parse_profile_yaml(payload) def test_profile_yaml_rejects_duplicate_mapping_keys() -> None: # Given payload = b"duplicate: first\nduplicate: second\n" # When / Then with pytest.raises(MaintenancePreparationError, match="profile_config"): _ = parse_profile_yaml(payload) @pytest.mark.parametrize( "payload", ( b"defaults: &defaults\n enabled: true\n", b"defaults: &defaults\n enabled: true\ncopy: *defaults\n", ), ) def test_profile_yaml_rejects_yaml_anchors_and_aliases(payload: bytes) -> None: # Given / When / Then with pytest.raises(MaintenancePreparationError, match="profile_config"): _ = parse_profile_yaml(payload) def test_profile_yaml_rejects_documents_over_maximum_bytes() -> None: # Given prefix = b"padding: " payload = prefix + b"x" * (MAX_PROFILE_YAML_BYTES + 1 - len(prefix)) # When / Then with pytest.raises(MaintenancePreparationError, match="profile_config"): _ = parse_profile_yaml(payload) def test_current_profile_yaml_parses_weekly_capacity_and_route_read_only( tmp_path: Path, ) -> None: # Given live_config = Path("/home/cube/.hermes/profiles/dualcoachtest/config.yaml") if not live_config.is_file(): pytest.skip("current profile is unavailable") config_path = tmp_path / "config.yaml" _ = config_path.write_bytes(live_config.read_bytes()) config_path.chmod(0o444) # When document = parse_profile_yaml(config_path.read_bytes()) # Then platforms = document.get("platforms") assert isinstance(platforms, dict) telegram = platforms.get("telegram") assert isinstance(telegram, dict) extra = telegram.get("extra") assert isinstance(extra, dict) nutrition = extra.get("nutrition_coaching") assert isinstance(nutrition, dict) assert isinstance(nutrition.get("weekly_operations_capacity"), int) weekly = nutrition.get("weekly_operations") assert isinstance(weekly, dict) assert weekly.get("enabled") is True route = nutrition.get("operator_review") assert isinstance(route, dict) assert route.get("topic_id") in {"59", 59} def test_current_profile_preparer_derives_projection_and_oracle_disposably( tmp_path: Path, ) -> None: # Given live_config = Path("/home/cube/.hermes/profiles/dualcoachtest/config.yaml") live_ledger = live_config.parent / "data/weekly-operations-topic59.jsonl" if not live_config.is_file() or not live_ledger.is_file(): pytest.skip("current profile is unavailable") profile = tmp_path / "profile" config_path = profile / "config.yaml" config_path.parent.mkdir(parents=True) _ = config_path.write_bytes(live_config.read_bytes()) config_path.chmod(0o444) candidate = tmp_path / "candidate" candidate.mkdir() candidate_digest = "1" * 64 _ = (candidate / "manifest.json").write_text( json.dumps({"candidate_digest": candidate_digest}) + "\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 for path in copy_r71b_evidence(arguments, candidate): path.chmod(0o444) rows = [ _OBJECT.validate_json(line) for line in live_ledger.read_bytes().splitlines() if line ] target = next( row for row in rows if row.get("state") == "prepared" and row.get("kst_day") == "2026-08-31" ) target["candidate_digest"] = candidate_digest target["kst_day"] = _window().kst_day.isoformat() target_payload = target.get("payload") assert isinstance(target_payload, dict) target_payload["kst_day"] = _window().kst_day.isoformat() data = profile / "data" data.mkdir() _ = (data / "weekly-operations-topic59.jsonl").write_text( "\n".join(json.dumps(row, separators=(",", ":")) for row in rows) + "\n", encoding="utf-8", ) _ = (data / "scheduled-deliveries.jsonl").write_bytes(b"") cron = profile / "cron" cron.mkdir() _ = (cron / "jobs.json").write_text( json.dumps({"jobs": [{ "id": KNOWN_R70_CRON_JOB_ID, "name": KNOWN_R70_CRON_JOB_NAME, "schedule_display": KNOWN_R70_CRON_SCHEDULE, "last_error": KNOWN_R70_CRON_ERROR, "enabled": True, }]}) + "\n", encoding="utf-8", ) window = _window() 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": candidate_digest}, "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 collected = collect_authoritative_r71b_preflight(candidate, profile, window) prepared = prepare_r71b_maintenance( collected.preflight, window, collected.facts, seed ) # Then assert collected.preflight.successor_config_digest == target["config_digest"] assert collected.facts == NoSendFacts(0, 0, 0, 0, 1) assert prepared.oracle["target_topic59_projection_count"] == 1 assert len(prepared.artifacts.final_package_digest) == 64 assert not (tmp_path / "package").exists()