Files
Genarrative/scripts/test_gitea_cache_maintenance.py
T
lhk229 4380d453d7
Project CI / Backend tests (pull_request) Has been cancelled
Project CI / Native shell tests (pull_request) Has been cancelled
Project CI / Frontend tests (pull_request) Has been cancelled
Project CI / Repository checks (pull_request) Has been cancelled
Project CI / AI game creator shell web tests (pull_request) Has been cancelled
Project CI / AI game creator shell Rust crates (pull_request) Has been cancelled
Project CI / AI game creator shell Rust lane 1/2 (pull_request) Has been cancelled
Project CI / AI game creator shell Rust lane 2/2 (pull_request) Has been cancelled
Project CI / AI game creator shell Rust smoke (pull_request) Has been cancelled
Project CI / AI game creator shell Rust lane 1/2 (push) Has been cancelled
Project CI / AI game creator shell Rust lane 2/2 (push) Has been cancelled
Project CI / AI game creator shell Rust smoke (push) Has been cancelled
Project CI / AI game creator shell Rust crates (push) Has been cancelled
Project CI / Backend tests (push) Has been cancelled
Project CI / Native shell tests (push) Has been cancelled
Project CI / Frontend tests (push) Has been cancelled
Project CI / Repository checks (push) Has been cancelled
Project CI / AI game creator shell web tests (push) Has been cancelled
修复 CI 缓存对象误判与产物下载完整性检查
按 ZIP 成员内容和元数据识别等价缓存对象,保留真实冲突拒绝规则。
核对产物大小与 ZIP 完整性,为截断和临时传输失败增加有界重试。
补充回归测试并同步部署文档与共享记忆。
通过 80 项缓存测试、17 项工作流测试及真实产物和 HTTP 截断验证。
2026-09-22 14:40:55 +00:00

761 lines
38 KiB
Python

#!/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
import zipfile
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'
def zip_bytes(*, size=None) -> bytes:
"""Make a valid stored ZIP, optionally padded to an exact download size."""
content = b"x" * 40000 if size else b"cache artifact"
output = io.BytesIO()
with zipfile.ZipFile(output, "w", zipfile.ZIP_STORED) as archive:
archive.writestr("snapshot.tar", content)
value = output.getvalue()
if size is None:
return value
if not len(value) < size <= len(value) + 65535:
raise AssertionError("requested ZIP size cannot be represented by its comment")
output = io.BytesIO(value)
with zipfile.ZipFile(output, "a") as archive:
archive.comment = b"p" * (size - len(value))
value = output.getvalue()
if len(value) != size:
raise AssertionError("ZIP comment did not produce the requested size")
return value
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", "size_in_bytes": 10, "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)
self.assertEqual({item["size_in_bytes"] for item in result["exports"]}, {10})
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, expected_size: 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"
archive = zip_bytes()
seen = []
target = ["https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"]
class Response(io.BytesIO):
def __init__(self, value):
super().__init__(value)
self.headers = {"Content-Length": str(len(value))}
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 Response(archive)
with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()):
api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive))
self.assertEqual(destination.read_bytes(), 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", len(archive))
self.assertFalse((self.root / "rejected.zip").exists())
self.assertEqual(len(seen), 3) # invalid redirect is deterministic and is not retried
def test_download_retries_transport_and_truncated_response_with_fresh_signed_urls(self):
api = maintenance_module.Api(self.config["api_url"], self.token)
destination = self.root / "artifact.zip"
archive = zip_bytes(size=100000)
seen, signed_responses = [], []
origin_requests = 0
target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"
class Response(io.BytesIO):
def __init__(self, value):
super().__init__(value)
self.headers = {"Content-Length": "100000"}
class Opener:
def open(self, request, timeout):
nonlocal origin_requests
seen.append(request)
if not isinstance(request, str):
origin_requests += 1
if origin_requests == 1:
raise maintenance_module.urllib.error.URLError("connection reset")
raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None)
signed_responses.append(request)
return Response(b"x" * 512 if len(signed_responses) == 1 else archive)
with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
patch.object(maintenance_module.time, "sleep") as sleep:
api.download("repos/team/project/actions/artifacts/1/zip", destination, 100000)
self.assertEqual(destination.read_bytes(), archive)
self.assertEqual(len(signed_responses), 2)
self.assertEqual(origin_requests, 3)
self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
def test_download_rejects_malformed_zip_after_retries(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 Response(io.BytesIO):
headers = {"Content-Length": "9"}
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}, None)
return Response(b"not a zip")
with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
patch.object(maintenance_module.time, "sleep") as sleep:
with self.assertRaisesRegex(RuntimeError, "incomplete after retries"):
api.download("repos/team/project/actions/artifacts/1/zip", destination, 9)
self.assertFalse(destination.exists())
self.assertEqual(sum(not isinstance(item, str) for item in seen), 3)
self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
def test_download_rejects_valid_zip_when_its_size_differs_from_metadata(self):
api = maintenance_module.Api(self.config["api_url"], self.token)
destination = self.root / "artifact.zip"
archive = zip_bytes()
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}, None)
return io.BytesIO(archive)
with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \
patch.object(maintenance_module.time, "sleep") as sleep:
with self.assertRaisesRegex(RuntimeError, "incomplete after retries"):
api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive) + 1)
self.assertFalse(destination.exists())
self.assertEqual(sum(not isinstance(item, str) for item in seen), 3)
self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)])
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()