"""V14 pre-reservation, import, and service-lifecycle safety regressions.""" from __future__ import annotations import hashlib import json import os from pathlib import Path from typing import final from unittest.mock import patch import pytest from pydantic import JsonValue, TypeAdapter import scripts.nutricoach_v150_sealed_controller as sealed_controller from scripts.nutricoach_v150_concrete_host import ConcreteLiveHost from scripts.nutricoach_v150_detached_bootstrap import ( BootstrapDenied, verify_closure, verified_pythonpath, ) from scripts.nutricoach_v150_host_operations import ( apply_migrations, capacity_after, service_state, switch_postimages, ) from scripts.nutricoach_v150_sealed_controller import ( APPROVAL_PHRASE, DisposableService, SealedControllerError, recover_pending, execute_disposable, ) from scripts.nutricoach_v150_phase_journal import PhaseJournal from scripts.nutricoach_v150_runtime_ops import ( dependency_snapshot_digest, install_dependency_snapshot, ) from scripts.nutricoach_v150_sealed_authority import GlobalLedger from scripts.nutricoach_v150_sealed_target import HostPaths, SystemdService from checkin_cli.multi_customer_admission_migration_models import ( AdmissionMigrationProposal, ) from tests.test_nutricoach_v150_sealed_controller import target_fixture _OBJECT = TypeAdapter(dict[str, JsonValue]) @final class ObservedService: def __init__(self, *, sticky_stop: bool = False) -> None: self.running = True self.sticky_stop = sticky_stop def stop(self) -> None: if not self.sticky_stop: self.running = False def reload(self) -> None: return None def start(self) -> None: self.running = True def observe(self) -> dict[str, str]: return { "ActiveState": "active" if self.running else "inactive", "SubState": "running" if self.running else "dead", "MainPID": "401" if self.running else "0", "ExecStart": "/observed/runtime/bin/python -m gateway.run", "ExecMainStartTimestampMonotonic": "700", } def test_detached_bootstrap_verifies_wheels_before_source(tmp_path: Path) -> None: wheels: list[dict[str, str]] = [] for name in ("hermes.whl", "profile.whl"): path = tmp_path / name _ = path.write_bytes(name.encode()) path.chmod(0o444) wheels.append({ "path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), }) target = tmp_path / "sealed-target.json" _ = target.write_text(json.dumps({"wheels": wheels}) + "\n") source = tmp_path / "source" pythonpath = verified_pythonpath(target, source) assert pythonpath.split(os.pathsep) == [ str(tmp_path / "hermes.whl"), str(tmp_path / "profile.whl"), str(source), ] hermes = tmp_path / "hermes.whl" hermes.chmod(0o644) _ = hermes.write_bytes(b"tampered") hermes.chmod(0o444) with pytest.raises(BootstrapDenied, match="wheel_integrity"): _ = verified_pythonpath(target, source) def test_detached_bootstrap_rejects_owner_writable_wheel(tmp_path: Path) -> None: wheel = tmp_path / "profile.whl" _ = wheel.write_bytes(b"wheel") wheel.chmod(0o644) target = tmp_path / "sealed-target.json" record = { "path": str(wheel), "sha256": hashlib.sha256(wheel.read_bytes()).hexdigest(), } _ = target.write_text(json.dumps({"wheels": [record, record]}) + "\n") with pytest.raises(BootstrapDenied, match="wheel_integrity"): _ = verified_pythonpath(target, tmp_path / "source") def test_closure_rejects_unmanifested_executable_file(tmp_path: Path) -> None: source = tmp_path / "source" source.mkdir() declared = source / "controller.py" _ = declared.write_text("pass\n") manifest = tmp_path / "closure.json" _ = manifest.write_text( json.dumps({ "files": { "controller.py": hashlib.sha256(declared.read_bytes()).hexdigest() } }) + "\n" ) _ = (source / "injected.pyc").write_bytes(b"bytecode") with pytest.raises(BootstrapDenied, match="closure_inventory"): _ = verify_closure(manifest, source) def test_expired_weekly_authority_is_denied_before_ledger_reservation( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) with ( patch.object( sealed_controller, "WEEKLY_AUTHORITY_EXPIRES_AT", "2020-01-01T00:00:00+00:00", ), pytest.raises(SealedControllerError, match="weekly_authority_window"), ): _ = execute_disposable(APPROVAL_PHRASE, tmp_path, host) assert not host.ledger_reserved() assert not host.ledger_consumed() def test_service_state_preserves_manager_observed_exec_start( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) state = service_state(ObservedService(), host.paths) assert state["ExecStart"] == "/observed/runtime/bin/python -m gateway.run" def test_host_stop_denies_service_that_remains_running(tmp_path: Path) -> None: _ = target_fixture(tmp_path) disposable = ConcreteLiveHost.disposable(tmp_path, DisposableService()) host = ConcreteLiveHost( disposable.paths, ObservedService(sticky_stop=True), live=False, ) with pytest.raises(RuntimeError, match="service_stop"): host.stop() def test_capacity_postimage_uses_explicit_sealed_candidate(tmp_path: Path) -> None: candidate = "b" * 64 registry = tmp_path / "registry.json" _ = registry.write_text( json.dumps({ "version": 1, "owner": {"user_id": "1", "chat_id": "1", "topic_id": "0"}, "customers": [], }) + "\n" ) postimage = _OBJECT.validate_json(capacity_after(registry, candidate)) policy = postimage["admission_policy"] assert isinstance(policy, dict) assert policy["candidate_digest"] == candidate def test_capacity_apply_uses_receipt_preserving_customer_migration( tmp_path: Path, ) -> None: candidate = "b" * 64 _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) after = capacity_after(host.paths.registry, candidate) calls: list[tuple[Path, AdmissionMigrationProposal, str]] = [] def commit_migration( profile_root: Path, proposal: AdmissionMigrationProposal, approval_phrase: str, ) -> str: calls.append((profile_root, proposal, approval_phrase)) return hashlib.sha256(after).hexdigest() with patch( "checkin_cli.customer_admin.commit_multi_customer_admission_migration", side_effect=commit_migration, ): apply_migrations( host.paths, after, candidate, receipt_preserving=True, ) assert len(calls) == 1 profile_root, proposal, approval_phrase = calls[0] assert profile_root == host.paths.profile assert proposal.candidate_digest == candidate assert proposal.max_enabled_customers == 5 assert approval_phrase == proposal.approval_phrase def test_switch_rebinds_task26_candidate_credential_name(tmp_path: Path) -> None: candidate = "b" * 64 _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) current_authority = host.paths.current_runtime.parent / "runtime-authority" current_authority.mkdir(parents=True) old_credential = current_authority / "task26-candidate-digest" _ = old_credential.write_text("old-candidate\n", encoding="utf-8") _ = host.paths.dropin.write_text( f"LoadCredential=task26-candidate-digest:{old_credential}\n", encoding="utf-8", ) host.successor_root.mkdir(parents=True) _ = switch_postimages(host.paths, candidate) rebound = ( host.paths.successor_runtime.parent / "runtime-authority/task26-candidate-digest" ) assert rebound.read_text(encoding="utf-8") == f"{candidate}\n" def test_dependency_snapshot_installs_exact_runtime_dependencies( tmp_path: Path, ) -> None: snapshot = tmp_path / "snapshot" package = snapshot / "croniter" dist_info = snapshot / "croniter-6.0.0.dist-info" package.mkdir(parents=True) dist_info.mkdir() _ = (package / "__init__.py").write_text("VERSION = '6.0.0'\n") _ = (dist_info / "METADATA").write_text("Name: croniter\nVersion: 6.0.0\n") telegram = snapshot / "telegram" telegram_dist_info = snapshot / "python_telegram_bot-22.6.dist-info" telegram.mkdir() telegram_dist_info.mkdir() _ = (telegram / "__init__.py").write_text("__version__ = '22.6'\n") _ = (telegram_dist_info / "METADATA").write_text( "Name: python-telegram-bot\nVersion: 22.6\n" ) site = tmp_path / "site-packages" site.mkdir() install_dependency_snapshot(site, snapshot) assert (site / "croniter/__init__.py").read_bytes() == ( package / "__init__.py" ).read_bytes() assert (site / "croniter-6.0.0.dist-info/METADATA").read_bytes() == ( dist_info / "METADATA" ).read_bytes() assert (site / "telegram/__init__.py").read_bytes() == ( telegram / "__init__.py" ).read_bytes() assert (site / "python_telegram_bot-22.6.dist-info/METADATA").read_bytes() == ( telegram_dist_info / "METADATA" ).read_bytes() def test_live_host_loads_candidate_and_wheels_from_sealed_target( tmp_path: Path, ) -> None: candidate = "b" * 64 hermes = tmp_path / "hermes.whl" profile = tmp_path / "profile.whl" _ = hermes.write_bytes(b"hermes") _ = profile.write_bytes(b"profile") dependency = tmp_path / "dependency" (dependency / "croniter").mkdir(parents=True) (dependency / "croniter-6.0.0.dist-info").mkdir() (dependency / "telegram").mkdir() (dependency / "python_telegram_bot-22.6.dist-info").mkdir() _ = (dependency / "croniter/__init__.py").write_text("VERSION = '6.0.0'\n") _ = (dependency / "croniter-6.0.0.dist-info/METADATA").write_text( "Name: croniter\nVersion: 6.0.0\n" ) _ = (dependency / "telegram/__init__.py").write_text("__version__ = '22.6'\n") _ = (dependency / "python_telegram_bot-22.6.dist-info/METADATA").write_text( "Name: python-telegram-bot\nVersion: 22.6\n" ) target = tmp_path / "sealed-target.json" _ = target.write_text( json.dumps({ "candidate_digest": candidate, "profile_root": str(tmp_path / "live-profile"), "service_name": "sealed.service", "registry_sha256": "c" * 64, "unit": str(tmp_path / "gateway.service"), "dropin": str(tmp_path / "authority.conf"), "current_runtime": str(tmp_path / "current/venv"), "successor_runtime": str(tmp_path / "successor/venv"), "execution_root": str(tmp_path / "execution"), "global_approval_ledger": str(tmp_path / "ledger"), "protected_inventory": str(tmp_path / "inventory.json"), "protected_inventory_sha256": "d" * 64, "dependency_snapshot": str(dependency), "dependency_snapshot_sha256": dependency_snapshot_digest(dependency), "wheels": [ { "path": str(hermes), "sha256": hashlib.sha256(hermes.read_bytes()).hexdigest(), }, { "path": str(profile), "sha256": hashlib.sha256(profile.read_bytes()).hexdigest(), }, ], }) + "\n" ) host = ConcreteLiveHost.live_target(target) assert host.candidate_digest == candidate assert host.hermes_wheel == hermes assert host.profile_wheel == profile assert host.dependency_snapshot == dependency assert host.paths.registry == tmp_path / "live-profile/customers/registry.json" def test_missing_canonical_events_file_is_rollback_owned(tmp_path: Path) -> None: _ = target_fixture(tmp_path) events = next((tmp_path / "profile/data/customers").glob("*/wizard/events.jsonl")) events.unlink() host = ConcreteLiveHost.disposable( tmp_path, DisposableService(), fault=ValueError("post-fence"), fault_stage="post_fence", ) with pytest.raises(ValueError, match="post-fence"): _ = execute_disposable(APPROVAL_PHRASE, tmp_path, host) assert not events.exists() def test_capacity_receipt_created_by_migration_is_rollback_owned( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) registry = _OBJECT.validate_json( (tmp_path / "profile/customers/registry.json").read_bytes() ) customers = registry["customers"] assert isinstance(customers, list) enabled = next( row for row in customers if isinstance(row, dict) and row.get("enabled") is True ) customer_key = enabled["customer_key"] assert isinstance(customer_key, str) receipt = ( tmp_path / "profile/data/customer-activation-receipts" / f"{customer_key}.json" ) host = ConcreteLiveHost.disposable( tmp_path, DisposableService(), receipt_preserving_migration=True, ) host.capture_preflight() host.write_recovery_manifest() receipt.parent.mkdir(mode=0o700) _ = receipt.write_text("{}\n", encoding="utf-8") receipt.chmod(0o600) host.remove_created() assert not receipt.exists() assert not receipt.parent.exists() def test_recovery_restarts_predecessor_before_snapshot_exists(tmp_path: Path) -> None: _ = target_fixture(tmp_path) service = DisposableService() host = ConcreteLiveHost.disposable(tmp_path, service) host.capture_preflight() host.write_recovery_manifest() ledger = GlobalLedger(host.paths.ledger_root, host.candidate_digest) ledger.reserve() journal = PhaseJournal(host.paths.execution_root / "phase.json") journal.advance("STOPPING") host.stop() recovered = ConcreteLiveHost.disposable(tmp_path, service) with pytest.raises(SealedControllerError, match="rollback_recovery_completed"): recover_pending(recovered) assert recovered.service.running assert journal.phase() == "ROLLED_BACK" assert ledger.outcome() == "FAILED" def test_recovery_rolls_back_preparing_without_snapshot(tmp_path: Path) -> None: _ = target_fixture(tmp_path) service = DisposableService() host = ConcreteLiveHost.disposable(tmp_path, service) host.capture_preflight() host.write_recovery_manifest() ledger = GlobalLedger(host.paths.ledger_root, host.candidate_digest) ledger.reserve() journal = PhaseJournal(host.paths.execution_root / "phase.json") journal.advance("PREPARING") recovered = ConcreteLiveHost.disposable(tmp_path, service) with pytest.raises(SealedControllerError, match="rollback_recovery_completed"): recover_pending(recovered) assert recovered.service.running assert journal.phase() == "ROLLED_BACK" assert ledger.outcome() == "FAILED" def test_rejected_replay_preserves_rolled_back_phase(tmp_path: Path) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable( tmp_path, DisposableService(), fault=ValueError("stop"), fault_stage="stop", ) journal = PhaseJournal(host.paths.execution_root / "phase.json") with pytest.raises(ValueError, match="stop"): _ = execute_disposable(APPROVAL_PHRASE, tmp_path, host) assert journal.phase() == "ROLLED_BACK" with pytest.raises(SealedControllerError, match="already_used"): _ = execute_disposable(APPROVAL_PHRASE, tmp_path, host) assert journal.phase() == "ROLLED_BACK" def test_recovery_consumes_reservation_without_phase_or_manifest( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) ledger = GlobalLedger(host.paths.ledger_root, host.candidate_digest) ledger.reserve() journal = PhaseJournal(host.paths.execution_root / "phase.json") with pytest.raises(SealedControllerError, match="rollback_recovery_completed"): recover_pending(host) assert journal.phase() == "ROLLED_BACK" assert ledger.outcome() == "FAILED" def test_recovery_finalizes_committing_success_without_rollback( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost.disposable(tmp_path, DisposableService()) ledger = GlobalLedger(host.paths.ledger_root, host.candidate_digest) ledger.reserve() ledger.consume("SUCCEEDED") journal = PhaseJournal(host.paths.execution_root / "phase.json") journal.advance("COMMITTING") recover_pending(host) assert host.service.running assert journal.phase() == "COMMITTED" assert ledger.outcome() == "SUCCEEDED" def test_systemd_service_uses_only_bound_service_identity() -> None: service = SystemdService("sealed.service") assert service.name == "sealed.service" def test_registry_drift_is_denied_before_ledger_reservation(tmp_path: Path) -> None: _ = target_fixture(tmp_path) host = ConcreteLiveHost( ConcreteLiveHost.disposable(tmp_path, DisposableService()).paths, DisposableService(), live=False, registry_sha256="0" * 64, ) with pytest.raises(RuntimeError, match="registry_drift"): _ = execute_disposable(APPROVAL_PHRASE, tmp_path, host) assert not host.ledger_reserved() assert not host.ledger_consumed() def test_protected_inventory_drift_is_denied_before_preflight( tmp_path: Path, ) -> None: _ = target_fixture(tmp_path) paths = ConcreteLiveHost.disposable(tmp_path, DisposableService()).paths inventory = tmp_path / "inventory.json" _ = inventory.write_text('{"profiles":{"stable":[]}}\n') host = ConcreteLiveHost( HostPaths( paths.profile, paths.registry, paths.config, paths.unit, paths.dropin, paths.current_runtime, paths.successor_runtime, paths.execution_root, paths.ledger_root, inventory, ), DisposableService(), live=False, protected_inventory_sha256="0" * 64, ) with pytest.raises(RuntimeError, match="protected_inventory_drift"): host.capture_preflight()