207 lines
8.5 KiB
Python
207 lines
8.5 KiB
Python
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"
|