import importlib.util import json from pathlib import Path import tempfile import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location( "lifecycle", Path(__file__).resolve().parents[1] / "scripts/wiki-image-lifecycle.py") m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) NOW = m.timestamp("2026-10-29T12:00:00Z") A, B, C, D = [letter * 40 for letter in "abcd"] def image(tags=None, labels=None, created="2026-10-20T12:00:00Z"): return {"Id": "sha256:example", "RepoTags": tags or [], "RepoDigests": [], "Created": created, "Config": {"Labels": labels or {}}} class RetentionTests(unittest.TestCase): def decide(self, item, referenced=(), retained=(A, B, C), ready=True, completed=None): return m.decision(item, set(referenced), set(retained), ready, NOW, completed)[0] def test_current_and_two_previous_survive(self): for sha in (A, B, C): self.assertEqual(self.decide(image([f"waterfall-wiki:{sha}"])), "keep") self.assertEqual(self.decide(image([f"waterfall-wiki:{D}"])), "delete") def test_retained_release_test_images_remain_available_to_studio_tools(self): for sha in (A, B, C): self.assertEqual(self.decide(image([f"waterfall-wiki-tests:{sha}"]), completed=sha), "keep") def test_stopped_container_reference_survives(self): self.assertEqual(self.decide(image([f"waterfall-wiki:{D}"]), referenced=("sha256:example",)), "keep") def test_all_tags_of_shared_retained_image_survive(self): self.assertEqual(self.decide(image([f"waterfall-wiki:{A}", f"waterfall-wiki:{D}"])), "keep") def test_foreign_and_legacy_tags_survive(self): for tag in (f"prompt:{D}", f"other-app:{D}", "waterfall-wiki:local"): self.assertEqual(self.decide(image([tag, f"waterfall-wiki:{D}"])), "keep") def test_bootstrap_never_deletes_runtime(self): self.assertEqual(self.decide(image([f"waterfall-wiki:{D}"]), ready=False), "keep") def test_failed_job_debug_window(self): for repo in ("waterfall-wiki", "waterfall-wiki-tests"): self.assertEqual(self.decide(image([f"{repo}:{D}"], created="2026-10-28T12:00:00Z")), "keep") def test_successful_job_test_image_can_be_removed_immediately(self): item = image([f"waterfall-wiki-tests:{D}"], created="2026-10-29T11:00:00Z") self.assertEqual(self.decide(item, completed=D), "delete") self.assertEqual(self.decide(item), "keep") def test_expired_tests_dont_require_release_ledger(self): self.assertEqual(self.decide(image([f"waterfall-wiki-tests:{D}"]), ready=False), "delete") def test_only_owned_dangling_images_are_candidates(self): self.assertEqual(self.decide(image()), "keep") self.assertEqual(self.decide(image(labels={m.PROJECT: "other"})), "keep") self.assertEqual(self.decide(image(labels={m.PROJECT: "waterfall-wiki"})), "delete") def test_registry_digest_is_conservatively_retained(self): item = image([f"waterfall-wiki:{D}"]) item["RepoDigests"] = ["registry.example/wiki@sha256:abc"] self.assertEqual(self.decide(item), "keep") def test_local_containerd_digest_does_not_disable_expiry(self): item = image([f"waterfall-wiki-tests:{D}"]) item["RepoDigests"] = ["waterfall-wiki-tests@sha256:example"] self.assertEqual(self.decide(item), "delete") def test_failed_container_requires_ownership_age_and_no_mounts(self): item = {"Config": {"Labels": {m.PROJECT: "waterfall-wiki", "io.entroia.wiki.build": "true"}}, "State": {"Status": "exited"}, "Mounts": [], "Created": "2026-10-20T12:00:00Z"} self.assertTrue(m.container_expired(item, NOW)) for key, value in (("Mounts", [{"Type": "volume"}]), ("State", {"Status": "running"}), ("Created", "2026-10-29T11:00:00Z"), ("Config", {"Labels": {}})): self.assertFalse(m.container_expired(dict(item, **{key: value}), NOW)) def test_failed_ci_preserves_original_exit_and_debug_images(self): error = m.subprocess.CalledProcessError(7, ["docker", "build"]) with patch.object(m, "run", side_effect=error), patch.object(m, "cleanup") as cleanup: self.assertEqual(m.ci(Path("/unused"), D), 7) cleanup.assert_called_once_with(Path("/unused"), apply=True, completed_test=None) def test_cleanup_error_does_not_turn_failed_ci_into_success(self): error = m.subprocess.CalledProcessError(7, ["docker", "build"]) with patch.object(m, "run", side_effect=error), \ patch.object(m, "cleanup", side_effect=RuntimeError("busy")): self.assertEqual(m.ci(Path("/unused"), D), 7) def test_success_only_cleans_tests_after_deployment(self): events = [] with patch.object(m, "run", side_effect=lambda args, **kw: events.append(args)), \ patch.object(m, "cleanup", side_effect=lambda *a, **kw: events.append(kw)): self.assertEqual(m.ci(Path("/unused"), D), 0) self.assertEqual(events[-2], ["bash", "scripts/deploy.sh", D]) self.assertEqual(events[-1], {"apply": True, "completed_test": D}) def test_busy_lock_refuses_work(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "lock" with m.lock(path): with self.assertRaisesRegex(RuntimeError, "Busy"): with m.lock(path): self.fail("Second holder entered lock") def test_only_current_verified_sha_can_be_recorded_and_order_is_unique(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "current").mkdir() (root / "current/release.env").write_text(f"RELEASE_SHA={A}\n") (root / "successful-releases").write_text(f"{B}\n{A}\n{C}\n") m.record(root, A) self.assertEqual(m.history(root), [A, B, C]) with self.assertRaises(RuntimeError): m.record(root, D) (root / "successful-releases").write_text("bad-ledger\n") with self.assertRaises(RuntimeError): m.history(root) def test_preview_has_no_docker_mutations(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) item = image([f"waterfall-wiki-tests:{D}"]) with patch.object(m, "history", return_value=[A, B, C]), \ patch.object(m, "current_sha", return_value=A), \ patch.object(m, "containers", return_value=[]), \ patch.object(m, "image_ids", return_value=[item["Id"]]), \ patch.object(m, "output", return_value=json.dumps([item])), \ patch.object(m, "run") as run, patch.object(m.subprocess, "run") as mutate: m.cleanup(root) run.assert_not_called() mutate.assert_not_called() def test_cleanup_rechecks_container_reference_before_removal(self): with tempfile.TemporaryDirectory() as directory: item = image([f"waterfall-wiki-tests:{D}"]) with patch.object(m, "history", return_value=[A, B, C]), \ patch.object(m, "current_sha", return_value=A), \ patch.object(m, "containers", side_effect=[[], [], [{"Image": item["Id"]}]]), \ patch.object(m, "image_ids", return_value=[item["Id"]]), \ patch.object(m, "output", return_value=json.dumps([item])), \ patch.object(m.subprocess, "run") as mutate: m.cleanup(Path(directory), apply=True) mutate.assert_not_called() def test_cleanup_never_forces_or_removes_volumes(self): with tempfile.TemporaryDirectory() as directory: item = image([f"waterfall-wiki-tests:{D}"]) with patch.object(m, "history", return_value=[A, B, C]), \ patch.object(m, "current_sha", return_value=A), \ patch.object(m, "containers", return_value=[]), \ patch.object(m, "image_ids", return_value=[item["Id"]]), \ patch.object(m, "output", return_value=json.dumps([item])), \ patch.object(m.subprocess, "run") as mutate: mutate.return_value.returncode = 0 m.cleanup(Path(directory), apply=True) mutate.assert_called_once_with(["docker", "image", "rm", f"waterfall-wiki-tests:{D}"], text=True) if __name__ == "__main__": unittest.main()