"""No Docker/network calls: exercise crash phases and fail-closed recovery gates.""" import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from storage import clone_cold, digest, dotenv, tree_manifest from transaction import (Production, STEPS, candidate_project, recover_transaction, run_transaction, validate_runtime_safety) class Crash(BaseException): pass class FakeOperations: def __init__(self, fail=None, crash=False): self.journal = {"phase": "snapshotted"} self.fail, self.crash = fail, crash self.original_running, self.renamed, self.candidate = True, False, False self.state, self.config = "original", "original" self.clones = False self.foreign = False self.saved = [] def save(self, phase): self.saved.append(phase) self.journal["phase"] = phase def event(self, name): if self.fail == name: raise Crash() if self.crash else ValueError("fixture failure") def stopping(self): self.original_running = False self.event("after_stop") def copying(self): assert not self.original_running self.clones = True self.event("after_copy") def renaming(self): self.renamed = True self.event("after_rename") def deploying(self): self.candidate = True self.event("after_deploy") def verifying(self): self.event("during_verify") def committing(self): self.state = "candidate" self.event("after_receipt") self.config = "candidate" self.event("after_config") def recover(self): if self.foreign: raise ValueError("foreign canonical container") self.candidate = False self.state = self.config = "original" self.renamed = False self.original_running = True def verify_committed(self): assert self.state == self.config == "candidate" def original(self, healthy=False): assert self.original_running def http_health(self): pass def activate(self): assert self.journal["phase"] == "committed" self.event("during_activation") self.journal["activated"] = True class TransactionTests(unittest.TestCase): def test_runtime_safety_parity_rejects_each_security_or_device_drift(self): host = {"Devices": [{"PathOnHost": "/dev/example", "PathInContainer": "/dev/example", "CgroupPermissions": "rwm"}], "CapAdd": ["CAP_CHOWN"], "CapDrop": ["ALL"], "ReadonlyRootfs": True, "SecurityOpt": ["no-new-privileges:true"], "Tmpfs": {"/tmp": "size=128m"}} service = {"devices": [{"source": "/dev/example", "target": "/dev/example", "permissions": "rwm"}], "cap_add": ["CHOWN"], "cap_drop": ["ALL"], "read_only": True, "security_opt": ["no-new-privileges:true"], "tmpfs": ["/tmp:size=128m"]} validate_runtime_safety(host, service) for key, value in (("devices", []), ("cap_add", []), ("cap_drop", []), ("read_only", False), ("privileged", True), ("security_opt", []), ("tmpfs", [])): with self.subTest(key=key), self.assertRaises(ValueError): validate_runtime_safety(host, {**service, key: value}) def test_activation_failure_never_rolls_back_and_recovery_resumes(self): ops = FakeOperations(fail="during_activation") with self.assertRaises(ValueError): run_transaction(ops) self.assertEqual(ops.journal["phase"], "committed") self.assertTrue(ops.candidate) self.assertFalse(ops.original_running) ops.fail = None self.assertEqual(recover_transaction(ops), "committed") self.assertTrue(ops.journal["activated"]) def test_real_marker_release_requires_matching_nonce_and_resumes_after_unlink(self): with tempfile.TemporaryDirectory() as temporary: ops = Production.__new__(Production) ops.workspace = Path(temporary).resolve() root = ops.workspace / "local" root.mkdir() ops.slug = "vacatureradar" ops.plan = {"candidateMounts": [{"destination": "/app/local", "source": str(root)}]} ops.journal = {"phase": "committed", "nonce": "correct", "candidateContainerId": "candidate"} ops.verify_committed = lambda: None ops.assert_candidate = lambda value: None ops.save = lambda phase: None marker = ops.hold_path() marker.write_bytes(b"wrong") with patch("transaction.private_read", side_effect=lambda path: path.read_bytes()): with self.assertRaises(ValueError): ops.activate() self.assertTrue(marker.exists()) marker.write_bytes(ops.hold_bytes()) with patch("transaction.os.open", return_value=123), patch("transaction.os.fsync"), patch("transaction.os.close"): ops.activate() self.assertFalse(marker.exists()) self.assertTrue(ops.journal["activationIntent"]) ops.activate() self.assertTrue(ops.journal["activated"]) ops.journal.pop("activationIntent") with self.assertRaises(ValueError): ops.activate() def test_intent_precedes_each_successful_phase(self): ops = FakeOperations() run_transaction(ops) self.assertEqual(ops.saved, [*STEPS, "committed"]) self.assertFalse(ops.original_running) self.assertTrue(ops.renamed and ops.clones and ops.candidate) self.assertEqual(recover_transaction(ops), "committed") self.assertTrue(ops.candidate, "Recovery cannot discard writes after a committed migration") def test_each_failure_restores_original_and_keeps_clones(self): for failure in ("after_stop", "after_copy", "after_rename", "after_deploy", "during_verify", "after_receipt", "after_config"): with self.subTest(failure=failure): ops = FakeOperations(fail=failure) with self.assertRaisesRegex(RuntimeError, "original runtime restored"): run_transaction(ops) self.assertTrue(ops.original_running) self.assertEqual(ops.state, "original") self.assertEqual(ops.config, "original") self.assertFalse(ops.candidate) self.assertEqual(ops.clones, failure != "after_stop") self.assertEqual(recover_transaction(ops), "rolled_back") def test_abrupt_crash_after_stop_rename_and_receipt_requires_explicit_recovery(self): for failure in ("after_stop", "after_rename", "after_receipt", "after_config"): with self.subTest(failure=failure): ops = FakeOperations(fail=failure, crash=True) with self.assertRaises(Crash): run_transaction(ops) self.assertNotEqual(ops.journal["phase"], "committed") self.assertEqual(recover_transaction(ops), "rolled_back") self.assertTrue(ops.original_running) def test_foreign_runtime_blocks_automatic_recovery(self): ops = FakeOperations(fail="during_verify") ops.foreign = True with self.assertRaisesRegex(RuntimeError, "requires explicit recovery"): run_transaction(ops) self.assertEqual(ops.journal["phase"], "recovery_failed") self.assertTrue(ops.candidate) def test_production_candidate_gate_rejects_wrong_nonce_project_and_replaced_id(self): ops = Production.__new__(Production) ops.prepared = {"originalContainerId": "original"} ops.plan = {"candidateComposeProject": "lumaops-managed"} ops.journal = {"nonce": "expected", "candidateContainerId": "candidate"} ops.runner = object() for fields in (["candidate", "image", "lumaops-managed", "wrong"], ["candidate", "image", "foreign-project", "expected"], ["original", "image", "lumaops-managed", "expected"], ["replacement", "image", "lumaops-managed", "expected"]): with self.subTest(fields=fields), patch("transaction.inspect_fields", return_value=fields): with self.assertRaises(ValueError): ops.assert_candidate(fields[0]) def test_future_broker_metadata_remains_dynamic(self): original = {"source_dir": "/old", "compose_files": ["docker-compose.unraid.yml"], "clean_untracked_paths": ["old"], "deploy_script": "old.sh"} prepared = {"source": "/exact/new"} plan = {"repository": "jens/vacatureradar", "candidateComposeProject": "vacatureradar-managed", "candidateMounts": [{"source": "/clone/local", "destination": "/app/local"}]} project = candidate_project(original, prepared, plan, "nonce", Path("/private/runtime.env")) self.assertTrue(project["exact_source"] and project["verify_oci_revision"]) self.assertNotIn("deploy_script", project) self.assertEqual(project["environment"]["MIGRATION_SOURCE_REVISION"], "{revision}") self.assertEqual(project["environment"]["MIGRATION_SOURCE_TREE"], "{tree}") self.assertEqual(project["environment"]["MIGRATION_BUILD_ID"], "{build_id}") self.assertEqual(project["environment"]["MIGRATION_BUILD_DATE"], "{build_date}") def test_dotenv_is_literal_and_rejects_unsupported_escaping(self): self.assertEqual(dotenv(["KEY=has $literal #value"]), b"KEY='has $literal #value'\n") for entries in (["KEY=one", "KEY=two"], ["BAD KEY=value"], ["KEY=has'quote"], ["KEY=line\nbreak"]): with self.assertRaises(ValueError): dotenv(entries) def test_manifest_detects_content_or_mode_change(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) file = root / "data" file.write_bytes(b"fixture-only") before = digest(json.dumps(tree_manifest(root), sort_keys=True).encode()) file.write_bytes(b"changed") after = digest(json.dumps(tree_manifest(root), sort_keys=True).encode()) self.assertNotEqual(before, after) def test_cold_copy_uses_archive_mode_and_rejects_metadata_mismatch(self): with tempfile.TemporaryDirectory() as temporary: source, target = Path(temporary) / "source", Path(temporary) / "target" source.mkdir() calls = [] identical = {".": {"mode": 448, "uid": 999, "gid": 999}} with patch("storage.tree_manifest", return_value=identical): clone_cold(source, target, lambda command, **kwargs: calls.append(command)) self.assertEqual(calls[0][:4], ["cp", "-a", "--reflink=auto", "--"]) with patch("storage.tree_manifest", side_effect=[identical, {}]): with self.assertRaisesRegex(ValueError, "content/metadata"): clone_cold(source, target, lambda *args, **kwargs: None) def test_real_recovery_adapter_refuses_foreign_container_before_remove(self): with tempfile.TemporaryDirectory() as temporary: workspace = Path(temporary) config, state = b"{}", b'{"status":"failed"}' for name, value in [("original-projects.json", config), ("original-state.json", state), ("projects.json", config), ("state.json", state)]: (workspace / name).write_bytes(value) ops = Production.__new__(Production) ops.workspace = workspace ops.broker = type("Broker", (), {"CONFIG_PATH": workspace / "projects.json"})() ops.state_path = workspace / "state.json" ops.prepared = {"originalContainerId": "original"} ops.plan = {"originalName": "lumaops", "candidateComposeProject": "lumaops-managed"} ops.journal = {"originalConfigSha256": digest(config), "candidateConfigSha256": "other", "originalStateSha256": digest(state), "nonce": "expected"} ops.original = lambda **kwargs: {"running": False} ops.canonical_id = lambda name: "foreign" calls = [] ops.runner = lambda command, **kwargs: calls.append(command) with patch("transaction.inspect_fields", return_value=["foreign", "image", "lumaops-managed", "wrong"]): with self.assertRaisesRegex(ValueError, "unrelated"): ops.recover() self.assertEqual(calls, []) def test_verification_binds_normal_broker_receipt_to_prepared_image(self): ops = Production.__new__(Production) ops.journal = {"candidateContainerId": "candidate"} ops.slug = "lumaops" ops.plan = {"originalName": "lumaops", "candidateMounts": []} ops.prepared = {"metadata": {"fixture": "metadata"}, "candidateImageId": "immutable", "revision": "rev"} ops.project, ops.repository, ops.runner = {}, "jens/lumaops", None ops.assert_candidate = lambda value: None ops.http_health = lambda: None ops.save = lambda value: None seen = [] class Broker: def verify_deployment(self, project, metadata, **kwargs): seen.append(kwargs) return {"fixture": "real-broker-receipt"} def validate_recorded_receipt(self, *args): return None ops.broker = Broker() with patch("transaction.mounts", return_value=[]), patch("transaction.inspect_fields", return_value=[{"lumaops_default": {}}]): ops.verifying() self.assertEqual(seen, [{"expected_image_ids": {"lumaops": "immutable"}}]) self.assertEqual(ops.journal["receipt"], {"fixture": "real-broker-receipt"}) if __name__ == "__main__": unittest.main()