diff --git a/README.md b/README.md index 03447bc..0191c99 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,7 @@ versioning features it can drive **end-to-end workflows**: | `pkgmgr worktree flow clear [PATH] [--json]` | Removes the merge flow | | `pkgmgr worktree cleanup [--apply]` | Removes clean worktrees whose commits are merged or pushed: pinned `worktree-fini` hook, compose stack, checkout; the branch stays | | `pkgmgr branch cleanup [--apply]` | Removes merged branches that no worktree holds, locally and on the remotes | +| `pkgmgr branch cleanup --local [--apply] [--json] [--merged-into REF]...` | Deletes only local branches whose tip is an ancestor of the remote-tracking default or of a `--merged-into` REF, each with `git update-ref --no-deref -d` against the planned sha and then its `branch.` section in the repository's own config file (`git config --file /config`; a section it cannot remove is noted on the done item), also when the branch has an upstream or remote copy. Every git call of `--local` runs without any inherited `GIT_*` variable but `GIT_CEILING_DIRECTORIES`, which only narrows repository discovery, so a directory below a ceiling is never resolved to the repository above it; only the config reads of `deleteMerged` and of the upstreams run twice, both with the system config: once with the caller's `GIT_CONFIG_GLOBAL`, `GIT_CONFIG_SYSTEM`, `GIT_CONFIG_PARAMETERS` and `GIT_CONFIG_COUNT` with its `GIT_CONFIG_KEY_` and `GIT_CONFIG_VALUE_`, once without any of them, without `XDG_CONFIG_HOME` and with `HOME` set to the passwd home, and a branch either read keeps stays, so the environment can add keeps but never remove one. A `--merged-into` REF must name exactly one branch or remote-tracking ref, as `git rev-parse --symbolic-full-name` resolves it, in every scanned repository git can read; `X~0`, `X^0`, `X@{0}`, a sha, a range or an unknown ref stops the run. The default comes from `origin`, or from the only remote when there is no `origin`: the ref `refs/remotes//HEAD`, else `main`, else `master` resolves to, every symref step inside `refs/remotes//`, so never a local branch or `HEAD`; without one, for example with several remotes and no `origin`, the repository is refused, also under `--merged-into`; so is one whose `refs`, `packed-refs`, `reftable`, `logs` or `logs/refs` is a symlink (git-new-workdir), one with a symlink anywhere below `refs/`, `reftable/` or `logs/refs/` (for example `refs/heads/team` linked to another clone's heads), one whose git dir is neither its git common dir nor directly under `/worktrees/` (a registration moved elsewhere), or one whose git output is not UTF-8. The repository is the one the current directory belongs to, also in a submodule or with a separate git dir; one whose main worktree belongs to another repository (`core.worktree`), whose git dir is neither its git common dir nor directly under `/worktrees/`, or whose path is not UTF-8 is refused with exit 2. Skips a symbolic branch, every branch a symbolic ref anywhere under `refs/` (also a tag or remote-tracking one) or under the main worktree's or any registration's `refs/worktree/` points to, a branch that is itself a `--merged-into` target, a branch a worktree holds or that the HEAD of the main repository or of any `worktrees/` registration names (also one `git worktree list` omits or whose id is no valid ref name, read through `git --git-dir= symbolic-ref HEAD`, so also under reftable), one a rebase or bisect in any worktree works on, one merged only by patch or PR, and, as `git branch --delete-merged` keeps them, one whose `branch..deleteMerged` is false in the config of any worktree (also its `config.worktree` or an `includeIf "onbranch:..."` file) or in the system config file, and the local upstream, as git's `%(upstream)` resolves it in any worktree, of any branch that stays, also of one that `--branch` or `--exclude` leaves out, each shown as kept with its reason; these holds are read through the git common dir, whichever worktree is scanned, and a registration, symbolic ref, HEAD or state file they cannot read refuses the repository; when a repository vanishes, moves or stops being readable, or its default or a `--merged-into` target stops resolving mid-run, its later items are skipped with that reason and the plan is still printed. A repository git or the filesystem fails on while planning (for example a dangling `.git` file under `--all`) is refused with that reason and the run goes on. The plan, refused repositories included, is printed (as JSON with `--json`), and the run exits 1 when an item failed, else 2 when a repository was refused, else 0. It stops with a message on stderr, no plan and exit 2 in exactly these cases: without `--all`, the current directory is not inside a git repository, its git dir is neither its git common dir nor directly under `/worktrees/`, its main worktree belongs to another repository (`core.worktree`), or its path or git output is not UTF-8; `--apply` outside the host PID namespace; a `--merged-into` REF refused as above; a `--branch` that matches no item (also one that exists only on a remote). Meant for working clones and their worktrees, not for a bare server repository. Not mirrored from `git branch --delete-merged`: git also keeps a branch whose push would update its own upstream; `--local` does not. A branch is deleted before its planned local upstream; right before each delete its holds and then its `deleteMerged` and the local upstreams of every branch still present are read again, and a branch one of them now keeps is skipped as changed since the plan (so is each branch of a local upstream cycle). Known residuals: a worktree, rebase or symbolic ref that starts holding a branch while its config is re-read, or a config change between that read and its `update-ref -d`, is not seen; an entry under `/worktrees/` that is no git dir (for example an empty leftover) refuses the repository. Reads no forge token and never fetches, pushes, deletes a remote branch, runs docker or repository hooks, touches a worktree, compose stack, volume or image, or refreshes the branch tree file (rerun `pkgmgr branch tree` for that) | | `pkgmgr branch tree [--format json\|yaml] [--output FILE]` | Writes every managed repository's branches and worktrees with their cleanup state, without fetching | | `pkgmgr gc [--all] [--apply]` | Both cleanups in one pass, for the current repository or every managed one | | `pkgmgr gc --report [--all] [--root DIR] [--threshold-mb MB] [--docker] [--json]` | Read-only disk inventory: worktree sizes, changes a cleanup would keep, `--root` children, core dumps, docker usage, free space per filesystem | diff --git a/src/pkgmgr/actions/branch/gc.py b/src/pkgmgr/actions/branch/gc.py index 1d51eff..9c4d7bd 100644 --- a/src/pkgmgr/actions/branch/gc.py +++ b/src/pkgmgr/actions/branch/gc.py @@ -8,9 +8,10 @@ import sys import time from collections.abc import Callable from dataclasses import asdict, dataclass, field +from graphlib import CycleError, TopologicalSorter from pkgmgr.actions.worktree.hooks import FINI_HOOK, hook_command, run_hook -from pkgmgr.core.git.errors import GitRunError +from pkgmgr.core.git.errors import GitBaseError, GitRunError from pkgmgr.core.git.queries import ( FILTER_CONFIG, UNSAFE_CONFIG, @@ -18,7 +19,13 @@ from pkgmgr.core.git.queries import ( local_config, resolve_base_branch, ) -from pkgmgr.core.git.run import END_OF_OPTIONS, SCAN_FLAGS, run +from pkgmgr.core.git.run import ( + CONFIG_VIEWS, + END_OF_OPTIONS, + SCAN_FLAGS, + run, + strict_env, +) from .gc_forge import ForgeFacts, fetch_forge_facts from .nofollow import canonical, remove_tree @@ -53,8 +60,24 @@ REFUSAL = ( "Refusing: not in the host PID namespace (e.g. a sandbox), so running " "processes and worktree locks cannot be verified. Run it on the host." ) -SCOPES = ("worktree", "branch", "all", "purge") +SCOPES = ("worktree", "branch", "all", "purge", "local") NOT_LINKED = "not a linked worktree of this repository" +NO_REMOTE_DEFAULT = ( + "refused: no remote-tracking default (origin, else the only remote: " + "refs/remotes//HEAD, main or master, resolving inside " + "refs/remotes//), so --local cannot tell which branches are merged" +) +NOT_UTF8 = "refused: git output (ref, remote or path names) is not UTF-8" +DEFAULT_NAMES = ("HEAD", "main", "master") +HELD_FILES = ( + ("rebase-merge/head-name", "rebase in progress"), + ("rebase-merge/update-refs", "rebase in progress"), + ("rebase-apply/head-name", "rebase in progress"), + ("BISECT_START", "bisect in progress"), +) +DELETE_MERGED_KEY = r"^branch\..*\.deletemerged$" +SHARED_STORES = ("refs", "packed-refs", "reftable", "logs", "logs/refs") +WALKED_STORES = ("refs", "reftable", "logs/refs") @dataclass @@ -136,6 +159,10 @@ class Repo: warnings: list[str] = field(default_factory=list) items: list[Item] = field(default_factory=list) unsafe: list[str] = field(default_factory=list) + refused: str | None = None + common: str = "" + opted_out: set[str] = field(default_factory=set) + upstreams: set[tuple[str, str]] = field(default_factory=set) def host_pid_namespace() -> bool: @@ -149,10 +176,93 @@ def main_repo_root(cwd: str) -> str: return os.path.dirname(_common_dir(cwd)) +def local_repo_root(cwd: str) -> str: + """ + Resolve the main worktree of the repository cwd belongs to, for scope + `local`. + + Args: + cwd: a directory inside a checkout, also of a submodule or a + repository with a separate git dir. + + Returns: + `git rev-parse --show-toplevel` when cwd is in the main worktree, + else the first `git worktree list` entry; either way of cwd's own + repository. + + Raises: + GitBaseError: when that root's git common dir is not cwd's, when + cwd's git dir is not its repository's own (see _own_git_dir), + when either is no repository, or when git output is not UTF-8. + """ + try: + with strict_env(): + common, git_dir = _own_git_dir(cwd) + in_main = git_dir == common + root = (in_main and _toplevel(cwd)) or _worktrees(cwd)[0].path + root_common = os.path.realpath(_common_dir(root)) + except UnicodeDecodeError as exc: + raise GitBaseError(f"{NOT_UTF8} for {cwd!r}") from exc + except OSError as exc: + raise GitBaseError(f"{cwd}: {exc}") from exc + if root_common != common: + raise GitBaseError( + f"{cwd}: main worktree {root} belongs to another repository than {common}" + ) + return root + + +def _own_git_dir(cwd: str) -> tuple[str, str]: + """ + Resolve cwd's git common dir and git dir, refusing a foreign git dir. + + Args: + cwd: a directory inside a checkout or a git dir. + + Returns: + The realpaths of `git rev-parse --git-common-dir` and + `--absolute-git-dir`. + + Raises: + GitBaseError: when the git dir is neither the common dir nor directly + under `/worktrees/`. + """ + common = os.path.realpath(_common_dir(cwd)) + git_dir = os.path.realpath(_git(["rev-parse", "--absolute-git-dir"], cwd)) + if git_dir != common and os.path.dirname(git_dir) != os.path.join( + common, "worktrees" + ): + raise GitBaseError( + f"{cwd}: git dir {git_dir} is neither {common} nor directly under " + f"{common}/worktrees" + ) + return common, git_dir + + +def _toplevel(cwd: str) -> str: + """ + Read the work tree root of cwd's checkout. + + Args: + cwd: a directory inside a checkout. + + Returns: + `git rev-parse --show-toplevel`, or "" in a bare repository. + """ + try: + return _git(["rev-parse", "--show-toplevel"], cwd) + except GitRunError: + return "" + + def _git(args: list[str], cwd: str) -> str: return run(args, cwd=cwd, scan=True) +def _config_read(args: list[str], cwd: str, view: str) -> str: + return run(args, cwd=cwd, scan=True, config_view=view) + + def _common_dir(cwd: str) -> str: return _git(["rev-parse", "--path-format=absolute", "--git-common-dir"], cwd) @@ -415,6 +525,26 @@ def human_size(size: int) -> str: return f"{value / 1024:.1f}T" +def _needs_forge(item: Item, tree: Worktree | None) -> bool: + """ + Whether deciding the item needs the forge's PR and protection facts. + + Args: + item: the classified item. + tree: the worktree holding its branch, or None. + + Returns: + False only for a branch that no worktree holds, without upstream and + remote branch, merged into the default by ancestry; True otherwise. + """ + return not ( + tree is None + and item.merged_by == "ancestor" + and item.upstream is None + and not item.remotes + ) + + def _skip( repo: Repo, item: Item, @@ -425,12 +555,14 @@ def _skip( number = repo.forge.open_heads.get(item.branch) if item.branch else None if number: return f"open-pr #{number}" - if repo.forge.unavailable: + if repo.forge.unavailable and _needs_forge(item, tree): return "forge unavailable" if tree is None: return None if tree.main: return "checked out in main worktree" + if repo.scope == "local": + return f"checked out in worktree {tree.path}" if not checkout: return "worktree checkout missing" if repo.containers is None: @@ -641,6 +773,8 @@ def _actions( dirty: int, registrations: Registrations, ) -> list[str]: + if repo.scope == "local": + return _local_actions(repo, item, tree) tree_part: list[str] = [] if tree: if ( @@ -679,6 +813,38 @@ def _actions( return actions +def _local_actions(repo: Repo, item: Item, tree: Worktree | None) -> list[str]: + """ + Settle the actions of an item planned with scope `local`. + + Args: + repo: the planned repository; its default and merged-into lines are + the refs the branch tip must be an ancestor of. + item: the classified item; a merged-clean one whose tip is an + ancestor of none of them gets that as its skip. + tree: the worktree holding its branch, or None. + + Returns: + `local-branch` for a merged-clean local branch that no worktree, + worktree registration, rebase, bisect or symbolic ref holds and whose + tip is an ancestor of the default or a merged-into line, else + nothing; never a worktree, stack or remote action. + """ + if item.state != "merged-clean" or not item.local_sha or tree is not None: + return [] + held = _held_branches(repo.repo).get(item.branch) + if held: + item.skip = item.skip or held + return [] + targets = [repo.default, *repo.lines] + if any(_is_ancestor(item.local_sha, ref, repo.repo) for ref in targets): + return ["local-branch"] + item.skip = item.skip or ( + f"merged by {item.merged_by}, tip is no ancestor of {', '.join(targets)}" + ) + return [] + + def _snapshot(repo: Repo) -> tuple[list[Worktree], Snapshot]: local, remote, tracking = _refs(repo.repo, repo.remotes) cwds = _process_cwds() if repo.host_ns else {} @@ -695,12 +861,15 @@ def _items(repo: Repo) -> list[Item]: trees, snapshot = _snapshot(repo) local, remote, *_ = snapshot by_branch = {t.branch: t for t in trees if t.branch} - names = ( - sorted(by_branch) - if repo.scope == "worktree" - else sorted(set(local) | set(remote) | set(by_branch)) - ) - items = [_build(repo, name, by_branch.get(name), snapshot) for name in names] + names = { + "worktree": set(by_branch), + "local": set(local) | set(by_branch), + }.get(repo.scope, set(local) | set(remote) | set(by_branch)) + items = [ + _build(repo, name, by_branch.get(name), snapshot) for name in sorted(names) + ] + if repo.scope == "local": + return items items += [ _build(repo, None, tree, snapshot) for tree in trees @@ -727,6 +896,188 @@ def _stale(repo_root: str) -> list[Item]: return items +def _default_remote(remotes: list[str]) -> str | None: + """ + Pick the one remote whose default scope `local` may measure against. + + Args: + remotes: the configured remote names. + + Returns: + origin when configured, else the only remote, else None. + """ + if "origin" in remotes: + return "origin" + return remotes[0] if len(remotes) == 1 else None + + +def _symbolic(repo_root: str, prefix: str, git_dir: str = "") -> dict[str, str]: + """ + List the resolvable refs under a prefix with their symref targets. + + Args: + repo_root: main worktree root of the repository. + prefix: a ref prefix such as `refs/heads` or `refs/remotes/origin`. + git_dir: the git common dir (shared refs plus the main worktree's own) + or a `worktrees/` dir whose refs to list, or "" for the refs + as repo_root's worktree sees them. + + Returns: + Full ref name -> the ref it points to, or "" for a plain ref. + """ + args = ["for-each-ref", "--format=%(refname)%00%(symref)", prefix] + out = _git([f"--git-dir={git_dir}", *args] if git_dir else args, repo_root) + return dict(line.split("\x00", 1) for line in out.splitlines()) + + +def _resolve_within(refs: dict[str, str], ref: str) -> str | None: + """ + Follow a ref's symref chain without leaving a set of refs. + + Args: + refs: refs under one prefix, as _symbolic lists them. + ref: the full ref name to resolve. + + Returns: + The plain ref the chain ends at, or None when a step leaves refs, + loops or ref is missing. + """ + seen: set[str] = set() + while ref in refs and ref not in seen: + seen.add(ref) + if not refs[ref]: + return ref + ref = refs[ref] + return None + + +def _remote_default(repo_root: str, remotes: list[str]) -> str | None: + """ + Pick the remote-tracking ref that scope `local` measures merges against. + + Args: + repo_root: main worktree root of the repository. + remotes: the configured remote names. + + Returns: + For the remote _default_remote picks, the plain ref its HEAD, else + main, else master resolves to while every step stays inside + `refs/remotes//`; else None. + """ + remote = _default_remote(remotes) + if remote is None: + return None + refs = _symbolic(repo_root, f"refs/remotes/{remote}") + for name in DEFAULT_NAMES: + target = _resolve_within(refs, f"refs/remotes/{remote}/{name}") + if target: + return target + return None + + +def _worktree_heads(repo_root: str, git_dirs: list[str]) -> dict[str, str]: + """ + Read the branch every worktree's HEAD names, on any ref backend. + + Args: + repo_root: main worktree root of the repository. + git_dirs: the git common dir first, then each `worktrees/`. + + Returns: + Branch name -> skip reason, from `git --git-dir= symbolic-ref + HEAD` for each of them, whatever its id. + + Raises: + GitBaseError: when a HEAD is neither readable nor detached. + """ + heads: dict[str, str] = {} + for git_dir in git_dirs: + try: + target = _git( + [f"--git-dir={git_dir}", "symbolic-ref", "--quiet", "HEAD"], repo_root + ) + except GitRunError as exc: + if exc.returncode == 1: + continue + raise + if target.startswith("refs/heads/"): + heads.setdefault( + target.removeprefix("refs/heads/"), f"checked out in {git_dir}" + ) + return heads + + +def _git_dirs(repo_root: str) -> list[str]: + """ + List the git dirs of every worktree of the repository. + + Args: + repo_root: main worktree root of the repository. + + Returns: + The git common dir, then each entry of `/worktrees/`, whatever + its id and whether `git worktree list` shows it. + """ + common = _common_dir(repo_root) + base = os.path.join(common, "worktrees") + entries = sorted(os.listdir(base)) if os.path.isdir(base) else [] + return [common, *(os.path.join(base, entry) for entry in entries)] + + +def _held_branches(repo_root: str) -> dict[str, str]: + """ + Find the local branches that something besides a listed worktree holds. + + Args: + repo_root: main worktree root of the repository. + + Returns: + Branch name -> skip reason: a symbolic branch, the branch any symbolic + ref under `refs/` (also a tag or remote-tracking one) or under the + main worktree's or any registration's `refs/worktree/` names, the HEAD + of the main repository and of every `worktrees/` registration, + also one `git worktree list` omits, each rebase-merge and rebase-apply + head-name, an interactive rebase's update-refs and BISECT_START; all + read through the git common dir, whichever worktree repo_root is. + + Raises: + GitBaseError: when git cannot read a registration, symref or HEAD. + OSError: when a state file exists but cannot be read. + ValueError: when a state file is not UTF-8. + """ + git_dirs = _git_dirs(repo_root) + held: dict[str, str] = {} + symrefs = _symbolic(repo_root, "refs", git_dirs[0]) + for git_dir in git_dirs[1:]: + for ref, target in _symbolic(repo_root, "refs/worktree", git_dir).items(): + symrefs[f"worktrees/{os.path.basename(git_dir)}/{ref}"] = target + for ref, target in symrefs.items(): + if target and ref.startswith("refs/heads/"): + held.setdefault( + ref.removeprefix("refs/heads/"), f"symbolic ref to {target}" + ) + if target.startswith("refs/heads/"): + held.setdefault( + target.removeprefix("refs/heads/"), f"target of symbolic ref {ref}" + ) + for branch, reason in _worktree_heads(repo_root, git_dirs).items(): + held.setdefault(branch, reason) + for git_dir in git_dirs: + for name, operation in HELD_FILES: + try: + with open(os.path.join(git_dir, name), encoding="utf-8") as fh: + lines = fh.read().splitlines() + except FileNotFoundError: + continue + for line in lines: + ref = line.strip().removeprefix("ref: ") + if ref.startswith("refs/heads/") or name == "BISECT_START": + held.setdefault( + ref.removeprefix("refs/heads/"), f"{operation} in {git_dir}" + ) + return held + + def _lines( repo_root: str, refs: list[str], remotes: list[str], warnings: list[str] ) -> tuple[dict[str, tuple[str, set[str]]], set[str]]: @@ -758,6 +1109,116 @@ def _lines( return lines, names +def _branch_names(repo_root: str, ref: str, remotes: list[str]) -> set[str]: + """ + Name the branch a merged-into ref is, whatever form it was given in. + + Args: + repo_root: main worktree root of the repository. + ref: the ref as given, e.g. `refs/heads/x`, `x` or `origin/x`. + remotes: the configured remote names. + + Returns: + `x` for a ref that resolves to `refs/heads/x` or + `refs/remotes//x`, else nothing. + """ + out = _git(["rev-parse", "--symbolic-full-name", END_OF_OPTIONS, ref], repo_root) + full = out.splitlines()[-1] if out else "" + prefixes = ["refs/heads/", *(f"refs/remotes/{r}/" for r in remotes)] + return {full.removeprefix(p) for p in prefixes if full.startswith(p)} + + +def _raise(exc: OSError) -> None: + raise exc + + +def _shared_refs(repo_root: str) -> str: + """ + Detect a ref store that other checkouts share through a symlink. + + Args: + repo_root: main worktree root of the repository. + + Returns: + The refusal reason when `refs`, `packed-refs`, `reftable`, `logs` or + `logs/refs` in the git common dir is a symlink (git-new-workdir), or + any entry below `refs/`, `reftable/` or `logs/refs/` is, else "". + + Raises: + OSError: when a directory below them cannot be listed. + """ + common = _common_dir(repo_root) + shared = [ + name for name in SHARED_STORES if os.path.islink(os.path.join(common, name)) + ] + stores = [] if shared else [os.path.join(common, s) for s in WALKED_STORES] + for store in filter(os.path.isdir, stores): + for parent, dirs, files in os.walk(store, onerror=_raise): + shared += [ + os.path.relpath(os.path.join(parent, name), common) + for name in sorted(dirs + files) + if os.path.islink(os.path.join(parent, name)) + ] + if not shared: + return "" + return ( + f"refused: {', '.join(shared)} in {common} is a symlink, so checkouts " + "sharing these refs hold branches git cannot list" + ) + + +def _refused( + repo_root: str, + containers: list[Container] | None, + host_ns: bool, + reason: str, + warnings: list[str], + *, + scope: str, + sizes: bool, + min_age: float | None, + activity: bool, + **_options: object, +) -> Repo: + """ + Build a repository that scope `local` refuses: no items, nothing deleted. + + Args: + repo_root: main worktree root of the repository. + containers: as plan_repo takes them. + host_ns: as plan_repo takes it. + reason: why the repository is refused; also appended to warnings. + warnings: warnings gathered so far. + scope: the planned scope. + sizes: as plan_repo takes it. + min_age: as plan_repo takes it. + activity: as plan_repo takes it. + _options: the other plan_repo options, unused. + + Returns: + The refused repository. + """ + return Repo( + repo=repo_root, + default="", + remotes=[], + protected=set(), + first_parent=set(), + urls={}, + forge=ForgeFacts(), + containers=containers, + host_ns=host_ns, + lines={}, + sizes=sizes, + min_age=min_age, + scope=scope, + activity=activity, + hooks={}, + warnings=[*warnings, reason], + refused=reason, + ) + + def plan_repo( repo_root: str, containers: list[Container] | None, @@ -795,6 +1256,8 @@ def plan_repo( """ if scope not in SCOPES: raise ValueError(f"unknown scope {scope!r}") + if scope == "local": + _own_git_dir(repo_root) warnings: list[str] = [] unsafe = local_config(repo_root, UNSAFE_CONFIG) if unsafe: @@ -834,9 +1297,25 @@ def plan_repo( ) except GitRunError: continue - default = next(iter(heads.values()), None) or resolve_base_branch( - "main", "master", cwd=repo_root, scan=True - ) + if scope == "local": + default = _remote_default(repo_root, remotes) + reason = _shared_refs(repo_root) or (NO_REMOTE_DEFAULT if not default else "") + if reason: + return _refused( + repo_root, + containers, + host_ns, + reason, + warnings, + scope=scope, + sizes=sizes, + min_age=min_age, + activity=activity, + ) + else: + default = next(iter(heads.values()), None) or resolve_base_branch( + "main", "master", cwd=repo_root, scan=True + ) head_names = {ref[len(remote) + 1 :] for remote, ref in heads.items()} urls = {r: _git(["remote", "get-url", "--", r], repo_root) for r in remotes} @@ -844,6 +1323,10 @@ def plan_repo( if forge.unavailable: warnings.append(f"forge unavailable: {forge.unavailable}") lines, line_names = _lines(repo_root, merged_into, remotes, warnings) + if scope == "local": + line_names |= { + name for ref in lines for name in _branch_names(repo_root, ref, remotes) + } repo = Repo( repo=repo_root, @@ -875,9 +1358,140 @@ def plan_repo( unsafe=unsafe, ) repo.items = _items(repo) + if scope == "local": + repo.opted_out = _opted_out(repo_root) + repo.upstreams = _local_upstreams(repo_root) return repo +def _opted_out(repo_root: str) -> set[str]: + """ + Find the branches whose `branch..deleteMerged` is false. + + Args: + repo_root: main worktree root of the repository. + + Returns: + Each branch that `git --git-dir= config` sets false in, for each + of _git_dirs (so also config.worktree and includeIf onbranch) in each + of CONFIG_VIEWS, so the union of what the caller's config variables + and the default system and global config give. + + Raises: + GitBaseError: when git cannot read one of these configs. + """ + names: set[str] = set() + for git_dir in _git_dirs(repo_root): + for view in CONFIG_VIEWS: + try: + out = _config_read( + [ + f"--git-dir={git_dir}", + "config", + "--null", + "--type=bool", + "--get-regexp", + DELETE_MERGED_KEY, + ], + repo_root, + view, + ) + except GitRunError as exc: + if exc.returncode == 1: + continue + raise + for record in out.split("\0"): + key, _, value = record.partition("\n") + if value == "false": + names.add(key.removeprefix("branch.").removesuffix(".deletemerged")) + return names + + +def _local_upstreams(repo_root: str) -> set[tuple[str, str]]: + """ + Read each local branch's upstream as git resolves it in any worktree. + + Args: + repo_root: main worktree root of the repository. + + Returns: + (branch, upstream branch) for each `%(upstream)` under `refs/heads/` + that `git --git-dir= for-each-ref` gives for any of _git_dirs in + any of CONFIG_VIEWS. + """ + pairs: set[tuple[str, str]] = set() + for git_dir in _git_dirs(repo_root): + for view in CONFIG_VIEWS: + out = _config_read( + [ + f"--git-dir={git_dir}", + "for-each-ref", + "--format=%(refname)%00%(upstream)", + "refs/heads", + ], + repo_root, + view, + ) + for line in out.splitlines(): + ref, _, upstream = line.partition("\0") + if upstream.startswith("refs/heads/"): + pairs.add( + ( + ref.removeprefix("refs/heads/"), + upstream.removeprefix("refs/heads/"), + ) + ) + return pairs + + +def _keep_by_config(repo: Repo, chosen: Callable[[Item], bool]) -> None: + """ + Keep, as `git branch --delete-merged` does, each branch whose + `branch..deleteMerged` is false and each local upstream of a branch + that stays, also of one kept so or one the filters leave out. + + Args: + repo: a repository planned with scope `local`; each kept item that + would be deleted loses its actions and gets the reason as its skip. + chosen: the --branch and --exclude filter; only an eligible chosen + item is deleted, every other branch stays. + """ + pending = { + item.branch: item for item in repo.items if item.eligible and chosen(item) + } + for name, reason in _config_keeps( + repo.opted_out, repo.upstreams, set(pending) + ).items(): + item = pending[name] + item.actions = [] + item.skip = item.skip or reason + + +def _config_keeps( + opted_out: set[str], upstreams: set[tuple[str, str]], pending: set[str] +) -> dict[str, str]: + """ + Args: + opted_out: branches whose `branch..deleteMerged` is false. + upstreams: (branch, local upstream branch) pairs. + pending: the branches that would be deleted; every other stays. + + Returns: + Pending branch -> keep reason, for each one opted out or the local + upstream of a branch that stays, also of one kept so. + """ + reasons = {name: f"kept: branch.{name}.deleteMerged is false" for name in opted_out} + leaving = set(pending) + while True: + for child, parent in sorted(upstreams): + if child not in leaving: + reasons.setdefault(parent, f"kept: local upstream of {child}") + kept = reasons.keys() & leaving + if not kept: + return {name: reasons[name] for name in pending - leaving} + leaving -= kept + + def _pins( hooks: dict[str, dict[str, list[str]]], repo_root: str, warnings: list[str] ) -> dict[str, list[str]]: @@ -985,6 +1599,32 @@ def _changed_by_hook(repo: Repo, item: Item) -> str | None: return _busy(_process_cwds(), path) if repo.host_ns else None +def _drop_branch_config(repo_root: str, branch: str) -> str | None: + """ + Remove a deleted branch's `branch.` section, as `branch -D` does, + from the config file in the repository's own git common dir. + + Args: + repo_root: main worktree root of the repository. + branch: the deleted branch; a missing section is no error. + + Returns: + None, or a note why the section was kept, e.g. a locked config. + """ + config = os.path.join(_common_dir(repo_root), "config") + try: + _git( + ["config", "--file", config, "--remove-section", f"branch.{branch}"], + repo_root, + ) + except GitRunError as exc: + stderr = (exc.stderr or "").strip() + if "no such section" in stderr.lower(): + return None + return f"config branch.{branch} kept: {(stderr or str(exc)).splitlines()[0]}" + return None + + def execute(repo: Repo, item: Item) -> None: fresh = _rebuild(repo, item) if _fingerprint(fresh) != _fingerprint(item): @@ -1100,6 +1740,153 @@ def _render( ) +def _plan( + root: str, containers: list[Container] | None, host_ns: bool, options: dict +) -> Repo: + """ + Plan one repository; under scope `local` refuse, instead of raising, one + that git or the filesystem fails on or whose git output is not UTF-8. + + Args: + root: main worktree root of the repository. + containers: as plan_repo takes them. + host_ns: as plan_repo takes it. + options: the plan_repo keyword options. + + Returns: + The planned repository, under scope `local` with its realpath git + common dir, or the refused one. + """ + if options["scope"] != "local": + return plan_repo(root, containers, host_ns, **options) + try: + repo = plan_repo(root, containers, host_ns, **options) + repo.common = os.path.realpath(_common_dir(root)) + return repo + except UnicodeDecodeError: + reason = NOT_UTF8 + except (GitBaseError, OSError, ValueError) as exc: + reason = f"refused: {_describe(exc)}" + return _refused(root, containers, host_ns, reason, [], **options) + + +def _gone(repo: Repo) -> str: + """ + Check that a repository planned with scope `local` is still the one + planned and that its targets still resolve. + + Args: + repo: the planned repository. + + Returns: + Why its remaining items must be skipped, else "". + """ + try: + if os.path.realpath(_common_dir(repo.repo)) != repo.common: + return f"repository {repo.repo} is no longer the planned one" + gone = [ + ref + for ref in (repo.default, *repo.lines) + if _sha(repo.repo, f"{ref}^{{commit}}") is None + ] + except (GitBaseError, OSError, ValueError) as exc: + return f"repository {repo.repo} unreadable: {_describe(exc)}" + return f"target {', '.join(gone)} is gone" if gone else "" + + +def _delete_order(repo: Repo, items: list[Item]) -> list[Item]: + """ + Args: + repo: a repository planned with scope `local`. + items: its eligible chosen items, in plan order. + + Returns: + The items with each branch before its local upstream, as the plan + read them; plan order on an upstream cycle, whose members the + re-check of _delete_local then keeps. + """ + by_name = {item.branch: item for item in items} + graph: dict[str, set[str]] = {name: set() for name in by_name} + for child, parent in sorted(repo.upstreams): + if child != parent and child in by_name and parent in by_name: + graph[parent].add(child) + try: + return [by_name[name] for name in TopologicalSorter(graph).static_order()] + except CycleError: + return items + + +def _delete_local(repo: Repo, item: Item) -> None: + """ + Args: + repo: a repository planned with scope `local`. + item: an eligible item, deleted after every planned branch whose + local upstream it is; it ends done, or skipped when its holds, + then its `deleteMerged` and the upstreams of every branch still + present, re-read right before the delete, keep it. + + Raises: + GitBaseError: when git cannot re-read the repository or the sha moved. + OSError: when a state file cannot be read. + ValueError: when a state file is not UTF-8. + """ + fresh = _rebuild(repo, item) + if _fingerprint(fresh) != _fingerprint(item): + item.result = f"skipped: changed since plan ({fresh.skip or fresh.reason})" + return + kept = _config_keeps( + _opted_out(repo.repo), _local_upstreams(repo.repo), {item.branch} + ).get(item.branch) + if kept: + item.result = f"skipped: changed since plan ({kept})" + return + _git( + ["update-ref", "--no-deref", "-d", f"refs/heads/{item.branch}", item.local_sha], + repo.repo, + ) + note = _drop_branch_config(repo.repo, item.branch) + item.result = "; ".join(["done", *([note] if note else [])]) + + +def _apply(repo: Repo, item: Item) -> None: + """ + Execute one planned item; under scope `local` without raising. + + Args: + repo: the planned repository. + item: an eligible item; under scope `local` it ends with a done, + failed or skipped result, skipped as _gone tells. + """ + if repo.scope != "local": + execute(repo, item) + return + gone = _gone(repo) + if gone: + item.result = f"skipped: {gone}" + return + try: + _delete_local(repo, item) + except (RuntimeError, OSError, ValueError) as exc: + item.result = f"failed: {_describe(exc)}" + + +def _exit_code(repos: list[Repo], apply: bool) -> int: + """ + Settle the exit code of a cleanup run. + + Args: + repos: the planned and, with apply, executed repositories. + apply: whether the plan was executed. + + Returns: + 1 when an applied item failed, else 2 when a repository was + refused, else 0. + """ + if apply and _finish(repos): + return 1 + return 2 if any(repo.refused for repo in repos) else 0 + + def prepare( repo_roots: list[str], include: list[str], @@ -1107,8 +1894,8 @@ def prepare( options: dict, ) -> tuple[list[Repo], bool, Callable[[Item], bool]] | None: host_ns = host_pid_namespace() - containers = list_compose_containers() - repos = [plan_repo(root, containers, host_ns, **options) for root in repo_roots] + containers = [] if options["scope"] == "local" else list_compose_containers() + repos = [_plan(root, containers, host_ns, options) for root in repo_roots] unknown = [ name for name in include @@ -1123,13 +1910,16 @@ def prepare( item.matches(n) for n in exclude ) + for repo in repos: + if repo.scope == "local": + _keep_by_config(repo, chosen) return repos, host_ns, chosen def _finish(repos: list[Repo]) -> int: results = [item.result for repo in repos for item in repo.items if item.result] if any(result.startswith("done") for result in results) and any( - repo.containers is not None for repo in repos + repo.containers is not None and repo.scope != "local" for repo in repos ): for args in (["image", "prune", "-f"], ["builder", "prune", "-f"]): try: @@ -1160,7 +1950,16 @@ def cleanup( scope: `worktree` removes clean worktrees whose commits are merged or pushed (hook, stack, checkout; the branch stays); `branch` removes merged branches without a worktree, local and remote; `all` does - both in one pass. + both in one pass; `local` deletes, with `update-ref -d` against + the planned sha, each local branch that no worktree, rebase or + bisect holds, that _keep_by_config does not keep, both re-checked + right before its delete, which follows the deletes of the + branches it is the local upstream of, and whose tip is an + ancestor of the + remote-tracking + default or a merged_into line, refuses a repository without such + a default, and reads no forge token, runs no fetch, push, docker, + hook or prune. apply: execute the plan instead of only printing it. as_json: print the plan as JSON instead of text lines. include: only items whose branch or worktree path matches one of these. @@ -1172,9 +1971,10 @@ def cleanup( CLI; a worktree-fini hook runs only for a root pinned there. Returns: - 0 on success, 1 when an applied item failed, 2 when refused. + As _exit_code settles it: 1 when an applied item failed, else 2 when + refused (also a repository scope `local` refused), else 0. """ - if scope not in ("worktree", "branch", "all"): + if scope not in ("worktree", "branch", "all", "local"): raise ValueError(f"unknown cleanup scope {scope!r}") if apply and not host_pid_namespace(): print(REFUSAL, file=sys.stderr) @@ -1186,7 +1986,76 @@ def cleanup( "scope": scope, "activity": False, "hooks": hooks, + "fetch": scope != "local", + "forge_source": (lambda _urls: ForgeFacts()) if scope == "local" else None, } + if scope != "local": + return _run(repo_roots, options, include, exclude, apply, as_json) + with strict_env(): + bad = _bad_lines(repo_roots, merged_into) + if bad: + print(f"Refusing: {bad}", file=sys.stderr) + return 2 + return _run(repo_roots, options, include, exclude, apply, as_json) + + +def _bad_lines(repo_roots: list[str], merged_into: list[str]) -> str: + """ + Check that every merged-into ref of scope `local` names one branch. + + Args: + repo_roots: main worktree roots of the repositories to scan; one git + cannot read is left to the planning, which refuses it. + merged_into: the refs as given. + + Returns: + Why a ref is refused, when in some repository it is unknown or + `rev-parse --symbolic-full-name` does not name exactly one ref under + `refs/heads/` or `refs/remotes/`, else "". + """ + for root in repo_roots: + for ref in merged_into: + try: + out = _git( + ["rev-parse", "--symbolic-full-name", END_OF_OPTIONS, ref], root + ) + except GitRunError: + return f"--merged-into {ref} names no ref in {root}" + except (GitBaseError, OSError, ValueError): + break + names = [line for line in out.splitlines() if line != END_OF_OPTIONS] + if len(names) != 1 or not names[0].startswith( + ("refs/heads/", "refs/remotes/") + ): + return ( + f"--merged-into {ref} is not exactly one branch or " + f"remote-tracking ref in {root}" + ) + return "" + + +def _run( + repo_roots: list[str], + options: dict, + include: list[str], + exclude: list[str], + apply: bool, + as_json: bool, +) -> int: + """ + Plan, print and with apply execute a cleanup, as cleanup describes it. + + Args: + repo_roots: main worktree roots of the repositories to scan. + options: the plan_repo keyword options. + include: as cleanup takes it. + exclude: as cleanup takes it. + apply: as cleanup takes it. + as_json: as cleanup takes it. + + Returns: + 2 for an unknown include, else as _exit_code settles it. + """ prepared = prepare(repo_roots, include, exclude, options) if prepared is None: return 2 @@ -1194,15 +2063,16 @@ def cleanup( if apply: for repo in repos: - for item in repo.items: - if item.eligible and chosen(item): - execute(repo, item) - print( - f"[{scope}] {item.branch or item.worktree}: {item.result}", - file=sys.stderr, - ) + todo = [item for item in repo.items if item.eligible and chosen(item)] + for item in _delete_order(repo, todo) if repo.scope == "local" else todo: + _apply(repo, item) + print( + f"[{options['scope']}] {item.branch or item.worktree}: " + f"{item.result}", + file=sys.stderr, + ) _render(repos, host_ns, chosen, as_json) - return _finish(repos) if apply else 0 + return _exit_code(repos, apply) def branch_purge( diff --git a/src/pkgmgr/cli/commands/branch.py b/src/pkgmgr/cli/commands/branch.py index a178ea7..099531f 100644 --- a/src/pkgmgr/cli/commands/branch.py +++ b/src/pkgmgr/cli/commands/branch.py @@ -11,7 +11,7 @@ from pkgmgr.actions.branch import ( open_branch, tree, ) -from pkgmgr.actions.branch.gc import main_repo_root +from pkgmgr.actions.branch.gc import local_repo_root, main_repo_root from pkgmgr.cli.context import CLIContext from pkgmgr.core.git.errors import GitBaseError from pkgmgr.core.repository.dir import get_repo_dir @@ -60,14 +60,23 @@ def refresh_tree(roots: list[str]) -> None: print(f"WARN tree file not rewritten for {root}: {exc}", file=sys.stderr) -def scan_repo_roots(args, ctx: CLIContext) -> list[str]: +def scan_repo_roots(args, ctx: CLIContext, local: bool = False) -> list[str]: """ Resolve the repositories a scan covers: the current one, or every pkgmgr-managed one with --all. Exits with 2 outside a repository. + + Args: + args: parsed arguments; `all` selects every managed repository. + ctx: CLI context. + local: resolve the current repository as scope `local` does, also in + a submodule or with a separate git dir, refusing a mismatch. + + Returns: + The main worktree roots to scan. """ if not args.all: try: - return [main_repo_root(".")] + return [(local_repo_root if local else main_repo_root)(".")] except GitBaseError as exc: print(f"Not inside a git repository: {exc}", file=sys.stderr) sys.exit(2) @@ -84,7 +93,7 @@ def scan_options(args) -> dict: def run_cleanup(args, ctx: CLIContext, scope: str) -> None: - roots = scan_repo_roots(args, ctx) + roots = scan_repo_roots(args, ctx, local=scope == "local") code = cleanup( roots, scope=scope, @@ -94,7 +103,7 @@ def run_cleanup(args, ctx: CLIContext, scope: str) -> None: hooks=pinned_hooks(ctx), **scan_options(args), ) - if args.apply: + if args.apply and scope != "local": refresh_tree(roots) sys.exit(code) @@ -107,8 +116,9 @@ def handle_branch(args, ctx: CLIContext) -> None: - pkgmgr branch open [] [--base ] - pkgmgr branch close [] [--base ] [--force|-f] - pkgmgr branch drop [] [--base ] [--force|-f] - - pkgmgr branch cleanup [--apply] [--json] [--all] [--branch N]... [--exclude N]... - [--merged-into REF]... [--sizes] [--min-age HOURS] + - pkgmgr branch cleanup [--apply] [--json] [--local] [--all] [--branch N]... + [--exclude N]... [--merged-into REF]... [--sizes] + [--min-age HOURS] - pkgmgr branch purge [--all] [--branch N]... [--exclude N]... [--merged-into REF]... [--sizes] [--min-age HOURS] - pkgmgr branch tree [--format json|yaml] [--output FILE] @@ -124,7 +134,7 @@ def handle_branch(args, ctx: CLIContext) -> None: ) if args.subcommand == "cleanup": - run_cleanup(args, ctx, "branch") + run_cleanup(args, ctx, "local" if args.local else "branch") if args.subcommand == "purge": roots = scan_repo_roots(args, ctx) diff --git a/src/pkgmgr/cli/parser/branch_cmd.py b/src/pkgmgr/cli/parser/branch_cmd.py index 2dc8104..682fc38 100644 --- a/src/pkgmgr/cli/parser/branch_cmd.py +++ b/src/pkgmgr/cli/parser/branch_cmd.py @@ -105,6 +105,41 @@ def add_branch_subparsers( ), ) add_apply_options(branch_cleanup) + branch_cleanup.add_argument( + "--local", + action="store_true", + help=( + "Delete only local branches whose tip is an ancestor of origin's " + "remote-tracking default (else the only remote's) or a --merged-into " + "REF, against the planned sha; refuse a repository without such a " + "default, one git fails on, one whose holds cannot be read, one " + "whose git dir is neither its common dir nor directly under " + "/worktrees/ and one with a symlink in or below refs, " + "packed-refs, reftable or logs/refs, and still print the plan; skip " + "a symbolic branch, a branch any symbolic ref of any worktree names, " + "a branch a worktree, worktree registration, rebase or bisect holds, " + "a --merged-into target, a branch with branch..deleteMerged " + "false in any worktree's or the system config and the local " + "upstream of a branch that stays, also of one --branch or --exclude " + "leaves out; " + "run git without inherited GIT_* variables but " + "GIT_CEILING_DIRECTORIES; the keep config reads run once with the " + "caller's GIT_CONFIG_* variables and once with the default system " + "and passwd-home config, and either read's keep counts; read no " + "forge token; " + "never fetch, push, delete a remote branch, run docker or hooks, or " + "refresh the branch tree file; exit 1 when an item failed, else 2 " + "when a repository was refused, else 0; exit 2 without a plan when, " + "without --all, the current directory is in no git repository, its " + "git dir is neither its common dir nor directly under " + "/worktrees/, its main worktree belongs to another " + "repository or its path or git output is not UTF-8, on --apply " + "outside the host PID namespace, on a --merged-into REF that is not " + "exactly one branch or remote-tracking ref (X~0, X^0, X@{0}, a sha), " + "and on a --branch " + "that matches nothing; meant for working clones" + ), + ) branch_purge = branch_subparsers.add_parser( "purge", diff --git a/src/pkgmgr/core/git/run.py b/src/pkgmgr/core/git/run.py index 06ee94d..eec5475 100644 --- a/src/pkgmgr/core/git/run.py +++ b/src/pkgmgr/core/git/run.py @@ -1,7 +1,12 @@ from __future__ import annotations +import contextvars import os +import pwd +import re import subprocess +from collections.abc import Iterator +from contextlib import contextmanager from .errors import GitNotRepositoryError, GitRunError @@ -52,10 +57,20 @@ SCAN_UNSET = ( "GIT_PAGER", "PAGER", ) +STRICT_KEEP = ("GIT_CEILING_DIRECTORIES",) +CONFIG_READ_KEEP = re.compile( + r"GIT_CONFIG_(GLOBAL|SYSTEM|PARAMETERS|COUNT|KEY_\d+|VALUE_\d+)" +) +CALLER_CONFIG = "caller" +DEFAULT_CONFIG = "default" +CONFIG_VIEWS = (CALLER_CONFIG, DEFAULT_CONFIG) SCAN_TIMEOUT = 60 # ref and remote names come from repositories an agent may write: one named # like --output= would otherwise make rev-list create that file END_OF_OPTIONS = "--end-of-options" +_STRICT_ENV: contextvars.ContextVar[bool] = contextvars.ContextVar( + "pkgmgr_git_strict_env", default=False +) def _is_not_repo_error(stderr: str) -> bool: @@ -63,9 +78,61 @@ def _is_not_repo_error(stderr: str) -> bool: return "not a git repository" in msg -def _scan_env() -> dict[str, str]: - env = {key: value for key, value in os.environ.items() if key not in SCAN_UNSET} - return {**env, **SCAN_ENV} +@contextmanager +def strict_env() -> Iterator[None]: + token = _STRICT_ENV.set(True) + try: + yield + finally: + _STRICT_ENV.reset(token) + + +def _strict_keeps(key: str, caller: bool) -> bool: + return key in STRICT_KEEP or ( + caller and CONFIG_READ_KEEP.fullmatch(key) is not None + ) + + +def _passwd_home() -> str | None: + try: + return pwd.getpwuid(os.getuid()).pw_dir + except KeyError: + return None + + +def _scan_env(strict: bool = False, view: str | None = None) -> dict[str, str]: + """ + Build the environment of a scan or strict git call. + + Args: + strict: drop every inherited `GIT_*` variable but STRICT_KEEP. + view: None, CALLER_CONFIG or DEFAULT_CONFIG, as run takes it. + + Returns: + The inherited environment without SCAN_UNSET, plus SCAN_ENV; for a + config view with the system config; for DEFAULT_CONFIG also without + any `GIT_*` but STRICT_KEEP and XDG_CONFIG_HOME, HOME set to the + passwd home (unset without a passwd entry). + """ + caller = view == CALLER_CONFIG + scrub = strict or view == DEFAULT_CONFIG + env = { + key: value + for key, value in os.environ.items() + if key not in SCAN_UNSET + and not (scrub and key.startswith("GIT_") and not _strict_keeps(key, caller)) + } + env = {**env, **SCAN_ENV} + if view is None: + return env + del env["GIT_CONFIG_NOSYSTEM"] + if view == DEFAULT_CONFIG: + env.pop("XDG_CONFIG_HOME", None) + env.pop("HOME", None) + home = _passwd_home() + if home: + env["HOME"] = home + return env def run( @@ -75,6 +142,7 @@ def run( preview: bool = False, scan: bool = False, input_text: str | None = None, + config_view: str | None = None, ) -> str: """ Run a Git command and return its stdout as a stripped string. @@ -85,9 +153,17 @@ def run( preview: print the command instead of running it. scan: read a repository an agent may have written: no fsmonitor, hook, pager, lazy fetch, replace object, system config or prompt, stdin - closed, a timeout. + closed, a timeout. Inside strict_env() every call, scan or not, + also runs without any inherited `GIT_*` variable but + `GIT_CEILING_DIRECTORIES`, which only narrows discovery. input_text: text fed to stdin, e.g. for `cat-file --batch-check`; replaces the closed stdin of scan mode. + config_view: for a pure config read, the config it sees, both with + the system config: CALLER_CONFIG also keeps, inside strict_env(), + `GIT_CONFIG_GLOBAL`, `GIT_CONFIG_SYSTEM`, `GIT_CONFIG_PARAMETERS`, + `GIT_CONFIG_COUNT` and its `GIT_CONFIG_KEY_` and + `GIT_CONFIG_VALUE_`; DEFAULT_CONFIG reads the default system + and passwd-home global config whatever the environment says. Raises GitRunError (or a subclass) if execution fails or times out. """ @@ -98,10 +174,15 @@ def run( print(f"[PREVIEW] Would run in {cwd!r}: {cmd_str}") return "" + strict = _STRICT_ENV.get() hardened: dict = ( - {"env": _scan_env(), "stdin": subprocess.DEVNULL, "timeout": SCAN_TIMEOUT} + { + "env": _scan_env(strict, config_view), + "stdin": subprocess.DEVNULL, + "timeout": SCAN_TIMEOUT, + } if scan - else {} + else ({"env": _scan_env(True, config_view)} if strict else {}) ) if input_text is not None: hardened.pop("stdin", None) diff --git a/tests/integration/test_branch_cleanup_local.py b/tests/integration/test_branch_cleanup_local.py new file mode 100644 index 0000000..2e9fb80 --- /dev/null +++ b/tests/integration/test_branch_cleanup_local.py @@ -0,0 +1,1965 @@ +from __future__ import annotations + +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from collections.abc import Callable, Iterator +from contextlib import contextmanager, redirect_stderr, redirect_stdout +from types import SimpleNamespace +from unittest.mock import patch + +from pkgmgr.actions.branch import gc, gc_forge, tree +from pkgmgr.cli.commands.branch import handle_branch, scan_repo_roots +from pkgmgr.cli.parser import create_parser +from pkgmgr.core.credentials.types import NoCredentialsError +from pkgmgr.core.git.errors import GitBaseError, GitRunError +from pkgmgr.core.git.run import CALLER_CONFIG, DEFAULT_CONFIG + +GITHUB_URL = "https://github.com/example/demo.git" +AUTHOR = { + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@example.org", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@example.org", +} +FAILING_DOCKER = """#!{python} +import json, os, sys +with open(os.environ["FAKE_DOCKER_LOG"], "a", encoding="utf-8") as fh: + fh.write(json.dumps(sys.argv[1:]) + "\\n") +sys.exit(1) +""" +MARKING_HOOK = """#!/bin/sh +echo "$@" >> "{marker}" +""" +BRANCH_CLI = """import sys +from types import SimpleNamespace +from pkgmgr.cli.commands.branch import handle_branch +from pkgmgr.cli.parser import create_parser +ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) +handle_branch(create_parser("test").parse_args(sys.argv[1:]), ctx=ctx) +""" + + +def clean_env() -> dict[str, str]: + """ + Build the environment of every git the tests run. + + Returns: + The current environment without any `GIT_*` variable, plus a fixed + author and committer. + """ + env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} + return {**env, **AUTHOR} + + +def git(cwd: str, *args: str) -> str: + return subprocess.run( + ["git", "-c", "commit.gpgsign=false", *args], + cwd=cwd, + env=clean_env(), + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +class TestBranchCleanupLocal(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = os.path.realpath(self._tmp.name) + self.addCleanup(os.chdir, os.getcwd()) + os.chdir(self.root) + self.home = os.path.join(self.root, "home") + os.makedirs(self.home) + for patcher in ( + patch.dict(os.environ, {"HOME": self.home}), + patch("pkgmgr.core.git.run._passwd_home", return_value=self.home), + ): + patcher.start() + self.addCleanup(patcher.stop) + self.origin = os.path.join(self.root, "origin.git") + self.repo = os.path.join(self.root, "repo") + git(self.root, "init", "--bare", "-b", "main", self.origin) + git(self.root, "init", "-b", "main", self.repo) + git(self.repo, "remote", "add", "origin", f"file://{self.origin}") + self._commit("init") + git(self.repo, "push", "-u", "origin", "main") + + git(self.repo, "checkout", "-b", "done", "main") + self._commit("done") + git(self.repo, "push", "-u", "origin", "done") + git(self.repo, "push", "origin", "done:refs/heads/remote-done") + git(self.repo, "checkout", "-b", "picked", "main") + self._commit("picked") + git(self.repo, "checkout", "main") + git(self.repo, "merge", "--no-ff", "-m", "merge done", "done") + git(self.repo, "cherry-pick", "picked") + git(self.repo, "push", "origin", "main") + git(self.repo, "checkout", "-b", "wip", "main") + self._commit("wip") + git(self.repo, "checkout", "main") + git(self.repo, "fetch", "origin") + git(self.repo, "update-ref", "--no-deref", "-d", "refs/remotes/origin/HEAD") + git(self.repo, "branch", "local-done", "done") + git(self.repo, "branch", "held-done", "done") + self.held = os.path.join(self.root, "wt-held") + git(self.repo, "worktree", "add", self.held, "held-done") + git(self.repo, "remote", "add", "hub", GITHUB_URL) + + self.bin = os.path.join(self.root, "bin") + os.makedirs(self.bin) + docker = os.path.join(self.bin, "docker") + with open(docker, "w", encoding="utf-8") as fh: + fh.write(FAILING_DOCKER.format(python=sys.executable)) + os.chmod(docker, 0o755) + self.docker_log = os.path.join(self.root, "docker.log") + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _set_head(self) -> None: + git(self.repo, "remote", "set-head", "origin", "main") + + def _commit(self, name: str) -> None: + with open(os.path.join(self.repo, f"{name}.txt"), "w") as fh: + fh.write(name) + git(self.repo, "add", f"{name}.txt") + git(self.repo, "commit", "-m", name) + + def _refs(self, cwd: str, prefix: str) -> dict[str, str]: + out = git(cwd, "for-each-ref", "--format=%(refname) %(objectname)", prefix) + return dict(line.split(" ") for line in out.splitlines()) + + @contextmanager + def _guarded(self) -> Iterator[SimpleNamespace]: + """ + Patch the host checks, drop every `GIT_*` variable from the + environment and watch every token, forge and docker access. + + Returns: + The git run spy, the token resolver, HTTP client and Secret + Service mocks, as attributes git, token, http and secret. + """ + env = { + **clean_env(), + "PATH": f"{self.bin}{os.pathsep}{os.environ['PATH']}", + "FAKE_DOCKER_LOG": self.docker_log, + "GIT_ALLOW_PROTOCOL": "file", + } + with ( + patch.dict(os.environ, env, clear=True), + patch.object(gc, "host_pid_namespace", return_value=True), + patch.object(gc, "_process_cwds", return_value={}), + patch.object(gc, "run", wraps=gc.run) as git_run, + patch.object(gc_forge, "TokenResolver") as resolver, + patch.object(gc_forge, "HttpClient") as http, + patch.object( + tree, "secret_service_token", return_value=(None, "test") + ) as secret, + ): + resolver.return_value.get_token.side_effect = NoCredentialsError("none") + yield SimpleNamespace(git=git_run, token=resolver, http=http, secret=secret) + + def _cleanup( + self, + scope: str, + apply: bool, + merged_into: tuple[str, ...] = (), + roots: tuple[str, ...] = (), + leak: dict[str, str] | None = None, + include: tuple[str, ...] = (), + exclude: tuple[str, ...] = (), + ) -> SimpleNamespace: + """ + Run gc.cleanup on the fixture under _guarded. + + Args: + scope: the cleanup scope. + apply: execute the plan. + merged_into: integration lines that also count as merged. + roots: the repositories to scan; the fixture repository if empty. + leak: variables set in the environment gc.cleanup sees, after + _guarded dropped every `GIT_*` one. + include: the --branch filter. + exclude: the --exclude filter. + + Returns: + The exit code, the first repository's items by branch and + warnings, every repository record, the stderr text and the + _guarded spies, as attributes code, items, warnings, repos, stderr + and guard; no repositories when nothing was printed. + """ + out, err = io.StringIO(), io.StringIO() + with ( + self._guarded() as guard, + patch.dict(os.environ, leak or {}), + redirect_stdout(out), + redirect_stderr(err), + ): + code = gc.cleanup( + list(roots) or [self.repo], + scope=scope, + apply=apply, + as_json=True, + include=list(include), + exclude=list(exclude), + merged_into=list(merged_into), + sizes=False, + min_age=None, + hooks={}, + ) + repos = json.loads(out.getvalue())["repos"] if out.getvalue() else [] + first = repos[0] if repos else {"items": [], "warnings": []} + return SimpleNamespace( + code=code, + items={i["branch"]: i for i in first["items"]}, + warnings=first["warnings"], + repos=repos, + stderr=err.getvalue(), + guard=guard, + ) + + def _branch_config(self, name: str) -> list[str]: + return subprocess.run( + ["git", "config", "--get-regexp", f"^branch\\.{name}\\."], + cwd=self.repo, + env=clean_env(), + check=False, + capture_output=True, + text=True, + ).stdout.split()[::2] + + def _results(self, run: SimpleNamespace) -> dict[str, str]: + return {name: i["result"] for name, i in run.items.items() if i["result"]} + + def assertLocalOnly(self, guard: SimpleNamespace) -> None: + self.assertFalse(os.path.exists(self.docker_log)) + guard.token.assert_not_called() + guard.http.assert_not_called() + guard.secret.assert_not_called() + for call in guard.git.call_args_list: + with self.subTest(argv=call.args[0]): + self.assertNotIn("push", call.args[0]) + self.assertNotIn("fetch", call.args[0]) + + def test_dry_run_plans_only_ancestor_tips(self) -> None: + self._set_head() + run = self._cleanup("local", apply=False) + items = run.items + + self.assertEqual(run.code, 0) + self.assertEqual(run.warnings, []) + self.assertEqual( + set(items), {"main", "done", "local-done", "held-done", "picked", "wip"} + ) + self.assertEqual(items["done"]["upstream"], "origin/done") + self.assertEqual(list(items["done"]["remotes"]), ["origin"]) + for name in ("done", "local-done"): + with self.subTest(deletes=name): + self.assertIsNone(items[name]["skip"]) + self.assertEqual(items[name]["actions"], ["local-branch"]) + self.assertEqual( + items["held-done"]["skip"], f"checked out in worktree {self.held}" + ) + self.assertEqual( + items["picked"]["skip"], + "merged by patch, tip is no ancestor of refs/remotes/origin/main", + ) + self.assertEqual(items["wip"]["state"], "unmerged") + self.assertEqual(items["main"]["state"], "protected") + for name in ("held-done", "picked", "wip", "main"): + with self.subTest(keeps=name): + self.assertEqual(items[name]["actions"], []) + self.assertLocalOnly(run.guard) + + def test_apply_deletes_only_local_refs_against_the_planned_sha(self) -> None: + self._set_head() + heads = self._refs(self.repo, "refs/heads") + tracking = self._refs(self.repo, "refs/remotes") + origin = self._refs(self.origin, "refs") + trees = git(self.repo, "worktree", "list", "--porcelain") + self.assertEqual( + self._branch_config("done"), ["branch.done.remote", "branch.done.merge"] + ) + self.assertEqual(self._branch_config("local-done"), []) + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 0) + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(self._branch_config("done"), []) + self.assertEqual( + self._branch_config("main"), ["branch.main.remote", "branch.main.merge"] + ) + after = self._refs(self.repo, "refs/heads") + self.assertEqual( + set(heads) - set(after), {"refs/heads/done", "refs/heads/local-done"} + ) + self.assertEqual(set(after) - set(heads), set()) + self.assertEqual(self._refs(self.repo, "refs/remotes"), tracking) + self.assertEqual(self._refs(self.origin, "refs"), origin) + self.assertEqual(git(self.repo, "worktree", "list", "--porcelain"), trees) + deletes = [c.args[0] for c in run.guard.git.call_args_list if "-d" in c.args[0]] + self.assertEqual( + deletes, + [ + [ + "update-ref", + "--no-deref", + "-d", + f"refs/heads/{name}", + heads[f"refs/heads/{name}"], + ] + for name in ("done", "local-done") + ], + ) + self.assertLocalOnly(run.guard) + + def test_branch_moved_after_the_check_is_not_deleted(self) -> None: + planned = git(self.repo, "rev-parse", "done") + moved = git( + self.repo, + "commit-tree", + "-p", + planned, + "-m", + "moved", + f"{planned}^{{tree}}", + ) + + def race(_repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "done": + git(self.repo, "update-ref", "refs/heads/done", moved) + return item + + with patch.object(gc, "_rebuild", side_effect=race): + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 1) + self.assertTrue(run.items["done"]["result"].startswith("failed: ")) + self.assertEqual(run.items["local-done"]["result"], "done") + self.assertEqual(git(self.repo, "rev-parse", "refs/heads/done"), moved) + + def test_cherry_picked_branch_stays_with_only_a_file_remote(self) -> None: + git(self.repo, "remote", "remove", "hub") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.items["picked"]["state"], "merged-clean") + self.assertEqual(run.items["picked"]["merged_by"], "patch") + self.assertEqual( + run.items["picked"]["skip"], + "merged by patch, tip is no ancestor of refs/remotes/origin/main", + ) + self.assertNotIn("picked", self._results(run)) + self.assertIn("refs/heads/picked", self._refs(self.repo, "refs/heads")) + + def test_target_is_the_default_ref_not_head(self) -> None: + init = git(self.repo, "rev-list", "--max-parents=0", "main") + git(self.repo, "checkout", "-b", "head-only", init) + self._commit("head-only") + git(self.repo, "checkout", "--detach", init) + git(self.repo, "merge", "--no-ff", "-m", "merge head-only", "head-only") + self.assertNotEqual( + git(self.repo, "rev-parse", "HEAD"), + git(self.repo, "rev-parse", "origin/main"), + ) + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(run.items["head-only"]["state"], "unmerged") + self.assertIn("refs/heads/head-only", self._refs(self.repo, "refs/heads")) + + def test_merged_into_line_counts_only_when_given(self) -> None: + git(self.repo, "checkout", "-b", "line-only", "main") + self._commit("line-only") + git(self.repo, "checkout", "-b", "staging", "main") + git(self.repo, "merge", "--no-ff", "-m", "merge line-only", "line-only") + git(self.repo, "push", "origin", "staging") + git(self.repo, "checkout", "main") + git(self.repo, "branch", "-D", "staging") + + without = self._cleanup("local", apply=False) + run = self._cleanup("local", apply=True, merged_into=("origin/staging",)) + + self.assertEqual(without.items["line-only"]["state"], "unmerged") + self.assertEqual(without.items["line-only"]["actions"], []) + self.assertEqual(run.items["line-only"]["merged_by"], "line:origin/staging") + self.assertEqual(run.items["line-only"]["result"], "done") + self.assertNotIn("refs/heads/line-only", self._refs(self.repo, "refs/heads")) + self.assertLocalOnly(run.guard) + + def test_without_remote_head_a_local_main_merge_counts_for_nothing(self) -> None: + git(self.repo, "checkout", "-b", "feat", "main") + self._commit("feat") + git(self.repo, "checkout", "main") + git(self.repo, "merge", "--no-ff", "-m", "merge feat", "feat") + head = subprocess.run( + ["git", "symbolic-ref", "refs/remotes/origin/HEAD"], + cwd=self.repo, + env=clean_env(), + check=False, + capture_output=True, + ) + self.assertNotEqual(head.returncode, 0) + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 0) + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(run.items["feat"]["state"], "unmerged") + self.assertIn("refs/heads/feat", self._refs(self.repo, "refs/heads")) + + def test_without_remote_default_the_repository_is_refused(self) -> None: + git(self.repo, "update-ref", "-d", "refs/remotes/origin/main") + git(self.repo, "branch", "-m", "main", "trunk") + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True, merged_into=("origin/done",)) + + self.assertEqual(run.code, 2) + self.assertEqual(run.warnings, [gc.NO_REMOTE_DEFAULT]) + self.assertEqual(run.items, {}) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + self.assertLocalOnly(run.guard) + + def test_rebase_and_bisect_hold_their_branches(self) -> None: + init = git(self.repo, "rev-list", "--max-parents=0", "main") + rebasing = os.path.join(self.root, "wt-rebase") + git(self.repo, "worktree", "add", rebasing, "local-done") + subprocess.run( + ["git", "rebase", "-f", "--exec", "false", "HEAD~1"], + cwd=rebasing, + env=clean_env(), + check=False, + capture_output=True, + ) + bisecting = os.path.join(self.root, "wt-bisect") + git(self.repo, "worktree", "add", bisecting, "done") + git(bisecting, "bisect", "start", "done", init) + git(bisecting, "checkout", "--detach") + for path, state in ((rebasing, "rebase-merge"), (bisecting, "BISECT_START")): + with self.subTest(precondition=state): + state_path = git(path, "rev-parse", "--git-path", state) + self.assertTrue(os.path.exists(os.path.join(path, state_path))) + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 0) + self.assertEqual(self._results(run), {}) + for name, operation in (("local-done", "rebase"), ("done", "bisect")): + with self.subTest(held=name): + self.assertIsNone(run.items[name]["worktree"]) + self.assertTrue( + run.items[name]["skip"].startswith(f"{operation} in progress in ") + ) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def _local_merge(self, branch: str) -> None: + git(self.repo, "checkout", "-b", branch, "main") + self._commit(branch) + git(self.repo, "checkout", "main") + git(self.repo, "merge", "--no-ff", "-m", f"merge {branch}", branch) + + def test_symbolic_ref_and_its_target_are_held(self) -> None: + git(self.repo, "symbolic-ref", "refs/heads/alias", "refs/heads/local-done") + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done"}) + self.assertEqual( + run.items["alias"]["skip"], "symbolic ref to refs/heads/local-done" + ) + self.assertEqual( + run.items["local-done"]["skip"], "target of symbolic ref refs/heads/alias" + ) + self.assertEqual( + git(self.repo, "symbolic-ref", "refs/heads/alias"), "refs/heads/local-done" + ) + self.assertEqual( + self._refs(self.repo, "refs/heads")["refs/heads/local-done"], + heads["refs/heads/local-done"], + ) + + def test_delete_never_follows_a_symref_made_after_the_check(self) -> None: + planned = git(self.repo, "rev-parse", "held-done") + + def race(_repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "done": + git(self.repo, "update-ref", "-d", "refs/heads/done") + git( + self.repo, "symbolic-ref", "refs/heads/done", "refs/heads/held-done" + ) + return item + + with patch.object(gc, "_rebuild", side_effect=race): + self._cleanup("local", apply=True) + + self.assertEqual(git(self.repo, "rev-parse", "refs/heads/held-done"), planned) + self.assertEqual(git(self.held, "symbolic-ref", "HEAD"), "refs/heads/held-done") + + def test_registration_and_state_files_hold_their_branches(self) -> None: + git(self.repo, "branch", "stacked", "done") + ghost = os.path.join(self.root, "wt-ghost") + git(self.repo, "worktree", "add", ghost, "local-done") + with open(os.path.join(ghost, "uncommitted.txt"), "w") as fh: + fh.write("work") + meta = os.path.join(self.repo, ".git", "worktrees") + os.remove(os.path.join(meta, "wt-ghost", "gitdir")) + sha = git(self.repo, "rev-parse", "stacked") + states = { + "rebase-apply/head-name": "refs/heads/done\n", + "rebase-merge/update-refs": f"refs/heads/stacked\n{sha}\n{sha}\n", + } + for name, content in states.items(): + path = os.path.join(meta, "wt-held", name) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(content) + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {}) + self.assertIsNone(run.items["local-done"]["worktree"]) + for name, reason in ( + ("local-done", "checked out in "), + ("done", "rebase in progress in "), + ("stacked", "rebase in progress in "), + ): + with self.subTest(held=name): + self.assertTrue(run.items[name]["skip"].startswith(reason)) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + self.assertTrue(os.path.isfile(os.path.join(ghost, "uncommitted.txt"))) + + def test_remote_head_selects_the_default_over_main(self) -> None: + init = git(self.repo, "rev-list", "--max-parents=0", "main") + git(self.repo, "checkout", "-b", "on-trunk", init) + self._commit("on-trunk") + git(self.repo, "checkout", "-b", "trunk", init) + git(self.repo, "merge", "--no-ff", "-m", "merge on-trunk", "on-trunk") + git(self.repo, "push", "origin", "trunk") + git(self.repo, "checkout", "main") + git(self.repo, "branch", "-D", "trunk") + git( + self.repo, + "symbolic-ref", + "refs/remotes/origin/HEAD", + "refs/remotes/origin/trunk", + ) + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"on-trunk": "done"}) + self.assertEqual(run.items["done"]["state"], "unmerged") + + def test_remote_head_into_a_local_branch_is_ignored(self) -> None: + self._local_merge("feat") + git(self.repo, "symbolic-ref", "refs/remotes/origin/HEAD", "refs/heads/main") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(run.items["feat"]["state"], "unmerged") + self.assertIn("refs/heads/feat", self._refs(self.repo, "refs/heads")) + + def test_remote_main_into_a_local_branch_is_refused(self) -> None: + self._local_merge("feat") + git(self.repo, "update-ref", "-d", "refs/remotes/origin/main") + git(self.repo, "symbolic-ref", "refs/remotes/origin/main", "refs/heads/main") + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertEqual(run.warnings, [gc.NO_REMOTE_DEFAULT]) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def _fork_main(self, merged: str) -> None: + init = git(self.repo, "rev-list", "--max-parents=0", "main") + git(self.repo, "checkout", "-b", merged, init) + self._commit(merged) + git(self.repo, "checkout", "--detach", init) + git(self.repo, "merge", "--no-ff", "-m", f"merge {merged}", merged) + sha = git(self.repo, "rev-parse", "HEAD") + git(self.repo, "checkout", "main") + git(self.repo, "remote", "add", "fork", "file:///nonexistent/fork.git") + git(self.repo, "update-ref", "refs/remotes/fork/main", sha) + git( + self.repo, + "symbolic-ref", + "refs/remotes/fork/HEAD", + "refs/remotes/fork/main", + ) + + def test_origin_wins_over_another_remote_head(self) -> None: + self._fork_main("feat4") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(run.items["feat4"]["state"], "unmerged") + + def test_several_remotes_without_origin_are_refused(self) -> None: + self._fork_main("feat4") + git(self.repo, "remote", "remove", "hub") + git(self.repo, "remote", "rename", "origin", "up") + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertEqual(run.warnings, [gc.NO_REMOTE_DEFAULT]) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def _raw_heads(self, cwd: str) -> bytes: + return subprocess.run( + ["git", "for-each-ref", "--format=%(refname) %(objectname)", "refs/heads"], + cwd=cwd, + env=clean_env(), + check=True, + capture_output=True, + ).stdout + + def test_submodule_cleanup_stays_in_the_submodule(self) -> None: + sub_origin = os.path.join(self.root, "sub-origin.git") + seed = os.path.join(self.root, "seed") + git(self.root, "init", "--bare", "-b", "main", sub_origin) + git(self.root, "init", "-b", "main", seed) + with open(os.path.join(seed, "s.txt"), "w") as fh: + fh.write("s") + git(seed, "add", "s.txt") + git(seed, "commit", "-m", "s") + git(seed, "push", f"file://{sub_origin}", "main") + git( + self.repo, + "-c", + "protocol.file.allow=always", + "submodule", + "add", + f"file://{sub_origin}", + "sub", + ) + sub = os.path.join(self.repo, "sub") + heads = self._refs(self.repo, "refs/heads") + args = create_parser("test").parse_args( + ["branch", "cleanup", "--local", "--apply", "--json"] + ) + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + os.chdir(sub) + out = io.StringIO() + with ( + self._guarded(), + redirect_stdout(out), + redirect_stderr(io.StringIO()), + self.assertRaises(SystemExit), + ): + handle_branch(args, ctx=ctx) + + self.assertEqual(gc.local_repo_root(sub), sub) + self.assertEqual(json.loads(out.getvalue())["repos"][0]["repo"], sub) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_main_worktree_of_another_repository_is_refused(self) -> None: + other = os.path.join(self.root, "other") + git(self.root, "init", "-b", "main", other) + git(other, "config", "core.worktree", self.repo) + + with self.assertRaises(GitBaseError): + gc.local_repo_root(other) + + def test_core_worktree_shape_keeps_the_head_root_outside_local(self) -> None: + parent = os.path.join(self.root, "cwt-a") + plain = os.path.join(self.root, "cwt-b") + git(self.root, "init", "-b", "main", parent) + os.makedirs(plain) + git(parent, "config", "core.worktree", plain) + parse = create_parser("test").parse_args + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + os.chdir(parent) + + self.assertEqual(gc.main_repo_root("."), parent) + self.assertEqual(scan_repo_roots(parse(["branch", "cleanup"]), ctx), [parent]) + local = parse(["branch", "cleanup", "--local"]) + with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as stop: + scan_repo_roots(local, ctx, local=True) + self.assertEqual(stop.exception.code, 2) + + def test_non_utf8_repository_path_is_refused(self) -> None: + bad = os.fsdecode(os.path.join(os.fsencode(self.root), b"bad\xffdir")) + subprocess.run( + ["git", "clone", f"file://{self.origin}", bad], + env=clean_env(), + check=True, + capture_output=True, + ) + before = self._raw_heads(bad) + args = create_parser("test").parse_args( + ["branch", "cleanup", "--local", "--apply", "--json"] + ) + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + os.chdir(bad) + err = io.StringIO() + with ( + self._guarded(), + redirect_stdout(io.StringIO()), + redirect_stderr(err), + self.assertRaises(SystemExit) as stop, + ): + handle_branch(args, ctx=ctx) + + self.assertEqual(stop.exception.code, 2) + self.assertIn(gc.NOT_UTF8, err.getvalue()) + self.assertEqual(self._raw_heads(bad), before) + + def test_reftable_registration_head_holds_its_branch(self) -> None: + rt = os.path.join(self.root, "rt") + made = subprocess.run( + ["git", "init", "-b", "main", "--ref-format=reftable", rt], + env=clean_env(), + check=False, + capture_output=True, + ) + if made.returncode: + self.skipTest("git without reftable support") + git(rt, "remote", "add", "origin", f"file://{self.origin}") + git(rt, "fetch", "origin") + git(rt, "branch", "feat", "origin/done") + ghost = os.path.join(self.root, "rt-ghost") + git(rt, "worktree", "add", ghost, "feat") + os.remove(os.path.join(rt, ".git", "worktrees", "rt-ghost", "gitdir")) + + run = self._cleanup("local", apply=True, roots=(rt,)) + + self.assertIsNone(run.items["feat"]["worktree"]) + self.assertTrue(run.items["feat"]["skip"].startswith("checked out in ")) + self.assertIn("refs/heads/feat", self._refs(rt, "refs/heads")) + + def test_merged_into_branch_target_is_never_deleted(self) -> None: + git(self.repo, "branch", "a-line", "done") + + run = self._cleanup("local", apply=True, merged_into=("refs/heads/a-line",)) + + self.assertEqual(run.items["a-line"]["state"], "protected") + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertIn("refs/heads/a-line", self._refs(self.repo, "refs/heads")) + + def test_vanished_target_skips_the_rest_and_still_reports(self) -> None: + git(self.repo, "checkout", "-b", "line-only", "main") + self._commit("line-only") + git(self.repo, "checkout", "-b", "staging", "main") + git(self.repo, "merge", "--no-ff", "-m", "merge line-only", "line-only") + git(self.repo, "push", "origin", "staging") + git(self.repo, "checkout", "main") + git(self.repo, "branch", "-D", "staging") + rebuild = gc._rebuild + + def vanish(repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "line-only": + git(self.repo, "update-ref", "-d", "refs/remotes/origin/staging") + return rebuild(repo, item) + + with patch.object(gc, "_rebuild", side_effect=vanish): + run = self._cleanup("local", apply=True, merged_into=("origin/staging",)) + + self.assertEqual(run.code, 1) + self.assertEqual(run.items["done"]["result"], "done") + self.assertTrue(run.items["line-only"]["result"].startswith("failed: ")) + self.assertEqual( + run.items["local-done"]["result"], + "skipped: target origin/staging is gone", + ) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/line-only", heads) + self.assertIn("refs/heads/local-done", heads) + + def test_failed_item_outranks_a_refused_repository(self) -> None: + other = os.path.join(self.root, "other") + git(self.root, "init", "-b", "main", other) + moved = git( + self.repo, "commit-tree", "-p", "done", "-m", "moved", "done^{tree}" + ) + + def race(_repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "done": + git(self.repo, "update-ref", "refs/heads/done", moved) + return item + + with patch.object(gc, "_rebuild", side_effect=race): + run = self._cleanup("local", apply=True, roots=(self.repo, other)) + + self.assertEqual(run.code, 1) + self.assertTrue(run.items["done"]["result"].startswith("failed: ")) + self.assertEqual(run.items["local-done"]["result"], "done") + self.assertEqual(run.repos[1]["warnings"], [gc.NO_REMOTE_DEFAULT]) + + def test_non_utf8_ref_name_refuses_the_repository(self) -> None: + git(self.repo, "branch", os.fsdecode(b"bad\xff"), "done") + before = self._raw_heads(self.repo) + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertEqual(run.warnings, [gc.NOT_UTF8]) + self.assertEqual(run.items, {}) + self.assertEqual(self._raw_heads(self.repo), before) + + def test_refs_shared_through_a_symlink_refuse_the_repository(self) -> None: + git(self.repo, "checkout", "local-done") + shared = os.path.join(self.root, "new") + os.makedirs(os.path.join(shared, ".git", "logs")) + for name in ("config", "refs", "objects", "info", "hooks", "packed-refs"): + source = os.path.join(self.repo, ".git", name) + if os.path.exists(source): + os.symlink(source, os.path.join(shared, ".git", name)) + os.symlink( + os.path.join(self.repo, ".git", "logs", "refs"), + os.path.join(shared, ".git", "logs", "refs"), + ) + with open(os.path.join(shared, ".git", "HEAD"), "w", encoding="utf-8") as fh: + fh.write("ref: refs/heads/main\n") + git(shared, "checkout", "-f", "main") + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True, roots=(shared,)) + + self.assertEqual(run.code, 2) + self.assertTrue(run.warnings[-1].startswith("refused: refs")) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + self.assertEqual( + git(self.repo, "symbolic-ref", "HEAD"), "refs/heads/local-done" + ) + + def test_symrefs_outside_heads_hold_their_targets(self) -> None: + git(self.repo, "symbolic-ref", "refs/tags/latest", "refs/heads/done") + git( + self.repo, + "symbolic-ref", + "refs/remotes/origin/alias", + "refs/heads/local-done", + ) + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {}) + self.assertEqual( + run.items["done"]["skip"], "target of symbolic ref refs/tags/latest" + ) + self.assertEqual( + run.items["local-done"]["skip"], + "target of symbolic ref refs/remotes/origin/alias", + ) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/done", heads) + self.assertIn("refs/heads/local-done", heads) + + def _clone_with(self, path: str, *branches: str) -> None: + git(self.root, "clone", f"file://{self.origin}", path) + for name in branches: + git(path, "branch", name, "origin/done") + + def test_vanished_repository_skips_its_rest_and_still_prints(self) -> None: + other = os.path.join(self.root, "other") + self._clone_with(other, "a-first", "local-done") + rebuild = gc._rebuild + + def vanish(repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "a-first": + os.rename(other, f"{other}-moved") + return rebuild(repo, item) + + with patch.object(gc, "_rebuild", side_effect=vanish): + run = self._cleanup("local", apply=True, roots=(other, self.repo)) + + self.assertEqual(run.code, 1) + self.assertTrue(run.items["a-first"]["result"].startswith("failed: ")) + self.assertTrue( + run.items["local-done"]["result"].startswith( + f"skipped: repository {other} unreadable: " + ) + ) + mine = {i["branch"]: i["result"] for i in run.repos[1]["items"]} + self.assertEqual(mine["done"], "done") + self.assertIn( + "refs/heads/local-done", self._refs(f"{other}-moved", "refs/heads") + ) + + def test_repository_replaced_mid_run_is_never_touched(self) -> None: + nested = os.path.join(self.repo, "nested") + self._clone_with(nested, "a-first", "local-done") + rebuild = gc._rebuild + + def replace(repo: gc.Repo, item: gc.Item) -> gc.Item: + if item.branch == "a-first": + shutil.rmtree(os.path.join(nested, ".git")) + return rebuild(repo, item) + + with patch.object(gc, "_rebuild", side_effect=replace): + run = self._cleanup("local", apply=True, roots=(nested,)) + + self.assertEqual( + run.items["local-done"]["result"], + f"skipped: repository {nested} is no longer the planned one", + ) + self.assertIn("refs/heads/local-done", self._refs(self.repo, "refs/heads")) + + def test_non_local_plan_keeps_a_merged_into_branch_unprotected(self) -> None: + run = self._cleanup( + "branch", apply=False, merged_into=("refs/heads/local-done",) + ) + + self.assertEqual(run.items["local-done"]["state"], "merged-clean") + self.assertEqual(run.items["local-done"]["actions"], ["local-branch"]) + + def test_worktree_symref_holds_its_target(self) -> None: + pinned = os.path.join(self.root, "wt-pin") + git(self.repo, "worktree", "add", "-b", "pin-base", pinned, "main") + git(pinned, "symbolic-ref", "refs/worktree/pin", "refs/heads/local-done") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done"}) + self.assertEqual( + run.items["local-done"]["skip"], + "target of symbolic ref worktrees/wt-pin/refs/worktree/pin", + ) + self.assertIn("refs/heads/local-done", self._refs(self.repo, "refs/heads")) + + def test_unreadable_repository_is_refused_and_the_run_goes_on(self) -> None: + stale = os.path.join(self.root, "stale") + os.makedirs(stale) + with open(os.path.join(stale, ".git"), "w", encoding="utf-8") as fh: + fh.write(f"gitdir: {self.root}/gone/.git/worktrees/stale\n") + + run = self._cleanup("local", apply=True, roots=(self.repo, stale)) + + self.assertEqual(run.code, 2) + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual(run.repos[1]["items"], []) + self.assertTrue(run.repos[1]["warnings"][-1].startswith("refused: ")) + + def _state(self, cwd: str) -> tuple[str, str, bytes]: + with open(os.path.join(cwd, ".git", "config"), "rb") as fh: + config = fh.read() + refs = git(cwd, "for-each-ref", "--format=%(refname) %(objectname)") + return refs, git(cwd, "remote", "-v"), config + + def test_git_dir_in_the_environment_never_reaches_a_repository(self) -> None: + sentinel = os.path.join(self.root, "sentinel") + git(self.root, "init", "-b", "trunk", sentinel) + git(sentinel, "commit", "--allow-empty", "-m", "sentinel") + before = self._state(sentinel) + dot_git = os.path.join(sentinel, ".git") + leak = { + "GIT_DIR": dot_git, + "GIT_COMMON_DIR": dot_git, + "GIT_WORK_TREE": sentinel, + } + + with patch.dict(os.environ, leak): + git(self.repo, "branch", "leak-done", "done") + self._commit("leak") + run = self._cleanup("local", apply=True) + + self.assertEqual(self._state(sentinel), before) + self.assertEqual(run.items["leak-done"]["result"], "done") + + def test_the_only_remote_gives_the_default_without_origin(self) -> None: + git(self.repo, "remote", "remove", "hub") + git(self.repo, "remote", "rename", "origin", "up") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.repos[0]["default"], "refs/remotes/up/main") + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + + def test_merged_into_remote_tracking_ref_protects_its_branch(self) -> None: + git(self.repo, "branch", "a-line", "done") + git(self.repo, "push", "origin", "a-line") + + run = self._cleanup( + "local", apply=True, merged_into=("refs/remotes/origin/a-line",) + ) + + self.assertEqual(run.items["a-line"]["state"], "protected") + self.assertIn("refs/heads/a-line", self._refs(self.repo, "refs/heads")) + + def test_reftable_shared_through_a_symlink_refuses_the_repository(self) -> None: + rt = os.path.join(self.root, "rt-shared") + made = subprocess.run( + ["git", "init", "-b", "main", "--ref-format=reftable", rt], + env=clean_env(), + check=False, + capture_output=True, + ) + if made.returncode: + self.skipTest("git without reftable support") + git(rt, "remote", "add", "origin", f"file://{self.origin}") + git(rt, "fetch", "origin") + git(rt, "branch", "feat", "origin/done") + store = os.path.join(self.root, "reftable-store") + os.rename(os.path.join(rt, ".git", "reftable"), store) + os.symlink(store, os.path.join(rt, ".git", "reftable")) + + run = self._cleanup("local", apply=True, roots=(rt,)) + + self.assertEqual(run.code, 2) + self.assertTrue(run.warnings[-1].startswith("refused: reftable")) + self.assertIn("refs/heads/feat", self._refs(rt, "refs/heads")) + + def test_main_worktree_rebase_and_bisect_hold_their_branches(self) -> None: + common = os.path.join(self.repo, ".git") + os.makedirs(os.path.join(common, "rebase-merge")) + with open(os.path.join(common, "rebase-merge", "head-name"), "w") as fh: + fh.write("refs/heads/local-done\n") + with open(os.path.join(common, "BISECT_START"), "w") as fh: + fh.write("done\n") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {}) + self.assertEqual( + run.items["local-done"]["skip"], f"rebase in progress in {common}" + ) + self.assertEqual(run.items["done"]["skip"], f"bisect in progress in {common}") + + def test_remote_master_is_the_last_default(self) -> None: + git(self.repo, "update-ref", "refs/remotes/origin/master", "origin/main") + git(self.repo, "update-ref", "-d", "refs/remotes/origin/main") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.repos[0]["default"], "refs/remotes/origin/master") + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + + def test_bare_repository_head_holds_its_branch(self) -> None: + bare = os.path.join(self.root, "bare.git") + git(self.root, "clone", "--bare", f"file://{self.origin}", bare) + git( + bare, "config", "remote.origin.fetch", "+refs/heads/*:refs/remotes/origin/*" + ) + git(bare, "fetch", "origin") + git(bare, "symbolic-ref", "HEAD", "refs/heads/done") + + run = self._cleanup("local", apply=True, roots=(bare,)) + + self.assertEqual(run.items["done"]["skip"], f"checked out in {bare}") + self.assertIn("refs/heads/done", self._refs(bare, "refs/heads")) + + def test_linked_worktree_resolves_to_the_main_worktree(self) -> None: + self.assertEqual(gc.local_repo_root(self.held), self.repo) + + def test_git_variables_never_reach_the_local_git_calls(self) -> None: + sentinel = os.path.join(self.root, "sentinel") + git(self.root, "init", "-b", "trunk", sentinel) + git(sentinel, "commit", "--allow-empty", "-m", "sentinel") + git(sentinel, "config", "branch.done.remote", "origin") + git(sentinel, "config", "branch.local-done.description", "keep me") + config = os.path.join(sentinel, ".git", "config") + with open(config, "rb") as fh: + before = fh.read() + leak = { + "GIT_CONFIG": config, + "GIT_OBJECT_DIRECTORY": os.path.join(sentinel, ".git", "objects"), + "GIT_NAMESPACE": "foo", + } + + run = self._cleanup("local", apply=True, leak=leak) + + self.assertEqual(run.code, 0) + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + with open(config, "rb") as fh: + self.assertEqual(fh.read(), before) + self.assertEqual(self._branch_config("done"), []) + own = os.path.join(self.repo, ".git", "config") + self.assertIn( + ["config", "--file", own, "--remove-section", "branch.done"], + [call.args[0] for call in run.guard.git.call_args_list], + ) + + def test_merged_into_must_name_exactly_one_branch(self) -> None: + forms = ( + "picked~0", + "picked^0", + "picked@{0}", + git(self.repo, "rev-parse", "picked"), + ) + heads = self._refs(self.repo, "refs/heads") + + for form in forms: + with self.subTest(form=form): + run = self._cleanup("local", apply=True, merged_into=(form,)) + + self.assertEqual(run.code, 2) + self.assertEqual(run.repos, []) + self.assertIn(f"--merged-into {form} ", run.stderr) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_registration_with_any_id_holds_its_branch(self) -> None: + ghost = os.path.join(self.root, "wt-ghost") + git(self.repo, "worktree", "add", ghost, "local-done") + with open(os.path.join(ghost, "work.txt"), "w") as fh: + fh.write("uncommitted") + meta = os.path.join(self.repo, ".git", "worktrees") + os.rename(os.path.join(meta, "wt-ghost"), os.path.join(meta, "wt ghost")) + os.remove(os.path.join(meta, "wt ghost", "gitdir")) + with open(os.path.join(ghost, ".git"), "w", encoding="utf-8") as fh: + fh.write(f"gitdir: {meta}/wt ghost\n") + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done"}) + self.assertEqual( + run.items["local-done"]["skip"], f"checked out in {meta}/wt ghost" + ) + self.assertIn("refs/heads/local-done", self._refs(self.repo, "refs/heads")) + self.assertTrue(os.path.isfile(os.path.join(ghost, "work.txt"))) + + def test_unreadable_registration_refuses_the_repository(self) -> None: + broken = os.path.join(self.repo, ".git", "worktrees", "broken") + os.makedirs(broken) + for name, content in (("HEAD", "garbage\n"), ("commondir", "../..\n")): + with open(os.path.join(broken, name), "w", encoding="utf-8") as fh: + fh.write(content) + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertEqual(run.items, {}) + self.assertTrue(run.warnings[-1].startswith("refused: ")) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_unreadable_state_file_refuses_the_repository(self) -> None: + os.makedirs(os.path.join(self.repo, ".git", "BISECT_START")) + heads = self._refs(self.repo, "refs/heads") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertTrue(run.warnings[-1].startswith("refused: ")) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_main_worktree_symref_holds_with_a_linked_root(self) -> None: + git(self.repo, "symbolic-ref", "refs/worktree/pin", "refs/heads/local-done") + + run = self._cleanup("local", apply=True, roots=(self.held,)) + + self.assertEqual( + run.items["local-done"]["skip"], "target of symbolic ref refs/worktree/pin" + ) + self.assertIn("refs/heads/local-done", self._refs(self.repo, "refs/heads")) + + def _cli_apply( + self, cwd: str, leak: dict[str, str] | None = None + ) -> tuple[int, str, str]: + """ + Run `branch cleanup --local --apply --json` in cwd under _guarded. + + Args: + cwd: the directory to run in. + leak: variables set in the environment the CLI sees, after + _guarded dropped every `GIT_*` one. + + Returns: + The exit code, stdout and stderr. + """ + args = create_parser("test").parse_args( + ["branch", "cleanup", "--local", "--apply", "--json"] + ) + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + os.chdir(cwd) + out, err = io.StringIO(), io.StringIO() + with ( + self._guarded(), + patch.dict(os.environ, leak or {}), + redirect_stdout(out), + redirect_stderr(err), + self.assertRaises(SystemExit) as stop, + ): + handle_branch(args, ctx=ctx) + return stop.exception.code, out.getvalue(), err.getvalue() + + def test_ceiling_stops_discovery_of_the_repository_above(self) -> None: + below = os.path.join(self.repo, "not-a-repo") + os.makedirs(below) + heads = self._refs(self.repo, "refs/heads") + + code, out, err = self._cli_apply( + below, leak={"GIT_CEILING_DIRECTORIES": self.repo} + ) + + self.assertEqual(code, 2) + self.assertIn("Not inside a git repository", err) + self.assertEqual(out, "") + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_git_variables_reaching_the_cli_never_touch_another_repo(self) -> None: + git(self.repo, "remote", "remove", "hub") + sentinel = os.path.join(self.root, "sentinel") + git(self.root, "init", "-b", "trunk", sentinel) + git(sentinel, "commit", "--allow-empty", "-m", "sentinel") + before = self._state(sentinel) + dot_git = os.path.join(sentinel, ".git") + env = { + **clean_env(), + "PATH": f"{self.bin}{os.pathsep}{os.environ['PATH']}", + "FAKE_DOCKER_LOG": self.docker_log, + "GIT_ALLOW_PROTOCOL": "file", + "GIT_DIR": dot_git, + "GIT_WORK_TREE": sentinel, + "GIT_INDEX_FILE": os.path.join(dot_git, "index"), + } + + for argv, extra in ( + (["--json"], {}), + (["--local", "--json"], {"GIT_COMMON_DIR": dot_git}), + ): + with self.subTest(argv=argv): + done = subprocess.run( + [sys.executable, "-c", BRANCH_CLI, "branch", "cleanup", *argv], + cwd=self.repo, + env={**env, **extra}, + check=False, + capture_output=True, + text=True, + ) + + self.assertEqual(done.returncode, 0, done.stderr) + self.assertEqual(json.loads(done.stdout)["repos"][0]["repo"], self.repo) + self.assertEqual(self._state(sentinel), before) + + def test_registration_outside_the_common_dir_refuses_the_repository( + self, + ) -> None: + moved = os.path.join(self.root, "wt-moved") + git(self.repo, "worktree", "add", moved, "local-done") + with open(os.path.join(moved, "work.txt"), "w") as fh: + fh.write("uncommitted") + common = os.path.join(self.repo, ".git") + reg = os.path.join(common, "worktrees", "wt-moved") + heads = self._refs(self.repo, "refs/heads") + + for target in ( + os.path.join(self.root, "reg"), + os.path.join(common, "worktrees", "nest", "wt-moved"), + ): + with self.subTest(target=target): + os.renames(reg, target) + reg = target + with open(os.path.join(reg, "commondir"), "w", encoding="utf-8") as fh: + fh.write(common + "\n") + with open(os.path.join(moved, ".git"), "w", encoding="utf-8") as fh: + fh.write(f"gitdir: {reg}\n") + + run = self._cleanup("local", apply=True, roots=(moved,)) + code, out, err = self._cli_apply(moved) + + self.assertEqual(run.code, 2) + self.assertEqual(run.items, {}) + self.assertIn(f"git dir {reg} is neither", run.warnings[-1]) + self.assertEqual(code, 2) + self.assertIn(f"git dir {reg} is neither", err) + self.assertEqual(out, "") + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + self.assertEqual( + git(moved, "symbolic-ref", "HEAD"), "refs/heads/local-done" + ) + + def test_symlink_below_refs_refuses_the_repository(self) -> None: + other = os.path.join(self.root, "other") + self._clone_with(other, "shared-done") + git(other, "checkout", "shared-done") + theirs = os.path.join(other, ".git", "refs", "heads") + own = os.path.join(self.repo, ".git") + before = self._refs(other, "refs/heads") + heads = self._refs(self.repo, "refs/heads") + git(self.repo, "pack-refs", "--all") + links = ( + ("refs/heads/team", theirs), + ("logs/refs/heads/team", os.path.join(other, ".git", "logs", "refs")), + ("refs/heads", theirs), + ) + + for link, target in links: + with self.subTest(link=link): + path = os.path.join(own, link) + if link == "refs/heads": + shutil.rmtree(path, ignore_errors=True) + os.makedirs(os.path.dirname(path), exist_ok=True) + os.symlink(target, path) + + run = self._cleanup("local", apply=True) + + os.remove(path) + self.assertEqual(run.code, 2) + self.assertTrue(run.warnings[-1].startswith(f"refused: {link} in ")) + self.assertEqual(self._refs(other, "refs/heads"), before) + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_delete_merged_opt_out_and_local_upstreams_are_kept(self) -> None: + for name, start in ( + ("topic", "done"), + ("inner", "done"), + ("base", "done"), + ("child", "wip"), + ("stem", "done"), + ("leaf", "done"), + ): + git(self.repo, "branch", name, start) + for key, value in ( + ("branch.topic.deleteMerged", "false"), + ("branch.child.remote", "."), + ("branch.child.merge", "refs/heads/base"), + ("branch.base.remote", "."), + ("branch.base.merge", "refs/heads/inner"), + ("branch.leaf.remote", "."), + ("branch.leaf.merge", "refs/heads/stem"), + ): + git(self.repo, "config", key, value) + + run = self._cleanup("local", apply=True) + + self.assertEqual( + self._results(run), + {"done": "done", "local-done": "done", "leaf": "done", "stem": "done"}, + ) + self.assertEqual( + {name: run.items[name]["skip"] for name in ("topic", "base", "inner")}, + { + "topic": "kept: branch.topic.deleteMerged is false", + "base": "kept: local upstream of child", + "inner": "kept: local upstream of base", + }, + ) + self.assertEqual(self._branch_config("topic"), ["branch.topic.deletemerged"]) + heads = self._refs(self.repo, "refs/heads") + for name in ("topic", "inner", "base", "child"): + self.assertIn(f"refs/heads/{name}", heads) + + def test_filtered_out_branch_keeps_its_local_upstream(self) -> None: + git(self.repo, "branch", "x", "done") + git(self.repo, "branch", "y", "done") + git(self.repo, "config", "branch.x.remote", ".") + git(self.repo, "config", "branch.x.merge", "refs/heads/y") + + for include, exclude in ((("y",), ()), ((), ("x",))): + with self.subTest(include=include, exclude=exclude): + run = self._cleanup( + "local", apply=True, include=include, exclude=exclude + ) + + self.assertEqual(run.items["y"]["skip"], "kept: local upstream of x") + self.assertIsNone(run.items["y"]["result"]) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/x", heads) + self.assertIn("refs/heads/y", heads) + + def test_local_upstream_is_resolved_as_git_does(self) -> None: + forms = {"x1": "heads/y1", "x2": "y2", "x3": "refs/heads/y3"} + for child, merge in forms.items(): + parent = f"y{child[1:]}" + git(self.repo, "branch", parent, "done") + git(self.repo, "branch", child, "wip") + git(self.repo, "config", f"branch.{child}.remote", ".") + git(self.repo, "config", f"branch.{child}.merge", merge) + self.assertEqual( + git(self.repo, "rev-parse", "--symbolic-full-name", f"{child}@{{u}}"), + f"refs/heads/{parent}", + ) + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertEqual( + {name: run.items[name]["skip"] for name in ("y1", "y2", "y3")}, + { + "y1": "kept: local upstream of x1", + "y2": "kept: local upstream of x2", + "y3": "kept: local upstream of x3", + }, + ) + + def test_worktree_config_opt_out_keeps_the_branch_from_every_root(self) -> None: + for name in ("y", "q"): + git(self.repo, "branch", name, "done") + w = os.path.join(self.root, "w") + git(self.repo, "worktree", "add", "-b", "feat/w", w, "wip") + git(self.repo, "config", "extensions.worktreeConfig", "true") + git(w, "config", "--worktree", "branch.y.deleteMerged", "false") + include = os.path.join(self.root, "keep.inc") + with open(include, "w", encoding="utf-8") as fh: + fh.write('[branch "q"]\n\tdeleteMerged = false\n') + git(self.repo, "config", "includeIf.onbranch:feat/**.path", include) + kept = { + "y": "kept: branch.y.deleteMerged is false", + "q": "kept: branch.q.deleteMerged is false", + } + + for root in (self.repo, w): + with self.subTest(root=root): + run = self._cleanup("local", apply=False, roots=(root,)) + + self.assertEqual({n: run.items[n]["skip"] for n in kept}, kept) + code, out, _ = self._cli_apply(w) + + items = {i["branch"]: i for i in json.loads(out)["repos"][0]["items"]} + self.assertEqual(code, 0) + self.assertEqual({n: items[n]["skip"] for n in kept}, kept) + self.assertEqual( + {n: i["result"] for n, i in items.items() if i["result"]}, + {"done": "done", "local-done": "done"}, + ) + + def test_system_config_opt_out_keeps_the_branch(self) -> None: + system = os.path.join(self.root, "system.cfg") + with open(system, "w", encoding="utf-8") as fh: + fh.write('[branch "local-done"]\n\tdeleteMerged = false\n') + + run = self._cleanup("local", apply=True, leak={"GIT_CONFIG_SYSTEM": system}) + + self.assertEqual(self._results(run), {"done": "done"}) + self.assertEqual( + run.items["local-done"]["skip"], + "kept: branch.local-done.deleteMerged is false", + ) + + def test_config_variables_reach_only_the_keep_reads(self) -> None: + global_config = os.path.join(self.root, "global.cfg") + with open(global_config, "w", encoding="utf-8") as fh: + fh.write('[branch "local-done"]\n\tdeleteMerged = false\n') + git(self.repo, "branch", "kid", "wip") + shapes = ( + ({"GIT_CONFIG_GLOBAL": global_config}, "local-done", "deleteMerged"), + ( + { + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "branch.local-done.deleteMerged", + "GIT_CONFIG_VALUE_0": "false", + }, + "local-done", + "deleteMerged", + ), + ( + { + "GIT_CONFIG_COUNT": "2", + "GIT_CONFIG_KEY_0": "branch.kid.remote", + "GIT_CONFIG_VALUE_0": ".", + "GIT_CONFIG_KEY_1": "branch.kid.merge", + "GIT_CONFIG_VALUE_1": "refs/heads/done", + }, + "done", + "upstream", + ), + ) + reasons = { + "deleteMerged": "kept: branch.local-done.deleteMerged is false", + "upstream": "kept: local upstream of kid", + } + + for leak, branch, rule in shapes: + with self.subTest(names=sorted(leak), rule=rule): + run = self._cleanup("local", apply=False, leak=leak) + + self.assertEqual(run.items[branch]["skip"], reasons[rule]) + self.assertEqual(run.items[branch]["actions"], []) + views = { + (call.args[0][1], call.kwargs["config_view"]) + for call in run.guard.git.call_args_list + if call.kwargs.get("config_view") + } + self.assertEqual( + views, + { + (command, view) + for command in ("config", "for-each-ref") + for view in (CALLER_CONFIG, DEFAULT_CONFIG) + }, + ) + + def _kid_tracks_par(self) -> None: + git(self.repo, "branch", "par", "done") + git(self.repo, "branch", "kid", "wip") + + def test_system_config_upstream_keeps_its_parent(self) -> None: + self._kid_tracks_par() + system = os.path.join(self.root, "system.cfg") + include = os.path.join(self.root, "inc.cfg") + kid_up = '[branch "kid"]\n\tremote = .\n\tmerge = refs/heads/par\n' + with open(include, "w", encoding="utf-8") as fh: + fh.write(kid_up) + shapes = { + "s1_whole_upstream": kid_up, + "s2_merge_line_only": '[branch "kid"]\n\tmerge = refs/heads/par\n', + "s3_include": f"[include]\n\tpath = {include}\n", + "s5_includeif_onbranch": f'[includeIf "onbranch:main"]\n\tpath = {include}\n', + } + + for shape, text in shapes.items(): + with self.subTest(shape=shape): + with open(system, "w", encoding="utf-8") as fh: + fh.write(text) + if shape.startswith("s2"): + git(self.repo, "config", "branch.kid.remote", ".") + + run = self._cleanup( + "local", apply=False, leak={"GIT_CONFIG_SYSTEM": system} + ) + + self.assertEqual( + run.items["par"]["skip"], "kept: local upstream of kid" + ) + self.assertEqual(run.items["par"]["actions"], []) + + def test_injected_config_never_removes_a_keep(self) -> None: + self._kid_tracks_par() + git(self.repo, "config", "branch.kid.remote", ".") + git(self.repo, "config", "branch.kid.merge", "refs/heads/par") + git(self.repo, "config", "branch.local-done.deleteMerged", "false") + include = os.path.join(self.root, "inc.cfg") + with open(include, "w", encoding="utf-8") as fh: + fh.write('[branch "kid"]\n\tremote = /nonexistent/url\n') + count = {"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "branch.kid.remote"} + leaks = { + "e1_count_remote": {**count, "GIT_CONFIG_VALUE_0": "origin"}, + "e2_params_remote": { + "GIT_CONFIG_PARAMETERS": "'branch.kid.remote'='origin'" + }, + "e3_params_include": { + "GIT_CONFIG_PARAMETERS": f"'include.path'='{include}'" + }, + "e4_count_url": { + **count, + "GIT_CONFIG_VALUE_0": "https://example.invalid/x.git", + }, + "e5_params_delete_merged": { + "GIT_CONFIG_PARAMETERS": "'branch.local-done.deletemerged'='true'" + }, + } + + for shape, leak in leaks.items(): + with self.subTest(shape=shape): + run = self._cleanup("local", apply=False, leak=leak) + + self.assertEqual( + run.items["par"]["skip"], "kept: local upstream of kid" + ) + self.assertEqual( + run.items["local-done"]["skip"], + "kept: branch.local-done.deleteMerged is false", + ) + run = self._cleanup("local", apply=True, leak=leaks["e1_count_remote"]) + + self.assertEqual(self._results(run), {"done": "done"}) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/par", heads) + self.assertIn("refs/heads/local-done", heads) + + def test_hostile_home_never_hides_the_passwd_home_keep(self) -> None: + self._kid_tracks_par() + home = os.path.join(self.root, "passwd-home") + os.makedirs(home) + with open(os.path.join(home, ".gitconfig"), "w", encoding="utf-8") as fh: + fh.write( + '[branch "kid"]\n\tremote = .\n\tmerge = refs/heads/par\n' + '[branch "local-done"]\n\tdeleteMerged = false\n' + ) + hostile = os.path.join(self.root, "hostile-home") + os.makedirs(hostile) + + with patch("pkgmgr.core.git.run._passwd_home", return_value=home): + run = self._cleanup( + "local", + apply=True, + leak={ + "HOME": hostile, + "XDG_CONFIG_HOME": hostile, + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_SYSTEM": os.devnull, + }, + ) + + self.assertEqual(self._results(run), {"done": "done"}) + self.assertEqual(run.items["par"]["skip"], "kept: local upstream of kid") + self.assertEqual( + run.items["local-done"]["skip"], + "kept: branch.local-done.deleteMerged is false", + ) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/par", heads) + self.assertIn("refs/heads/local-done", heads) + + def test_local_upstream_from_a_worktree_config_keeps_its_parent(self) -> None: + for name in ("par1", "par2", "par3"): + git(self.repo, "branch", name, "done") + for name in ("kid2", "kid3"): + git(self.repo, "branch", name, "wip") + w1 = os.path.join(self.root, "w1") + w2 = os.path.join(self.root, "w2") + git(self.repo, "worktree", "add", "-b", "kid1", w1, "wip") + git(self.repo, "worktree", "add", "-b", "feat/w2", w2, "wip") + git(self.repo, "config", "extensions.worktreeConfig", "true") + for checkout, child, parent in ((w1, "kid1", "par1"), (w2, "kid2", "par2")): + git(checkout, "config", "--worktree", f"branch.{child}.remote", ".") + git( + checkout, + "config", + "--worktree", + f"branch.{child}.merge", + f"refs/heads/{parent}", + ) + include = os.path.join(self.root, "up.inc") + with open(include, "w", encoding="utf-8") as fh: + fh.write('[branch "kid3"]\n\tremote = .\n\tmerge = refs/heads/par3\n') + git(self.repo, "config", "includeIf.onbranch:feat/**.path", include) + kept = {f"par{n}": f"kept: local upstream of kid{n}" for n in (1, 2, 3)} + + run = self._cleanup("local", apply=False) + code, out, _ = self._cli_apply(w2) + + self.assertEqual({n: run.items[n]["skip"] for n in kept}, kept) + items = {i["branch"]: i for i in json.loads(out)["repos"][0]["items"]} + self.assertEqual(code, 0) + self.assertEqual({n: items[n]["skip"] for n in kept}, kept) + self.assertEqual( + {n: i["result"] for n, i in items.items() if i["result"]}, + {"done": "done", "local-done": "done"}, + ) + heads = self._refs(self.repo, "refs/heads") + for name in kept: + self.assertIn(f"refs/heads/{name}", heads) + + def test_inherited_git_variables_never_move_the_local_root(self) -> None: + sentinel = os.path.join(self.root, "sentinel") + git(self.root, "init", "-b", "trunk", sentinel) + dot_git = os.path.join(sentinel, ".git") + + with patch.dict(os.environ, {"GIT_DIR": dot_git, "GIT_COMMON_DIR": dot_git}): + self.assertEqual(gc.local_repo_root(self.held), self.repo) + + def test_failed_head_read_refuses_the_repository(self) -> None: + heads = self._refs(self.repo, "refs/heads") + real = gc._git + + def timed_out(args: list[str], cwd: str) -> str: + if args[1:] == ["symbolic-ref", "--quiet", "HEAD"]: + err = GitRunError("Git command timed out") + err.returncode = -1 + raise err + return real(args, cwd) + + with patch.object(gc, "_git", side_effect=timed_out): + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 2) + self.assertEqual(run.items, {}) + self.assertEqual(run.warnings[-1], "refused: Git command timed out") + self.assertEqual(self._refs(self.repo, "refs/heads"), heads) + + def test_locked_config_is_a_note_on_a_done_item(self) -> None: + with open(os.path.join(self.repo, ".git", "config.lock"), "w") as fh: + fh.write("") + + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 0) + self.assertTrue( + run.items["done"]["result"].startswith("done; config branch.done kept: ") + ) + self.assertTrue(run.items["local-done"]["result"].startswith("done")) + self.assertNotIn("refs/heads/done", self._refs(self.repo, "refs/heads")) + self.assertEqual( + self._branch_config("done"), ["branch.done.remote", "branch.done.merge"] + ) + + def test_repository_hooks_never_run(self) -> None: + marker = os.path.join(self.root, "hook.log") + hooks = os.path.join(self.repo, ".git", "hooks") + os.makedirs(hooks, exist_ok=True) + for name in ("reference-transaction", "post-checkout"): + path = os.path.join(hooks, name) + with open(path, "w", encoding="utf-8") as fh: + fh.write(MARKING_HOOK.format(marker=marker)) + os.chmod(path, 0o755) + + run = self._cleanup("local", apply=True) + + self.assertEqual(self._results(run), {"done": "done", "local-done": "done"}) + self.assertFalse(os.path.exists(marker)) + + def test_config_keep_made_after_the_plan_skips_the_delete(self) -> None: + real = gc._rebuild + + def late_keep(repo: gc.Repo, item: gc.Item) -> gc.Item: + fresh = real(repo, item) + if item.branch == "done": + git(self.repo, "branch", "kid", "wip") + git(self.repo, "config", "branch.kid.remote", ".") + git(self.repo, "config", "branch.kid.merge", "refs/heads/done") + if item.branch == "local-done": + git(self.repo, "config", "branch.local-done.deleteMerged", "false") + return fresh + + with patch.object(gc, "_rebuild", side_effect=late_keep): + run = self._cleanup("local", apply=True) + + self.assertEqual(run.code, 0) + self.assertEqual( + self._results(run), + { + "done": "skipped: changed since plan (kept: local upstream of kid)", + "local-done": ( + "skipped: changed since plan " + "(kept: branch.local-done.deleteMerged is false)" + ), + }, + ) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/done", heads) + self.assertIn("refs/heads/local-done", heads) + self.assertEqual( + self._branch_config("done"), ["branch.done.remote", "branch.done.merge"] + ) + + def _family(self) -> None: + self._set_head() + git(self.repo, "branch", "a-base", "done") + git(self.repo, "branch", "z-kid", "done") + git(self.repo, "config", "branch.z-kid.remote", ".") + git(self.repo, "config", "branch.z-kid.merge", "refs/heads/a-base") + + def _apply_after(self, change: Callable[[], None]) -> SimpleNamespace: + """ + Args: + change: run once after the plan, before the first re-check of + `--local --apply --branch a-base --branch z-kid`. + + Returns: + As _cleanup returns it. + """ + real = gc._rebuild + changes = [change] + + def late(repo: gc.Repo, item: gc.Item) -> gc.Item: + if changes: + changes.pop()() + return real(repo, item) + + with patch.object(gc, "_rebuild", side_effect=late): + return self._cleanup("local", apply=True, include=("a-base", "z-kid")) + + def assertFamilyKept(self, run: SimpleNamespace) -> None: + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/a-base", heads) + self.assertIn("refs/heads/z-kid", heads) + self.assertFalse((run.items["a-base"]["result"] or "").startswith("done")) + self.assertEqual( + git(self.repo, "config", "--get", "branch.z-kid.merge"), "refs/heads/a-base" + ) + + def test_kid_held_before_the_plan_keeps_its_base(self) -> None: + self._family() + git(self.repo, "worktree", "add", os.path.join(self.root, "wt-kid"), "z-kid") + + run = self._cleanup("local", apply=True, include=("a-base", "z-kid")) + + self.assertEqual(run.items["a-base"]["skip"], "kept: local upstream of z-kid") + self.assertFamilyKept(run) + + def test_kid_held_after_the_plan_keeps_its_base(self) -> None: + self._family() + kid = os.path.join(self.root, "wt-kid") + + run = self._apply_after(lambda: git(self.repo, "worktree", "add", kid, "z-kid")) + + self.assertEqual(run.code, 0) + self.assertTrue(run.items["z-kid"]["result"].startswith("skipped: changed")) + self.assertEqual( + run.items["a-base"]["result"], + "skipped: changed since plan (kept: local upstream of z-kid)", + ) + self.assertFamilyKept(run) + + def test_kid_moved_after_the_plan_keeps_its_base(self) -> None: + self._family() + + def move() -> None: + tree = os.path.join(self.root, "wt-tmp") + git(self.repo, "worktree", "add", "--detach", tree, "z-kid") + with open(os.path.join(tree, "kid.txt"), "w", encoding="utf-8") as fh: + fh.write("kid") + git(tree, "add", "kid.txt") + git(tree, "commit", "-m", "kid") + moved = git(tree, "rev-parse", "HEAD") + git(self.repo, "update-ref", "refs/heads/z-kid", moved) + git(self.repo, "worktree", "remove", "--force", tree) + + run = self._apply_after(move) + + self.assertTrue(run.items["z-kid"]["result"].startswith("skipped: changed")) + self.assertFamilyKept(run) + + def test_unchanged_kid_is_deleted_before_its_base(self) -> None: + self._family() + + run = self._cleanup("local", apply=True, include=("a-base", "z-kid")) + + self.assertEqual(run.code, 0) + self.assertEqual(self._results(run), {"a-base": "done", "z-kid": "done"}) + deleted = [ + c.args[0][3] + for c in run.guard.git.call_args_list + if c.args[0][:3] == ["update-ref", "--no-deref", "-d"] + ] + self.assertEqual(deleted, ["refs/heads/z-kid", "refs/heads/a-base"]) + + def test_local_upstream_cycle_is_kept_on_apply(self) -> None: + self._set_head() + for name in ("c1", "c2"): + git(self.repo, "branch", name, "done") + for name, upstream in (("c1", "c2"), ("c2", "c1")): + git(self.repo, "config", f"branch.{name}.remote", ".") + git(self.repo, "config", f"branch.{name}.merge", f"refs/heads/{upstream}") + + run = self._cleanup("local", apply=True, include=("c1", "c2")) + + self.assertEqual( + self._results(run), + { + name: f"skipped: changed since plan (kept: local upstream of {other})" + for name, other in (("c1", "c2"), ("c2", "c1")) + }, + ) + heads = self._refs(self.repo, "refs/heads") + self.assertIn("refs/heads/c1", heads) + self.assertIn("refs/heads/c2", heads) + + def _hub_call(self, *branches: str) -> SimpleNamespace: + """ + Args: + branches: the --branch values of `branch cleanup --local --apply + --json`, run in the fixture repository under _guarded with a + compose stack listed in the held worktree. + + Returns: + The exit code, stdout, stderr, the _guarded spies and the + container listing and docker mocks, as attributes code, out, + err, guard, compose and docker. + """ + argv = ["branch", "cleanup", "--local", "--apply", "--json"] + args = create_parser("test").parse_args( + [*argv, *(f"--branch={name}" for name in branches)] + ) + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + stack = [gc.Container("proj", os.path.join(self.held, "app"), "img")] + os.chdir(self.repo) + out, err = io.StringIO(), io.StringIO() + with ( + self._guarded() as guard, + patch.object(gc, "list_compose_containers", return_value=stack) as ps, + patch.object(gc, "_docker", side_effect=AssertionError) as docker, + redirect_stdout(out), + redirect_stderr(err), + self.assertRaises(SystemExit) as stop, + ): + handle_branch(args, ctx=ctx) + return SimpleNamespace( + code=stop.exception.code, + out=out.getvalue(), + err=err.getvalue(), + guard=guard, + compose=ps, + docker=docker, + ) + + def test_hub_call_deletes_only_a_merged_branch_no_worktree_holds(self) -> None: + self._set_head() + heads = self._refs(self.repo, "refs/heads") + tracking = self._refs(self.repo, "refs/remotes") + origin = self._refs(self.origin, "refs") + trees = git(self.repo, "worktree", "list", "--porcelain") + + refused = self._hub_call("remote-done") + run = self._hub_call("held-done", "wip", "local-done") + + self.assertEqual(refused.code, 2) + self.assertEqual(refused.out, "") + self.assertIn("Unknown --branch: remote-done", refused.err) + self.assertEqual(run.code, 0) + items = {i["branch"]: i for i in json.loads(run.out)["repos"][0]["items"]} + self.assertEqual(set(items), {"held-done", "wip", "local-done"}) + held = items["held-done"] + self.assertEqual(held["skip"], f"checked out in worktree {self.held}") + self.assertEqual(held["worktree"], self.held) + self.assertEqual(held["compose_projects"], []) + self.assertEqual(items["wip"]["state"], "unmerged") + for name in ("held-done", "wip"): + with self.subTest(keeps=name): + self.assertEqual(items[name]["actions"], []) + self.assertIsNone(items[name]["result"]) + self.assertEqual(items["local-done"]["result"], "done") + after = self._refs(self.repo, "refs/heads") + self.assertEqual(set(heads) - set(after), {"refs/heads/local-done"}) + self.assertEqual(self._refs(self.repo, "refs/remotes"), tracking) + self.assertEqual(self._refs(self.origin, "refs"), origin) + self.assertEqual(git(self.repo, "worktree", "list", "--porcelain"), trees) + self.assertTrue(os.path.isfile(os.path.join(self.held, "done.txt"))) + for call in (refused, run): + call.compose.assert_not_called() + call.docker.assert_not_called() + self.assertLocalOnly(call.guard) + + def test_branch_scope_still_needs_the_forge_for_a_remote_copy(self) -> None: + run = self._cleanup("branch", apply=False) + + self.assertEqual(run.code, 0) + self.assertIn("forge unavailable: no GitHub token (none)", run.warnings) + self.assertEqual(run.items["done"]["skip"], "forge unavailable") + self.assertEqual(run.items["remote-done"]["skip"], "forge unavailable") + self.assertIsNone(run.items["local-done"]["skip"]) + self.assertEqual(run.items["local-done"]["actions"], ["local-branch"]) + + def test_cli_local_apply_reads_no_token_and_keeps_the_tree_file(self) -> None: + tree_path = os.path.join(self.root, "branch-tree.json") + with open(tree_path, "w", encoding="utf-8") as fh: + json.dump( + {"repos": [{"path": self.repo, "known": True, "scanned_at": ""}]}, fh + ) + with open(tree_path, "rb") as fh: + before = fh.read() + args = create_parser("test").parse_args( + ["branch", "cleanup", "--local", "--apply", "--json"] + ) + ctx = SimpleNamespace(config_merged={"worktree_hooks": {}, "repositories": []}) + out = io.StringIO() + with ( + self._guarded() as guard, + patch.dict(os.environ, {"PKGMGR_TREE_FILE": tree_path}), + patch("pkgmgr.cli.commands.branch.local_repo_root", return_value=self.repo), + redirect_stdout(out), + redirect_stderr(io.StringIO()), + self.assertRaises(SystemExit) as exit_ctx, + ): + handle_branch(args, ctx=ctx) + + self.assertEqual(exit_ctx.exception.code, 0) + items = { + i["branch"]: i for i in json.loads(out.getvalue())["repos"][0]["items"] + } + self.assertEqual(items["done"]["result"], "done") + self.assertNotIn("refs/heads/done", self._refs(self.repo, "refs/heads")) + with open(tree_path, "rb") as fh: + self.assertEqual(fh.read(), before) + self.assertLocalOnly(guard) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_branch_gc.py b/tests/integration/test_branch_gc.py index 1748d88..46c1f5f 100644 --- a/tests/integration/test_branch_gc.py +++ b/tests/integration/test_branch_gc.py @@ -6,17 +6,18 @@ import os import subprocess import tempfile import unittest -from contextlib import redirect_stdout +from contextlib import redirect_stderr, redirect_stdout from types import SimpleNamespace from unittest.mock import patch -from pkgmgr.actions.branch import gc +from pkgmgr.actions.branch import gc, gc_forge from pkgmgr.actions.branch.gc_forge import ForgeFacts from pkgmgr.cli.commands.branch import handle_branch from pkgmgr.cli.parser import create_parser +from pkgmgr.core.credentials.types import NoCredentialsError -GIT_ENV = { - **os.environ, +GITHUB_URL = "https://github.com/example/demo.git" +AUTHOR = { "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.org", "GIT_COMMITTER_NAME": "t", @@ -28,7 +29,7 @@ def git(cwd: str, *args: str) -> str: return subprocess.run( ["git", "-c", "commit.gpgsign=false", *args], cwd=cwd, - env=GIT_ENV, + env={**os.environ, **AUTHOR}, check=True, capture_output=True, text=True, @@ -46,7 +47,16 @@ def docker_available() -> bool: class GcFixture: def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) self.root = os.path.realpath(self._tmp.name) + self.addCleanup(os.chdir, os.getcwd()) + os.chdir(self.root) + home = os.path.join(self.root, "home") + os.makedirs(home) + scratch = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} + patcher = patch.dict(os.environ, {**scratch, "HOME": home}, clear=True) + patcher.start() + self.addCleanup(patcher.stop) self.origin = os.path.join(self.root, "origin.git") self.repo = os.path.join(self.root, "repo") git(self.root, "init", "--bare", "-b", "main", self.origin) @@ -81,9 +91,6 @@ class GcFixture: merged_heads={self.squashed_sha: 7}, open_heads={"wip": 3} ) - def tearDown(self) -> None: - self._tmp.cleanup() - def _commit(self, name: str) -> None: with open(os.path.join(self.repo, f"{name}.txt"), "w") as fh: fh.write(name) @@ -241,6 +248,80 @@ class TestBranchGc(GcFixture, unittest.TestCase): self.assertEqual(items["merged"].skip, "forge unavailable") self.assertIsNone(items["squashed"].merged_by) + def _github_without_token(self) -> None: + git(self.repo, "branch", "local-done", "merged") + git(self.repo, "checkout", "-b", "local-wip", "main") + self._commit("local-wip") + git(self.repo, "checkout", "main") + git(self.repo, "branch", "--track", "tracked", "origin/merged") + git(self.repo, "push", "origin", "merged:refs/heads/remote-done") + git(self.repo, "remote", "set-url", "origin", GITHUB_URL) + + def _cleanup_without_token(self, apply: bool) -> tuple[int, dict]: + out = io.StringIO() + with ( + patch.dict(os.environ, {"GIT_ALLOW_PROTOCOL": "file"}), + patch.object(gc, "host_pid_namespace", return_value=True), + patch.object(gc, "list_compose_containers", return_value=[]), + patch.object(gc, "_process_cwds", return_value={}), + patch.object(gc_forge, "TokenResolver") as resolver, + patch.object(gc_forge, "HttpClient") as http, + redirect_stdout(out), + redirect_stderr(io.StringIO()), + ): + resolver.return_value.get_token.side_effect = NoCredentialsError("none") + code = gc.cleanup( + [self.repo], + scope="branch", + apply=apply, + as_json=True, + include=[], + exclude=[], + merged_into=[], + sizes=False, + min_age=None, + hooks={}, + ) + resolver.return_value.get_token.assert_called_once() + http.assert_not_called() + return code, json.loads(out.getvalue())["repos"][0] + + def test_local_only_merged_branch_needs_no_forge(self) -> None: + self._github_without_token() + code, plan = self._cleanup_without_token(apply=False) + items = {i["branch"]: i for i in plan["items"] if i["branch"]} + + self.assertEqual(code, 0) + self.assertIn("forge unavailable: no GitHub token (none)", plan["warnings"]) + self.assertIsNone(items["local-done"]["skip"]) + self.assertEqual(items["local-done"]["merged_by"], "ancestor") + self.assertEqual(items["local-done"]["actions"], ["local-branch"]) + self.assertEqual(items["remote-done"]["state"], "merged-clean") + self.assertEqual(items["tracked"]["state"], "merged-clean") + self.assertEqual(items["local-wip"]["state"], "unmerged") + for name in ("merged", "remote-done", "tracked", "local-wip", "squashed"): + with self.subTest(branch=name): + self.assertEqual(items[name]["skip"], "forge unavailable") + + def test_apply_without_token_deletes_only_the_local_only_merged_branch( + self, + ) -> None: + self._github_without_token() + before = set(git(self.repo, "for-each-ref", "--format=%(refname)").split()) + code, plan = self._cleanup_without_token(apply=True) + after = set(git(self.repo, "for-each-ref", "--format=%(refname)").split()) + + self.assertEqual(code, 0) + self.assertEqual( + [(i["branch"], i["result"]) for i in plan["items"] if i["result"]], + [("local-done", "done")], + ) + self.assertEqual(before - after, {"refs/heads/local-done"}) + self.assertEqual(after - before, set()) + self.assertEqual( + git(self.origin, "branch", "--list", "remote-done"), "remote-done" + ) + def test_stale_metadata_is_listed_not_pruned(self) -> None: stub = os.path.join(self.repo, ".git", "worktrees", "stub") os.makedirs(stub) diff --git a/tests/unit/pkgmgr/core/git/test_run.py b/tests/unit/pkgmgr/core/git/test_run.py index a10ea6e..393b4d1 100644 --- a/tests/unit/pkgmgr/core/git/test_run.py +++ b/tests/unit/pkgmgr/core/git/test_run.py @@ -1,8 +1,12 @@ +from __future__ import annotations + +import os +import pwd import unittest from unittest.mock import MagicMock, patch from pkgmgr.core.git.errors import GitRunError -from pkgmgr.core.git.run import run +from pkgmgr.core.git.run import CALLER_CONFIG, DEFAULT_CONFIG, run, strict_env class TestGitRun(unittest.TestCase): @@ -55,6 +59,98 @@ class TestGitRun(unittest.TestCase): self.assertNotIn("stdin", kwargs) self.assertIn("timeout", kwargs) + OVERRIDES = ( + "GIT_DIR", + "GIT_WORK_TREE", + "GIT_INDEX_FILE", + "GIT_COMMON_DIR", + "GIT_CONFIG", + "GIT_CEILING_DIRECTORIES", + "GIT_CONFIG_GLOBAL", + "GIT_CONFIG_SYSTEM", + "GIT_CONFIG_PARAMETERS", + "GIT_CONFIG_COUNT", + "GIT_CONFIG_KEY_0", + "GIT_CONFIG_VALUE_0", + "GIT_CONFIG_KEY_X", + ) + CONFIG_READ = ( + "GIT_CEILING_DIRECTORIES", + "GIT_CONFIG_COUNT", + "GIT_CONFIG_GLOBAL", + "GIT_CONFIG_KEY_0", + "GIT_CONFIG_PARAMETERS", + "GIT_CONFIG_SYSTEM", + "GIT_CONFIG_VALUE_0", + ) + + def _env( + self, scan: bool, strict: bool, view: str | None = None + ) -> dict[str, str] | None: + completed = MagicMock() + completed.stdout = "" + planted = { + **dict.fromkeys(self.OVERRIDES, "/elsewhere"), + "HOME": "/elsewhere", + "XDG_CONFIG_HOME": "/elsewhere", + } + with ( + patch.dict("os.environ", planted), + patch( + "pkgmgr.core.git.run.subprocess.run", return_value=completed + ) as mock_run, + ): + if strict: + with strict_env(): + run(["status"], scan=scan, config_view=view) + else: + run(["status"], scan=scan, config_view=view) + return mock_run.call_args.kwargs.get("env") + + def _kept(self, env: dict[str, str]) -> list[str]: + return sorted(name for name in self.OVERRIDES if name in env) + + def test_scan_drops_repository_overrides_from_the_environment(self) -> None: + unset = ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE") + for view in (None, CALLER_CONFIG): + with self.subTest(view=view): + env = self._env(scan=True, strict=False, view=view) + + self.assertEqual( + self._kept(env), + sorted(name for name in self.OVERRIDES if name not in unset), + ) + self.assertEqual(env.get("GIT_CONFIG_NOSYSTEM"), None if view else "1") + self.assertIsNone(self._env(scan=False, strict=False, view=view)) + + def test_strict_env_keeps_only_the_ceiling(self) -> None: + for scan in (True, False): + with self.subTest(scan=scan): + env = self._env(scan=scan, strict=True) + + self.assertEqual(self._kept(env), ["GIT_CEILING_DIRECTORIES"]) + self.assertEqual(env.get("GIT_CONFIG_NOSYSTEM"), "1") + + def test_strict_caller_config_also_keeps_the_config_variables(self) -> None: + for scan in (True, False): + with self.subTest(scan=scan): + env = self._env(scan=scan, strict=True, view=CALLER_CONFIG) + + self.assertEqual(self._kept(env), list(self.CONFIG_READ)) + self.assertIsNone(env.get("GIT_CONFIG_NOSYSTEM")) + self.assertEqual(env.get("HOME"), "/elsewhere") + + def test_default_config_ignores_the_environment(self) -> None: + home = pwd.getpwuid(os.getuid()).pw_dir + for strict in (True, False): + with self.subTest(strict=strict): + env = self._env(scan=True, strict=strict, view=DEFAULT_CONFIG) + + self.assertEqual(self._kept(env), ["GIT_CEILING_DIRECTORIES"]) + self.assertIsNone(env.get("GIT_CONFIG_NOSYSTEM")) + self.assertEqual(env.get("HOME"), home) + self.assertFalse("XDG_CONFIG_HOME" in env) + def test_failure_raises_giterror_with_details(self) -> None: # Build a CalledProcessError with stdout/stderr populated import subprocess as sp