mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-01 07:33:45 -06:00
919 lines
42 KiB
Python
919 lines
42 KiB
Python
"""autorun: the runner every unattended job goes through, and its first jobs.
|
|
|
|
Everything here is offline. Jobs are fake commands, git repositories are
|
|
built in tmp/tests with a bare repository standing in for origin, and gh is
|
|
a function returning canned CI runs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import unittest.mock
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import autorun as ar # noqa: E402
|
|
import bios_integrity as bib # noqa: E402
|
|
import cleanup_disk as cd # noqa: E402
|
|
import sync_release_assets as sra # noqa: E402
|
|
|
|
PY = "python"
|
|
_ENV_KEYS = ("GIT_CONFIG_GLOBAL", "GIT_CONFIG_NOSYSTEM")
|
|
_saved_env: dict[str, str | None] = {}
|
|
_config_dir: Path | None = None
|
|
|
|
|
|
def setUpModule() -> None:
|
|
global _config_dir
|
|
_config_dir = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests"))
|
|
config = _config_dir / "gitconfig"
|
|
config.write_text(
|
|
"[user]\n\tname = Test\n\temail = test@example.invalid\n"
|
|
"[init]\n\tdefaultBranch = main\n"
|
|
"[commit]\n\tgpgsign = false\n",
|
|
encoding="utf-8",
|
|
)
|
|
for key in _ENV_KEYS:
|
|
_saved_env[key] = os.environ.get(key)
|
|
os.environ["GIT_CONFIG_GLOBAL"] = str(config)
|
|
os.environ["GIT_CONFIG_NOSYSTEM"] = "1"
|
|
|
|
|
|
def tearDownModule() -> None:
|
|
for key, value in _saved_env.items():
|
|
if value is None:
|
|
os.environ.pop(key, None)
|
|
else:
|
|
os.environ[key] = value
|
|
if _config_dir is not None:
|
|
shutil.rmtree(_config_dir, ignore_errors=True)
|
|
|
|
|
|
def run_git(cwd: Path, *args: str) -> str:
|
|
return subprocess.run(
|
|
["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True,
|
|
).stdout
|
|
|
|
|
|
class FakeHost(ar.Host):
|
|
def __init__(self, loads: list[float] | None = None, cpus: int = 4) -> None:
|
|
self.loads = list(loads or [0.0])
|
|
self.now = 0.0
|
|
self.slept: list[float] = []
|
|
super().__init__(
|
|
loadavg=self._load, sleep=self._sleep, clock=lambda: self.now,
|
|
cpu_count=cpus, poll=10.0,
|
|
)
|
|
|
|
def _load(self) -> tuple[float, float, float]:
|
|
value = self.loads.pop(0) if len(self.loads) > 1 else self.loads[0]
|
|
return (value, value, value)
|
|
|
|
def _sleep(self, seconds: float) -> None:
|
|
self.slept.append(seconds)
|
|
self.now += seconds
|
|
|
|
|
|
class Scratch(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests"))
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
|
|
def context(self, jobs: dict, **kwargs) -> ar.Context:
|
|
repo = self.tmp / "repo"
|
|
repo.mkdir(exist_ok=True)
|
|
parsed = ar.parse_jobs({"jobs": jobs})
|
|
kwargs.setdefault("host", FakeHost())
|
|
return ar.Context(repo=repo, jobs=parsed, lock_path=self.tmp / "auto.lock", **kwargs)
|
|
|
|
def state(self, ctx: ar.Context) -> dict:
|
|
return json.loads(ctx.state_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
class Registry(Scratch):
|
|
def test_the_committed_registry_loads(self):
|
|
jobs = ar.load_jobs()
|
|
self.assertIn("nightly", jobs)
|
|
self.assertTrue(jobs["nightly"].commits)
|
|
self.assertFalse(jobs["disk-check"].lock)
|
|
|
|
def test_an_unknown_key_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
ar.parse_jobs({"jobs": {"x": {"steps": [], "schedul": "daily"}}})
|
|
|
|
def test_a_subject_the_hook_refuses_is_refused_at_load(self):
|
|
for subject in ("chore: fix the broken manifests",
|
|
"update things",
|
|
"chore: " + "a" * 60):
|
|
with self.subTest(subject=subject), self.assertRaises(ValueError):
|
|
ar.parse_jobs({"jobs": {"x": {"commits": [
|
|
{"subject": subject, "paths": ["a"]}]}}})
|
|
|
|
def test_a_job_outside_the_worktree_cannot_commit(self):
|
|
with self.assertRaises(ValueError):
|
|
ar.parse_jobs({"jobs": {"x": {"worktree": False, "commits": [
|
|
{"subject": "chore: add a", "paths": ["a"]}]}}})
|
|
|
|
def test_a_twice_entry_names_its_output(self):
|
|
with self.assertRaises(ValueError):
|
|
ar.parse_jobs({"jobs": {"x": {"twice": [
|
|
{"artifact": "a.json", "run": [PY, "gen.py"]}]}}})
|
|
|
|
|
|
class Lock(Scratch):
|
|
def test_a_second_holder_is_refused(self):
|
|
path = self.tmp / "auto.lock"
|
|
with ar.global_lock(path):
|
|
self.assertTrue(ar.lock_busy(path))
|
|
with self.assertRaises(ar.LockBusy), ar.global_lock(path):
|
|
pass
|
|
self.assertFalse(ar.lock_busy(path))
|
|
|
|
def test_a_job_waiting_on_the_lock_is_deferred_and_recorded(self):
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0,
|
|
"steps": [[PY, "-c", "open('ran', 'w')"]]}})
|
|
with ar.global_lock(ctx.lock_path):
|
|
result, detail = ar.run_job(ctx, "x")
|
|
self.assertEqual(result, ar.DEFERRED)
|
|
self.assertIn("held", detail)
|
|
self.assertFalse((ctx.repo / "ran").exists())
|
|
self.assertEqual(self.state(ctx)["jobs"]["x"]["result"], ar.DEFERRED)
|
|
|
|
def test_a_job_that_takes_no_lock_runs_beside_a_holder(self):
|
|
ctx = self.context({"x": {"worktree": False, "lock": False, "min_free_gb": 0,
|
|
"steps": [[PY, "-c", "open('ran', 'w')"]]}})
|
|
with ar.global_lock(ctx.lock_path):
|
|
self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK)
|
|
self.assertTrue((ctx.repo / "ran").exists())
|
|
|
|
|
|
class Guards(Scratch):
|
|
def test_the_load_guard_waits_then_lets_go(self):
|
|
host = FakeHost(loads=[3.0, 3.0, 0.5], cpus=4)
|
|
self.assertTrue(ar.wait_for_load(host, max_wait=600))
|
|
self.assertEqual(host.slept, [10.0, 10.0])
|
|
|
|
def test_the_load_guard_gives_up_at_its_deadline(self):
|
|
host = FakeHost(loads=[8.0], cpus=4)
|
|
self.assertFalse(ar.wait_for_load(host, max_wait=25))
|
|
self.assertEqual(sum(host.slept), 25)
|
|
|
|
def test_a_busy_host_defers_the_job(self):
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "max_load_wait": 0,
|
|
"steps": [[PY, "-c", "open('ran', 'w')"]]}},
|
|
host=FakeHost(loads=[64.0]))
|
|
self.assertEqual(ar.run_job(ctx, "x")[0], ar.DEFERRED)
|
|
self.assertFalse((ctx.repo / "ran").exists())
|
|
|
|
def test_a_full_disk_refuses_the_job(self):
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 10 ** 9,
|
|
"steps": [[PY, "-c", "open('ran', 'w')"]]}})
|
|
result, detail = ar.run_job(ctx, "x")
|
|
self.assertEqual(result, ar.REFUSED)
|
|
self.assertIn("GB free", detail)
|
|
self.assertFalse((ctx.repo / "ran").exists())
|
|
|
|
def test_a_failure_counts_and_a_success_resets(self):
|
|
ctx = self.context({
|
|
"bad": {"worktree": False, "min_free_gb": 0, "steps": [[PY, "-c", "raise SystemExit(3)"]]},
|
|
})
|
|
ar.run_job(ctx, "bad")
|
|
ar.run_job(ctx, "bad")
|
|
record = self.state(ctx)["jobs"]["bad"]
|
|
self.assertEqual(record["result"], ar.FAILED)
|
|
self.assertEqual(record["consecutive_failures"], 2)
|
|
self.assertIsNone(record["last_success"])
|
|
|
|
|
|
class StepEnvironment(Scratch):
|
|
"""A step can run third-party code: it never inherits a token or a password."""
|
|
|
|
HOST = MappingProxyType({
|
|
"PATH": "/usr/bin", "HOME": "/home/u", "LANG": "C.UTF-8", "LC_ALL": "C", "TZ": "UTC",
|
|
"TMPDIR": "/t", "GITHUB_TOKEN": "t1", "GH_TOKEN": "t2", "ROMSTATION_USER": "u",
|
|
"ROMSTATION_PASS": "p", "NPM_TOKEN": "t3", "AWS_SECRET_ACCESS_KEY": "s",
|
|
"DB_PASSWORD": "p2", "ROMFETCH_CONTENT_DIR": "/c", "EDITOR": "vi"})
|
|
|
|
def test_the_allowlist(self):
|
|
env = ar.safe_env(["ROMFETCH_CONTENT_DIR", "GITHUB_TOKEN"], {"RETROBIOS_AUTO": "1"},
|
|
source=dict(self.HOST))
|
|
self.assertEqual(env, {"PATH": "/usr/bin", "HOME": "/home/u", "LANG": "C.UTF-8",
|
|
"LC_ALL": "C", "TZ": "UTC", "TMPDIR": "/t",
|
|
"ROMFETCH_CONTENT_DIR": "/c", "RETROBIOS_AUTO": "1"})
|
|
|
|
def test_a_step_naming_a_secret_is_refused_at_load(self):
|
|
for name in ("GITHUB_TOKEN", "ROMSTATION_PASS", "MY_SECRET"):
|
|
with self.subTest(name=name), self.assertRaises(ValueError):
|
|
ar.parse_jobs({"jobs": {"x": {"steps": [{"run": [PY, "-V"], "env": [name]}]}}})
|
|
|
|
def test_a_network_job_step_sees_no_token(self):
|
|
dump = "import json, os; print(json.dumps(dict(os.environ)))"
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "network": True,
|
|
"steps": [{"run": [PY, "-c", dump], "stdout": "env.json",
|
|
"env": ["ROMFETCH_CONTENT_DIR"]}]}})
|
|
with unittest.mock.patch.dict(os.environ, {"GITHUB_TOKEN": "t1", "GH_TOKEN": "t2",
|
|
"ROMSTATION_PASS": "p",
|
|
"ROMFETCH_CONTENT_DIR": "/c", "EDITOR": "vi"}):
|
|
self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK)
|
|
given = ar.token_env()
|
|
seen = json.loads((ctx.repo / "env.json").read_text())
|
|
self.assertFalse({"GITHUB_TOKEN", "GH_TOKEN", "ROMSTATION_PASS", "EDITOR"} & set(seen))
|
|
self.assertEqual((seen["ROMFETCH_CONTENT_DIR"], seen["RETROBIOS_AUTO"]), ("/c", "1"))
|
|
self.assertEqual(seen[ar.LOCK_HELD_ENV], "1")
|
|
# The push and gh are the phases that get it.
|
|
self.assertEqual(given["GITHUB_TOKEN"], "t1")
|
|
|
|
|
|
class ReportAlert(Scratch):
|
|
def test_a_report_alert_is_raised_then_cleared(self):
|
|
write = "import json, sys; json.dump({'alert': sys.argv[1]} if sys.argv[1] else {}, open('r.json', 'w'))"
|
|
for message, expected in (("no sandbox", True), ("", False)):
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "report": "r.json",
|
|
"steps": [[PY, "-c", write, message]]}})
|
|
ar.run_job(ctx, "x")
|
|
alerts = self.state(ctx).get("alerts", {})
|
|
self.assertEqual("job:x" in alerts, expected)
|
|
if expected:
|
|
self.assertEqual(alerts["job:x"]["detail"], "no sandbox")
|
|
|
|
|
|
class Retry(Scratch):
|
|
def _runs(self, *outcomes: ar.Outcome):
|
|
queue = list(outcomes)
|
|
calls = []
|
|
|
|
def run() -> ar.Outcome:
|
|
calls.append(1)
|
|
return queue.pop(0)
|
|
return run, calls
|
|
|
|
def test_a_segfault_is_retried_once_on_an_idle_host(self):
|
|
run, calls = self._runs(ar.Outcome(139), ar.Outcome(0))
|
|
waited = []
|
|
outcome = ar.run_with_retry(run, (0,), lambda: waited.append(1))
|
|
self.assertEqual((outcome.returncode, outcome.retried, len(calls), waited),
|
|
(0, True, 2, [1]))
|
|
|
|
def test_a_second_segfault_is_a_failure(self):
|
|
run, calls = self._runs(ar.Outcome(-11), ar.Outcome(-11), ar.Outcome(0))
|
|
outcome = ar.run_with_retry(run, (0,), lambda: None)
|
|
self.assertEqual((outcome.returncode, len(calls)), (-11, 2))
|
|
|
|
def test_a_host_misread_message_is_retried(self):
|
|
run, calls = self._runs(ar.Outcome(1, "SystemError: unknown opcode"), ar.Outcome(0))
|
|
self.assertEqual(ar.run_with_retry(run, (0,), lambda: None).returncode, 0)
|
|
self.assertEqual(len(calls), 2)
|
|
|
|
def test_an_ordinary_failure_is_not_retried(self):
|
|
run, calls = self._runs(ar.Outcome(1, "KeyError: 'x'"), ar.Outcome(0))
|
|
self.assertEqual(ar.run_with_retry(run, (0,), lambda: None).returncode, 1)
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_a_real_segfault_runs_again_and_the_job_passes(self):
|
|
crash = ("import os, signal, pathlib\n"
|
|
"p = pathlib.Path('crashed')\n"
|
|
"if not p.exists():\n"
|
|
" p.touch()\n"
|
|
" os.kill(os.getpid(), signal.SIGSEGV)\n")
|
|
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0,
|
|
"steps": [[PY, "-c", crash]]}})
|
|
self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK)
|
|
self.assertEqual(len(self.state(ctx)["jobs"]["x"]["retried"]), 1)
|
|
|
|
|
|
class DoubleComputation(Scratch):
|
|
GEN = ("import json, random, sys\n"
|
|
"json.dump({'generated_at': str(random.random()), 'value': %s},"
|
|
" open(sys.argv[1], 'w'))\n")
|
|
|
|
def _run(self, cwd: Path):
|
|
def run(argv):
|
|
return ar.Outcome(subprocess.run(argv, cwd=cwd, check=False).returncode)
|
|
return run
|
|
|
|
def test_timestamps_aside_two_equal_runs_agree(self):
|
|
script = self.tmp / "gen.py"
|
|
script.write_text(self.GEN % "1", encoding="utf-8")
|
|
diff = ar.twice([PY, str(script), "{out}"], self.tmp / "a.json",
|
|
self.tmp / "b.json", self._run(self.tmp))
|
|
self.assertEqual(diff, [])
|
|
|
|
def test_two_runs_that_disagree_are_named(self):
|
|
script = self.tmp / "gen.py"
|
|
script.write_text(self.GEN % "random.random()", encoding="utf-8")
|
|
diff = ar.twice([PY, str(script), "{out}"], self.tmp / "a.json",
|
|
self.tmp / "b.json", self._run(self.tmp))
|
|
self.assertEqual(diff, ["a.json"])
|
|
|
|
def test_directories_compare_file_by_file_within_the_glob(self):
|
|
a, b = self.tmp / "a", self.tmp / "b"
|
|
(a / "targets").mkdir(parents=True)
|
|
b.mkdir()
|
|
(a / "x.json").write_text('{"generated": "1", "n": 1}')
|
|
(b / "x.json").write_text('{"generated": "2", "n": 1}')
|
|
(a / "targets" / "t.json").write_text("{}")
|
|
self.assertEqual(ar.differences(a, b, "*.json"), [])
|
|
(b / "y.json").write_text("{}")
|
|
self.assertEqual(ar.differences(a, b, "*.json"), ["y.json"])
|
|
|
|
def test_a_job_whose_artifact_disagrees_fails_with_an_alert(self):
|
|
script = self.tmp / "repo" / "gen.py"
|
|
script.parent.mkdir(parents=True, exist_ok=True)
|
|
script.write_text(self.GEN % "random.random()", encoding="utf-8")
|
|
ctx = self.context({"x": {
|
|
"worktree": False, "min_free_gb": 0, "max_load_wait": 0,
|
|
"gate": [[PY, "gen.py", "db.json"]],
|
|
"twice": [{"artifact": "db.json", "run": [PY, "gen.py", "{out}"]}],
|
|
}})
|
|
result, detail = ar.run_job(ctx, "x")
|
|
self.assertEqual(result, ar.FAILED)
|
|
self.assertIn("two computations disagree", detail)
|
|
self.assertIn("host-integrity", self.state(ctx)["alerts"])
|
|
|
|
def test_a_job_whose_artifact_agrees_passes_and_clears_the_alert(self):
|
|
script = self.tmp / "repo" / "gen.py"
|
|
script.parent.mkdir(parents=True, exist_ok=True)
|
|
script.write_text(self.GEN % "1", encoding="utf-8")
|
|
ctx = self.context({"x": {
|
|
"worktree": False, "min_free_gb": 0,
|
|
"gate": [[PY, "gen.py", "db.json"]],
|
|
"twice": [{"artifact": "db.json", "run": [PY, "gen.py", "{out}"]}],
|
|
}})
|
|
ar.save_state(ctx.state_path, {"alerts": {"host-integrity": {"since": "x"}}})
|
|
self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK)
|
|
self.assertNotIn("host-integrity", self.state(ctx).get("alerts", {}))
|
|
|
|
|
|
def ci_run(sha: str, conclusion: str | None, created: str, name: str = "Deploy Site") -> dict:
|
|
return {"workflowName": name, "headSha": sha, "createdAt": created,
|
|
"status": "completed" if conclusion else "in_progress",
|
|
"conclusion": conclusion or ""}
|
|
|
|
|
|
class CircuitBreaker(Scratch):
|
|
def _gh(self, by_commit: dict, on_main: list):
|
|
calls = []
|
|
|
|
def gh(args):
|
|
calls.append(args)
|
|
if "--commit" in args:
|
|
return by_commit.get(args[args.index("--commit") + 1], [])
|
|
return on_main
|
|
return gh, calls
|
|
|
|
def test_a_red_run_on_the_pushed_commit_opens_it(self):
|
|
state = {"push": {"sha": "a" * 40, "ci": "pending"}}
|
|
gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, "success", "2026-10-27T02:10:00Z"),
|
|
ci_run("a" * 40, "failure", "2026-10-27T02:10:00Z",
|
|
"Validate")]}, [])
|
|
ar.refresh_breaker(state, gh, "now")
|
|
self.assertTrue(state["breaker"]["open"])
|
|
self.assertEqual(state["breaker"]["reason"], "Validate")
|
|
self.assertIn("ci", state["alerts"])
|
|
self.assertEqual(ar.may_push(state), "circuit breaker open")
|
|
|
|
def test_runs_still_going_keep_the_push_pending(self):
|
|
state = {"push": {"sha": "a" * 40, "ci": "pending"}}
|
|
gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, None, "2026-10-27T02:10:00Z")]}, [])
|
|
ar.refresh_breaker(state, gh, "now")
|
|
self.assertFalse(state["breaker"]["open"])
|
|
self.assertIn("has not ended green", ar.may_push(state))
|
|
|
|
def test_a_later_green_run_on_main_closes_it(self):
|
|
state = {"push": {"sha": "a" * 40, "ci": "failure"},
|
|
"breaker": {"open": True, "sha": "a" * 40,
|
|
"failed_run_at": "2026-10-27T02:10:00Z"},
|
|
"alerts": {"ci": {"since": "x"}}}
|
|
pending = [ci_run("b" * 40, None, "2026-10-27T09:00:00Z")]
|
|
gh, _ = self._gh({}, pending + [ci_run("a" * 40, "failure", "2026-10-27T02:10:00Z")])
|
|
ar.refresh_breaker(state, gh, "now")
|
|
self.assertTrue(state["breaker"]["open"])
|
|
green = [ci_run("c" * 40, "success", "2026-10-27T10:00:00Z"),
|
|
ci_run("c" * 40, "skipped", "2026-10-27T10:00:01Z", "Validate")]
|
|
gh, _ = self._gh({}, green + pending)
|
|
ar.refresh_breaker(state, gh, "later")
|
|
self.assertFalse(state["breaker"]["open"])
|
|
self.assertNotIn("ci", state["alerts"])
|
|
|
|
def test_an_open_breaker_blocks_a_committing_job_before_any_git(self):
|
|
gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, "failure", "t")]}, [])
|
|
ctx = self.context({"x": {"min_free_gb": 0, "commits": [
|
|
{"subject": "chore: add a", "paths": ["a"]}]}}, gh=gh)
|
|
ar.save_state(ctx.state_path, {"push": {"sha": "a" * 40, "ci": "pending"}})
|
|
result, _ = ar.run_job(ctx, "x")
|
|
self.assertEqual(result, ar.BLOCKED)
|
|
self.assertFalse(ctx.worktree.exists())
|
|
|
|
|
|
class Origin(Scratch):
|
|
"""A bare origin, a main checkout cloned from it, and a second clone
|
|
standing for another contributor."""
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
self.origin = self.tmp / "origin.git"
|
|
run_git(self.tmp, "init", "--quiet", "--bare", "-b", "main", str(self.origin))
|
|
self.repo = self.tmp / "repo"
|
|
run_git(self.tmp, "clone", "--quiet", str(self.origin), str(self.repo))
|
|
(self.repo / ".gitignore").write_text(".cache/\ntmp/\n", encoding="utf-8")
|
|
(self.repo / "database.json").write_text("{}\n", encoding="utf-8")
|
|
(self.repo / "install").mkdir()
|
|
(self.repo / "install" / "a.json").write_text("{}\n", encoding="utf-8")
|
|
run_git(self.repo, "add", ".gitignore", "database.json", "install/a.json")
|
|
run_git(self.repo, "commit", "--quiet", "-m", "chore: start")
|
|
run_git(self.repo, "push", "--quiet", "origin", "main")
|
|
|
|
def other_commit(self, name: str) -> None:
|
|
other = self.tmp / "other"
|
|
if not other.exists():
|
|
run_git(self.tmp, "clone", "--quiet", str(self.origin), str(other))
|
|
run_git(other, "pull", "--quiet", "origin", "main")
|
|
(other / name).write_text(name, encoding="utf-8")
|
|
run_git(other, "add", name)
|
|
run_git(other, "commit", "--quiet", "-m", f"chore: add {name}")
|
|
run_git(other, "push", "--quiet", "origin", "main")
|
|
|
|
def worktree(self) -> Path:
|
|
wt = self.repo / ".cache" / "autorun" / "worktree"
|
|
ar.ensure_worktree(self.repo, wt)
|
|
return wt
|
|
|
|
|
|
class Commits(Origin):
|
|
def test_globs_stay_inside_one_segment(self):
|
|
self.assertTrue(ar.path_matches("install/*.json", "install/a.json"))
|
|
self.assertFalse(ar.path_matches("install/*.json", "install/targets/a.json"))
|
|
self.assertFalse(ar.path_matches("install/*.json", "x/install/a.json"))
|
|
|
|
def test_the_dry_run_names_only_matching_files_and_stages_nothing(self):
|
|
wt = self.worktree()
|
|
(wt / "database.json").write_text('{"n": 1}\n', encoding="utf-8")
|
|
(wt / "install" / "b.json").write_text("{}\n", encoding="utf-8")
|
|
(wt / "notes.txt").write_text("x", encoding="utf-8")
|
|
ctx = ar.Context(repo=self.repo, lock_path=self.tmp / "l", dry_run=True,
|
|
jobs=ar.parse_jobs({"jobs": {"n": {"commits": [{
|
|
"subject": "chore: regenerate collection records",
|
|
"paths": ["database.json", "install/*.json"]}]}}}))
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(ar.run_job(ctx, "n"), (ar.OK, "dry run"))
|
|
lines = ar.describe(ctx.jobs["n"], ctx)
|
|
would = sorted(l.split("would commit ")[1] for l in lines if "would commit" in l)
|
|
self.assertEqual(would, ["database.json", "install/b.json"])
|
|
self.assertEqual(run_git(wt, "diff", "--cached", "--name-only"), "")
|
|
self.assertFalse(ctx.state_path.exists())
|
|
|
|
def test_a_commit_takes_its_paths_and_nothing_staged_beside(self):
|
|
wt = self.worktree()
|
|
(wt / "database.json").write_text('{"n": 1}\n', encoding="utf-8")
|
|
(wt / "other.txt").write_text("x", encoding="utf-8")
|
|
run_git(wt, "add", "other.txt")
|
|
rule = ar.CommitRule("chore: regenerate collection records", ("database.json",))
|
|
files = ar.plan_commit(ar.changed_files(wt), rule.paths)
|
|
ar.commit_paths(wt, files, rule)
|
|
shown = run_git(wt, "show", "--name-only", "--format=%s", "HEAD").split()
|
|
self.assertEqual(shown[-1], "database.json")
|
|
self.assertNotIn("other.txt", shown)
|
|
self.assertEqual(run_git(wt, "diff", "--cached", "--name-only").split(), ["other.txt"])
|
|
|
|
|
|
class Worktree(Origin):
|
|
def test_it_is_created_on_its_own_branch_and_reused(self):
|
|
wt = self.worktree()
|
|
self.assertEqual(run_git(wt, "branch", "--show-current").strip(), ar.BRANCH)
|
|
self.assertEqual(run_git(self.repo, "branch", "--show-current").strip(), "main")
|
|
ar.ensure_worktree(self.repo, wt)
|
|
self.assertEqual(len(run_git(self.repo, "worktree", "list").splitlines()), 2)
|
|
|
|
def test_it_follows_origin_and_stops_on_leftovers(self):
|
|
wt = self.worktree()
|
|
self.other_commit("x.txt")
|
|
base = ar.sync_worktree(wt)
|
|
self.assertTrue((wt / "x.txt").exists())
|
|
self.assertEqual(base, run_git(wt, "rev-parse", "HEAD").strip())
|
|
(wt / "database.json").write_text("changed", encoding="utf-8")
|
|
with self.assertRaises(ar.AutorunError) as caught:
|
|
ar.sync_worktree(wt)
|
|
self.assertEqual(caught.exception.result, ar.BLOCKED)
|
|
self.assertEqual((wt / "database.json").read_text(), "changed")
|
|
|
|
def test_the_hard_links_share_bytes_with_the_main_checkout(self):
|
|
(self.repo / ".gitignore").write_text(".cache/\ntmp/\nbios/big.bin\n")
|
|
(self.repo / "bios").mkdir()
|
|
(self.repo / "bios" / "big.bin").write_bytes(b"x" * 10)
|
|
(self.repo / ".cache" / "large").mkdir(parents=True)
|
|
(self.repo / ".cache" / "large" / "a.zip").write_bytes(b"z")
|
|
wt = self.worktree()
|
|
seeded = ar.seed_worktree(self.repo, wt)
|
|
self.assertEqual(sorted(seeded), [".cache/large", "bios/big.bin"])
|
|
self.assertTrue(os.path.samefile(wt / "bios" / "big.bin", self.repo / "bios" / "big.bin"))
|
|
self.assertEqual(ar.seed_worktree(self.repo, wt), [])
|
|
|
|
|
|
class Push(Origin):
|
|
def _commit(self, wt: Path) -> None:
|
|
(wt / "database.json").write_text('{"n": 2}\n', encoding="utf-8")
|
|
ar.commit_paths(wt, ["database.json"],
|
|
ar.CommitRule("chore: regenerate collection records", ("database.json",)))
|
|
|
|
def test_a_batch_on_an_unmoved_origin_pushes_without_a_second_gate(self):
|
|
wt = self.worktree()
|
|
base = ar.sync_worktree(wt)
|
|
self._commit(wt)
|
|
gates = []
|
|
pushed = ar.push_batch(wt, base, lambda: gates.append(1) or True)
|
|
self.assertEqual(gates, [])
|
|
self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed)
|
|
|
|
def test_a_moved_origin_rebases_and_gates_again(self):
|
|
wt = self.worktree()
|
|
base = ar.sync_worktree(wt)
|
|
self._commit(wt)
|
|
self.other_commit("y.txt")
|
|
gates = []
|
|
pushed = ar.push_batch(wt, base, lambda: gates.append(1) or True)
|
|
self.assertEqual(gates, [1])
|
|
log = run_git(self.origin, "log", "--format=%s", "main").splitlines()
|
|
self.assertEqual(log[:2], ["chore: regenerate collection records", "chore: add y.txt"])
|
|
self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed)
|
|
|
|
def test_a_failed_second_gate_pushes_nothing(self):
|
|
wt = self.worktree()
|
|
base = ar.sync_worktree(wt)
|
|
self._commit(wt)
|
|
self.other_commit("y.txt")
|
|
before = run_git(self.origin, "rev-parse", "main")
|
|
with self.assertRaises(ar.AutorunError):
|
|
ar.push_batch(wt, base, lambda: False)
|
|
self.assertEqual(run_git(self.origin, "rev-parse", "main"), before)
|
|
|
|
def test_a_conflicting_batch_is_kept_and_nothing_pushed(self):
|
|
wt = self.worktree()
|
|
base = ar.sync_worktree(wt)
|
|
self._commit(wt)
|
|
other = self.tmp / "other"
|
|
run_git(self.tmp, "clone", "--quiet", str(self.origin), str(other))
|
|
(other / "database.json").write_text('{"n": 3}\n', encoding="utf-8")
|
|
run_git(other, "commit", "--quiet", "-am", "chore: other database")
|
|
run_git(other, "push", "--quiet", "origin", "main")
|
|
mine = run_git(wt, "rev-parse", "HEAD")
|
|
with self.assertRaises(ar.AutorunError) as caught:
|
|
ar.push_batch(wt, base, lambda: True)
|
|
self.assertEqual(caught.exception.result, ar.BLOCKED)
|
|
self.assertEqual(run_git(wt, "rev-parse", "HEAD"), mine)
|
|
self.assertEqual(ar.changed_files(wt), [])
|
|
|
|
|
|
class RunnerPush(Origin):
|
|
def _ctx(self, value: int, push: bool) -> ar.Context:
|
|
write = "open('database.json', 'w').write('{\"n\": %d}')" % value
|
|
jobs = {"n": {"min_free_gb": 0, "steps": [[PY, "-c", write]], "commits": [
|
|
{"subject": "chore: regenerate collection records", "paths": ["database.json"]}]}}
|
|
return ar.Context(repo=self.repo, jobs=ar.parse_jobs({"jobs": jobs}),
|
|
lock_path=self.tmp / "auto.lock", host=FakeHost(),
|
|
gh=lambda _args: [], push=push)
|
|
|
|
def test_without_push_the_commit_stays_local(self):
|
|
before = run_git(self.origin, "rev-parse", "main")
|
|
result, detail = ar.run_job(self._ctx(5, push=False), "n")
|
|
self.assertEqual((result, detail), (ar.OK, "1 commits"))
|
|
self.assertEqual(run_git(self.origin, "rev-parse", "main"), before)
|
|
self.assertEqual(ar.unpushed(self.repo / ".cache" / "autorun" / "worktree"), 1)
|
|
|
|
def test_a_push_waits_for_the_previous_push_to_end_green(self):
|
|
ctx = self._ctx(5, push=True)
|
|
result, detail = ar.run_job(ctx, "n")
|
|
self.assertEqual(result, ar.OK)
|
|
self.assertIn("pushed", detail)
|
|
pushed = run_git(self.origin, "rev-parse", "main").strip()
|
|
self.assertEqual(self.state(ctx)["push"]["sha"], pushed)
|
|
result, detail = ar.run_job(self._ctx(6, push=True), "n")
|
|
self.assertEqual(result, ar.OK)
|
|
self.assertIn("push held", detail)
|
|
self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed)
|
|
|
|
|
|
class Units(Scratch):
|
|
def test_one_service_template_and_a_timer_per_scheduled_job(self):
|
|
jobs = ar.parse_jobs({"jobs": {
|
|
"nightly": {"schedule": "*-*-* 02:00"}, "manual": {}}})
|
|
units = ar.render_units(jobs, Path("/srv/retrobios"), "/usr/bin/python3")
|
|
self.assertEqual(sorted(units), ["retrobios-nightly.timer", "retrobios@.service"])
|
|
service = units["retrobios@.service"]
|
|
self.assertIn("ExecStart=/usr/bin/python3 /srv/retrobios/scripts/autorun.py run %i\n",
|
|
service)
|
|
self.assertIn("WorkingDirectory=/srv/retrobios\n", service)
|
|
self.assertIn("SuccessExitStatus=75\n", service)
|
|
timer = units["retrobios-nightly.timer"]
|
|
self.assertIn("OnCalendar=*-*-* 02:00\n", timer)
|
|
self.assertIn("Persistent=true\n", timer)
|
|
self.assertIn("Unit=retrobios@nightly.service\n", timer)
|
|
pushed = ar.render_units(jobs, Path("/r"), "py", push=True)["retrobios@.service"]
|
|
self.assertIn("run %i --push\n", pushed)
|
|
|
|
def test_units_are_written_and_enabled_only_when_asked(self):
|
|
units = ar.render_units(ar.load_jobs(), Path("/r"), "py")
|
|
calls = []
|
|
timers = ar.install_timers(self.tmp, units, False, lambda a: calls.append(a) or 0)
|
|
self.assertEqual(calls, [])
|
|
self.assertEqual(sorted(p.name for p in self.tmp.iterdir()), sorted(units))
|
|
ar.install_timers(self.tmp, units, True, lambda a: calls.append(a) or 0)
|
|
self.assertEqual(calls, [["daemon-reload"], ["enable", "--now", *timers]])
|
|
|
|
|
|
class Cleanup(Origin):
|
|
HOURS = 72
|
|
|
|
def _worktree(self, name: str, age_hours: float) -> Path:
|
|
path = self.repo / "sessions" / name
|
|
run_git(self.repo, "worktree", "add", "--quiet", "-b", f"wt-{name}", str(path))
|
|
stamp = time.time() - age_hours * 3600
|
|
os.utime(path, (stamp, stamp))
|
|
return path
|
|
|
|
def _decisions(self, busy=()) -> dict[str, cd.Decision]:
|
|
return {d.path.name: d for d in cd.plan(self.repo, busy, self.HOURS, time.time())}
|
|
|
|
def test_worktree_and_tmp_rules(self):
|
|
old_clean = self._worktree("old-clean", 100)
|
|
old_code = self._worktree("old-code", 100)
|
|
(old_code / "install" / "a.json").write_text("generated", encoding="utf-8")
|
|
(old_code / ".gitignore").write_text("edited", encoding="utf-8")
|
|
old_generated = self._worktree("old-generated", 100)
|
|
(old_generated / "database.json").write_text("generated", encoding="utf-8")
|
|
self._worktree("recent", 10)
|
|
busy = self._worktree("busy", 100)
|
|
tmp = self.repo / "tmp"
|
|
for name in ("acquire", "old", "fresh", "used"):
|
|
(tmp / name).mkdir(parents=True)
|
|
(tmp / "main.bundle").write_bytes(b"")
|
|
stamp = time.time() - 100 * 3600
|
|
for name in ("acquire", "old", "used", "main.bundle"):
|
|
os.utime(tmp / name, (stamp, stamp))
|
|
# The worktree mtimes moved when files were written into them.
|
|
for path in (old_clean, old_code, old_generated, busy):
|
|
os.utime(path, (stamp, stamp))
|
|
|
|
decisions = self._decisions(busy=[str(busy / "scripts"), str(tmp / "used" / "f")])
|
|
actions = {name: d.action for name, d in decisions.items()}
|
|
self.assertEqual(actions, {
|
|
"old-clean": "remove", "old-code": "keep", "old-generated": "remove",
|
|
"recent": "keep", "busy": "keep",
|
|
"acquire": "keep", "main.bundle": "keep", "old": "remove",
|
|
"fresh": "keep", "used": "keep",
|
|
})
|
|
self.assertIn("uncommitted code: 1 files", decisions["old-code"].reason)
|
|
|
|
failures = cd.apply(self.repo, decisions.values())
|
|
self.assertEqual(failures, [])
|
|
self.assertFalse(old_clean.exists())
|
|
self.assertFalse(old_generated.exists())
|
|
self.assertTrue(old_code.exists())
|
|
self.assertFalse((tmp / "old").exists())
|
|
self.assertTrue((tmp / "acquire").exists())
|
|
branches = run_git(self.repo, "branch", "--format=%(refname:short)").split()
|
|
self.assertIn("wt-old-clean", branches)
|
|
self.assertIn("wt-old-generated", branches)
|
|
|
|
def test_a_name_prefix_of_a_busy_path_is_not_in_use(self):
|
|
self.assertTrue(cd.in_use(Path("/r/tmp/a"), ["/r/tmp/a/x"]))
|
|
self.assertFalse(cd.in_use(Path("/r/tmp/a"), ["/r/tmp/a2/x"]))
|
|
|
|
def test_the_automation_worktree_is_never_a_candidate(self):
|
|
self.worktree()
|
|
self.assertEqual(cd.session_worktrees(self.repo), [])
|
|
|
|
def test_a_worktree_outside_the_checkout_is_never_a_candidate(self):
|
|
outside = self.tmp / "elsewhere"
|
|
run_git(self.repo, "worktree", "add", "--quiet", "-b", "away", str(outside))
|
|
self.assertEqual(cd.session_worktrees(self.repo), [])
|
|
|
|
def test_a_tmp_entry_holding_a_worktree_with_code_survives(self):
|
|
holder = self.repo / "tmp" / "holder"
|
|
holder.mkdir(parents=True)
|
|
work = holder / "wt"
|
|
run_git(self.repo, "worktree", "add", "--quiet", "-b", "held", str(work))
|
|
(work / ".gitignore").write_text("edited", encoding="utf-8")
|
|
stamp = time.time() - 100 * 3600
|
|
for path in (holder, work):
|
|
os.utime(path, (stamp, stamp))
|
|
decisions = self._decisions()
|
|
self.assertEqual((decisions["wt"].action, decisions["holder"].action), ("keep", "keep"))
|
|
self.assertEqual(decisions["holder"].reason, "holds a worktree")
|
|
cd.apply(self.repo, decisions.values())
|
|
self.assertEqual((work / ".gitignore").read_text(), "edited")
|
|
|
|
def test_low_space_raises_the_alert_line(self):
|
|
line, low = cd.disk_line(self.tmp, alert_gb=10 ** 9)
|
|
self.assertTrue(low)
|
|
self.assertIn("ALERT: free space under", line)
|
|
self.assertFalse(cd.disk_line(self.tmp, alert_gb=0)[1])
|
|
|
|
|
|
class AssetSync(Scratch):
|
|
GITIGNORE = "bios/A/big.bin\nbios/B/big.bin\nbios/C/solo.iso\n"
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
self.root = self.tmp
|
|
for rel, data in (("bios/A/big.bin", b"a"), ("bios/C/solo.iso", b"c")):
|
|
(self.root / rel).parent.mkdir(parents=True, exist_ok=True)
|
|
(self.root / rel).write_bytes(data)
|
|
self.db = {"files": {
|
|
sra.sha1_of(self.root / "bios/A/big.bin"): {"path": "bios/A/big.bin", "size": 1},
|
|
"0" * 40: {"path": "bios/C/solo.iso", "size": 1},
|
|
}}
|
|
|
|
def test_only_a_file_whose_sha1_matches_twice_is_cleared(self):
|
|
findings = [("missing", "bios/A/big.bin", 1, None), ("size", "bios/C/solo.iso", 1, 5),
|
|
("notes", "release description", 0, None)]
|
|
plans = sra.plan_uploads(self.root, self.db, self.GITIGNORE, findings)
|
|
self.assertEqual([(p.path, p.asset, p.verdict) for p in plans], [
|
|
("bios/A/big.bin", "A--big.bin", "upload"),
|
|
("bios/C/solo.iso", "solo.iso", "refused"),
|
|
])
|
|
self.assertIn("not the database's", plans[1].reason)
|
|
|
|
def test_two_reads_that_disagree_refuse_the_upload(self):
|
|
reads = iter(["1" * 40, "2" * 40])
|
|
plans = sra.plan_uploads(self.root, self.db, self.GITIGNORE,
|
|
[("content", "bios/A/big.bin", 1, 1)],
|
|
hasher=lambda _p: next(reads))
|
|
self.assertEqual(plans[0].verdict, "refused")
|
|
self.assertIn("two reads disagree", plans[0].reason)
|
|
|
|
def test_an_upload_goes_under_its_asset_name_and_is_tried_twice(self):
|
|
plans = [sra.Upload("bios/A/big.bin", "A--big.bin", "missing", "upload"),
|
|
sra.Upload("bios/C/solo.iso", "solo.iso", "size", "refused", "x")]
|
|
seen = []
|
|
|
|
def upload(path: Path) -> int:
|
|
seen.append(path.name)
|
|
return 1
|
|
self.assertEqual(sra.upload_all(self.root, plans, upload), ["bios/A/big.bin"])
|
|
self.assertEqual(seen, ["A--big.bin", "A--big.bin"])
|
|
self.assertFalse((self.root / sra.STAGING / "A--big.bin").exists())
|
|
|
|
|
|
class BiosIntegrity(Scratch):
|
|
"""A flipped bit keeps the size and the mtime, so git status shows a
|
|
clean tree. Only the rehash against database.json finds it."""
|
|
|
|
OLD = time.time() - 10 * 86400
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
self.repo = self.tmp / "repo"
|
|
self.repo.mkdir()
|
|
run_git(self.repo, "init", "--quiet", "-b", "main")
|
|
files = {
|
|
"bios/Aristocrat/aristmk6.zip": bytes(range(256)) * 4,
|
|
"bios/Sony/scph5501.bin": b"\x55" * 512,
|
|
"bios/Sony/edited.bin": b"\x01" * 64,
|
|
"bios/Big/large.iso": b"\x99" * 2048,
|
|
}
|
|
(self.repo / ".gitignore").write_text("bios/Big/large.iso\n", encoding="utf-8")
|
|
self.db = {"files": {}}
|
|
for rel, data in files.items():
|
|
path = self.repo / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
# An old mtime keeps git from re-reading the file as racily clean.
|
|
os.utime(path, (self.OLD, self.OLD))
|
|
sha1 = bib.sha1_of(path)
|
|
self.db["files"][sha1] = {"path": rel, "sha1": sha1}
|
|
run_git(self.repo, "add", ".gitignore", *[r for r in files if "Big" not in r])
|
|
run_git(self.repo, "commit", "--quiet", "-m", "chore: start")
|
|
|
|
def flip(self, rel: str, offset: int = 100) -> bytes:
|
|
"""Flip one bit in place, size and mtime unchanged."""
|
|
path = self.repo / rel
|
|
data = bytearray(path.read_bytes())
|
|
original = bytes(data)
|
|
data[offset] ^= 0x10
|
|
path.write_bytes(bytes(data))
|
|
os.utime(path, (self.OLD, self.OLD))
|
|
return original
|
|
|
|
def test_the_flip_is_invisible_to_git_and_restored(self):
|
|
rel = "bios/Aristocrat/aristmk6.zip"
|
|
original = self.flip(rel)
|
|
self.assertEqual(run_git(self.repo, "status", "--porcelain"), "")
|
|
report = bib.check(self.repo, self.db, apply=False)
|
|
self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.FLAGGED)])
|
|
self.assertNotEqual((self.repo / rel).read_bytes(), original)
|
|
|
|
report = bib.check(self.repo, self.db, apply=True)
|
|
self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.RESTORED)])
|
|
self.assertEqual((self.repo / rel).read_bytes(), original)
|
|
self.assertTrue(report.clean)
|
|
self.assertEqual(bib.check(self.repo, self.db, apply=True).findings, [])
|
|
|
|
def test_a_flip_in_a_gitignored_file_is_flagged_never_fixed(self):
|
|
rel = "bios/Big/large.iso"
|
|
self.flip(rel)
|
|
flipped = (self.repo / rel).read_bytes()
|
|
report = bib.check(self.repo, self.db, apply=True)
|
|
self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.FLAGGED)])
|
|
self.assertIn("gitignored", report.findings[0].reason)
|
|
self.assertEqual((self.repo / rel).read_bytes(), flipped)
|
|
self.assertFalse(report.clean)
|
|
|
|
def test_a_working_change_git_sees_is_left_to_its_session(self):
|
|
rel = "bios/Sony/edited.bin"
|
|
(self.repo / rel).write_bytes(b"\x02" * 80)
|
|
report = bib.check(self.repo, self.db, apply=True)
|
|
self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.SKIPPED)])
|
|
self.assertEqual((self.repo / rel).read_bytes(), b"\x02" * 80)
|
|
|
|
def test_a_mismatch_the_second_read_does_not_confirm_is_a_misread(self):
|
|
reads = {}
|
|
expected = {e["path"]: e["sha1"] for e in self.db["files"].values()}
|
|
|
|
def hasher(path: Path) -> str:
|
|
rel = path.relative_to(self.repo).as_posix()
|
|
reads[rel] = reads.get(rel, 0) + 1
|
|
if rel == "bios/Sony/scph5501.bin" and reads[rel] == 1:
|
|
return "f" * 40
|
|
return expected[rel]
|
|
report = bib.check(self.repo, self.db, apply=True, hasher=hasher)
|
|
self.assertEqual([(f.path, f.verdict) for f in report.findings],
|
|
[("bios/Sony/scph5501.bin", bib.MISREAD)])
|
|
self.assertTrue(report.clean)
|
|
|
|
def test_a_blob_that_is_not_the_database_content_is_not_written(self):
|
|
rel = "bios/Sony/scph5501.bin"
|
|
self.flip(rel)
|
|
db = json.loads(json.dumps(self.db))
|
|
for entry in db["files"].values():
|
|
if entry["path"] == rel:
|
|
entry["sha1"] = "0" * 40
|
|
flipped = (self.repo / rel).read_bytes()
|
|
report = bib.check(self.repo, db, apply=True)
|
|
self.assertEqual(report.findings[0].verdict, bib.FLAGGED)
|
|
self.assertIn("other bytes", report.findings[0].reason)
|
|
self.assertEqual((self.repo / rel).read_bytes(), flipped)
|
|
|
|
def test_the_runner_keeps_the_report_in_the_job_status(self):
|
|
rel = "bios/Aristocrat/aristmk6.zip"
|
|
self.flip(rel)
|
|
Path(self.db_path).write_text(json.dumps(self.db), encoding="utf-8")
|
|
script = str(REPO_ROOT / "scripts" / "bios_integrity.py")
|
|
ctx = ar.Context(
|
|
repo=self.repo, lock_path=self.tmp / "auto.lock", host=FakeHost(),
|
|
jobs=ar.parse_jobs({"jobs": {"bios-integrity": {
|
|
"worktree": False, "min_free_gb": 0,
|
|
"steps": [{"run": [PY, script, "--apply", "--json", "--db", self.db_path],
|
|
"stdout": "tmp/autorun/bios-integrity.json"}],
|
|
"report": "tmp/autorun/bios-integrity.json"}}}))
|
|
self.assertEqual(ar.run_job(ctx, "bios-integrity")[0], ar.OK)
|
|
report = self.state(ctx)["jobs"]["bios-integrity"]["report"]
|
|
self.assertEqual(report["counts"], {"restored": 1})
|
|
self.assertEqual(report["findings"][0]["path"], rel)
|
|
self.assertIn("1 restored", ar.render_status(ar.status(ctx)))
|
|
|
|
@property
|
|
def db_path(self) -> str:
|
|
return str(self.tmp / "database.json")
|
|
|
|
|
|
class SubjectsFollowTheRepositoryRules(unittest.TestCase):
|
|
"""The commit rules live in scripts/commit_rules.py, which the local hook
|
|
calls: a copy in the hook alone let CI check only the shape."""
|
|
|
|
def test_the_runner_reads_the_tracked_rules(self):
|
|
import commit_rules # noqa: PLC0415
|
|
|
|
for subject in ("chore: fix the broken manifests", "update things", "chore: " + "a" * 60):
|
|
with self.subTest(subject=subject):
|
|
self.assertEqual(ar.check_subject(subject), commit_rules.problems(subject))
|
|
self.assertTrue(ar.check_subject(subject))
|
|
self.assertEqual(ar.check_subject("chore: tidy the cache"), [])
|
|
|
|
def test_the_rules_run_as_a_hook(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
message = Path(tmp) / "MSG"
|
|
script = REPO_ROOT / "scripts" / "commit_rules.py"
|
|
for text, code in (("fix: keep the cache\n", 0), ("fix: a broken cache\n", 1),
|
|
("fix: keep it\n\nCo-Authored-By: x\n", 1)):
|
|
message.write_text(text)
|
|
done = subprocess.run([sys.executable, str(script), str(message)],
|
|
capture_output=True, text=True, check=False)
|
|
with self.subTest(text=text):
|
|
self.assertEqual(done.returncode, code, done.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|