libretro/tests/test_autorun.py

"""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 SubjectsFollowTheRepositoryHook(unittest.TestCase):
    """The vocabulary rules live in the commit-msg hook alone: a copy in the
    runner put the very words the history refuses into a tracked file."""

    def test_a_refusal_comes_from_the_hook(self):
        with tempfile.TemporaryDirectory() as tmp:
            hook = Path(tmp) / "commit-msg"
            hook.write_text('#!/bin/sh\ngrep -q forbiddenword "$1" && echo refused && exit 1\nexit 0\n')
            hook.chmod(0o755)
            real = subprocess.run
            def fake(argv, *args, **kwargs):
                if argv[:3] == ["git", "rev-parse", "--git-path"]:
                    return subprocess.CompletedProcess(argv, 0, stdout=str(hook) + "\n", stderr="")
                return real(argv, *args, **kwargs)
            with unittest.mock.patch.object(ar.subprocess, "run", side_effect=fake):
                self.assertEqual(ar.check_subject("chore: tidy the cache"), [])
                self.assertEqual(ar.check_subject("chore: add forbiddenword"), ["refused"])

    def test_the_shape_is_checked_without_a_hook(self):
        problems = ar.check_subject("not a subject that follows the type rule at all, too long")
        self.assertTrue(any("50 at most" in p for p in problems))
        self.assertTrue(any("type: description" in p for p in problems))


if __name__ == "__main__":
    unittest.main()