codex-bot
dbc3ad8c69
Wiki image retention checks / retention-policy (pull_request) Successful in 52s
169 lines
8.5 KiB
Python
169 lines
8.5 KiB
Python
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()
|