import importlib import json import subprocess from pathlib import Path import pytest from test_privatemuse_f84_adoption import ( _install_fixture, _post_adoption_evidence, _run_adoption, adoption, diagnosis, ) def _setup(monkeypatch, tmp_path): module_path = Path(adoption.__file__).with_name("finalize_privatemuse_adoption_health.py") assert module_path.exists(), "The approved baseline needs a verified health finalization command" finalizer = importlib.import_module("finalize_privatemuse_adoption_health") state_path, archive, *_rest, controller = _install_fixture(monkeypatch, tmp_path) state = json.loads(state_path.read_text()) state["live_status"] = "unknown" state_path.write_bytes(adoption._json_bytes(state)) _run_adoption() state = json.loads(state_path.read_text()) state.pop("f84_runtime_adoption") state["message"] = finalizer.FAILURE state["last_attempt"] = { "revision": finalizer.REVISION, "status": "failed", "message": finalizer.FAILURE, "metadata": {"revision": finalizer.REVISION}, } state["updated_at_epoch"] += 1 state_path.write_bytes(adoption._json_bytes(state)) monkeypatch.setattr(finalizer, "JOURNAL", tmp_path / "health-finalization") monkeypatch.setattr(finalizer, "_main_revision", lambda *_: finalizer.REVISION) monkeypatch.setattr(diagnosis, "collect_live_evidence_under_lock", lambda **_: _post_adoption_evidence()) return finalizer, state_path, archive, controller def _run(module): return module.finalize(approval_reference=adoption._APPROVAL_REFERENCE, apply=True) def test_finalization_restores_usable_health_without_rewriting_failed_attempt(monkeypatch, tmp_path): module, path, archive, _ = _setup(monkeypatch, tmp_path) before = path.read_bytes() old_archive = {p.name: p.read_bytes() for p in archive.iterdir()} result = _run(module) after = json.loads(path.read_bytes()) expected = json.loads(before) expected["live_status"] = "healthy" assert after == expected assert result["status"] == "finalized" assert (module.JOURNAL / "before.json").read_bytes() == before assert {p.name: p.read_bytes() for p in archive.iterdir()} == old_archive assert _run(module)["status"] == "already_finalized" @pytest.mark.parametrize("change", ["main", "attempt", "receipt", "configuration", "archive"]) def test_unproven_baseline_never_changes_state(monkeypatch, tmp_path, change): module, path, archive, controller = _setup(monkeypatch, tmp_path) if change == "main": monkeypatch.setattr(module, "_main_revision", lambda *_: "f" * 40) elif change == "attempt": state = json.loads(path.read_bytes()) state["last_attempt"]["message"] = "A different failure" path.write_bytes(adoption._json_bytes(state)) elif change == "receipt": controller.current_receipt["containers"]["privatemuse-backend-1"]["container_id"] = "f" * 64 elif change == "configuration": evidence = _post_adoption_evidence() evidence["backend_runtime_mount_network_port_and_env_match_current_compose"] = False monkeypatch.setattr(diagnosis, "collect_live_evidence_under_lock", lambda **_: evidence) else: (archive / "approval-audit.json").write_text("{}") before = path.read_bytes() with pytest.raises(adoption.AdoptionError): _run(module) assert path.read_bytes() == before def test_failure_after_write_restores_exact_state(monkeypatch, tmp_path): module, path, _, _ = _setup(monkeypatch, tmp_path) before = path.read_bytes() real_verify = module._verify_runtime calls = 0 def fail_post(*args): nonlocal calls calls += 1 if calls == 3: raise adoption.AdoptionError("runtime changed after write") return real_verify(*args) monkeypatch.setattr(module, "_verify_runtime", fail_post) with pytest.raises(adoption.AdoptionError, match="runtime changed"): _run(module) assert path.read_bytes() == before def test_concurrent_state_is_preserved(monkeypatch, tmp_path): module, path, _, _ = _setup(monkeypatch, tmp_path) original_verify = module._verify_runtime calls = 0 outsider = b'{"other":"operator state"}\n' def change_post(*args): nonlocal calls calls += 1 if calls == 3: path.write_bytes(outsider) raise adoption.AdoptionError("concurrent update") return original_verify(*args) monkeypatch.setattr(module, "_verify_runtime", change_post) with pytest.raises(adoption.AdoptionError, match="diverged"): _run(module) assert path.read_bytes() == outsider def test_journal_tampering_and_missing_approval_fail_closed(monkeypatch, tmp_path): module, path, _, _ = _setup(monkeypatch, tmp_path) before = path.read_bytes() with pytest.raises(adoption.AdoptionError): module.finalize(approval_reference="other", apply=True) assert path.read_bytes() == before _run(module) healthy = path.read_bytes() (module.JOURNAL / "after.json").write_text("{}") with pytest.raises(adoption.AdoptionError): _run(module) assert path.read_bytes() == healthy def test_main_change_after_write_restores_unknown_state(monkeypatch, tmp_path): module, path, _, _ = _setup(monkeypatch, tmp_path) before = path.read_bytes() calls = 0 def moving_main(*_): nonlocal calls calls += 1 return module.REVISION if calls < 5 else "f" * 40 monkeypatch.setattr(module, "_main_revision", moving_main) with pytest.raises(adoption.AdoptionError, match="main changed"): _run(module) assert path.read_bytes() == before def test_canonical_bare_repository_needs_no_origin(monkeypatch, tmp_path): import finalize_privatemuse_adoption_health as module repo = tmp_path / "canonical.git" subprocess.run(["git", "init", "--bare", str(repo)], check=True, capture_output=True) tree = subprocess.check_output(["git", f"--git-dir={repo}", "mktree"], input=b"").decode().strip() revision = subprocess.check_output([ "git", f"--git-dir={repo}", "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit-tree", tree, "-m", "baseline", ]).decode().strip() subprocess.run(["git", f"--git-dir={repo}", "update-ref", "refs/heads/main", revision], check=True) monkeypatch.setattr(module, "REVISION", revision) class Controller: def repository_path(self, *_): return repo assert module._main_revision(Controller(), {}) == revision subprocess.run(["git", f"--git-dir={repo}", "update-ref", "-d", "refs/heads/main"], check=True) with pytest.raises(adoption.AdoptionError): module._main_revision(Controller(), {}) def test_finalized_baseline_passes_real_rollback_capture_gate(monkeypatch, tmp_path): from test_deploy_broker import BROKER_ROOT, _load_module deploy = _load_module("privatemuse_health_consumer_test", BROKER_ROOT / "deploy.py") module, path, _, fake = _setup(monkeypatch, tmp_path) receipt = fake.current_receipt names = list(receipt["containers"]) project = {"slug": "privatemuse", "compose_project": "privatemuse", "containers": names, "_repository_name": "jens/privatemuse"} monkeypatch.setattr(deploy, "read_state", lambda _: json.loads(path.read_bytes())) monkeypatch.setattr(deploy, "compose_project_container_names", lambda _: set(names)) monkeypatch.setattr(deploy, "inspect_container", lambda name: { "Id": receipt["containers"][name]["container_id"], "Image": receipt["containers"][name]["image_id"], "Config": {"Image": receipt["containers"][name]["image_ref"], "Labels": {"com.docker.compose.project": "privatemuse"}}, "State": {"Running": True}, }) monkeypatch.setattr(deploy, "inspect_image", lambda _: {"present": True}) monkeypatch.setattr(deploy, "verify_deployment", lambda *_, **__: receipt) monkeypatch.setattr(deploy, "run", lambda _: None) monkeypatch.setattr(deploy, "cleanup_rollback_image_tags", lambda *_: None) with pytest.raises(deploy.DeployError, match="valid healthy live deployment state"): deploy.capture_compose_rollback(project) _run(module) snapshot = deploy.capture_compose_rollback(project) assert snapshot["revision"] == diagnosis._LIVE_REVISION assert snapshot["receipt"] == receipt assert snapshot["compose_containers"] == sorted(names) assert json.loads(path.read_bytes())["status"] == "failed"