#!/usr/bin/env python3 """High-risk state-transition tests for the host-only Rust cache maintainer.""" from __future__ import annotations import importlib.util import io import json from pathlib import Path import re import tempfile import unittest from unittest.mock import patch SCRIPT = Path(__file__).with_name("maintain-gitea-rust-cache.py") SPEC = importlib.util.spec_from_file_location("gitea_cache_maintenance", SCRIPT) assert SPEC and SPEC.loader maintenance_module = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(maintenance_module) IMAGE = "sha256:" + "1" * 64 OLD_IMAGE = "sha256:" + "2" * 64 BASE_IMAGE = "sha256:" + "3" * 64 SHA = "a" * 40 def runner_config(image: str) -> str: return f'runners:\n - "genarrative-ci:docker://{image}"\n' class FakeApi: def __init__(self): self.requests: list[str] = [] def request(self, path, *, method, body=None, raw=False): self.requests.append(path) raise AssertionError(f"unexpected API request: {path}") class GiteaCacheMaintenanceTest(unittest.TestCase): def setUp(self): self.temporary_directory = tempfile.TemporaryDirectory() self.root = Path(self.temporary_directory.name) self.token = self.root / "token" self.token.write_text("test-token\n", encoding="utf-8") self.config = { "state_dir": str((self.root / "state").resolve()), "api_url": "https://gitea.example.test/api/v1", "token_file": str(self.token), "repository": "team/project", "repository_url": "http://gitea:3000/team/project.git", "runner_container": "gitea-runner", "gate_socket": str(self.root / "gate.sock"), "gate_url": "https://gitea.example.test", "clone_url": "https://gitea.example.test/team/project.git", } def tearDown(self): self.temporary_directory.cleanup() def test_api_requires_explicit_method_and_preserves_it_with_body(self): api = maintenance_module.Api(self.config["api_url"], self.token) with self.assertRaises(TypeError): api.request("example", body={"value": 1}) with self.assertRaises(TypeError): api.request("example") with patch.object(maintenance_module.urllib.request, "build_opener") as build_opener: for method in ("GET", "POST", "PUT", "PATCH", "DELETE"): with self.subTest(method=method): build_opener.return_value.open.return_value = io.BytesIO(b'{}') body = {"value": 1} if method in {"POST", "PUT", "PATCH"} else None self.assertEqual(api.request("example", method=method, body=body), {}) request = build_opener.return_value.open.call_args.args[0] self.assertEqual(request.get_method(), method) self.assertEqual(request.data, None if body is None else json.dumps(body).encode()) def test_operation_logs_duration_and_build_log_without_exception_secrets(self): build_log = self.root / "artifacts" / SHA / "build.log" with patch.object(maintenance_module.time, "monotonic", side_effect=[10.0, 13.25]), \ patch("builtins.print") as printed: with self.assertRaisesRegex(RuntimeError, "test failure"): with maintenance_module.operation("merge cache artifacts", build_log=build_log): raise RuntimeError("test failure") messages = [call.args[0] for call in printed.call_args_list] self.assertEqual(messages[0], f"[cache-maintenance] merge cache artifacts: started; build log={build_log}") self.assertEqual(messages[1], f"[cache-maintenance] merge cache artifacts: failed after 3.2s; build log={build_log}") def test_workflow_jobs_and_cache_producers_match_maintenance_contract(self): workflow = (SCRIPT.parent.parent / ".gitea/workflows/project-ci.yml").read_text(encoding="utf-8") # 沿用 workflow 的显式 job/step 格式，枚举实际 job，避免另一份名单漏掉新增项。 sections = re.split(r"(?m)^ ([\w-]+):\s*$", workflow.split("\njobs:\n", 1)[1]) jobs, producers = {}, {} for job_id, section in zip(sections[1::2], sections[2::2]): names = re.findall(r"(?m)^ name: (.+)$", section) self.assertEqual(len(names), 1, f"{job_id}: expected one explicit job name") name = names[0].strip().strip("\"'") self.assertNotIn(name, jobs, f"duplicate workflow job name: {name}") jobs[name] = job_id if "run: python3 scripts/export-gitea-rust-cache.py" in section: self.assertIn(f" - name: {maintenance_module.EXPORT_STEP}\n", section) producers[name] = job_id self.assertEqual(set(jobs), maintenance_module.JOBS, "workflow job names drifted from maintainer JOBS") self.assertEqual(producers, maintenance_module.RUST_JOB_IDS, "workflow cache producer names/IDs drifted from maintainer RUST_JOB_IDS") def record(self, image, *, source=SHA, owned=True, verified_run=None, base=BASE_IMAGE): return { "image": image, "tag": ( "genarrative/gitea-project-ci:rust-cache-auto-" + source if source is not None else None ), "source": source, "inputs": "inputs", "base": base, "owned": owned, "activated": "2026-01-01T00:00:00+00:00", "verified_run": verified_run, "artifact": str(self.root / "state" / "artifacts" / (source or "manual")), } def maintenance(self, state): state_root = Path(self.config["state_dir"]) state_root.mkdir(parents=True, exist_ok=True) (state_root / "state.json").write_text(json.dumps(state), encoding="utf-8") instance = maintenance_module.Maintenance(self.config) instance.api = FakeApi() return instance def activation_maintenance(self, *, idle_values, switch=None): old = self.record(OLD_IMAGE, owned=False, verified_run=1) candidate = self.record(IMAGE) state = { "versions": [old, candidate], "current": OLD_IMAGE, "rollback": None, "candidate": IMAGE, } if switch is not None: state["switch"] = switch instance = self.maintenance(state) config = {"value": runner_config(OLD_IMAGE)} writes: list[str] = [] restarts: list[tuple] = [] statuses = 0 instance.check_gate_route = lambda: None instance.read_config = lambda: config["value"] instance.write_config = lambda value: (writes.append(value), config.__setitem__("value", value)) instance.image_info = lambda image, inner=False: {"Id": image} def docker(*args, inner=False, **kwargs): if args[:1] == ("restart",): restarts.append(args) return "" if args[:2] == ("inspect", "--format"): if args[2] == "{{.State.StartedAt}}": return "2026-01-01T00:00:00+00:00" if args[2] == "{{.State.Status}}": return "running" if args[:1] == ("logs",): return "declare successfully" raise AssertionError(f"unexpected docker call: {args}, inner={inner}") instance.docker = docker iterator = iter(idle_values) instance.idle = lambda: next(iterator) def gate(action): nonlocal statuses if action == "pause": return {"paused": True} if action == "resume": return {"paused": False} if action == "status": statuses += 1 return {"paused": False} if statuses == 1 else {"paused": True, "inflight": 0} raise AssertionError(action) instance.gate = gate return instance, writes, restarts def test_cache_inputs_keeps_openapi_and_embedded_skill_text(self): tree = "\n".join( [ "100644 blob a\tdocs/guide.md", "100644 blob b\tdocs/openapi/external.json", "100644 blob c\t.codex/skills/example/SKILL.md", "100644 blob d\tREADME.md", ] ) digest = maintenance_module.cache_inputs(tree) self.assertEqual(digest, maintenance_module.cache_inputs( tree.replace("guide.md", "【说明】中文文档.md").replace("\n", "\0") + "\0")) self.assertEqual(digest, maintenance_module.cache_inputs(tree.replace("guide.md", "other.md"))) self.assertNotEqual(digest, maintenance_module.cache_inputs(tree.replace("external.json", "changed.json"))) self.assertNotEqual(digest, maintenance_module.cache_inputs(tree.replace("SKILL.md", "PROMPT.md"))) def test_gate_route_requires_independent_checkout_and_live_runner_proof(self): instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None}) instance.read_config = lambda: runner_config(OLD_IMAGE) with self.assertRaisesRegex(RuntimeError, "GENARRATIVE_GITEA_REPOSITORY_URL"): instance.check_gate_route() instance.read_config = lambda: runner_config(OLD_IMAGE) + ( ' envs:\n GENARRATIVE_GITEA_REPOSITORY_URL: "http://gitea:3000/team/project.git"\n') started = "2026-01-01T00:00:00Z" def docker(*args, **kwargs): if args[0] == "exec": return json.dumps({"address": self.config["gate_url"]}) return json.dumps([{"State": {"StartedAt": started}, "NetworkSettings": { "Networks": {"internal": {"IPAddress": "10.0.0.2"}}}}]) instance.docker = docker proof = {"last_fetch_peer": "10.0.0.2", "last_fetch_at": maintenance_module.timestamp(started).timestamp() + 1, "active_tasks": 0} instance.gate = lambda action: proof instance.check_gate_route() proof["last_fetch_at"] -= 2 with self.assertRaisesRegex(RuntimeError, "this runner startup"): instance.check_gate_route() proof.update(last_fetch_peer="10.0.0.3", last_fetch_at=9999999999) with self.assertRaisesRegex(RuntimeError, "this runner startup"): instance.check_gate_route() def test_complete_nine_job_run_requires_real_rust_cache_statistics(self): run = {"status": "completed", "conclusion": "success"} jobs = [{"name": name, "conclusion": "success"} for name in maintenance_module.JOBS] self.assertTrue(maintenance_module.fully_passed(run, jobs)) self.assertFalse(maintenance_module.fully_passed(run, jobs[:-1])) good = ( f"worker image: {IMAGE}\n[rust-cache] mode=sccache\n" "Cache hits 7\nCache errors 0\nCache read errors 0\nCache write errors 0\n" ) self.assertTrue(maintenance_module.verifies_image({"name": "Backend tests"}, good, IMAGE)) self.assertFalse(maintenance_module.verifies_image({"name": "Backend tests"}, good.replace("Cache hits 7", "Cache hits 0"), IMAGE)) self.assertFalse(maintenance_module.verifies_image({"name": "Backend tests"}, good.replace("Cache write errors 0", "Cache write errors 1"), IMAGE)) def test_fetches_latest_master_without_ci_green_gate(self): instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None}) instance.repo.mkdir(parents=True) (instance.repo / ".git").mkdir() (instance.repo / ".git" / "genarrative-cache-maintenance").write_text("owned source checkout\n") calls = [] def fake_command(*args, **kwargs): calls.append(args) if args[1:4] == ("remote", "get-url", "origin"): return instance.config["clone_url"] + "\n" if args[1:3] == ("rev-parse", "FETCH_HEAD^{commit}"): return SHA + "\n" if args[1:3] == ("ls-tree", "-rz"): return "100644 blob x\tdocs/openapi/current.json\n" return "" with patch.object(maintenance_module, "command", fake_command): self.assertEqual(instance.fetch_source()[0], SHA) self.assertIn(("git", "remote", "get-url", "origin"), calls) self.assertIn(("git", "fetch", "--no-tags", "origin", "refs/heads/master"), calls) self.assertIn(("git", "clean", "-ffdx"), calls) self.assertFalse(any("runs" in " ".join(call) for call in calls)) def test_idle_race_after_pause_restores_without_restart(self): instance, writes, restarts = self.activation_maintenance(idle_values=[True, False]) with patch.object(maintenance_module.time, "sleep", lambda _: None): self.assertFalse(instance.activate()) self.assertEqual(restarts, []) self.assertEqual(writes, []) self.assertEqual(instance.state["current"], OLD_IMAGE) self.assertEqual(instance.state["candidate"], IMAGE) self.assertNotIn("pause_owned", instance.state) def test_idle_requires_task_ledger_and_all_live_container_states(self): instance = self.maintenance({"versions": [], "current": None}) state = {"uncertain": False, "active_tasks": 1} instance.gate = lambda _: state calls = [] instance.docker = lambda *args, **kw: calls.append((args, kw)) or "" self.assertFalse(instance.idle()) # assigned task before its first container self.assertEqual(calls, []) state["active_tasks"] = 0 self.assertTrue(instance.idle()) self.assertIn("status=created", calls[-1][0]) self.assertTrue(calls[-1][1]["inner"]) instance.docker = lambda *_a, **_k: "created-container\n" self.assertFalse(instance.idle()) del state["active_tasks"] with self.assertRaisesRegex(RuntimeError, "durable task tracking"): instance.idle() def source_fixture(self): old = self.record(OLD_IMAGE, source="b" * 40, owned=False, verified_run=1) instance = self.maintenance({"versions": [old], "current": OLD_IMAGE}) run = {"id": 44, "status": "completed", "conclusion": "failure", "event": "push", "path": "project-ci.yml@refs/heads/master", "head_branch": "master", "head_sha": SHA} jobs = [{"id": i, "name": name, "status": "completed", "conclusion": "failure", "head_sha": SHA, "run_attempt": 1, "steps": [{"name": maintenance_module.EXPORT_STEP, "conclusion": "success"}]} for i, name in enumerate(maintenance_module.RUST_JOB_IDS, start=1)] artifacts = [{"id": job["id"], "name": "rust-cache-v1-" + maintenance_module.RUST_JOB_IDS[job["name"]] + "-attempt-1", "expired": False, "workflow_run": run} for job in jobs] class Api: def pages(self, path, key): return iter({"workflow_runs": [run], "jobs": jobs, "artifacts": artifacts}[key]) def request(self, path, *, method, raw=False): job_id = int(path.split("/jobs/")[1].split("/")[0]) name = next(item["name"] for item in artifacts if item["id"] == job_id) return f"worker image: {OLD_IMAGE}\n[rust-cache] artifact={name} objects=1 bytes=10 complete=true\n" instance.api = Api() return instance, run, jobs, artifacts def select_source(self, instance): with patch.object(maintenance_module.subprocess, "run", return_value=type("Result", (), {"returncode": 0})()), \ patch.object(maintenance_module, "command", lambda *args, **kw: "100644 blob " + args[-1] + "\tsrc.rs\0"): return instance.source_run() def test_complete_failed_master_can_supply_cache_but_pr_and_missing_groups_cannot(self): instance, run, jobs, artifacts = self.source_fixture() result = self.select_source(instance) self.assertEqual(result["run_id"], 44) self.assertEqual(len(result["exports"]), 6) run["event"] = "pull_request" self.assertIsNone(self.select_source(instance)) run["event"] = "push" artifacts.pop() self.assertIsNone(self.select_source(instance)) def test_stale_attempt_cancelled_job_or_failed_export_cannot_supply_cache(self): instance, run, jobs, artifacts = self.source_fixture() run["conclusion"] = "cancelled" self.assertIsNone(self.select_source(instance)) # all six artifacts may already exist run["conclusion"] = "failure" jobs[0]["run_attempt"] = 2 self.assertIsNone(self.select_source(instance)) jobs[0]["run_attempt"] = 1 jobs[0]["conclusion"] = "cancelled" self.assertIsNone(self.select_source(instance)) jobs[0]["conclusion"] = "success" jobs[0]["steps"][0]["conclusion"] = "failure" self.assertIsNone(self.select_source(instance)) def test_exports_using_different_images_are_not_combined(self): instance, _, _, _ = self.source_fixture() original = instance.api.request instance.api.request = lambda path, **kwargs: original(path, **kwargs).replace(OLD_IMAGE, IMAGE) if '/jobs/1/' in path else original(path, **kwargs) self.assertIsNone(self.select_source(instance)) def test_completed_artifact_without_upload_completion_log_is_not_used(self): instance, _, _, _ = self.source_fixture() instance.api.request = lambda *_a, **_k: f"worker image: {OLD_IMAGE}\n" self.assertIsNone(self.select_source(instance)) def test_assembly_uses_ci_objects_and_trusted_binary_without_warming_compiler(self): instance, _, _, _ = self.source_fixture() source = self.select_source(instance) shell_calls, docker_calls = [], [] instance.build_command = lambda log, script, *args, **kw: shell_calls.append((script, args)) def info(image, inner=False): return {"Id": IMAGE if image.startswith("genarrative/") else image, "Config": {"Labels": { "world.genarrative.ci.rust-cache-source": "b" * 40, "world.genarrative.ci.rust-cache-base": BASE_IMAGE, "com.genarrative.ci.definition-sha256": "definition", }}} instance.image_info = info def docker(*args, **kw): docker_calls.append(args) if args[0] == "create": return "temporary-copy-container" if args[0] == "cp": root = Path(args[-1]) (root / "objects").mkdir() (root / "sccache").write_bytes(b"trusted binary") if args[-2:] == ("rustc", "-vV"): return "rustc test\n" return "" instance.docker = docker instance.api.download = lambda path, destination: destination.write_bytes(b"download") def merge(inputs, output, **kwargs): self.assertEqual(len(inputs), 6) self.assertEqual(kwargs["expected_inherited_source_sha"], "b" * 40) output.mkdir() (output / "objects").mkdir() return type("Merged", (), {"sccache_version": "sccache 0.18.0", "rustc": "rustc test\n", "workspace": "/workspace/team/project", "base_image": BASE_IMAGE})() with patch.object(maintenance_module, "command", return_value="definition\n"), \ patch.object(maintenance_module, "merge_snapshots", merge): instance.build(source) self.assertEqual(shell_calls, [("gitea-ci-job-image.sh", ("verify", "genarrative/gitea-project-ci:rust-cache-auto-" + SHA))]) self.assertIn(("create", OLD_IMAGE), docker_calls) self.assertIn(("rm", "--volumes", "temporary-copy-container"), docker_calls) self.assertEqual(instance.state["candidate"], IMAGE) self.assertEqual(instance.version(IMAGE)["run_id"], 44) def test_export_cleanup_only_deletes_collected_or_expired_owned_master_artifacts(self): current = self.record(OLD_IMAGE, owned=False, verified_run=1) current.update(staged=True, run_id=44, exports=[{"id": 1}]) instance = self.maintenance({"versions": [current], "current": OLD_IMAGE}) old_run = {"id": 9, "status": "completed", "path": "project-ci.yml@refs/heads/master", "event": "push", "head_branch": "master", "head_sha": SHA} artifact = {"id": 2, "name": "rust-cache-v1-backend-tests-attempt-1", "workflow_run": {"id": 9}, "created_at": "2020-01-01T00:00:00Z"} rows = [artifact, {**artifact, "id": 3, "name": "release-binary"}, {**artifact, "id": 4, "workflow_run": {"id": 10}}, {**artifact, "id": 5, "workflow_run": {"id": 11}}] deleted = [] instance.api.pages = lambda *_: iter(rows) def request(path, *, method): if method == "DELETE": return deleted.append((path, method)) return {9: old_run, 10: {**old_run, "event": "pull_request"}, 11: {**old_run, "status": "in_progress"}}[int(path.rsplit("/", 1)[1])] instance.api.request = request instance.cleanup_exports() self.assertEqual(deleted, [(instance.repo_api + "/actions/artifacts/1", "DELETE"), (instance.repo_api + "/actions/artifacts/2", "DELETE")]) self.assertTrue(instance.version(OLD_IMAGE)["exports"][0]["deleted"]) def test_signed_download_drops_token_and_rejects_other_origins(self): api = maintenance_module.Api(self.config["api_url"], self.token) destination = self.root / "artifact.zip" seen = [] target = ["https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"] class Opener: def open(self, request, timeout): seen.append(request) if not isinstance(request, str): raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target[0]}, None) return io.BytesIO(b"archive") with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()): api.download("repos/team/project/actions/artifacts/1/zip", destination) self.assertEqual(destination.read_bytes(), b"archive") self.assertEqual(seen[0].get_header("Authorization"), "token test-token") self.assertIsInstance(seen[1], str) # signed URL request has no Authorization header target[0] = "https://other.example.test/download" with self.assertRaisesRegex(RuntimeError, "configured HTTPS Gitea origin"): api.download("repos/team/project/actions/artifacts/1/zip", self.root / "rejected.zip") self.assertFalse((self.root / "rejected.zip").exists()) def test_pending_chunk_cleanup_requires_old_terminal_master_run(self): instance = self.maintenance({"versions": [], "current": None, "attempts": [{"run_id": 4}]}) instance.config["artifact_storage_dir"] = str(self.root / "storage") run = {"id": 1, "status": "completed", "path": "project-ci.yml@refs/heads/master", "event": "push", "head_branch": "master", "head_sha": SHA, "completed_at": "2020-01-01T00:00:00Z"} rows = [run, {**run, "id": 2, "event": "pull_request"}, {**run, "id": 3, "status": "in_progress"}, {**run, "id": 4}, {**run, "id": 5, "completed_at": "2999-01-01T00:00:00Z"}] instance.api.pages = lambda *_: iter(rows) with patch.object(maintenance_module, "cleanup_upload_chunks", return_value=0) as cleanup: instance.cleanup_pending_uploads() self.assertEqual(cleanup.call_args.args[1], {1}) def test_uncertain_or_inflight_gate_never_restarts(self): instance, _, restarts = self.activation_maintenance(idle_values=[True]) statuses = 0 def uncertain_gate(action): nonlocal statuses if action == "pause": return {"paused": True} if action == "resume": return {"paused": False} statuses += 1 return {"paused": False} if statuses == 1 else {"paused": True, "uncertain": True, "inflight": 1} instance.gate = uncertain_gate with patch.object(maintenance_module.time, "sleep", lambda _: None): with self.assertRaisesRegex(RuntimeError, "uncertain"): instance.activate() self.assertEqual(restarts, []) self.assertEqual(instance.state["candidate"], IMAGE) self.assertEqual(instance.state["current"], OLD_IMAGE) self.assertNotIn("pause_owned", instance.state) def test_inflight_gate_defers_without_restart_and_resumes(self): instance, _, restarts = self.activation_maintenance(idle_values=[True]) statuses = 0 def inflight_gate(action): nonlocal statuses if action == "pause": return {"paused": True} if action == "resume": return {"paused": False} statuses += 1 return {"paused": False} if statuses == 1 else {"paused": True, "inflight": 1} instance.gate = inflight_gate with patch.object(maintenance_module.time, "sleep", lambda _: None): self.assertFalse(instance.activate()) self.assertEqual(restarts, []) self.assertEqual(instance.state["candidate"], IMAGE) self.assertEqual(instance.state["current"], OLD_IMAGE) self.assertNotIn("pause_owned", instance.state) def test_interrupted_switch_recovers_current_candidate_and_rollback(self): backup = self.root / "state" / "backups" / "switch.yaml" backup.parent.mkdir(parents=True) backup.write_text(runner_config(OLD_IMAGE), encoding="utf-8") instance, _, restarts = self.activation_maintenance( idle_values=[True, True], switch={"old": OLD_IMAGE, "new": IMAGE, "backup": str(backup)}, ) with patch.object(maintenance_module.time, "sleep", lambda _: None): self.assertTrue(instance.recover_switch()) self.assertEqual(len(restarts), 1) self.assertEqual(instance.state["current"], IMAGE) self.assertIsNone(instance.state["candidate"]) self.assertEqual(instance.state["rollback"], OLD_IMAGE) self.assertNotIn("switch", instance.state) def test_protected_images_include_state_bases_config_and_container_references(self): current = self.record(OLD_IMAGE, owned=False, verified_run=1, base=BASE_IMAGE) candidate = self.record(IMAGE, base=BASE_IMAGE) container_image = "sha256:" + "4" * 64 instance = self.maintenance({"versions": [current, candidate], "current": OLD_IMAGE, "rollback": OLD_IMAGE, "candidate": IMAGE}) instance.read_config = lambda: runner_config(OLD_IMAGE) def docker(*args, inner=False, **kwargs): if args[:2] == ("ps", "-aq"): return "container-1\n" if args[:2] == ("inspect", "--format"): return container_image + "\n" raise AssertionError(args) instance.docker = docker protected = instance.protected_images() self.assertTrue({OLD_IMAGE, IMAGE, BASE_IMAGE, container_image}.issubset(protected)) def test_cleanup_keeps_manual_images_and_rejects_outside_artifacts(self): current = self.record(OLD_IMAGE, owned=False, verified_run=1) manual = self.record(IMAGE, owned=False) stale = self.record("sha256:" + "5" * 64, source="b" * 40) stale["artifact"] = str(self.root / "outside") instance = self.maintenance({"versions": [current, manual, stale], "current": OLD_IMAGE, "rollback": None, "candidate": None}) instance.protected_images = lambda: set() instance.remove_owned_image = lambda *_: True with self.assertRaisesRegex(RuntimeError, "outside"): instance.cleanup() self.assertIn(manual, instance.state["versions"]) self.assertIn(stale, instance.state["versions"]) def test_additional_tags_are_never_deleted(self): instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None}) removals = [] tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA instance.image_info = lambda image, inner=False: {"RepoTags": [tag, "operator:keep"]} instance.docker = lambda *args, inner=False, **kwargs: ( IMAGE + "\n" if args[:2] == ("image", "ls") else removals.append(args) or "" ) self.assertFalse(instance.remove_owned_image(IMAGE, tag, set())) self.assertEqual(removals, []) def test_owned_image_removal_handles_dangling_and_already_absent_images(self): instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None}) tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA present = {True: True, False: True} removals = [] def docker(*args, inner=False, **kwargs): if args[:2] == ("image", "ls"): return IMAGE + "\n" if present[inner] else "" if args[:3] == ("image", "rm", IMAGE): removals.append((args, inner)) present[inner] = False return IMAGE + "\n" raise AssertionError(args) instance.docker = docker instance.image_info = lambda image, inner=False: {"RepoTags": []} self.assertTrue(instance.remove_owned_image(IMAGE, tag, set())) self.assertEqual({inner for _, inner in removals}, {True, False}) calls = [] instance.docker = lambda *args, inner=False, **kwargs: calls.append((args, inner)) or "" self.assertTrue(instance.remove_owned_image( IMAGE, "genarrative/gitea-project-ci:rust-cache-auto-" + SHA, set())) self.assertFalse(any(args[:3] == ("image", "rm", IMAGE) for args, _ in calls)) def test_manual_current_without_cache_stats_can_be_verified(self): current = self.record(OLD_IMAGE, source=None, owned=False) instance = self.maintenance({"versions": [current], "current": OLD_IMAGE, "rollback": None, "candidate": None}) jobs = [ {"id": index, "name": name, "conclusion": "success"} for index, name in enumerate(sorted(maintenance_module.JOBS), start=1) ] run = { "id": 44, "status": "completed", "conclusion": "success", "event": "push", "path": "project-ci.yml@refs/heads/master", "head_sha": SHA, "started_at": "2026-01-02T00:00:00Z", } class VerificationApi: def pages(self, path, key): if path.endswith("/actions/runs?status=success&branch=master&event=push"): return iter([run]) if path.endswith("/actions/runs/44/jobs"): return iter(jobs) raise AssertionError(path) def request(self, path, *, method, body=None, raw=False): if not raw or "/actions/jobs/" not in path: raise AssertionError(path) return f"runner image: {OLD_IMAGE}\n" instance.api = VerificationApi() with patch.object(maintenance_module.subprocess, "run", return_value=type("Result", (), {"returncode": 0})()): self.assertTrue(instance.verify_current()) self.assertEqual(instance.version(OLD_IMAGE)["verified_run"], 44) self.assertEqual(instance.version(OLD_IMAGE)["verified_sha"], SHA) def test_recover_builds_adopts_only_pre_registered_attempt_tags(self): tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA base_tag = "genarrative/gitea-project-ci:base-auto-" + SHA artifact = self.root / "state" / "artifacts" / SHA attempt = {"source": SHA, "inputs": "inputs", "tag": tag, "base_tag": base_tag, "artifact": str(artifact)} instance = self.maintenance({ "versions": [], "current": None, "rollback": None, "candidate": None, "attempts": [attempt], }) queries = [] def docker(*args, inner=False, **kwargs): if args[:3] == ("image", "ls", "--no-trunc"): queries.append(args[-1]) return BASE_IMAGE + "\n" if args[-1] == base_tag else IMAGE + "\n" raise AssertionError(args) instance.docker = docker instance.image_info = lambda image, inner=False: { "Config": {"Labels": { "world.genarrative.ci.rust-cache-source": SHA, "world.genarrative.ci.rust-cache-base": BASE_IMAGE, }} } instance.recover_builds() self.assertEqual(queries, [base_tag, tag]) self.assertEqual(instance.state["attempts"], []) self.assertEqual(instance.state["bases"], {BASE_IMAGE: base_tag}) self.assertEqual(instance.state["versions"][0]["image"], IMAGE) self.assertTrue(instance.state["versions"][0]["owned"]) self.assertEqual(instance.state["candidate"], IMAGE) if __name__ == "__main__": unittest.main()