#!/usr/bin/env python3 """High-risk state-transition tests for the host-only Rust cache maintainer.""" from __future__ import annotations import importlib.util import json from pathlib import Path 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, disabled: bool = False): self.disabled = disabled self.requests: list[str] = [] def request(self, path, body=None, raw=False): self.requests.append(path) if path.startswith("admin/actions/runners/"): return {"disabled": self.disabled, "busy": False} 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_api_path": "admin/actions/runners/1", "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 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} 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_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, 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()