diff --git a/scripts/README.md b/scripts/README.md index 36c8dc99..409918e5 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -239,7 +239,7 @@ Every failure is a stop rather than a fallback, because each alternative closes What this trades away is stated rather than glossed. A mutation spelled as a `gh` command in a shell is read by the `gh-write-guard` hook and one this script performs is not. The script removes that failure at its source instead: every node ID comes from a live query in the same run. The owner check is also enforced in-process. The whole-source test guards against every other state-changing call and counts each reviewed mutation document. -`cleanup` deletes the head branch of a merged pull request, the verify-then-delete `drive-pr` and `merge-and-release` prescribe after a merge into `develop`. It is the one place holding the delete's checks, and no skill prescribes a hand-typed delete beside it. A delete typed by hand cannot be relied on to run from an agent seat in auto mode. The `gh-write-guard` hook denies `git push origin --delete` from a primary checkout, and Claude Code's auto-mode classifier refuses the ref API delete as a destructive git operation on some days and allows it on others. A command matching a narrow `permissions.allow` rule is settled before that classifier runs, and the host-safety installer already registers `Bash(python3 scripts/pr_review.py:*)`, so the delete runs here under that rule with no settings change. That rule approves any arguments, so every check lives in the subcommand. It takes the pull request number and `--repo`, never a branch name, and reads the branch from the pull request. It refuses unless the pull request is merged, its base is `develop`, its head repository is `--repo` itself, and the head branch is neither `develop`, `main`, nor the default branch read live. Past the same owner check `comment` and `reply` make, `handoff.py`'s own write-scope check, imported rather than copied, binds the owner to `registry/repos.json` and admits a repository only where it is registered or a fork. A branch any ruleset or branch protection covers is refused, since the hook's per-branch rule check never sees a delete made from inside this script. A branch that another open pull request in `--repo` has as its head or base is refused too, since the delete would close or retarget it. Every open pull request in `--repo` with the branch as its head is named, read page by page. A page that does not answer stops at exit `2`. Up to twenty with the branch as base are named, and any past that are counted. The ref is read by its exact qualified name, and the delete is sent only where that read's tip is the pull request's `headRefOid`, by the node id the same read returned. The delete carries no expected tip of its own, so this is verify-then-delete rather than a compare-and-swap. The delete is confirmed by reading the ref back rather than by the mutation's own answer, so a delete that reports a failure and landed still reads as done, and one that reports success and left the ref reads as unconfirmed. Exit `0` is the branch deleted and confirmed gone, or already gone. Exit `1` is a refusal, the registry check's among them, and so is a tip that moved after the merge, which names both commits. Exit `2` is a read that did not answer, the rule read included, which never reads as no rules, or a delete the read back could not confirm. A usage error also exits `2`, argparse's own code. Exit `64` is the owner check `comment` and `reply` make against this checkout's `origin`, which also exits `64` where that `origin` cannot be read, run before the registry and before any read. No `--dry-run` exists, since no other subcommand here has one. +`cleanup` deletes the head branch of a merged pull request, the verify-then-delete `drive-pr` and `merge-and-release` prescribe after a merge into `develop`. It is the one place holding the delete's checks, and no skill prescribes a hand-typed delete beside it. A delete typed by hand cannot be relied on to run from an agent seat in auto mode. The `gh-write-guard` hook denies `git push origin --delete` from a primary checkout, and Claude Code's auto-mode classifier refuses the ref API delete as a destructive git operation on some days and allows it on others. A command matching a narrow `permissions.allow` rule is settled before that classifier runs, and the host-safety installer already registers `Bash(python3 scripts/pr_review.py:*)`, so the delete runs here under that rule with no settings change. That rule approves any arguments, so every check lives in the subcommand. It takes the pull request number and `--repo`, never a branch name, and reads the branch from the pull request. It refuses unless the pull request is merged, its base is `develop`, its head repository is `--repo` itself, and the head branch is neither `develop`, `main`, nor the default branch read live. Past the same owner check `comment` and `reply` make, `handoff.py`'s own write-scope check, imported rather than copied, binds the owner to `registry/repos.json` and admits a repository only where it is registered or a fork. A branch any ruleset or branch protection covers is refused, since the hook's per-branch rule check never sees a delete made from inside this script. A branch that another open pull request in `--repo` has as its head or base is refused too, since the delete would close or retarget it. Every open pull request in `--repo` with the branch as its head is named, read page by page. A page that does not answer stops at exit `2`. Up to twenty with the branch as base are named, and any past that are counted. The ref is read by its exact qualified name, and the delete is sent only where that read's tip is the pull request's `headRefOid`, by the node id the same read returned. The delete carries no expected tip of its own, so this is verify-then-delete rather than a compare-and-swap. The delete is confirmed by reading the ref back rather than by the mutation's own answer, so a delete that reports a failure and landed still reads as done, and one that reports success and left the ref reads as unconfirmed. Exit `0` is the branch deleted and confirmed gone, or already gone. Exit `1` is a refusal, the registry check's among them, and so is a tip that moved after the merge, which names both commits. Exit `2` is a read that did not answer, the rule read included, which never reads as no rules, or a delete the read back could not confirm. A usage error also exits `2`, argparse's own code. Exit `64` is the owner check `comment` and `reply` make against this checkout's `origin`, which also exits `64` where that `origin` cannot be read, run before the registry and before any read. With `--checkout` set to a primary clone, a local half follows a remote half that exited `0`. The raw git steps it replaces are not on the permission allowlist, and this script is. Every refusal is decided before anything is deleted, the remote branch included. That includes the primary leaving the merged branch. It also includes a rebase or bisect paused on it elsewhere, and a tip that is neither the merged head nor behind it. A re-check after the delete can still refuse when a concurrent session changed the branch or a worktree in between. The output then names what already happened. A worktree entry whose directory is gone is removed. The local half removes the worktrees holding the branch without force. It deletes the local branch with `git branch -D` only at the merged head or an ancestor of it. A tip re-read must show it unchanged first. It switches the primary to `develop` only off the merged branch, then fast-forwards `develop`. Base outcomes are reported and never fail the run. `cleanup --sweep --checkout ` takes no pull request. It reports what is provably finished and removes it only with `--apply`. The report fetches and prunes the remote-tracking refs of `origin` and writes nothing else. An item must be clean, idle for `--min-age-hours` (default 24), and not mid-operation. A branch is finished when a pull request from `--repo` merged it at its exact tip or past it. It is also finished when `origin/develop` or `origin/main` contains it. A detached worktree is finished when any `origin` ref contains its commit. For a merged `develop` pull request the sweep also deletes the remote branch, and the report runs the same checks and names the result. The sweep never touches the primary except to fast-forward a tracked-clean `develop` to `origin/develop`, and the report names that step. It never touches permanent branches, locked worktrees, or worktrees with submodules. `claims` checks that a description does not contradict its own branch, and it exists for the same reason `sha-pin` now resolves: a reference that points at nothing is a silent failure caught by a reviewer or not at all. Three stale descriptions in one session generated six review findings between them, each a reviewer noticing that the body named a commit or a behavior the branch no longer carried. It reads the commits a body **claims** the branch carries and the `uses:` refs it quotes, confirms the head tree still carries each, and exits `70` where one does not. Prose claims stay **out of scope**, since judging those needs a similarity heuristic, which [`spec/section-model.md`][section-model] rejects for exactly the reason it would fail here. diff --git a/scripts/pr_review.py b/scripts/pr_review.py index 604913c9..249d1fd8 100755 --- a/scripts/pr_review.py +++ b/scripts/pr_review.py @@ -45,12 +45,37 @@ base, naming each. The ref is read exactly, its tip is checked against the pull request's `headRefOid` before the delete is sent, and the delete is confirmed by reading the ref back. The delete itself carries no expected tip, so this is verify-then-delete rather - than a compare-and-swap. No `--dry-run` exists, since no other subcommand here has - one. Exit 0 = deleted and confirmed gone, or already gone, 1 = refused, a - moved tip naming both commits among them, 2 = a read did not answer, a partial one - included, or the delete could not be confirmed, which argparse's own usage error - shares, 64 = the write scope could not be established or excludes the target, checked - before the registry and before any read. + than a compare-and-swap. With `--checkout `, a local half follows a + remote half that exited 0, on the pull request's own branch and head commit. Every + refusal is decided before anything is deleted, the remote branch included. A re-check + after the delete can still refuse when a concurrent session changed the branch or a + worktree in between. The output then names what already happened. A locked, dirty, or + mid-operation worktree refuses. So does a branch tip that is neither the merged head nor + behind it. So does a rebase or bisect paused on the branch in any worktree, the primary + included. A worktree entry whose directory is gone is removed. The primary leaves the + merged branch before the remote delete, and a failed checkout refuses. The local half + removes the worktrees holding the branch. It deletes the local branch only at the merged + head or an ancestor of it, after a tip re-read shows it unchanged. Then it fast-forwards + the clone to current `develop`, never a reset, stash, or force. A clone left behind is + reported and does not fail the run. The clone must be a primary checkout whose origin is + `--repo`. `cleanup --sweep --checkout ` takes no pull request. It reports + what is provably finished, one `would-remove` or `keep ` line per worktree or + branch, and removes it only with `--apply`. The report fetches and prunes the + remote-tracking refs of `origin` and writes nothing else. An item must sit untouched for + `--min-age-hours` (default 24) and be clean. A branch is finished when a pull request + from `--repo` merged it at its exact tip or past it. It is also finished when + `origin/develop` or `origin/main` contains it. A detached worktree is finished when any + `origin` ref contains its commit. For a merged develop pull request the sweep also + deletes the remote head branch through the same checks as `cleanup `. The report runs + those checks and names the result. The sweep never touches the primary checkout except + to fast-forward a tracked-clean `develop` to `origin/develop`, and the report names that + step. Permanent branches, locked worktrees, and worktrees mid-operation are never + touched. Exit 0 = deleted + and confirmed gone, or already gone, or a sweep that completed. Exit 1 = refused, + a moved tip naming both commits among them. Exit 2 = a read did not answer, a + partial one included. A delete or removal that could not be confirmed also exits 2, + as does argparse's own usage error. Exit 64 = the write scope could not be + established or excludes the target, checked before the registry and before any read. status One digest line, any unresolved threads, and any suppressed findings. Read-only. Exit 0 = no Copilot review covers the head yet, or every output shape is recognized and full diff coverage is stated for this head, by the round covering it or by the @@ -1085,6 +1110,15 @@ def strip_fences( repository(owner:$o,name:$r){ bases: pullRequests(states:OPEN,baseRefName:$b,first:20){ totalCount nodes{ number } } }} """ +Q_SWEEP_DEFAULT = """ +query($o:String!,$r:String!){ repository(owner:$o,name:$r){ defaultBranchRef{ name } }} +""" +Q_SWEEP_MERGED = """ +query($o:String!,$r:String!,$b:String!){ + repository(owner:$o,name:$r){ + merged: pullRequests(states:MERGED,headRefName:$b,first:50){ + totalCount nodes{ number baseRefName headRefOid headRepository{ nameWithOwner } } } }} +""" M_DELETE_REF = """ mutation($refId:ID!){ deleteRef(input:{refId:$refId}){ clientMutationId }} @@ -6285,42 +6319,18 @@ def branch_users(owner: str, repo: str, branch: str) -> list[str]: return users -def cleanup(owner: str, repo: str, num: int) -> int: - """Delete the head branch of merged pull request `num` once every check passes. - - Returns 0 where the branch is deleted and the read back confirms it, or is already gone, 1 on - a refusal, 2 where a read did not answer or the delete could not be confirmed, and 64 where - the write scope `comment` and `reply` check excludes the target. Past that check, the owner is - bound to the fleet registry and the repository must be registered or a fork, by `handoff.py`'s - own write-scope check, reused rather than copied so the two cannot drift apart. Its refusal - text is printed as `handoff.py` words it. - """ - ok, why = in_scope(owner) - if not ok: - print(f"status=OUT_OF_SCOPE nothing was deleted: {why}") - return 64 - target = f"{owner}/{repo}" - try: - handoff.require_write_scope(argparse.Namespace(cmd="cleanup", repo=target)) - except handoff.Refusal as exc: - print(f"status=REFUSED nothing was deleted, refused by handoff.py's write scope: {exc}") - return 1 - except handoff.Execution as exc: - print(f"status=UNREAD nothing was deleted: {exc}") - return 2 - try: - return _cleanup(owner, repo, num) - except CleanupUnread as exc: - print(f"status=UNREAD nothing was deleted: {exc}") - return 2 +class CleanupPlan(NamedTuple): + branch: str + head: str + ref: dict | None -def _cleanup(owner: str, repo: str, num: int) -> int: - """The checks, the delete, and the read back, raising `CleanupUnread` on any unanswered read. +def cleanup_check(owner: str, repo: str, num: int) -> CleanupPlan: + """Every read-only check on pull request `num` and its remote branch, with no write. - A branch anything protects is refused whatever that protection would allow, since the write - guard's per-branch rule check never sees a delete made from inside this script. A delete that - reports a failure may still have landed, so the read back decides the outcome either way. + Raises `CleanupRefused` carrying the whole refusal text, and `CleanupUnread` on any + unanswered read. A branch anything protects is refused whatever that protection would allow. + The write guard's per-branch rule check never sees a delete made from inside this script. """ target = f"{owner}/{repo}" data = cleanup_read(Q_CLEANUP_TARGET, o=owner, r=repo, n=num) @@ -6344,39 +6354,72 @@ def _cleanup(owner: str, repo: str, num: int) -> int: if branch in ("develop", "main", default): refusals.append(f"its head branch {branch!r} is a permanent branch") if refusals: - print(f"status=REFUSED nothing was deleted from {target} #{num}: {', and '.join(refusals)}") - return 1 + raise CleanupRefused( + "REFUSED", f"nothing was deleted from {target} #{num}: {', and '.join(refusals)}" + ) ref = cleanup_ref(owner, repo, branch) if ref is None: - print(f"status=ALREADY_GONE branch={branch} is not on {target}, so nothing was deleted") - return 0 + return CleanupPlan(branch, head, None) rules = branch_rules(owner, repo, branch) if rules: - print( - f"status=REFUSED nothing was deleted: {branch!r} on {target} is covered by " - f"{', '.join(rules)}, so its delete is the maintainer's call" + raise CleanupRefused( + "REFUSED", + f"nothing was deleted: {branch!r} on {target} is covered by " + f"{', '.join(rules)}, so its delete is the maintainer's call", ) - return 1 users = branch_users(owner, repo, branch) if users: - print( - f"status=REFUSED nothing was deleted: open pull requests use {branch!r} on {target}, " - f"{', '.join(users)}, and deleting it would close or retarget them" + raise CleanupRefused( + "REFUSED", + f"nothing was deleted: open pull requests use {branch!r} on {target}, " + f"{', '.join(users)}, and deleting it would close or retarget them", ) - return 1 tip = ref["target"]["oid"] if tip != head: - print( - f"status=TIP_MOVED nothing was deleted: {branch!r} on {target} is at {tip}, and " - f"#{num} merged {head}, so something reached the branch after the merge or the name " - "was reused" + raise CleanupRefused( + "TIP_MOVED", + f"nothing was deleted: {branch!r} on {target} is at {tip}, and #{num} merged " + f"{head}, so something reached the branch after the merge or the name was reused", ) - return 1 + return CleanupPlan(branch, head, ref) + + +def cleanup_fresh(owner: str, repo: str, plan: CleanupPlan) -> CleanupPlan: + """Read the remote branch again right before its delete, so the delete uses a fresh node id. + + The checks and the local preflight take time, and another session may have pushed since. + """ + if plan.ref is None: + return plan + ref = cleanup_ref(owner, repo, plan.branch) + if ref is None: + return CleanupPlan(plan.branch, plan.head, None) + tip = ref["target"]["oid"] + if tip != plan.head: + raise CleanupRefused( + "TIP_MOVED", + f"nothing was deleted: {plan.branch!r} on {owner}/{repo} is at {tip} just before " + f"its delete, and the pull request merged {plan.head}", + ) + return CleanupPlan(plan.branch, plan.head, ref) + +def cleanup_remote_delete(owner: str, repo: str, plan: CleanupPlan) -> int: + """Delete the remote branch `plan` verified and read it back. + + A delete that reports a failure may still have landed, so the read back decides the outcome + either way. + """ + target = f"{owner}/{repo}" + branch = plan.branch + if plan.ref is None: + print(f"status=ALREADY_GONE branch={branch} is not on {target}, so nothing was deleted") + return 0 + tip = plan.ref["target"]["oid"] print(f"deleting: {branch} on {target} at {tip}") try: - response = shared.request(M_DELETE_REF, {"refId": ref["id"]}).data + response = shared.request(M_DELETE_REF, {"refId": plan.ref["id"]}).data print(f" response: {json.dumps(response)[:400]}") except shared.GraphQLError as exc: print(f" the delete reported a failure: {exc}") @@ -6399,6 +6442,830 @@ def _cleanup(owner: str, repo: str, num: int) -> int: return 0 +def _cleanup(owner: str, repo: str, num: int) -> int: + """The checks, the delete, and the read back, raising `CleanupUnread` on any unanswered read.""" + try: + plan = cleanup_check(owner, repo, num) + except CleanupRefused as exc: + print(f"status={exc.status} {exc}") + return 1 + return cleanup_remote_delete(owner, repo, plan) + + +def cleanup_scope(owner: str, repo: str) -> int | None: + """The exit code where the write scope refuses or cannot be established, else None.""" + ok, why = in_scope(owner) + if not ok: + print(f"status=OUT_OF_SCOPE nothing was deleted: {why}") + return 64 + try: + handoff.require_write_scope(argparse.Namespace(cmd="cleanup", repo=f"{owner}/{repo}")) + except handoff.Refusal as exc: + print(f"status=REFUSED nothing was deleted, refused by handoff.py's write scope: {exc}") + return 1 + except handoff.Execution as exc: + print(f"status=UNREAD nothing was deleted: {exc}") + return 2 + return None + + +def cleanup(owner: str, repo: str, num: int, checkout: str | None = None) -> int: + """Delete the head branch of merged pull request `num` once every check passes. + + Returns 0 where the branch is deleted and confirmed gone, or is already gone, and 1 on a + refusal. Returns 2 where a read did not answer or the delete could not be confirmed. Returns + 64 where the write scope `comment` and `reply` check excludes the target. Past that check, + the owner is bound to the fleet registry. The repository must be registered or a fork, by + `handoff.py`'s own write-scope check, reused rather than copied so the two cannot drift + apart. Its refusal text is printed as `handoff.py` words it. + + With `checkout`, the local half follows a remote half that exited 0. The checkout and every + local refusal are judged before the remote delete. A local refusal then means nothing at all + was deleted, the remote branch included. A re-check after the delete can still refuse when a + concurrent session changed the branch or a worktree. The output then names what happened. + """ + refused = cleanup_scope(owner, repo) + if refused is not None: + return refused + target = f"{owner}/{repo}" + primary = None + try: + if checkout is not None: + primary = cleanup_primary(checkout, target) + plan = cleanup_check(owner, repo, num) + if primary is not None: + cleanup_preflight(primary, plan.branch, plan.head) + cleanup_leave_branch(primary, plan.branch) + plan = cleanup_fresh(owner, repo, plan) + except CleanupRefused as exc: + if exc.status in ("REFUSED", "TIP_MOVED", "BAD_CHECKOUT"): + print(f"status={exc.status} {exc}") + else: + print(f"status={exc.status} nothing was deleted, remote branch included: {exc}") + return 1 + except CleanupUnread as exc: + print(f"status=UNREAD nothing was deleted: {exc}") + return 2 + code = cleanup_remote_delete(owner, repo, plan) + if code != 0 or primary is None: + return code + try: + return cleanup_local(primary, plan.branch, plan.head) + except CleanupRefused as exc: + print(f"status={exc.status} {exc}") + return 1 + except CleanupUnread as exc: + print(f"status=UNREAD the local half stopped: {exc}") + return 2 + + +ON_WINDOWS = os.name == "nt" + + +class CleanupRefused(Exception): + """A local check refused, carrying the `status=` name to print, reported at exit 1.""" + + def __init__(self, status: str, message: str) -> None: + super().__init__(message) + self.status = status + + +class Worktree(NamedTuple): + path: str + head: str + branch: str | None + locked: bool + prunable: bool + + +GITHUB_REMOTE = re.compile( + r"^(?:https?://(?:[^@/]+@)?github\.com/|ssh://[^@/]+@github\.com/|[^@/]+@github\.com:)" + r"([^/]+)/([^/]+?)(?:\.git)?/?$", + re.IGNORECASE, +) + + +def cleanup_git( + root: str, *args: str, ok: tuple[int, ...] = (0,) +) -> subprocess.CompletedProcess[str]: + """One git call in `root` as an argv list, raising with its stderr on an unexpected code. + + A return code the caller lists in `ok` is an answer it reads. Any other code is a call that + did not do what was asked, so it is shown and ends the step rather than swallowed. + """ + try: + proc = subprocess.run( + ["git", "-C", root, *args], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=False, + timeout=300, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise CleanupUnread(f"`git {' '.join(args)}` did not run: {exc}") from exc + if proc.returncode not in ok: + sys.stderr.write(proc.stderr) + raise CleanupUnread(f"`git {' '.join(args)}` failed rc={proc.returncode}") + return proc + + +def cleanup_primary(checkout: str, target: str) -> str: + """The real path of the primary checkout at `checkout`, refusing anything else. + + A linked worktree is refused, since removing worktrees and moving branches belongs to the + clone that owns them. So is an origin naming another repository, since every local step + here derives from a pull request in `target`. + """ + probe = cleanup_git( + checkout, + "rev-parse", + "--absolute-git-dir", + "--git-common-dir", + "--show-toplevel", + ok=(0, 128), + ) + lines = probe.stdout.splitlines() + if probe.returncode != 0 or len(lines) != 3: + raise CleanupRefused("BAD_CHECKOUT", f"{checkout!r} is not a git checkout with a work tree") + git_dir, common, top = lines + if os.path.realpath(git_dir) != os.path.realpath(os.path.join(checkout, common)): + raise CleanupRefused( + "BAD_CHECKOUT", f"{checkout!r} is a linked worktree, not a primary one" + ) + url = cleanup_git(top, "config", "--get", "remote.origin.url", ok=(0, 1)).stdout.strip() + found = GITHUB_REMOTE.match(url) + if found is None or f"{found[1]}/{found[2]}".lower() != target.lower(): + shown = re.sub(r"//[^/@]*@", "//", url) or "unset" + raise CleanupRefused("BAD_CHECKOUT", f"origin of {top} is {shown}, not {target}") + return os.path.realpath(top) + + +def cleanup_worktrees(primary: str) -> list[Worktree]: + """Every worktree git lists for the clone, the primary first.""" + out = cleanup_git(primary, "worktree", "list", "--porcelain").stdout + trees = [] + for block in out.split("\n\n"): + fields = dict(line.partition(" ")[::2] for line in block.splitlines()) + if "worktree" not in fields: + continue + branch = fields.get("branch") + trees.append( + Worktree( + path=os.path.realpath(fields["worktree"]), + head=fields.get("HEAD", ""), + branch=branch.removeprefix("refs/heads/") if branch else None, + locked="locked" in fields, + prunable="prunable" in fields, + ) + ) + return trees + + +def cleanup_tip(primary: str, ref: str) -> str | None: + """The commit `ref` names, or None where it does not exist.""" + proc = cleanup_git(primary, "rev-parse", "--verify", "--quiet", ref, ok=(0, 1)) + return proc.stdout.strip() or None + + +def cleanup_dirty(path: str, untracked: bool) -> bool: + """Whether the work tree at `path` has changes, without refreshing its index. + + `--no-optional-locks` keeps the read from rewriting the index, which would make an untouched + tree look touched to the age rule of `--sweep`. + """ + flag = "--untracked-files=all" if untracked else "--untracked-files=no" + out = cleanup_git(path, "--no-optional-locks", "status", "--porcelain", flag).stdout + return bool(out.strip()) + + +IN_PROGRESS = ( + "rebase-merge", + "rebase-apply", + "BISECT_LOG", + "MERGE_HEAD", + "CHERRY_PICK_HEAD", + "REVERT_HEAD", +) + + +def cleanup_admin(path: str) -> str | None: + """The git directory holding the worktree at `path`'s state, or None where unreadable. + + A linked worktree names it in its `.git` file, and a primary checkout's `.git` is it. + """ + if os.path.isdir(os.path.join(path, ".git")): + return os.path.join(path, ".git") + try: + text = Path(path, ".git").read_text(encoding="utf-8").strip() + except OSError: + return None + if not text.startswith("gitdir:"): + return None + return os.path.join(path, text.removeprefix("gitdir:").strip()) + + +def cleanup_in_progress(path: str) -> bool: + """Whether a rebase, bisect, merge, cherry-pick, or revert is paused in the worktree.""" + admin = cleanup_admin(path) + return admin is not None and any(os.path.exists(os.path.join(admin, n)) for n in IN_PROGRESS) + + +def cleanup_operation_names(path: str) -> set[str]: + """The branch names a rebase or bisect paused in the worktree at `path` is working on. + + Either detaches HEAD, so the worktree no longer lists the branch it will return to. + """ + admin = cleanup_admin(path) + names: set[str] = set() + if admin is None: + return names + for name in ("rebase-merge/head-name", "rebase-apply/head-name", "BISECT_START"): + try: + text = Path(admin, name).read_text(encoding="utf-8").strip() + except OSError: + continue + names.add(text.removeprefix("refs/heads/")) + return names + + +def cleanup_cwd_inside(path: str) -> bool: + """Whether this process's working directory is the worktree at `path` or inside it.""" + try: + here = os.path.realpath(os.getcwd()) + except OSError: + return False + return here == path or here.startswith(path + os.sep) + + +def cleanup_remove_worktree(primary: str, path: str) -> bool: + """Remove the worktree at `path` without force and read the list back for it. + + Returns whether this process was standing inside it. The process moves to the primary + checkout first. Windows cannot free a directory the parent shell holds, so callers refuse + there before the first removal. + """ + inside = cleanup_cwd_inside(path) + if inside: + os.chdir(primary) + try: + cleanup_git(primary, "worktree", "remove", path) + except CleanupUnread: + if any(w.path == path for w in cleanup_worktrees(primary)): + raise + if os.path.exists(path): + raise CleanupUnread(f"git failed to remove {path}, which is unlisted but still on disk") + print(f" git reported an error, yet {path} is no longer listed, so it is already gone") + return inside + if any(w.path == path for w in cleanup_worktrees(primary)): + raise CleanupUnread(f"the worktree {path} is still listed after its removal") + return inside + + +def cleanup_moved_note() -> None: + print(" note: the shell's directory was that worktree and is gone, so change directory") + + +def cleanup_delete_branch(primary: str, branch: str, tip: str) -> bool: + """Delete the local branch while it still names `tip`, and read it back. + + Returns whether this call deleted it, False where it was already gone. The tip is read again + just before the delete. Git's own refusal then covers a branch that is checked out, being + rebased, or being bisected in any worktree. It is raised as a refusal. + """ + ref = f"refs/heads/{branch}" + now = cleanup_tip(primary, ref) + if now is None: + return False + if now != tip: + raise CleanupRefused( + "LOCAL_TIP_MOVED", f"{ref} reads {now} just before its delete, and {tip} was expected" + ) + proc = cleanup_git(primary, "branch", "-D", "--", branch, ok=(0, 1)) + if proc.returncode != 0: + if cleanup_tip(primary, ref) is None: + return False + raise CleanupRefused("WORKTREE_KEPT", f"git kept {ref}: {proc.stderr.strip()}") + if cleanup_tip(primary, ref) is not None: + raise CleanupUnread(f"{ref} still exists after its delete") + return True + + +def cleanup_base(primary: str, apply: bool = True) -> bool: + """Fast-forward the primary checkout to current `develop`, never reset, stash, or force. + + Returns whether the clone is current, printing a status line for each way it is not. A clone + on another branch, with tracked changes, or that cannot fast-forward is left alone. No + outcome here is an error, so callers report it and never fail a run on it. Without `apply` + it only reports the fast-forward it would make. + """ + try: + current = cleanup_git(primary, "symbolic-ref", "--quiet", "--short", "HEAD", ok=(0, 1)) + on = current.stdout.strip() + if on != "develop": + print(f"keep base-not-on-develop {primary} is on {on or 'a detached head'}") + return False + if cleanup_dirty(primary, False): + print(f"status=BASE_DIRTY {primary} has tracked changes, so it was left where it is") + return False + behind = cleanup_git( + primary, "merge-base", "--is-ancestor", "HEAD", "origin/develop", ok=(0, 1) + ) + if behind.returncode == 1: + print(f"status=BASE_NOT_FF develop in {primary} cannot fast-forward to origin/develop") + return False + if not apply and cleanup_tip(primary, "HEAD") != cleanup_tip( + primary, "refs/remotes/origin/develop" + ): + print(f"would-fast-forward develop in {primary}") + return True + if apply: + try: + cleanup_git(primary, "merge", "--ff-only", "origin/develop") + except CleanupUnread: + if cleanup_tip(primary, "HEAD") != cleanup_tip( + primary, "refs/remotes/origin/develop" + ): + raise + except CleanupUnread as exc: + print(f"status=BASE_UNREAD develop in {primary} could not be brought current: {exc}") + return False + print(f" develop in {primary} is at origin/develop") + return True + + +def cleanup_behind(primary: str, tip: str, head: str) -> bool: + """Whether commit `tip` is an ancestor of `head`, so it holds nothing `head` lacks. + + A `head` object this clone lacks proves nothing, and reads as not behind. + """ + proc = cleanup_git(primary, "merge-base", "--is-ancestor", tip, head, ok=(0, 1, 128)) + return proc.returncode == 0 + + +def cleanup_preflight(primary: str, branch: str, head: str) -> tuple[str | None, list[Worktree]]: + """Fetch, then decide every local refusal, returning the branch tip and the worktrees to remove. + + Nothing is removed here, so a refusal leaves all of it in place. `head` is the commit the + merged pull request carried, read live by the remote half. + """ + cleanup_git(primary, "fetch", "--prune", "origin") + ref = f"refs/heads/{branch}" + tip = cleanup_tip(primary, ref) + if tip is not None and tip != head and not cleanup_behind(primary, tip, head): + raise CleanupRefused( + "LOCAL_TIP_MOVED", + f"{ref} is at {tip}, which is neither the merged head {head} nor behind it, " + "so it may carry commits that were never pushed", + ) + mirror = cleanup_tip(primary, f"refs/remotes/origin/{branch}") + if mirror is not None and mirror != head: + raise CleanupRefused( + "TIP_MOVED", + f"nothing was deleted: origin/{branch} is at {mirror}, and the pull request merged " + f"{head}, so something reached the branch after the merge", + ) + linked = [] + trees = cleanup_worktrees(primary) + others = [w for w in trees if w.path != primary] + for w in trees: + if not w.prunable and branch in cleanup_operation_names(w.path): + raise CleanupRefused( + "WORKTREE_KEPT", f"{w.path} has a rebase or bisect in progress on {branch!r}" + ) + holds = any(w.path == primary and w.branch == branch for w in cleanup_worktrees(primary)) + if holds and cleanup_in_progress(primary): + raise CleanupRefused( + "WORKTREE_KEPT", f"{primary} has a merge, cherry-pick, revert, or bisect in progress" + ) + if holds and cleanup_dirty(primary, False): + raise CleanupRefused( + "BASE_DIRTY", f"{primary} has {branch!r} checked out with tracked changes" + ) + for w in (w for w in others if w.branch == branch): + why = ( + "it is locked" + if w.locked + else "a rebase, bisect, merge, cherry-pick, or revert is in progress in it" + if cleanup_in_progress(w.path) + else "it holds submodules, which only git's own check may remove" + if os.path.exists(os.path.join(w.path, ".gitmodules")) + else "run from a directory outside it, such as the base clone" + if ON_WINDOWS and cleanup_cwd_inside(w.path) + else None + ) + if why is not None: + raise CleanupRefused("WORKTREE_KEPT", f"{w.path}, {why}") + if not w.prunable and cleanup_dirty(w.path, True): + raise CleanupRefused("WORKTREE_DIRTY", f"{w.path} has uncommitted changes") + linked.append(w) + return tip, linked + + +def cleanup_leave_branch(primary: str, branch: str) -> None: + """Move the primary checkout to `develop` where it holds `branch`, refusing if git will not. + + This runs before the remote delete, so a checkout that fails still deletes nothing. + """ + if not any(w.path == primary and w.branch == branch for w in cleanup_worktrees(primary)): + return + try: + cleanup_git(primary, "checkout", "develop") + except CleanupUnread as exc: + raise CleanupRefused( + "WORKTREE_KEPT", f"{primary} could not leave {branch}, so the branch was kept" + ) from exc + print(f" switched {primary} off {branch} to develop") + + +def cleanup_local(primary: str, branch: str, head: str) -> int: + """Remove the finished branch's worktrees and local branch, then return the base clone. + + The refusals are decided again before the first removal, so a refused run leaves all of it + in place. The base clone's return is reported and never fails the run. The local branch is + deleted only at `head` or an ancestor of it, after a tip re-read shows it unchanged. + """ + try: + tip, linked = cleanup_preflight(primary, branch, head) + except CleanupRefused as exc: + raise CleanupRefused(exc.status, f"nothing local was removed: {exc}") from exc + for w in linked: + moved = cleanup_remove_worktree(primary, w.path) + print(f" removed worktree {w.path}") + if moved: + cleanup_moved_note() + if tip is None: + print(f" local branch {branch} is already gone") + else: + try: + deleted = cleanup_delete_branch(primary, branch, tip) + except CleanupUnread as exc: + print(f"status=LOCAL_DELETE_NOT_CONFIRMED {exc}") + return 2 + if deleted: + print(f" deleted local branch {branch} at {tip}, confirmed by reading the ref back") + else: + print(f" local branch {branch} is already gone") + current = cleanup_base(primary) + try: + first = cleanup_git(primary, "status", "--short", "--branch").stdout.splitlines()[:1] + print(f" {first[0] if first else 'status unavailable'}") + except CleanupUnread: + print(" status unavailable") + try: + print(f" worktrees listed: {len(cleanup_worktrees(primary))}") + except CleanupUnread: + print(" worktrees listed: unavailable") + base = "current" if current else "kept" + print(f"status=CLEANED branch={branch} oid={head} locally, base={base}") + return 0 + + +class CleanupSweep: + """One pass over a clone's worktrees and local branches, removing what is provably finished. + + Nothing here is a failure unless a read did not answer or a removal could not be confirmed. + `problems` counts those, so the run continues with the other items and still exits 2. + """ + + def __init__(self, owner: str, repo: str, primary: str, apply: bool, min_age: float) -> None: + self.owner = owner + self.repo = repo + self.target = f"{owner}/{repo}" + self.primary = primary + self.apply = apply + self.min_age = min_age + self.problems = 0 + self.permanent = {"develop", "main"} + self.common = "" + self.operating: set[str] = set() + + def keep(self, reason: str, what: str) -> None: + print(f"keep {reason} {what}") + + def failed(self, what: str, exc: CleanupUnread) -> None: + self.problems += 1 + print(f"unread {what}: {exc}") + + def newest_hours(self, paths: list[str]) -> float: + """Hours since the newest mtime among `paths`, infinite where none can be read.""" + stamps = [] + for path in paths: + try: + stamps.append(os.stat(path).st_mtime) + except OSError: + continue + return float("inf") if not stamps else (time.time() - max(stamps)) / 3600 + + def worktree_hours(self, w: Worktree) -> float | None: + """The idle hours of a linked worktree, or None where its admin directory is unreadable. + + Taken before any status runs on it, and from the admin files git itself writes on use. + The directory's own mtime moves only when an entry is added or removed. + """ + admin = cleanup_admin(w.path) + if admin is None: + return None + files = [os.path.join(admin, n) for n in ("HEAD", "index", os.path.join("logs", "HEAD"))] + return self.newest_hours([w.path, *files]) + + def finished(self, branch: str, tip: str) -> tuple[str, int | None] | None: + """How `branch` at `tip` is known finished, with the develop pull request's number if any. + + A merged pull request counts only from a head in this repository whose last commit is + `tip`. A reused branch name or a later commit then never rides on an old merge. + """ + data = cleanup_read(Q_SWEEP_MERGED, o=self.owner, r=self.repo, b=branch) + found = ((data.get("repository") or {}).get("merged")) or {} + nodes = found.get("nodes") + total = found.get("totalCount") + if not isinstance(nodes, list) or not isinstance(total, int) or total > len(nodes): + raise CleanupUnread(f"the merged pull request read for {branch!r} was incomplete") + matches = [ + n + for n in nodes + if isinstance(n, dict) + and in_repository(n, self.owner, self.repo) + and ( + n.get("headRefOid") == tip + or cleanup_behind(self.primary, tip, n.get("headRefOid") or "") + ) + ] + matches.sort(key=lambda n: n.get("baseRefName") != "develop") + if matches: + num = matches[0].get("number") + return f"merged PR #{num}", num if matches[0].get("baseRefName") == "develop" else None + for upstream in ("origin/develop", "origin/main"): + if cleanup_behind(self.primary, tip, f"refs/remotes/{upstream}"): + return f"contained in {upstream}", None + return None + + def remote(self, num: int | None, branch: str) -> None: + """Report, or with `apply` run, the remote half for a merged develop pull request. + + The remote half makes its own checks, so its refusal keeps only the remote branch. A + report runs those same checks and writes nothing. + """ + if num is None: + return + if not self.apply: + try: + plan = cleanup_check(self.owner, self.repo, num) + except CleanupRefused as exc: + print(f"keep-remote via #{num}: {exc}") + except CleanupUnread as exc: + self.failed(f"remote branch of #{num}", exc) + else: + if plan.ref is None: + print(f"remote already gone {branch} via #{num}") + else: + print(f"would-remove remote branch {branch} via #{num}") + return + print(f" remote half of #{num}:") + try: + code = _cleanup(self.owner, self.repo, num) + except CleanupUnread as exc: + self.failed(f"remote branch of #{num}", exc) + return + if code == 2: + self.problems += 1 + elif code != 0: + print(f"kept-remote #{num}, the local removal stands") + + def run(self) -> int: + cleanup_git(self.primary, "fetch", "--prune", "origin") + found = cleanup_read(Q_SWEEP_DEFAULT, o=self.owner, r=self.repo) + default = ((found.get("repository") or {}).get("defaultBranchRef") or {}).get("name") + if not default: + raise CleanupUnread(f"{self.target} returned no default branch") + self.permanent.add(default) + self.common = os.path.realpath( + os.path.join( + self.primary, + cleanup_git(self.primary, "rev-parse", "--git-common-dir").stdout.strip(), + ) + ) + trees = cleanup_worktrees(self.primary) + linked = [w for w in trees if w.path != self.primary] + for w in trees: + if w.path == self.primary and w.branch: + self.permanent.add(w.branch) + held = {w.branch for w in trees if w.branch} + for w in trees: + if not w.prunable: + self.operating |= cleanup_operation_names(w.path) + self.prune([w for w in linked if w.prunable and not w.locked]) + for w in linked: + if not w.prunable or w.locked: + self.linked(w) + listed = cleanup_git( + self.primary, "for-each-ref", "--format=%(refname)%00%(objectname)", "refs/heads/" + ).stdout.splitlines() + for line in listed: + name, _, tip = line.partition("\0") + branch = name.removeprefix("refs/heads/") + if branch not in held and branch not in self.permanent: + self.loose(branch, tip) + self.base() + return 2 if self.problems else 0 + + def prune(self, stale: list[Worktree]) -> None: + """Prune the entries whose directory is gone and whose record is old enough.""" + old = [] + for w in stale: + hours = self.entry_hours(w) + if hours < self.min_age: + self.keep("too-recent", f"prunable worktree entry {w.path} idle {hours:.1f}h") + else: + old.append(w) + if not old: + return + if not self.apply: + for w in old: + print(f"would-remove prunable worktree entry {w.path}") + return + expire = f"{int(self.min_age * 3600)}.seconds.ago" + try: + cleanup_git(self.primary, "worktree", "prune", "--expire", expire) + left = {w.path for w in cleanup_worktrees(self.primary)} + except CleanupUnread as exc: + self.failed("prunable worktree entries", exc) + return + for w in old: + if w.path in left: + self.problems += 1 + print(f"unread prunable worktree entry {w.path} is still listed after the prune") + else: + print(f"removed prunable worktree entry {w.path}") + + def entry_hours(self, w: Worktree) -> float: + """The idle hours of a missing worktree's record, from the `index` file prune reads. + + Git prunes by the age of that file, and by nothing once it is gone. An entry this cannot + match reads as brand new, so it is kept. + """ + base = os.path.join(self.common, "worktrees") + try: + names = os.listdir(base) + except OSError: + return 0.0 + for name in names: + try: + text = Path(base, name, "gitdir").read_text(encoding="utf-8").strip() + except OSError: + continue + if os.path.realpath(os.path.dirname(os.path.join(base, name, text))) == w.path: + return self.newest_hours([os.path.join(base, name, "index")]) + return 0.0 + + def linked(self, w: Worktree) -> None: + what = f"worktree {w.path}" + (f" branch={w.branch}" if w.branch else " detached") + try: + if w.locked: + self.keep("locked", what) + return + if cleanup_in_progress(w.path): + self.keep("in-progress-operation", what) + return + if os.path.exists(os.path.join(w.path, ".gitmodules")): + self.keep("submodules", what) + return + if ON_WINDOWS and cleanup_cwd_inside(w.path): + self.keep("cwd-inside", what) + return + if w.branch in self.permanent: + self.keep("permanent-branch", what) + return + if w.branch in self.operating: + self.keep("in-progress-operation", what) + return + hours = self.worktree_hours(w) + if hours is None: + self.keep("age-unreadable", what) + return + if hours < self.min_age: + self.keep("too-recent", f"{what} idle {hours:.1f}h of {self.min_age:g}h") + return + if cleanup_dirty(w.path, True): + self.keep("dirty", what) + return + if w.branch is None: + return self.detached(w, what) + tip = cleanup_tip(self.primary, f"refs/heads/{w.branch}") + if tip is None: + self.keep("branch-unreadable", what) + return + done = self.finished(w.branch, tip) + if done is None: + self.keep("unmerged-commits", what) + return + evidence, num = done + line = f"{what} tip={tip} evidence={evidence}" + if not self.apply: + print(f"would-remove {line}") + self.remote(num, w.branch) + return + if cleanup_remove_worktree(self.primary, w.path): + cleanup_moved_note() + print(f"removed worktree {w.path}") + try: + deleted = cleanup_delete_branch(self.primary, w.branch, tip) + except CleanupRefused as exc: + print(f"kept branch {w.branch} after removing its worktree: {exc}") + return + if deleted: + print(f"removed branch {w.branch} tip={tip} evidence={evidence}") + else: + print(f"branch {w.branch} was already gone") + self.remote(num, w.branch) + except CleanupUnread as exc: + self.failed(what, exc) + + def detached(self, w: Worktree, what: str) -> None: + contained = cleanup_git( + self.primary, + "for-each-ref", + "--contains", + w.head, + "refs/remotes/origin", + "--count=1", + "--format=%(refname)", + ).stdout.strip() + if not contained: + self.keep("unpushed-commits", what) + return + if not self.apply: + print(f"would-remove {what} head={w.head} evidence=contained in {contained}") + return + if cleanup_remove_worktree(self.primary, w.path): + cleanup_moved_note() + print(f"removed worktree {w.path} head={w.head} evidence=contained in {contained}") + + def loose(self, branch: str, tip: str) -> None: + """A local branch no worktree holds, judged by the age of its reflog.""" + what = f"branch {branch}" + try: + if branch in self.operating: + self.keep("in-progress-operation", what) + return + log = os.path.join(self.common, "logs", "refs", "heads", branch) + if not os.path.exists(log): + self.keep("age-unreadable", what) + return + hours = self.newest_hours([log]) + if hours < self.min_age: + self.keep("too-recent", f"{what} idle {hours:.1f}h of {self.min_age:g}h") + return + done = self.finished(branch, tip) + if done is None: + self.keep("unmerged-commits", what) + return + evidence, num = done + if not self.apply: + print(f"would-remove {what} tip={tip} evidence={evidence}") + self.remote(num, branch) + return + if cleanup_delete_branch(self.primary, branch, tip): + print(f"removed {what} tip={tip} evidence={evidence}") + else: + print(f"{what} was already gone") + self.remote(num, branch) + except CleanupRefused as exc: + reason = "tip-moved" if exc.status == "LOCAL_TIP_MOVED" else "checked-out-elsewhere" + self.keep(reason, f"{what}: {exc}") + except CleanupUnread as exc: + self.failed(what, exc) + + def base(self) -> None: + """Return the primary checkout to current `develop`, reporting what it could not do.""" + cleanup_base(self.primary, self.apply) + + +def cleanup_sweep(owner: str, repo: str, checkout: str, apply: bool, min_age: float) -> int: + """Report, or with `apply` remove, the clone's worktrees and branches that are provably done. + + Exits 0 where the pass completed. Exits 2 where a read did not answer or a removal could + not be confirmed, the other items having been tried either way. + """ + refused = cleanup_scope(owner, repo) + if refused is not None: + return refused + try: + primary = cleanup_primary(checkout, f"{owner}/{repo}") + sweep = CleanupSweep(owner, repo, primary, apply, min_age) + code = sweep.run() + except CleanupRefused as exc: + print(f"status={exc.status} nothing was removed: {exc}") + return 1 + except CleanupUnread as exc: + print(f"status=UNREAD the sweep stopped: {exc}") + return 2 + mode = "removed" if apply else "reported only, pass --apply to remove" + print(f"status=SWEPT {mode}, problems={sweep.problems}") + return code + + def utf8_console() -> None: """Write the console streams as UTF-8, whatever the host code page is. @@ -6414,10 +7281,11 @@ def utf8_console() -> None: def main(argv: list[str] | None = None) -> int: utf8_console() ap = argparse.ArgumentParser() + sweeping = "--sweep" in (sys.argv[1:] if argv is None else argv) ap.add_argument( "cmd", choices=["attest", "claims", "cleanup", "comment", "status", "reply", "wait"] ) - ap.add_argument("number", type=int) + ap.add_argument("number", type=int, nargs="?" if sweeping else None) # No default, because the wrong repository is the failure this argument has actually had. # A default names one repository, and every run from elsewhere silently reads that one. # The number resolves there, the digest renders, and nothing in the output disagrees. @@ -6512,7 +7380,26 @@ def main(argv: list[str] | None = None) -> int: "--checkout", metavar="DIR", help="attest: the checkout holding the pull request's head and its recorded local pass " - "(default the current directory)", + "(default the current directory), cleanup: the primary clone whose worktree, local " + "branch, and base are cleaned after the remote branch, and the one `--sweep` cleans", + ) + ap.add_argument( + "--sweep", + action="store_true", + help="cleanup: take no pull request, and report the finished worktrees and branches of " + "--checkout", + ) + ap.add_argument( + "--apply", + action="store_true", + help="cleanup --sweep: remove what the report lists, including the remote head branch " + "of a merged develop pull request, through the same checks as `cleanup `", + ) + ap.add_argument( + "--min-age-hours", + type=float, + default=None, + help="cleanup --sweep: hours an item must sit untouched before it is eligible (default 24)", ) a = ap.parse_args(argv) # Named for the command they belong to, since one silently ignored reads as one that took effect. @@ -6530,8 +7417,20 @@ def main(argv: list[str] | None = None) -> int: ap.error(f"{flag} belongs to `reply`, not `{a.cmd}`") if a.cmd != "wait" and a.request: ap.error(f"--request belongs to `wait`, not `{a.cmd}`") - if a.cmd != "attest" and a.checkout is not None: - ap.error(f"--checkout belongs to `attest`, not `{a.cmd}`") + if a.cmd not in ("attest", "cleanup") and a.checkout is not None: + ap.error(f"--checkout belongs to `attest` or `cleanup`, not `{a.cmd}`") + if a.cmd != "cleanup" and a.sweep: + ap.error(f"--sweep belongs to `cleanup`, not `{a.cmd}`") + if a.number is None and not a.sweep: + ap.error("the following arguments are required: number") + if a.sweep and a.number is not None: + ap.error("--sweep takes no pull request number") + if a.sweep and a.checkout is None: + ap.error("--sweep requires --checkout, the primary clone it sweeps") + if not a.sweep and (a.apply or a.min_age_hours is not None): + ap.error("--apply and --min-age-hours belong to `cleanup --sweep`") + if a.min_age_hours is not None and a.min_age_hours < 0: + ap.error("--min-age-hours cannot be negative") if a.cmd not in ("comment", "reply") and a.body is not None: ap.error(f"--body belongs to `comment` or `reply`, not `{a.cmd}`") required = ["--body"] + (["--match"] if a.cmd == "reply" else []) @@ -6568,8 +7467,13 @@ def main(argv: list[str] | None = None) -> int: if a.cmd == "claims": return check_claims(owner, repo, a.number) + if a.cmd == "cleanup" and a.sweep: + return cleanup_sweep( + owner, repo, a.checkout, a.apply, 24.0 if a.min_age_hours is None else a.min_age_hours + ) + if a.cmd == "cleanup": - return cleanup(owner, repo, a.number) + return cleanup(owner, repo, a.number, a.checkout) if a.cmd == "attest": return attest(owner, repo, a.number, a.checkout or ".") diff --git a/tests/test_pr_review.py b/tests/test_pr_review.py index 8bb0fda9..5e6c4ad3 100755 --- a/tests/test_pr_review.py +++ b/tests/test_pr_review.py @@ -19,6 +19,8 @@ import json import os import re +import shlex +import shutil import subprocess import sys import tempfile @@ -26,7 +28,7 @@ import unittest from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import ClassVar +from typing import Any, ClassVar from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) @@ -11941,5 +11943,913 @@ def test_an_unreadable_registry_stops_at_2(self) -> None: self.assertEqual([], self.docs) +EDIT_PICKS = ( + r"import sys,re,pathlib; p=pathlib.Path(sys.argv[1]); " + r"p.write_text(re.sub(r\"^pick\", \"edit\", p.read_text(), flags=re.M))" +) +SEQUENCE_EDITOR = shlex.quote(sys.executable) + ' -c "' + EDIT_PICKS + '"' + + +def run_git(cwd: str, *args: str) -> str: + """One git call in a throwaway repository, failing the test on a nonzero exit.""" + done = subprocess.run( + ["git", "-C", cwd, *args], capture_output=True, text=True, encoding="utf-8", check=True + ) + return done.stdout.strip() + + +class LocalCleanupCase(CleanupCase): + """A bare origin, a primary clone naming o/r, and linked worktrees, all real git in a temp dir. + + Only the GitHub reads and writes are mocked, so every local removal below is git's own. + """ + + def setUp(self) -> None: + super().setUp() + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.root = os.path.realpath(tmp.name) + self.enterContext( + mock.patch.dict( + os.environ, {"GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_SYSTEM": os.devnull} + ) + ) + self.bare = os.path.join(self.root, "origin.git") + self.primary = os.path.join(self.root, "primary") + self.seed = os.path.join(self.root, "seed") + subprocess.run( + ["git", "init", "--bare", "-b", "develop", self.bare], check=True, capture_output=True + ) + subprocess.run(["git", "init", "-b", "develop", self.seed], check=True, capture_output=True) + self.configure(self.seed) + self.commit(self.seed, "seed") + run_git(self.seed, "push", self.bare, "develop:develop", "develop:main") + subprocess.run(["git", "clone", self.bare, self.primary], check=True, capture_output=True) + self.configure(self.primary) + run_git(self.primary, "branch", "main", "origin/main") + run_git(self.primary, "config", f"url.{self.bare}.insteadOf", "https://github.com/o/r.git") + run_git(self.primary, "remote", "set-url", "origin", "https://github.com/o/r.git") + self.count = 0 + + def configure(self, repo: str) -> None: + run_git(repo, "config", "user.name", "Test User") + run_git(repo, "config", "user.email", "test@example.com") + run_git(repo, "config", "commit.gpgsign", "false") + + def commit(self, repo: str, message: str) -> str: + run_git(repo, "commit", "--allow-empty", "-m", message) + return run_git(repo, "rev-parse", "HEAD") + + def worktree(self, branch: str, commits: int = 1) -> tuple[str, str]: + """A linked worktree on a new local branch with `commits` unpushed commits.""" + path = os.path.join(self.root, "wt", branch.replace("/", "_")) + run_git(self.primary, "worktree", "add", "-b", branch, path, "origin/develop") + for _ in range(commits): + self.count += 1 + self.commit(path, f"work {self.count}") + return path, run_git(path, "rev-parse", "HEAD") + + def advance_origin(self) -> str: + """A new commit on origin's develop, as a merge into it would leave.""" + run_git(self.seed, "pull", "--ff-only", self.bare, "develop") + self.count += 1 + tip = self.commit(self.seed, f"merge {self.count}") + run_git(self.seed, "push", self.bare, "develop:develop") + return tip + + def branches(self) -> list[str]: + out = run_git(self.primary, "for-each-ref", "--format=%(refname:short)", "refs/heads") + return out.splitlines() + + def listed(self) -> list[str]: + out = run_git(self.primary, "worktree", "list", "--porcelain") + return [ + os.path.realpath(ln.split(" ", 1)[1]) + for ln in out.splitlines() + if ln.startswith("worktree ") + ] + + def run_local(self, checkout: str | None = None) -> int: + argv = ["cleanup", "7", "--repo", "o/r", "--checkout", checkout or self.primary] + return pr_review.main(argv) + + def pause_rebase(self, repo: str) -> None: + env = {**os.environ, "GIT_SEQUENCE_EDITOR": SEQUENCE_EDITOR} + argv = ["git", "-C", repo, "rebase", "-i", "HEAD~2"] + subprocess.run(argv, env=env, capture_output=True, check=True) + + def wire_merged(self, tip: str) -> None: + self.wire(cleanup_pr(headRefOid=tip), refs=[tip, tip, None]) + + +class TestCleanupLocalHalf(LocalCleanupCase): + def test_the_worktree_and_branch_go_and_the_base_clone_catches_up(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + merged = self.advance_origin() + self.wire_merged(tip) + self.assertEqual(0, self.run_local()) + self.assertFalse(os.path.exists(path)) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + self.assertEqual([self.primary], self.listed()) + self.assertEqual(merged, run_git(self.primary, "rev-parse", "HEAD")) + out = self.out.getvalue() + self.assertIn("status=DELETED", out) + self.assertIn("status=CLEANED", out) + self.assertIn("## develop...origin/develop", out) + + def test_without_a_checkout_nothing_local_is_touched(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire(cleanup_pr(headRefOid=tip), refs=[tip, None]) + self.assertEqual(0, pr_review.main(["cleanup", "7", "--repo", "o/r"])) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + self.assertNotIn("status=CLEANED", self.out.getvalue()) + + def test_a_branch_already_gone_locally_still_returns_the_base_clone(self) -> None: + merged = self.advance_origin() + self.wire(cleanup_pr(headRefOid=merged), refs=[merged, merged, None]) + self.assertEqual(0, self.run_local()) + self.assertIn("is already gone", self.out.getvalue()) + self.assertEqual(merged, run_git(self.primary, "rev-parse", "HEAD")) + + def test_a_moved_local_tip_refuses_and_removes_nothing(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.count += 1 + later = self.commit(path, f"unpushed {self.count}") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + out = self.out.getvalue() + self.assertIn("status=LOCAL_TIP_MOVED", out) + self.assertIn(tip, out) + self.assertIn(later, out) + self.assertFalse(self.deleted()) + self.assertIn("remote branch included", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertEqual(later, run_git(self.primary, "rev-parse", CLEANUP_BRANCH)) + + def test_a_dirty_worktree_refuses_and_removes_nothing(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + Path(path, "scratch.txt").write_text("x", encoding="utf-8") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_DIRTY", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn("remote branch included", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_locked_worktree_refuses_and_removes_nothing(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + run_git(self.primary, "worktree", "lock", path) + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn("remote branch included", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_checkout_whose_origin_is_another_repository_is_refused_before_the_remote_delete( + self, + ) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + run_git(self.primary, "remote", "set-url", "origin", "git@github.com:other/repo.git") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=BAD_CHECKOUT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertTrue(os.path.isdir(path)) + + def test_a_linked_worktree_is_not_a_primary_checkout(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + self.assertEqual(1, self.run_local(path)) + self.assertIn("status=BAD_CHECKOUT", self.out.getvalue()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_remote_refusal_skips_the_local_half(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire(cleanup_pr(headRefOid=tip, merged=False)) + self.assertEqual(1, self.run_local()) + self.assertNotIn("status=CLEANED", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_primary_holding_the_branch_moves_to_develop_before_the_delete(self) -> None: + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH) + tip = self.commit(self.primary, "primary work") + self.advance_origin() + self.wire_merged(tip) + self.assertEqual(0, self.run_local()) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + self.assertEqual("develop", run_git(self.primary, "symbolic-ref", "--short", "HEAD")) + out = self.out.getvalue() + self.assertLess(out.index("switched"), out.index("deleting:")) + + def test_a_rebase_paused_in_the_primary_on_the_branch_refuses(self) -> None: + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH) + for n in range(3): + self.commit(self.primary, f"primary work {n}") + tip = run_git(self.primary, "rev-parse", "HEAD") + self.pause_rebase(self.primary) + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_the_closing_informational_reads_never_fail_a_landed_cleanup(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_git + + def flaky(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:1] == ("status",) and "--branch" in args: + raise pr_review.CleanupUnread("status read failed") + return real(root, *args, **kw) + + listing = pr_review.cleanup_worktrees + + def failing_list(primary: str) -> list[pr_review.Worktree]: + if "status unavailable" in self.out.getvalue(): + raise pr_review.CleanupUnread("list read failed") + return listing(primary) + + with ( + mock.patch.object(pr_review, "cleanup_git", side_effect=flaky), + mock.patch.object(pr_review, "cleanup_worktrees", side_effect=failing_list), + ): + self.assertEqual(0, self.run_local()) + out = self.out.getvalue() + self.assertIn("status unavailable", out) + self.assertIn("worktrees listed: unavailable", out) + self.assertIn("status=CLEANED", out) + + def test_a_primary_stopped_mid_cherry_pick_on_the_branch_refuses(self) -> None: + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH) + tip = self.commit(self.primary, "primary work") + admin = Path(run_git(self.primary, "rev-parse", "--absolute-git-dir")) + (admin / "CHERRY_PICK_HEAD").write_text(tip + "\n", encoding="utf-8") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_remote_branch_that_moved_before_the_delete_is_not_deleted(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + self.wire(cleanup_pr(headRefOid=tip), refs=[tip, "c" * 40]) + self.assertEqual(1, self.run_local()) + self.assertIn("status=TIP_MOVED", self.out.getvalue()) + self.assertFalse(self.deleted()) + + def test_an_origin_branch_that_differs_from_the_merged_head_refuses(self) -> None: + path, _ = self.worktree(CLEANUP_BRANCH) + run_git(path, "push", "origin", CLEANUP_BRANCH) + self.count += 1 + tip = self.commit(path, f"after the push {self.count}") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=TIP_MOVED", self.out.getvalue()) + self.assertFalse(self.deleted()) + + def test_a_primary_holding_the_branch_with_tracked_changes_refuses(self) -> None: + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH) + Path(self.primary, "tracked.txt").write_text("a", encoding="utf-8") + run_git(self.primary, "add", "tracked.txt") + tip = run_git(self.primary, "rev-parse", "HEAD") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=BASE_DIRTY", self.out.getvalue()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_diverged_develop_is_reported_without_failing_the_removals(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.advance_origin() + run_git(self.primary, "fetch", "origin") + self.count += 1 + self.commit(self.primary, f"local only {self.count}") + self.wire_merged(tip) + self.assertEqual(0, self.run_local()) + out = self.out.getvalue() + self.assertLess(out.index("removed worktree"), out.index("status=BASE_NOT_FF")) + self.assertIn("base=kept", out) + self.assertFalse(os.path.exists(path)) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + + def test_the_shell_directory_inside_a_removed_worktree_is_handled_per_platform(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + before = os.getcwd() + self.addCleanup(os.chdir, before) + os.chdir(path) + if pr_review.ON_WINDOWS: + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + else: + self.assertEqual(0, self.run_local()) + self.assertIn("note: the shell's directory", self.out.getvalue()) + self.assertFalse(os.path.exists(path)) + + def test_a_shell_inside_the_worktree_refuses_on_windows(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + before = os.getcwd() + self.addCleanup(os.chdir, before) + os.chdir(path) + with mock.patch.object(pr_review, "ON_WINDOWS", True): + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_worktree_with_submodules_refuses_and_removes_nothing(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + Path(path, ".gitmodules").write_text("", encoding="utf-8") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + + def test_a_rerun_with_the_primary_elsewhere_is_done_with_the_base_kept(self) -> None: + run_git(self.primary, "checkout", "-b", "elsewhere") + self.wire(cleanup_pr(headRefOid=CLEANUP_HEAD), refs=[CLEANUP_HEAD, CLEANUP_HEAD, None]) + self.assertEqual(0, self.run_local()) + out = self.out.getvalue() + self.assertIn("is already gone", out) + self.assertIn("base=kept", out) + + def test_a_tip_that_moves_before_the_delete_is_a_refusal_after_the_removals(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_tip + heads = [] + + def moving(primary: str, ref: str) -> str | None: + if not ref.startswith("refs/heads/"): + return real(primary, ref) + heads.append(ref) + return "b" * 40 if len(heads) == 3 else real(primary, ref) + + with mock.patch.object(pr_review, "cleanup_tip", side_effect=moving): + self.assertEqual(1, self.run_local()) + out = self.out.getvalue() + self.assertLess(out.index("removed worktree"), out.index("status=LOCAL_TIP_MOVED")) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_an_unconfirmed_branch_delete_is_exit_2(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + with mock.patch.object(pr_review, "cleanup_tip", side_effect=itertools.repeat(tip)): + self.assertEqual(2, self.run_local()) + self.assertIn("status=LOCAL_DELETE_NOT_CONFIRMED", self.out.getvalue()) + self.assertFalse(os.path.exists(path)) + + def test_a_primary_on_another_branch_is_left_there(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + run_git(self.primary, "checkout", "-b", "elsewhere") + self.wire_merged(tip) + self.assertEqual(0, self.run_local()) + self.assertEqual("elsewhere", run_git(self.primary, "symbolic-ref", "--short", "HEAD")) + out = self.out.getvalue() + self.assertIn("keep base-not-on-develop", out) + self.assertIn("base=kept", out) + + def test_a_branch_being_bisected_in_another_worktree_is_kept(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH, commits=3) + run_git(path, "bisect", "start") + run_git(path, "bisect", "bad", "HEAD") + run_git(path, "bisect", "good", "HEAD~2") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + self.assertTrue(os.path.isdir(path)) + + def test_a_branch_being_rebased_in_another_worktree_is_kept(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH, commits=3) + self.pause_rebase(path) + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_prunable_entry_holding_the_branch_is_removed_without_force(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + shutil.rmtree(path) + self.wire_merged(tip) + self.assertEqual(0, self.run_local()) + self.assertEqual([self.primary], self.listed()) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + + def test_a_local_tip_behind_the_merged_head_is_removed(self) -> None: + path, _ = self.worktree(CLEANUP_BRANCH) + self.count += 1 + ahead = self.commit(path, f"merged tip {self.count}") + run_git(self.primary, "branch", "ahead", ahead) + run_git(path, "reset", "--hard", "HEAD~1") + self.wire_merged(ahead) + self.assertEqual(0, self.run_local()) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + + def test_a_partly_removed_worktree_is_an_unconfirmed_removal(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_git + + def partial(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:2] == ("worktree", "remove"): + real(root, *args) + os.makedirs(path) + raise pr_review.CleanupUnread("could not delete every file") + return real(root, *args, **kw) + + with mock.patch.object(pr_review, "cleanup_git", side_effect=partial): + self.assertEqual(2, self.run_local()) + self.assertIn(path, self.out.getvalue()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_branch_delete_git_refuses_is_a_kept_branch(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_git + + def refusing(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:2] == ("branch", "-D"): + return subprocess.CompletedProcess([], 1, "", "error: used by worktree") + return real(root, *args, **kw) + + with mock.patch.object(pr_review, "cleanup_git", side_effect=refusing): + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_worktree_mid_operation_refuses_and_removes_nothing(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + admin = Path(run_git(path, "rev-parse", "--absolute-git-dir")) + (admin / "MERGE_HEAD").write_text(tip + "\n", encoding="utf-8") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertTrue(os.path.isdir(path)) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_worktree_removed_by_someone_else_mid_run_is_done(self) -> None: + path, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_git + + def racing(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:2] == ("worktree", "remove"): + real(root, *args) + raise pr_review.CleanupUnread("another process got there first") + return real(root, *args, **kw) + + with mock.patch.object(pr_review, "cleanup_git", side_effect=racing): + self.assertEqual(0, self.run_local()) + self.assertIn("already gone", self.out.getvalue()) + self.assertFalse(os.path.exists(path)) + self.assertNotIn(CLEANUP_BRANCH, self.branches()) + + def test_a_branch_deleted_by_someone_else_before_the_delete_is_already_gone(self) -> None: + _, tip = self.worktree(CLEANUP_BRANCH) + self.wire_merged(tip) + real = pr_review.cleanup_tip + calls = [] + + def vanishing(primary: str, ref: str) -> str | None: + if ref.startswith("refs/heads/"): + calls.append(ref) + if len(calls) == 3 and ref.startswith("refs/heads/"): + run_git(primary, "branch", "-D", "--", CLEANUP_BRANCH) + return None + return real(primary, ref) + + with mock.patch.object(pr_review, "cleanup_tip", side_effect=vanishing): + self.assertEqual(0, self.run_local()) + self.assertIn("status=CLEANED", self.out.getvalue()) + + def test_a_primary_that_cannot_leave_the_branch_keeps_it_and_exits_1(self) -> None: + Path(self.seed, "clash.txt").write_text("x", encoding="utf-8") + run_git(self.seed, "add", "clash.txt") + self.commit(self.seed, "add clash") + run_git(self.seed, "push", self.bare, "develop:develop") + run_git(self.primary, "pull", "--ff-only", "origin", "develop") + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH, "develop~1") + Path(self.primary, "clash.txt").write_text("y", encoding="utf-8") + tip = run_git(self.primary, "rev-parse", "HEAD") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("could not leave", self.out.getvalue()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertIn("nothing was deleted, remote branch included", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + def test_a_develop_held_by_another_worktree_stops_the_primary_leaving(self) -> None: + run_git(self.primary, "checkout", "-b", CLEANUP_BRANCH) + tip = self.commit(self.primary, "primary work") + other = os.path.join(self.root, "wt", "develop-holder") + run_git(self.primary, "worktree", "add", other, "develop") + self.wire_merged(tip) + self.assertEqual(1, self.run_local()) + self.assertIn("status=WORKTREE_KEPT", self.out.getvalue()) + self.assertFalse(self.deleted()) + self.assertIn(CLEANUP_BRANCH, self.branches()) + + +class TestCleanupOriginParsing(unittest.TestCase): + def test_https_and_ssh_forms_name_the_repository_case_insensitively(self) -> None: + for url in ( + "https://github.com/O/R.git", + "https://github.com/o/r", + "https://user@github.com/o/r.git/", + "git@github.com:o/R.git", + "ssh://git@github.com/O/r", + ): + found = pr_review.GITHUB_REMOTE.match(url) + self.assertIsNotNone(found, url) + assert found is not None + self.assertEqual("o/r", f"{found[1]}/{found[2]}".lower()) + + def test_other_hosts_and_shapes_do_not_match(self) -> None: + for url in ("https://example.com/o/r.git", "/srv/o/r.git", "https://github.com/o", ""): + self.assertIsNone(pr_review.GITHUB_REMOTE.match(url), url) + + +class TestCleanupSweepArguments(CleanupCase): + def usage(self, *argv: str) -> None: + with self.assertRaises(SystemExit) as caught: + pr_review.main(list(argv)) + self.assertEqual(2, caught.exception.code) + + def test_a_missing_number_is_a_usage_error_outside_the_sweep(self) -> None: + self.usage("status", "--repo", "o/r") + self.usage("cleanup", "--repo", "o/r") + + def test_a_required_number_still_parses_ahead_of_the_options(self) -> None: + class Reached(Exception): + pass + + with ( + mock.patch.object(pr_review, "full_read", side_effect=Reached), + self.assertRaises(Reached), + ): + pr_review.main(["status", "--repo", "o/r", "7"]) + with mock.patch.object(pr_review, "cleanup", return_value=0) as cleaned: + self.assertEqual( + 0, pr_review.main(["cleanup", "--repo", "o/r", "7", "--checkout", "x"]) + ) + cleaned.assert_called_once_with("o", "r", 7, "x") + + def test_the_sweep_flags_are_checked_against_each_other(self) -> None: + self.usage("cleanup", "7", "--sweep", "--repo", "o/r", "--checkout", "/x") + self.usage("cleanup", "--sweep", "--repo", "o/r") + self.usage("cleanup", "7", "--repo", "o/r", "--apply") + self.usage("cleanup", "7", "--repo", "o/r", "--min-age-hours", "1") + self.usage("status", "7", "--repo", "o/r", "--sweep") + + +class TestCleanupSweep(LocalCleanupCase): + def setUp(self) -> None: + super().setUp() + self.merged_prs: dict[str, list[dict]] = {} + self.failing: set[str] = set() + + def age(self, path: str, hours: float) -> None: + """Set every file the sweep reads the age from to `hours` ago.""" + admin = Path(path, ".git").read_text(encoding="utf-8").split(":", 1)[1].strip() + stamp = time.time() - hours * 3600 + for p in (path, *(os.path.join(admin, n) for n in ("HEAD", "index", "logs/HEAD"))): + if os.path.exists(p): + os.utime(p, (stamp, stamp)) + + def age_branch(self, branch: str, hours: float) -> None: + log = os.path.join(self.primary, ".git", "logs", "refs", "heads", branch) + stamp = time.time() - hours * 3600 + os.utime(log, (stamp, stamp)) + + def merged_pr(self, branch: str, tip: str, base: str = "develop") -> None: + node = { + "number": 7, + "baseRefName": base, + "headRefOid": tip, + "headRepository": {"nameWithOwner": "o/r"}, + } + self.merged_prs[branch] = [node] + + def wire_sweep( + self, remote_tip: str | None = None, remote_gone: bool = False, **pr: object + ) -> None: + refs: list[object] = [remote_tip, None] if remote_tip else [] + if remote_gone: + refs = [None] + self.wire(cleanup_pr(headRefOid=remote_tip or CLEANUP_HEAD, **pr), refs=refs) + inner = pr_review.shared.request + + def fake(query: str, variables: dict, **kw: Any) -> pr_review.shared.Response: + if query == pr_review.Q_SWEEP_DEFAULT: + head = {"defaultBranchRef": {"name": "main"}} + return pr_review.shared.Response({"repository": head}, []) + if query == pr_review.Q_SWEEP_MERGED: + if variables["b"] in self.failing: + raise pr_review.shared.GraphQLError("gh graphql reported errors") + nodes = self.merged_prs.get(variables["b"], []) + page = {"merged": {"totalCount": len(nodes), "nodes": nodes}} + return pr_review.shared.Response({"repository": page}, []) + return inner(query, variables, **kw) + + self.enterContext(mock.patch.object(pr_review.shared, "request", side_effect=fake)) + + def sweep(self, *extra: str) -> int: + argv = ["cleanup", "--sweep", "--repo", "o/r", "--checkout", self.primary, *extra] + return pr_review.main(argv) + + def scenario(self) -> dict[str, str]: + """Old and finished items beside every kind of item that must stay.""" + made: dict[str, str] = {} + path, tip = self.worktree(CLEANUP_BRANCH) + self.merged_pr(CLEANUP_BRANCH, tip) + self.age(path, 48) + made["merged"] = path + detached = os.path.join(self.root, "wt", "detached") + run_git(self.primary, "worktree", "add", "--detach", detached, "origin/develop") + self.age(detached, 48) + made["detached"] = detached + gone = os.path.join(self.root, "wt", "gone") + run_git(self.primary, "worktree", "add", "--detach", gone, "origin/develop") + admin = Path(gone, ".git").read_text(encoding="utf-8").split(":", 1)[1].strip() + shutil.rmtree(gone) + old = time.time() - 48 * 3600 + os.utime(os.path.join(admin, "index"), (old, old)) + made["gone"] = gone + fresh = os.path.join(self.root, "wt", "fresh-gone") + run_git(self.primary, "worktree", "add", "--detach", fresh, "origin/develop") + shutil.rmtree(fresh) + made["fresh-gone"] = fresh + recent, recent_tip = self.worktree("recent-work") + self.merged_pr("recent-work", recent_tip) + made["recent"] = recent + unmerged, _ = self.worktree("unmerged-work") + self.age(unmerged, 48) + made["unmerged"] = unmerged + dirty, dirty_tip = self.worktree("dirty-work") + self.merged_pr("dirty-work", dirty_tip) + Path(dirty, "scratch.txt").write_text("x", encoding="utf-8") + self.age(dirty, 48) + made["dirty"] = dirty + locked, locked_tip = self.worktree("locked-work") + self.merged_pr("locked-work", locked_tip) + run_git(self.primary, "worktree", "lock", locked) + self.age(locked, 48) + made["locked"] = locked + run_git(self.primary, "branch", "loose-done", "origin/develop") + self.age_branch("loose-done", 48) + run_git(self.primary, "branch", "loose-recent", "origin/develop") + self.advance_origin() + return made + + def test_a_dry_run_reports_every_item_and_removes_nothing(self) -> None: + made = self.scenario() + before = (sorted(self.branches()), sorted(self.listed())) + self.wire_sweep(remote_tip=run_git(self.primary, "rev-parse", CLEANUP_BRANCH)) + self.assertEqual(0, self.sweep()) + self.assertEqual(before, (sorted(self.branches()), sorted(self.listed()))) + self.assertFalse(self.deleted()) + out = self.out.getvalue() + self.assertIn(f"would-remove worktree {made['merged']}", out) + self.assertIn(f"would-remove worktree {made['detached']} detached", out) + self.assertIn(f"would-remove prunable worktree entry {made['gone']}", out) + self.assertIn("would-remove branch loose-done", out) + self.assertIn(f"would-remove remote branch {CLEANUP_BRANCH} via #7", out) + self.assertIn(f"would-fast-forward develop in {self.primary}", out) + self.assertNotIn("removed ", out.replace("would-remove", "")) + + def test_apply_removes_what_is_finished_and_keeps_the_rest(self) -> None: + made = self.scenario() + tip = run_git(self.primary, "rev-parse", CLEANUP_BRANCH) + self.wire_sweep(remote_tip=tip) + self.assertEqual(0, self.sweep("--apply")) + self.assertIn(made["fresh-gone"], self.listed()) + for gone in ("merged", "detached", "gone"): + self.assertNotIn(made[gone], self.listed(), gone) + self.assertFalse(os.path.exists(made["merged"])) + for kept in ("recent", "unmerged", "dirty", "locked"): + self.assertTrue(os.path.isdir(made[kept]), kept) + self.assertIn(made[kept], self.listed(), kept) + branches = self.branches() + self.assertNotIn(CLEANUP_BRANCH, branches) + self.assertNotIn("loose-done", branches) + for kept in ("develop", "main", "loose-recent", "recent-work", "unmerged-work"): + self.assertIn(kept, branches) + self.assertTrue(self.deleted()) + out = self.out.getvalue() + self.assertIn("keep too-recent", out) + self.assertIn("keep unmerged-commits", out) + self.assertIn("keep dirty", out) + self.assertIn("keep locked", out) + self.assertEqual( + run_git(self.primary, "rev-parse", "origin/develop"), + run_git(self.primary, "rev-parse", "develop"), + ) + + def test_a_read_failure_continues_and_exits_2(self) -> None: + made = self.scenario() + early, _ = self.worktree("aa-failing") + self.age(early, 48) + self.failing.add("aa-failing") + tip = run_git(self.primary, "rev-parse", CLEANUP_BRANCH) + self.wire_sweep(remote_tip=tip) + self.assertEqual(2, self.sweep("--apply")) + out = self.out.getvalue() + self.assertLess(out.index("unread worktree"), out.index("removed worktree")) + self.assertNotIn(made["merged"], self.listed()) + self.assertIn(early, self.listed()) + + def test_the_remote_check_in_a_report_names_a_refusal(self) -> None: + self.scenario() + self.wire_sweep(remote_tip=run_git(self.primary, "rev-parse", CLEANUP_BRANCH), merged=False) + self.assertEqual(0, self.sweep()) + out = self.out.getvalue() + self.assertIn("keep-remote via #7", out) + self.assertNotIn("would-remove remote", out) + self.assertFalse(self.deleted()) + + def test_the_remote_check_in_a_report_names_a_branch_already_gone(self) -> None: + self.scenario() + self.wire_sweep(remote_gone=True) + self.assertEqual(0, self.sweep()) + self.assertIn(f"remote already gone {CLEANUP_BRANCH} via #7", self.out.getvalue()) + + def test_a_loose_branch_whose_tip_moved_is_kept_as_tip_moved(self) -> None: + run_git(self.primary, "branch", "loose-done", "origin/develop") + self.age_branch("loose-done", 48) + self.wire_sweep() + moved = pr_review.CleanupRefused("LOCAL_TIP_MOVED", "moved") + with mock.patch.object(pr_review, "cleanup_delete_branch", side_effect=moved): + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep tip-moved branch loose-done", self.out.getvalue()) + + def test_main_and_develop_survive_a_sweep_with_no_age_floor(self) -> None: + run_git(self.primary, "checkout", "-b", "primary-work") + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply", "--min-age-hours", "0")) + for kept in ("main", "develop", "primary-work"): + self.assertIn(kept, self.branches()) + + def test_the_primary_checkout_branch_is_never_touched(self) -> None: + run_git(self.primary, "checkout", "-b", "primary-work") + self.age_branch("primary-work", 48) + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("primary-work", self.branches()) + for line in self.out.getvalue().splitlines(): + if "primary-work" in line: + self.assertTrue(line.startswith("keep base-not-on-develop"), line) + + def test_a_branch_without_a_reflog_is_kept_as_age_unreadable(self) -> None: + run_git(self.primary, "branch", "loose-done", "origin/develop") + os.remove(os.path.join(self.primary, ".git", "logs", "refs", "heads", "loose-done")) + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep age-unreadable branch loose-done", self.out.getvalue()) + self.assertIn("loose-done", self.branches()) + + def test_a_failed_prune_is_counted_and_the_other_items_still_run(self) -> None: + made = self.scenario() + tip = run_git(self.primary, "rev-parse", CLEANUP_BRANCH) + self.wire_sweep(remote_tip=tip) + real = pr_review.cleanup_git + + def failing(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:2] == ("worktree", "prune"): + raise pr_review.CleanupUnread("prune failed") + return real(root, *args, **kw) + + with mock.patch.object(pr_review, "cleanup_git", side_effect=failing): + self.assertEqual(2, self.sweep("--apply")) + self.assertIn("unread prunable worktree entries", self.out.getvalue()) + self.assertNotIn(made["merged"], self.listed()) + + def test_a_stale_remote_ref_for_a_deleted_branch_is_not_containment(self) -> None: + stale = os.path.join(self.root, "wt", "stale-detached") + run_git(self.primary, "worktree", "add", "--detach", stale, "origin/develop") + self.count += 1 + self.commit(stale, f"only on a deleted branch {self.count}") + run_git(stale, "push", "origin", "HEAD:refs/heads/gone-branch") + run_git(self.primary, "fetch", "origin") + run_git(self.bare, "branch", "-D", "gone-branch") + self.age(stale, 48) + sha = run_git(stale, "rev-parse", "HEAD") + self.wire_sweep() + for extra in ((), ("--apply",)): + run_git(self.primary, "update-ref", "refs/remotes/origin/gone-branch", sha) + self.assertEqual(0, self.sweep(*extra)) + out = self.out.getvalue() + self.assertIn(f"keep unpushed-commits worktree {stale} detached", out) + self.assertNotIn(f"would-remove worktree {stale}", out) + self.assertIn(stale, self.listed()) + + def test_a_primary_paused_mid_rebase_keeps_the_branch_in_both_modes(self) -> None: + run_git(self.primary, "checkout", "-b", "rebasing-work") + for n in range(3): + self.commit(self.primary, f"primary work {n}") + tip = run_git(self.primary, "rev-parse", "HEAD") + self.pause_rebase(self.primary) + self.merged_pr("rebasing-work", tip) + self.age_branch("rebasing-work", 48) + self.wire_sweep() + for extra in ((), ("--apply",)): + self.assertEqual(0, self.sweep(*extra)) + self.assertIn("keep in-progress-operation branch rebasing-work", self.out.getvalue()) + self.assertIn("rebasing-work", self.branches()) + + def test_a_worktree_with_submodules_is_kept(self) -> None: + path, tip = self.worktree("modular-work") + self.merged_pr("modular-work", tip) + Path(path, ".gitmodules").write_text("", encoding="utf-8") + self.age(path, 48) + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep submodules", self.out.getvalue()) + self.assertIn(path, self.listed()) + + def test_a_shell_inside_a_worktree_keeps_it_on_windows(self) -> None: + path, tip = self.worktree("held-work") + self.merged_pr("held-work", tip) + self.age(path, 48) + self.wire_sweep() + before = os.getcwd() + self.addCleanup(os.chdir, before) + os.chdir(path) + with mock.patch.object(pr_review, "ON_WINDOWS", True): + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep cwd-inside", self.out.getvalue()) + self.assertIn(path, self.listed()) + + def test_a_worktree_mid_operation_is_kept(self) -> None: + path, tip = self.worktree("paused-work") + self.merged_pr("paused-work", tip) + admin = Path(run_git(path, "rev-parse", "--absolute-git-dir")) + (admin / "BISECT_LOG").write_text("\n", encoding="utf-8") + self.age(path, 48) + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep in-progress-operation", self.out.getvalue()) + self.assertIn(path, self.listed()) + self.assertIn("paused-work", self.branches()) + + def test_a_branch_being_bisected_elsewhere_is_kept_as_in_progress(self) -> None: + path, tip = self.worktree("bisected-work", commits=3) + self.merged_pr("bisected-work", tip) + run_git(path, "bisect", "start") + run_git(path, "bisect", "bad", "HEAD") + run_git(path, "bisect", "good", "HEAD~2") + self.age(path, 48) + self.age_branch("bisected-work", 48) + self.wire_sweep() + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep in-progress-operation branch bisected-work", self.out.getvalue()) + self.assertIn("bisected-work", self.branches()) + self.assertIn(path, self.listed()) + + def test_a_branch_delete_git_refuses_is_kept_as_checked_out_elsewhere(self) -> None: + run_git(self.primary, "branch", "loose-done", "origin/develop") + self.age_branch("loose-done", 48) + self.wire_sweep() + real = pr_review.cleanup_git + + def refusing(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str]: + if args[:2] == ("branch", "-D"): + return subprocess.CompletedProcess([], 1, "", "error: used by worktree") + return real(root, *args, **kw) + + with mock.patch.object(pr_review, "cleanup_git", side_effect=refusing): + self.assertEqual(0, self.sweep("--apply")) + self.assertIn("keep checked-out-elsewhere branch loose-done", self.out.getvalue()) + + def test_a_branch_behind_a_merged_head_counts_as_finished(self) -> None: + path, _ = self.worktree("behind-work") + self.count += 1 + ahead = self.commit(path, f"merged tip {self.count}") + run_git(self.primary, "branch", "ahead", ahead) + run_git(path, "reset", "--hard", "HEAD~1") + run_git(self.primary, "worktree", "remove", path) + self.merged_pr("behind-work", ahead) + self.age_branch("behind-work", 48) + self.wire_sweep(remote_tip=ahead) + self.assertEqual(0, self.sweep()) + self.assertIn("would-remove branch behind-work", self.out.getvalue()) + + def test_a_checkout_naming_another_repository_is_refused(self) -> None: + run_git(self.primary, "remote", "set-url", "origin", "https://github.com/other/repo.git") + self.wire_sweep() + self.assertEqual(1, self.sweep("--apply")) + self.assertIn("status=BAD_CHECKOUT", self.out.getvalue()) + + if __name__ == "__main__": unittest.main(verbosity=2)