skills/tests/test_hub_cleanup.py

from __future__ import annotations

import importlib.util
import json
import secrets
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path

from hub_case import HubCase
from hub_support import HUB, ORIGIN, key, signing, verifier

CLEANUP = HUB / "client" / "cleanup.py"
CLEANUP_SPEC = importlib.util.spec_from_file_location("kvb_orchestrator_cleanup_runner", CLEANUP)
host = importlib.util.module_from_spec(CLEANUP_SPEC)
CLEANUP_SPEC.loader.exec_module(host)
DONE_TIP = "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4"
OTHER_TIP = "1111111111111111111111111111111111111111"
# Param: format with python=<interpreter>; plan.json beside it is the dry-run plan, next-tree.json the rewritten tree.
FAKE_PKGMGR = """#!{python}
import json
import os
import sys
from pathlib import Path

here = Path(__file__).parent
with (here / "calls.jsonl").open("a", encoding="utf-8") as log:
    log.write(json.dumps({{"argv": sys.argv[1:], "cwd": os.getcwd()}}) + "\\n")
if sys.argv[1:3] == ["branch", "tree"]:
    following = here / "next-tree.json"
    rewritten = following.read_text(encoding="utf-8") if following.exists() else json.dumps({{"rewritten": True}})
    Path(sys.argv[sys.argv.index("--output") + 1]).write_text(rewritten, encoding="utf-8")
    raise SystemExit(0)
print(json.dumps({{"items": json.loads((here / "plan.json").read_text(encoding="utf-8"))}}))
"""


class TestHubCleanup(HubCase):
    def ready(self, ready: bool) -> dict:
        root = self.write_tree()
        data = json.loads(self.tree_file.read_text(encoding="utf-8"))
        data["delete_ready"], data["delete_reason"] = ready, "" if ready else "Secret Service locked"
        self.tree_file.write_text(json.dumps(data), encoding="utf-8")
        return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", f"{root}/worktrees/github.com/me/demo/feature-done", DONE_TIP)

    def branches(self) -> dict:
        status, _, body = self.get("/api/tree")
        self.assertEqual(status, 200, body)
        demo = next(repo for repo in json.loads(body)["repos"] if repo["name"] == "github.com/me/demo")
        return {entry["name"]: entry for entry in demo["branches"]}

    def sign(self, items: list[dict]) -> tuple[int, bytes]:
        status, _, body = self.post_json("/api/sign", {"items": items}, self.guard(self.host))
        return status, body

    def test_the_tree_offers_a_delete_only_for_merged_clean_branches_without_skip_reason(self):
        item = self.ready(False)
        branches = self.branches()
        done = branches["feature/done"]
        self.assertEqual(done["cleanup"], item)
        self.assertEqual(done["plan"], [
            "feature/done @ 5b1e6fd3a7c0: merged-clean, gemergt durch #12",
            f"entfernt den Worktree {item['worktree']}",
            "löscht den lokalen Branch feature/done",
            "bleibt: remote-branch (nur lokal)",
            "stoppt den Compose-Stack demo-feature-done ohne -v",
            "entfernt das Image demo:feature-done",
        ])
        self.assertEqual((done["merged_by"], done["merged_into"], done["gone"]), ("#12", "main", True))
        self.assertEqual({name: entry["cleanup"] for name, entry in branches.items() if name != "feature/done"}, {"feature": None, "feature/kept": None, "main": None})
        self.assertEqual(branches["feature/kept"]["skip"], "ignored files beyond known caches: notes.txt")

    def test_signing_a_delete_needs_delete_ready_and_exactly_an_offered_item(self):
        item = self.ready(False)
        status, body = self.sign([item])
        self.assertEqual((status, body.decode()), (400, "löschen gesperrt: Secret Service locked\n"))
        item = self.ready(True)
        kept = {**item, "branch": "feature/kept", "worktree": "", "tip": "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d"}
        for name, items in (("changed tip", [{**item, "tip": OTHER_TIP}]), ("skip reason", [kept]), ("pattern", [{**item, "branch": "feature/*"}]), ("twice", [item, item]), ("extra key", [{**item, "command": "rm -rf /"}])):
            with self.subTest(name=name):
                self.assertEqual(self.sign(items)[0], 400)
        status, body = self.sign([item])
        self.assertEqual(status, 200, body)
        self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(item)])

    def test_a_signed_delete_becomes_a_job_without_any_command(self):
        item = self.ready(True)
        signer = self.pin()
        proof = signer.proof(self.prepare([item])["payload"], self.origin)
        self.assertEqual(self.post_json("/api/cleanup", {"items": [item], "proof": None}, self.guard(self.host))[0], 400)
        self.assertEqual(self.post_json("/api/cleanup", {"items": [item], "proof": proof}, {"Host": self.host})[0], 403)
        status, _, body = self.post_json("/api/cleanup", {"items": [item], "proof": proof}, self.guard(self.host))
        self.assertEqual(status, 202, body)
        job_id = json.loads(body)["id"]
        written = self.cleanup / "jobs" / f"{job_id}.json"
        job = json.loads(written.read_text(encoding="utf-8"))
        self.assertEqual(sorted(job), ["created", "id", "items", "proof", "v"])
        self.assertEqual(job["items"], [{key_name: item[key_name] for key_name in ("repository", "branch", "worktree", "tip")}])
        self.assertEqual(job["proof"], proof)
        self.assertNotIn("command", written.read_text(encoding="utf-8"))
        self.assertEqual((written.stat().st_mode & 0o777, (self.cleanup / "jobs").stat().st_mode & 0o777), (0o600, 0o700))
        self.assertEqual(json.loads(self.get(f"/api/cleanup/{job_id}")[2])["status"], "queued")
        self.assertEqual(self.get("/api/cleanup/0000000000000000")[0], 404)
        result = {"id": job_id, "finished": time.time(), "items": [{**job["items"][0], "status": "done", "reason": ""}]}
        (self.cleanup / "results" / f"{job_id}.json").write_text(json.dumps(result), encoding="utf-8")
        self.assertEqual(json.loads(self.get(f"/api/cleanup/{job_id}")[2])["status"], "done")
        self.assertEqual(json.loads(self.get("/api/tree")[2])["cleanups"], [result])


class TestHostCleanup(unittest.TestCase):
    def setUp(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        base = Path(tmp.name)
        self.repo = base / "repos" / "demo"
        self.repo.mkdir(parents=True)
        self.jobs, self.results = base / "jobs", base / "results"
        self.jobs.mkdir()
        self.results.mkdir()
        self.fake = base / "bin" / "pkgmgr"
        self.fake.parent.mkdir()
        self.fake.write_text(FAKE_PKGMGR.format(python=sys.executable), encoding="utf-8")
        self.fake.chmod(0o755)
        self.calls = self.fake.parent / "calls.jsonl"
        self.tree = base / "tree" / "branch-tree.json"
        self.tree.parent.mkdir()
        self.tree.write_text(json.dumps({"repos": [{"path": str(self.repo), "known": True}, {"path": str(base / "stranger"), "known": False}]}), encoding="utf-8")
        self.approvers = base / "config" / "approvers.json"
        self.trust = verifier.Trust(approvers=self.approvers, replay=base / "replay" / "cleanup.json", origin=ORIGIN)
        self.signer = key(-7, "main")
        self.signer.pin(self.approvers, ORIGIN)
        self.item = {"repository": str(self.repo), "branch": "feature/done", "worktree": f"{self.repo}/.worktrees/feature-done", "tip": DONE_TIP}

    def proof(self, items: list[dict], signer) -> dict:
        hashes = [signing.item_hash(signing.cleanup_item(**item)) for item in items]
        return signer.proof(signing.approval_payload(hashes, secrets.token_hex(16), int(time.time()) + 600), ORIGIN)

    def entry(self, **changes: object) -> dict:
        return {"branch": self.item["branch"], "worktree": self.item["worktree"], "local_sha": DONE_TIP, "head": DONE_TIP, "state": "merged-clean", "skip": None, **changes}

    def run_job(self, items: list[dict], proof: object, plan: list[dict]) -> tuple[dict, list[dict]]:
        (self.fake.parent / "plan.json").write_text(json.dumps(plan), encoding="utf-8")
        job_id = secrets.token_hex(8)
        (self.jobs / f"{job_id}.json").write_text(json.dumps({"v": 1, "id": job_id, "created": "x", "items": items, "proof": proof}), encoding="utf-8")
        self.assertEqual(host.cycle(self.jobs, self.results, host.Cleanup(self.trust, self.tree, self.fake)), 1)
        self.assertEqual(list(self.jobs.iterdir()), [])
        calls = [json.loads(line) for line in self.calls.read_text(encoding="utf-8").splitlines()] if self.calls.exists() else []
        self.calls.unlink(missing_ok=True)
        return json.loads((self.results / f"{job_id}.json").read_text(encoding="utf-8")), calls

    def test_a_signed_merged_clean_item_is_dry_run_then_deleted_by_exact_name(self):
        result, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
        self.assertEqual([(entry["status"], entry["reason"]) for entry in result["items"]], [("done", "")])
        selector = ["--json", "--local", "--branch", "feature/done"]
        self.assertEqual(calls, [
            {"argv": ["branch", "cleanup", *selector], "cwd": str(self.repo)},
            {"argv": ["branch", "cleanup", "--apply", *selector], "cwd": str(self.repo)},
            {"argv": ["branch", "tree", "--format", "json", "--output", str(self.tree)], "cwd": str(self.tree.parent)},
        ])
        self.assertEqual(json.loads(self.tree.read_text(encoding="utf-8")), {"rewritten": True})
        self.assertEqual(result["tree"], {"code": 0, "error": ""})

    def test_every_doubtful_item_is_refused_and_changes_nothing(self):
        pattern = {**self.item, "branch": "feature/*"}
        option = {**self.item, "branch": "--all"}
        stranger = {**self.item, "repository": str(self.repo.parent.parent / "stranger")}
        rule = self.signer.proof(signing.rule_payload(secrets.token_hex(8), ["review"], "", "*", "*", int(time.time()) + 600), ORIGIN)
        cases = (
            ("bad signature", [self.item], self.proof([self.item], key(-7, "intruder")), [self.entry()], "no valid operator signature", 0),
            ("missing signature", [self.item], None, [self.entry()], "no signature", 0),
            ("rule instead of approval", [self.item], rule, [self.entry()], "no valid operator signature", 0),
            ("other item signed", [self.item], self.proof([{**self.item, "tip": OTHER_TIP}], self.signer), [self.entry()], "does not list this item", 0),
            ("unknown repository", [stranger], self.proof([stranger], self.signer), [self.entry()], "unknown repository", 0),
            ("pattern characters", [pattern], self.proof([pattern], self.signer), [self.entry(branch="feature/*")], "pattern or option characters", 0),
            ("option name", [option], self.proof([option], self.signer), [self.entry(branch="--all")], "pattern or option characters", 0),
            ("changed tip", [self.item], self.proof([self.item], self.signer), [self.entry(local_sha=OTHER_TIP)], "tip changed", 1),
            ("not merged-clean", [self.item], self.proof([self.item], self.signer), [self.entry(state="dirty")], "state is dirty", 1),
            ("skip reason", [self.item], self.proof([self.item], self.signer), [self.entry(skip="ignored files")], "gc skip reason", 1),
            ("other worktree", [self.item], self.proof([self.item], self.signer), [self.entry(worktree="/elsewhere")], "worktree changed", 1),
            ("pattern match in pkgmgr", [self.item], self.proof([self.item], self.signer), [self.entry(), self.entry(branch="feature/done-too")], "other branches too", 1),
        )
        for name, items, proof, plan, reason, dry_runs in cases:
            with self.subTest(name=name):
                result, calls = self.run_job(items, proof, plan)
                self.assertEqual([entry["status"] for entry in result["items"]], ["refused"])
                self.assertIn(reason, result["items"][0]["reason"])
                self.assertEqual([call["argv"] for call in calls], [["branch", "cleanup", "--json", "--local", "--branch", items[0]["branch"]]] * dry_runs)
                self.assertNotIn("tree", result)

    def test_a_signature_deletes_once_and_a_malformed_job_nothing(self):
        proof = self.proof([self.item], self.signer)
        self.assertEqual(self.run_job([self.item], proof, [self.entry()])[0]["items"][0]["status"], "done")
        again, calls = self.run_job([self.item], proof, [self.entry()])
        self.assertEqual((again["items"][0]["status"], calls), ("refused", []))
        self.assertIn("already used", again["items"][0]["reason"])
        for name, items in (("command smuggled in", [{**self.item, "command": "rm -rf ~"}]), ("relative repository", [{**self.item, "repository": "demo"}]), ("no items", [])):
            with self.subTest(name=name):
                result, calls = self.run_job(items, proof, [self.entry()])
                self.assertEqual((result["items"], calls), ([], []))
                self.assertIn("malformed job", result["error"])

    def test_the_unit_runs_the_script_once_over_the_job_directory(self):
        (self.fake.parent / "plan.json").write_text(json.dumps([self.entry()]), encoding="utf-8")
        job_id = secrets.token_hex(8)
        (self.jobs / f"{job_id}.json").write_text(json.dumps({"v": 1, "id": job_id, "created": "x", "items": [self.item], "proof": self.proof([self.item], self.signer)}), encoding="utf-8")
        (self.jobs / "stray.tmp").write_text("x", encoding="utf-8")
        trust = ["--approvers", str(self.approvers), "--replay", str(self.trust.replay)]
        command = [sys.executable, str(CLEANUP), "--port", ORIGIN.rsplit(":", 1)[1], "--jobs", str(self.jobs), "--results", str(self.results), *trust, "--tree", str(self.tree), "--pkgmgr", str(self.fake)]
        done = subprocess.run(command, capture_output=True, text=True, timeout=60, check=False)
        self.assertEqual(done.returncode, 0, done.stderr)
        self.assertEqual(list(self.jobs.iterdir()), [])
        result = json.loads((self.results / f"{job_id}.json").read_text(encoding="utf-8"))
        self.assertEqual(result["items"][0]["status"], "done")
        self.assertEqual((self.results / f"{job_id}.json").stat().st_mode & 0o777, 0o600)


class TestHostCleanupGitConfig(unittest.TestCase):
    def test_pkgmgr_runs_no_hook_the_repository_config_names(self):
        with tempfile.TemporaryDirectory() as tmp:
            base = Path(tmp)
            repo, hooks, marker = base / "repo", base / "hooks", base / "hook-ran"
            hooks.mkdir()
            (hooks / "reference-transaction").write_text(f"#!/bin/sh\necho path >> {marker}\n", encoding="utf-8")
            (hooks / "reference-transaction").chmod(0o755)
            git = ["git", "-c", "user.name=t", "-c", "user.email=t@example.invalid", "-c", "commit.gpgsign=false"]
            subprocess.run([*git, "init", "-q", str(repo)], check=True)
            subprocess.run([*git, "commit", "-q", "--allow-empty", "-m", "x"], cwd=repo, check=True)
            subprocess.run(["git", "branch", "control"], cwd=repo, check=True)
            subprocess.run(["git", "branch", "doomed"], cwd=repo, check=True)
            subprocess.run(["git", "config", "core.hooksPath", str(hooks)], cwd=repo, check=True)
            subprocess.run(["git", "config", "hook.spy.event", "reference-transaction"], cwd=repo, check=True)
            subprocess.run(["git", "config", "hook.spy.command", f"echo config >> {marker}"], cwd=repo, check=True)
            subprocess.run(["git", "branch", "-D", "control"], cwd=repo, check=True, capture_output=True)
            self.assertEqual({line.split()[0] for line in marker.read_text(encoding="utf-8").splitlines()}, {"path", "config"})
            marker.unlink()
            fake = base / "pkgmgr"
            fake.write_text(f"#!{sys.executable}\nimport subprocess\nsubprocess.run(['git', 'branch', '-D', '--', 'doomed'], check=True)\n", encoding="utf-8")
            fake.chmod(0o755)
            code, _, error = host.run(fake, ["branch", "cleanup", "--apply"], str(repo))
            self.assertEqual(code, 0, error)
            self.assertFalse(marker.exists())
            branches = subprocess.run(["git", "branch", "--format=%(refname:short)"], cwd=repo, capture_output=True, text=True, check=True).stdout.split()
            self.assertNotIn("doomed", branches)


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