Kevin Veen-BirkenbachandClaude Opus 5.5
7036488766
The pkgmgr wrappers replay renamed kvb-git-branch-gc: branch cleanup outside worktrees is now kvb-git-branch-cleanup, and removing merged branches with their worktrees is kvb-git-gc. The hub's held-by-worktree hint, the cleanup dialog and its test now name kvb-git-gc. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
363 lines
25 KiB
Python
363 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import secrets
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from hub_case import HubCase
|
|
from hub_support import GIT_ENV, 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)
|
|
INSTALLED_PKGMGR = shutil.which("pkgmgr")
|
|
DONE_TIP = "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4"
|
|
OTHER_TIP = "1111111111111111111111111111111111111111"
|
|
# Param: format with python=<interpreter>; parses argv as pkgmgr's branch_cmd.py does (argparse exits 2 on any other flag) and prints the plan as gc.py _render does, {"host_pid_namespace", "repos": [{"repo", "default", "merged_into", "warnings", "items"}]}; plan.json beside it is the plan's items, next-tree.json the rewritten tree, outcome.txt the result --apply reports per item (done without it), exit 0 also for a skipped item.
|
|
FAKE_PKGMGR = """#!{python}
|
|
import argparse
|
|
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")
|
|
parser = argparse.ArgumentParser(prog="pkgmgr")
|
|
branch = parser.add_subparsers(dest="command", required=True).add_parser("branch").add_subparsers(dest="subcommand", required=True)
|
|
tree = branch.add_parser("tree")
|
|
tree.add_argument("--format", choices=("json", "yaml"), default="json")
|
|
tree.add_argument("--output")
|
|
cleanup = branch.add_parser("cleanup")
|
|
for flag in ("--apply", "--json", "--all", "--sizes"):
|
|
cleanup.add_argument(flag, action="store_true")
|
|
for option in ("--branch", "--exclude", "--merged-into"):
|
|
cleanup.add_argument(option, action="append", default=[])
|
|
cleanup.add_argument("--min-age", type=float)
|
|
args = parser.parse_args()
|
|
if args.subcommand == "tree":
|
|
following = here / "next-tree.json"
|
|
rewritten = following.read_text(encoding="utf-8") if following.exists() else json.dumps({{"rewritten": True}})
|
|
Path(args.output).write_text(rewritten, encoding="utf-8")
|
|
raise SystemExit(0)
|
|
items = json.loads((here / "plan.json").read_text(encoding="utf-8"))
|
|
if args.apply:
|
|
outcome = here / "outcome.txt"
|
|
result = outcome.read_text(encoding="utf-8") if outcome.exists() else "done"
|
|
items = [{{**item, "result": result}} for item in items]
|
|
plan = {{"host_pid_namespace": True, "repos": [{{"repo": os.getcwd(), "default": "origin/main", "merged_into": [], "warnings": [], "items": items}}]}}
|
|
print(json.dumps(plan) if args.json else "\\n".join(str(item["branch"]) for item in items))
|
|
"""
|
|
|
|
|
|
class TestHubCleanup(HubCase):
|
|
def ready(self, ready: bool, held: bool) -> dict:
|
|
"""Args:
|
|
ready: delete_ready of the tree file.
|
|
held: keep feature/done in its worktree; otherwise it is a local
|
|
branch without one.
|
|
"""
|
|
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"
|
|
done = next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/done")
|
|
if not held:
|
|
done.update(worktree=None, worktree_state=None, actions=["local-branch", "remote-branch"], compose_projects=[], images=[])
|
|
self.tree_file.write_text(json.dumps(data), encoding="utf-8")
|
|
return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", done["worktree"] or "", 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, False)
|
|
branches = self.branches()
|
|
done = branches["feature/done"]
|
|
self.assertEqual((done["cleanup"], done["held"]), (item, ""))
|
|
self.assertEqual(done["plan"], [
|
|
"feature/done @ 5b1e6fd3a7c0: merged-clean, merged by #12",
|
|
"deletes the local branch feature/done",
|
|
"stays: remote-branch (local only; the host refuses a plan that deletes it)",
|
|
])
|
|
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, False)
|
|
status, body = self.sign([item])
|
|
self.assertEqual((status, body.decode()), (400, "delete blocked: Secret Service locked\n"))
|
|
item = self.ready(True, False)
|
|
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_branch_held_by_a_worktree_offers_no_delete_and_names_branch_gc(self):
|
|
item = self.ready(True, True)
|
|
done = self.branches()["feature/done"]
|
|
held = f"held by worktree {item['worktree']}: pkgmgr branch cleanup skips it, use branch gc (kvb-git-gc, pkgmgr gc)"
|
|
self.assertEqual((done["cleanup"], done["held"], done["plan"][1]), (None, held, held))
|
|
status, body = self.sign([item])
|
|
self.assertEqual((status, body.decode()), (400, f"feature/done in {item['repository']} is no merged-clean branch without skip reason at that tip in the tree file\n"))
|
|
page = (HUB / "templates" / "page-tree.js").read_text(encoding="utf-8")
|
|
self.assertIn('if (entry.held) {\n return h("span", {class: "hint held", title: entry.held}, chip("bi-folder-symlink", entry.held, "neutral"));', page)
|
|
self.assertIn("deleteDialog(items, locked, held)", page)
|
|
self.assertIn("held by a worktree: not deleted here", (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8"))
|
|
|
|
@unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
|
|
def test_every_volume_text_states_the_compose_down_argv_of_the_installed_pkgmgr(self):
|
|
found = subprocess.run([str(Path(INSTALLED_PKGMGR).parent / "python"), "-I", "-c", "import pkgmgr.actions.branch.gc as gc; print(gc.__file__)"], capture_output=True, text=True, timeout=60, check=True)
|
|
calls = [node for node in ast.walk(ast.parse(Path(found.stdout.strip()).read_text(encoding="utf-8"))) if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "_docker" and node.args and isinstance(node.args[0], ast.List)]
|
|
downs = [["docker", *(part.value if isinstance(part, ast.Constant) else "{project}" for part in call.args[0].elts)] for call in calls if any(isinstance(part, ast.Constant) and part.value == "down" for part in call.args[0].elts)]
|
|
self.assertEqual(len(downs), 1, downs)
|
|
command = " ".join(downs[0]).format(project="demo-feature-done")
|
|
self.ready(True, True)
|
|
plan = self.branches()["feature/done"]["plan"]
|
|
self.assertIn(f"branch gc would run {command}, which deletes the volumes of demo-feature-done", plan)
|
|
self.assertIn("-v", downs[0])
|
|
dialog = (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8")
|
|
self.assertIn("never touches a worktree, Compose stack or volume", dialog)
|
|
self.assertIn("its docker compose down -v deletes the stack's volumes", dialog)
|
|
texts = [*plan, dialog, (HUB / "kvb_orchestrator" / "tree.py").read_text(encoding="utf-8")]
|
|
self.assertFalse([text for text in texts if "without -v" in text])
|
|
|
|
def test_a_signed_delete_becomes_a_job_without_any_command(self):
|
|
item = self.ready(True, False)
|
|
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({"delete_ready": True, "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", "--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_an_apply_that_exits_0_but_skips_or_reports_no_result_is_failed_with_pkgmgrs_reason_and_keeps_the_tree(self):
|
|
outcome = self.fake.parent / "outcome.txt"
|
|
before = self.tree.read_text(encoding="utf-8")
|
|
cases = (("skipped", "skipped: changed since plan (dirty: 1 paths)", "skipped: changed since plan (dirty: 1 paths)"), ("no result", "", "pkgmgr reported no result for feature/done"))
|
|
for name, result, reason in cases:
|
|
with self.subTest(name=name):
|
|
outcome.write_text(result, encoding="utf-8")
|
|
done, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
|
|
self.assertEqual([(entry["status"], entry["reason"], entry["code"]) for entry in done["items"]], [("failed", reason, 0)])
|
|
self.assertEqual([call["argv"][:3] for call in calls], [["branch", "cleanup", "--json"], ["branch", "cleanup", "--apply"]])
|
|
self.assertNotIn("tree", done)
|
|
self.assertEqual(self.tree.read_text(encoding="utf-8"), before)
|
|
outcome.write_text("done; image kept: demo:feature-done", encoding="utf-8")
|
|
done, _ = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
|
|
self.assertEqual([(entry["status"], entry["reason"]) for entry in done["items"]], [("done", "")])
|
|
|
|
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),
|
|
("remote branch delete", [self.item], self.proof([self.item], self.signer), [self.entry(actions=["local-branch", "remote-branch:origin"])], "a signed delete is local only", 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", "--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)
|
|
|
|
|
|
@unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
|
|
class TestInstalledPkgmgrContract(unittest.TestCase):
|
|
def git(self, cwd: Path, *args: str) -> str:
|
|
return subprocess.run(["git", *args], cwd=cwd, env={**os.environ, **GIT_ENV}, check=True, capture_output=True, text=True).stdout.strip()
|
|
|
|
def test_the_installed_pkgmgr_takes_the_runners_dry_run_argv_and_prints_a_plan_the_runner_parses_and_accepts(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
base = Path(tmp)
|
|
origin, repo = base / "origin.git", base / "repo"
|
|
self.git(base, "init", "-q", "--bare", "-b", "main", str(origin))
|
|
self.git(base, "init", "-q", "-b", "main", str(repo))
|
|
self.git(repo, "remote", "add", "origin", origin.as_uri())
|
|
self.git(repo, "commit", "-q", "--allow-empty", "-m", "one")
|
|
self.git(repo, "checkout", "-q", "-b", "side")
|
|
self.git(repo, "commit", "-q", "--allow-empty", "-m", "side")
|
|
tip = self.git(repo, "rev-parse", "HEAD")
|
|
self.git(repo, "checkout", "-q", "main")
|
|
self.git(repo, "merge", "-q", "--no-ff", "-m", "merge side", "side")
|
|
self.git(origin, "fetch", "-q", str(repo), "main:main")
|
|
self.git(repo, "fetch", "-q", "origin")
|
|
self.git(repo, "remote", "set-head", "origin", "main")
|
|
code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
|
|
self.assertEqual(code, 0, stderr)
|
|
plan = host.plan_items(stdout)
|
|
self.assertIsNotNone(plan, stdout[:500])
|
|
self.assertEqual([(entry["branch"], entry["state"], entry["merged_by"], entry["merged_into"], entry["skip"], entry["actions"]) for entry in plan], [("side", "merged-clean", "ancestor", "origin/main", None, ["local-branch"])])
|
|
self.assertIsNone(host.mismatch(plan[0], {"tip": tip, "worktree": ""}))
|
|
self.assertTrue(signing.token_free(plan[0], "main"))
|
|
self.assertEqual(self.git(repo, "rev-parse", "refs/heads/side"), tip)
|
|
held = base / "held"
|
|
self.git(repo, "worktree", "add", "-q", str(held), "side")
|
|
code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
|
|
self.assertEqual(code, 0, stderr)
|
|
plan = host.plan_items(stdout)
|
|
self.assertEqual([(entry["worktree"], entry["actions"], bool(entry["skip"])) for entry in plan], [(os.path.realpath(held), [], True)], plan)
|
|
self.assertIn("gc skip reason", host.mismatch(plan[0], {"tip": tip, "worktree": os.path.realpath(held)}))
|
|
|
|
|
|
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()
|