From 54f0cbb37bfd93a0dafcda33d74d6f69a04ea066 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 7 Nov 2026 06:35:50 +0000 Subject: [PATCH] feat(branch): add pkgmgr branch tree, a hardened meta-tree of all repos MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pkgmgr branch tree --format json|yaml --output FILE scans every known repository with gc's own classification (plan_repo with fetch=False) through the hardened git runner and writes one document atomically (flock, mkstemp, fsync, os.replace): per repo path, remote, default, missing, forge status, scanned_at, warnings and every branch and worktree item, plus orphan shared worktrees under ~/.local/share/worktrees. Forge facts take the token only from an unlocked Secret Service item, never unlocking, and cache per repository for 600 s. open, close, drop and gc --apply/purge refresh only their repository's entry and never overwrite a newer scan. Scan hardening for repositories an agent may write: log.showSignature=false joins the scan flags; status is skipped for local or worktree filter.* config and for worktrees with gitlinks; --end-of-options precedes every ref argument and -- every remote, branch and du path, and worktree list runs with -z. Before, a remote named --output= made rev-list create that file on the host, and a worktree path with blank lines forged a record. MEDUSA acceptances (kvb-think-criticize 2026-11-07, 1 skeptic): EXA004 cli/parser/branch_cmd.py:68 by design (-f/--force help text on an operator CLI); B603 actions/branch/gc.py:146 false positive (docker argv list, no shell); B603 actions/branch/gc.py:311 fixed (du gets --, paths from worktree list -z); B603 core/git/run.py:83 by design (argv list, scan callers guard names). 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 104 % ⏱️ Time: 108 % 🛡️ Security: 100 % 🏆 Score: 79 🚦 Gate: ⚠️ cost 104 % > 100, time 108 % > 100: ask the operator before the commit Co-Authored-By: Claude Opus 5.5 --- src/pkgmgr/actions/branch/__init__.py | 3 + src/pkgmgr/actions/branch/gc.py | 269 ++++++++--- src/pkgmgr/actions/branch/gc_forge.py | 14 + src/pkgmgr/actions/branch/tree.py | 342 ++++++++++++++ src/pkgmgr/cli/commands/branch.py | 37 +- src/pkgmgr/cli/parser/branch_cmd.py | 19 + .../core/git/commands/delete_local_branch.py | 2 +- src/pkgmgr/core/git/queries/list_remotes.py | 6 +- .../core/git/queries/resolve_base_branch.py | 5 +- src/pkgmgr/core/git/run.py | 61 ++- tests/integration/test_branch_gc.py | 48 ++ tests/integration/test_branch_tree.py | 440 ++++++++++++++++++ .../branch/test_tree_secret_service.py | 64 +++ .../git/queries/test_resolve_base_branch.py | 4 +- 14 files changed, 1226 insertions(+), 88 deletions(-) create mode 100644 src/pkgmgr/actions/branch/tree.py create mode 100644 tests/integration/test_branch_tree.py create mode 100644 tests/unit/pkgmgr/actions/branch/test_tree_secret_service.py diff --git a/src/pkgmgr/actions/branch/__init__.py b/src/pkgmgr/actions/branch/__init__.py index aac1ee5..73cf94d 100644 --- a/src/pkgmgr/actions/branch/__init__.py +++ b/src/pkgmgr/actions/branch/__init__.py @@ -6,11 +6,14 @@ from .close_branch import close_branch from .drop_branch import drop_branch from .gc import branch_gc, branch_purge from .open_branch import open_branch +from .tree import branch_tree, refresh_tree_entry __all__ = [ "branch_gc", "branch_purge", + "branch_tree", "close_branch", "drop_branch", "open_branch", + "refresh_tree_entry", ] diff --git a/src/pkgmgr/actions/branch/gc.py b/src/pkgmgr/actions/branch/gc.py index d2c3fcf..2a4ab95 100644 --- a/src/pkgmgr/actions/branch/gc.py +++ b/src/pkgmgr/actions/branch/gc.py @@ -17,12 +17,25 @@ from pkgmgr.core.git.run import run from .gc_forge import ForgeFacts, fetch_forge_facts +Snapshot = tuple[ + dict[str, str], + dict[str, dict[str, str]], + dict[int, str], + dict[str, tuple[str | None, int | None, int | None, bool]], +] + # PROC_PID_INIT_INO: only the initial PID namespace sees every process of the host. INIT_PID_NAMESPACE = "pid:[4026531836]" CLEANUP_IMAGE = "alpine:3.20" COMPOSE_PROJECT = "com.docker.compose.project" COMPOSE_WORKING_DIR = "com.docker.compose.project.working_dir" LOCK_PID = re.compile(r"\bpid (\d+)") +TRACK = re.compile(r"(ahead|behind) (\d+)") +UNSAFE_LOCAL_CONFIG = re.compile(r"^(local|worktree)\t(filter\.\S+)") +GITLINK = "160000" +# 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" ALWAYS_PROTECTED = ("main", "master") LABELS = { "merged-clean": "DELETE", @@ -58,6 +71,17 @@ class Container: class Item: branch: str | None worktree: str | None = None + head: str | None = None + detached: bool = False + locked: str | None = None + worktree_state: str | None = None + staged: int | None = None + unstaged: int | None = None + untracked: int | None = None + upstream: str | None = None + ahead: int | None = None + behind: int | None = None + gone: bool = False local_sha: str | None = None remotes: dict[str, str] = field(default_factory=dict) compose_projects: list[str] = field(default_factory=list) @@ -88,6 +112,7 @@ class Repo: remotes: list[str] protected: set[str] first_parent: set[str] + urls: dict[str, str] forge: ForgeFacts containers: list[Container] | None host_ns: bool @@ -109,8 +134,12 @@ def main_repo_root(cwd: str) -> str: return os.path.dirname(_common_dir(cwd)) +def _git(args: list[str], cwd: str) -> str: + return run(args, cwd=cwd, scan=True) + + def _common_dir(cwd: str) -> str: - return run(["rev-parse", "--path-format=absolute", "--git-common-dir"], cwd=cwd) + return _git(["rev-parse", "--path-format=absolute", "--git-common-dir"], cwd) def _docker(args: list[str]) -> str: @@ -151,52 +180,71 @@ def _inside(path: str, root: str) -> bool: def _worktrees(repo_root: str) -> list[Worktree]: trees: list[Worktree] = [] - for block in run(["worktree", "list", "--porcelain"], cwd=repo_root).split("\n\n"): - fields: dict[str, str] = {} - for line in block.splitlines(): - key, _, value = line.partition(" ") + fields: dict[str, str] = {} + for part in _git(["worktree", "list", "--porcelain", "-z"], repo_root).split("\0"): + if part: + key, _, value = part.partition(" ") fields[key] = value - if "worktree" not in fields: continue - branch = fields.get("branch") + record, fields = fields, {} + if "worktree" not in record: + continue + branch = record.get("branch") trees.append( Worktree( - path=fields["worktree"], - head=fields.get("HEAD"), + path=record["worktree"], + head=record.get("HEAD"), branch=branch.removeprefix("refs/heads/") if branch else None, - locked=fields.get("locked"), - prunable="prunable" in fields, + locked=record.get("locked"), + prunable="prunable" in record, main=not trees, ) ) return trees +def _tracking( + upstream: str, track: str +) -> tuple[str | None, int | None, int | None, bool]: + if not upstream: + return None, None, None, False + if track == "gone": + return upstream, None, None, True + counts = dict(TRACK.findall(track)) + return upstream, int(counts.get("ahead", 0)), int(counts.get("behind", 0)), False + + def _refs( repo_root: str, remotes: list[str] -) -> tuple[dict[str, str], dict[str, dict[str, str]]]: - out = run( +) -> tuple[ + dict[str, str], + dict[str, dict[str, str]], + dict[str, tuple[str | None, int | None, int | None, bool]], +]: + out = _git( [ "for-each-ref", - "--format=%(refname) %(objectname)", + "--format=%(refname)%00%(objectname)%00%(upstream:short)%00%(upstream:track,nobracket)", "refs/heads", "refs/remotes", ], - cwd=repo_root, + repo_root, ) by_length = sorted(remotes, key=len, reverse=True) local: dict[str, str] = {} remote: dict[str, dict[str, str]] = {} + tracking: dict[str, tuple[str | None, int | None, int | None, bool]] = {} for line in out.splitlines(): - ref, sha = line.split(" ") + ref, sha, upstream, track = line.split("\x00") if ref.startswith("refs/heads/"): local[ref.removeprefix("refs/heads/")] = sha + tracking[ref.removeprefix("refs/heads/")] = _tracking(upstream, track) continue rest = ref.removeprefix("refs/remotes/") owner = next((r for r in by_length if rest.startswith(r + "/")), None) if owner and rest[len(owner) + 1 :] != "HEAD": remote.setdefault(rest[len(owner) + 1 :], {})[owner] = sha - return local, remote + return local, remote, tracking def _process_cwds() -> dict[int, str]: @@ -232,7 +280,7 @@ def _lock_skip(reason: str | None, host_ns: bool) -> str | None: def _is_ancestor(sha: str, ref: str, repo_root: str) -> bool: try: - run(["merge-base", "--is-ancestor", sha, ref], cwd=repo_root) + _git(["merge-base", "--is-ancestor", END_OF_OPTIONS, sha, ref], repo_root) except GitRunError as exc: if exc.returncode == 1: return False @@ -261,15 +309,13 @@ def _merged_by(repo: Repo, sha: str) -> str | None: def _du_bytes(path: str) -> int | None: out = subprocess.run( - ["du", "-sk", path], check=False, capture_output=True, text=True + ["du", "-sk", "--", path], check=False, capture_output=True, text=True ).stdout.split() return int(out[0]) * 1024 if out and out[0].isdigit() else None def _age_hours(repo: Repo, shas: list[str], tree: Worktree | None) -> float: - stamps = [ - int(run(["log", "-1", "--format=%ct", sha], cwd=repo.repo)) for sha in shas - ] + stamps = [int(_git(["log", "-1", "--format=%ct", sha], repo.repo)) for sha in shas] if tree and os.path.isfile(os.path.join(tree.path, ".git")): with open(os.path.join(tree.path, ".git"), encoding="utf-8") as fh: index = os.path.join(fh.read().strip().removeprefix("gitdir: "), "index") @@ -316,19 +362,83 @@ def _skip( return None +def _unsafe_config(path: str) -> list[str]: + try: + out = _git(["config", "--show-scope", "--get-regexp", r"^filter\."], path) + except GitRunError as exc: + if exc.returncode == 1: + return [] + raise + return [m.group(2) for m in map(UNSAFE_LOCAL_CONFIG.match, out.splitlines()) if m] + + +def _has_submodules(path: str) -> bool: + return any( + line.startswith(f"{GITLINK} ") + for line in _git(["ls-files", "--stage"], path).splitlines() + ) + + +def _status(path: str) -> tuple[int, int, int, int]: + lines = _git( + ["status", "--porcelain=v2", "--untracked-files=all"], path + ).splitlines() + codes = [line.split(" ", 2)[1] for line in lines if line[:1] in "12u"] + return ( + len(lines), + sum(code[0] != "." for code in codes), + sum(code[1] != "." for code in codes), + sum(line.startswith("? ") for line in lines), + ) + + +def _read_worktree(item: Item, checkout: bool) -> tuple[int, str | None]: + if item.worktree is None: + return 0, None + if not checkout: + item.worktree_state = "missing" + return 0, None + try: + if _has_submodules(item.worktree): + item.worktree_state = "changed" + return -1, "status skipped: submodules, whose config the scan cannot vet" + unsafe = _unsafe_config(item.worktree) + if unsafe: + item.worktree_state = "changed" + return -1, f"status skipped: {', '.join(unsafe)} in the repository config" + dirty, item.staged, item.unstaged, item.untracked = _status(item.worktree) + except GitRunError as exc: + item.worktree_state = "changed" + return -1, f"status failed: {_describe(exc)}" + item.worktree_state = "changed" if dirty else "clean" + return dirty, None + + def _build( repo: Repo, branch: str | None, tree: Worktree | None, - snapshot: tuple[dict[str, str], dict[str, dict[str, str]], dict[int, str]], + snapshot: Snapshot, ) -> Item: - local, remote, cwds = snapshot + local, remote, cwds, tracking = snapshot + upstream, ahead, behind, gone = tracking.get( + branch or "", (None, None, None, False) + ) item = Item( branch=branch, worktree=tree.path if tree else None, + head=tree.head if tree else None, + detached=tree is not None and branch is None, + locked=tree.locked if tree else None, + upstream=upstream, + ahead=ahead, + behind=behind, + gone=gone, local_sha=local.get(branch) if branch else None, remotes=dict(remote.get(branch, {})) if branch else {}, ) + checkout = tree is not None and not tree.prunable and os.path.isdir(tree.path) + dirty, unread = _read_worktree(item, checkout) if branch in repo.protected: item.state, item.reason = "protected", "default or protected branch" return item @@ -341,37 +451,30 @@ def _build( shas = [item.local_sha, *item.remotes.values(), tree.head if tree else None] verdicts = {sha: _merged_by(repo, sha) for sha in dict.fromkeys(shas) if sha} unmerged = [sha for sha, verdict in verdicts.items() if verdict is None] - checkout = tree is not None and not tree.prunable and os.path.isdir(tree.path) - try: - dirty = ( - len( - run( - ["status", "--porcelain", "--untracked-files=all"], cwd=tree.path - ).splitlines() - ) - if checkout and not tree.main - else 0 - ) - except GitRunError as exc: - item.state, item.reason = "dirty", f"status failed: {_describe(exc)}" - dirty = -1 - - if dirty > 0: + if unread: + item.state, item.reason = "dirty", unread + elif dirty > 0: item.state, item.reason = "dirty", f"dirty: {dirty} paths" - elif dirty == 0 and unmerged: - ahead = int( - run( - ["rev-list", "--count", f"{repo.default}..{unmerged[0]}"], cwd=repo.repo + elif unmerged: + own = int( + _git( + [ + "rev-list", + "--count", + END_OF_OPTIONS, + f"{repo.default}..{unmerged[0]}", + ], + repo.repo, ) ) item.state = "unmerged" item.reason = ( - f"unmerged: {ahead} commits not in {repo.default}" - if ahead + f"unmerged: {own} commits not in {repo.default}" + if own else "no own commits" ) - elif dirty == 0: + else: item.state = "merged-clean" item.merged_by = next( (verdict for verdict in verdicts.values() if verdict != "ancestor"), @@ -401,19 +504,15 @@ def _build( return item -def _snapshot( - repo: Repo, -) -> tuple[ - list[Worktree], tuple[dict[str, str], dict[str, dict[str, str]], dict[int, str]] -]: - local, remote = _refs(repo.repo, repo.remotes) +def _snapshot(repo: Repo) -> tuple[list[Worktree], Snapshot]: + local, remote, tracking = _refs(repo.repo, repo.remotes) cwds = _process_cwds() if repo.host_ns else {} - return _worktrees(repo.repo), (local, remote, cwds) + return _worktrees(repo.repo), (local, remote, cwds, tracking) def _items(repo: Repo) -> list[Item]: trees, snapshot = _snapshot(repo) - local, remote, _ = snapshot + local, remote, _, _ = snapshot by_branch = {t.branch: t for t in trees if t.branch} names = sorted(set(local) | set(remote) | set(by_branch)) items = [_build(repo, name, by_branch.get(name), snapshot) for name in names] @@ -450,14 +549,23 @@ def _lines( names: set[str] = set() for ref in refs: try: - tip = run( - ["rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}"], cwd=repo_root + tip = _git( + [ + "rev-parse", + "--verify", + "--quiet", + END_OF_OPTIONS, + f"{ref}^{{commit}}", + ], + repo_root, ) except GitRunError: warnings.append(f"merged-into ref {ref} not found") continue first_parent = set( - run(["rev-list", "--first-parent", ref], cwd=repo_root).splitlines() + _git( + ["rev-list", "--first-parent", END_OF_OPTIONS, ref], repo_root + ).splitlines() ) lines[ref] = (tip, first_parent) owner = next((r for r in remotes if ref.startswith(r + "/")), None) @@ -473,6 +581,8 @@ def plan_repo( merged_into: list[str], sizes: bool, min_age: float | None, + fetch: bool = True, + forge_source: Callable[[dict[str, str]], ForgeFacts] | None = None, ) -> Repo: """ Scan one repository and classify every branch and worktree. @@ -484,41 +594,44 @@ def plan_repo( merged_into: integration lines (refs) that also count as merged; each is protected. sizes: measure each worktree's disk usage into size_bytes. min_age: skip items whose newest commit or worktree index is younger (hours). + fetch: fetch every remote first; the tree scan reads the known refs only. + forge_source: remote name -> URL to forge facts; fetch_forge_facts by default. Returns: The repository with its classified items. """ warnings: list[str] = [] - try: - # Auto-maintenance runs worktree-prune, which deletes the metadata of - # sandboxed worktrees whose gitdir stub is missing. - run( - ["fetch", "--all", "--prune", "--quiet", "--no-auto-maintenance"], - cwd=repo_root, - ) - except GitRunError as exc: - warnings.append(f"fetch failed: {_describe(exc)}") + if fetch: + try: + # Auto-maintenance runs worktree-prune, which deletes the metadata of + # sandboxed worktrees whose gitdir stub is missing. + run( + ["fetch", "--all", "--prune", "--quiet", "--no-auto-maintenance"], + cwd=repo_root, + ) + except GitRunError as exc: + warnings.append(f"fetch failed: {_describe(exc)}") if containers is None: warnings.append("docker unavailable: worktree items are skipped") - remotes = sorted(list_remotes(cwd=repo_root), key=lambda r: (r != "origin", r)) + remotes = sorted( + list_remotes(cwd=repo_root, scan=True), key=lambda r: (r != "origin", r) + ) heads: dict[str, str] = {} for remote in remotes: try: - heads[remote] = run( - ["symbolic-ref", "--short", f"refs/remotes/{remote}/HEAD"], - cwd=repo_root, + heads[remote] = _git( + ["symbolic-ref", "--short", f"refs/remotes/{remote}/HEAD"], repo_root ) except GitRunError: continue default = next(iter(heads.values()), None) or resolve_base_branch( - "main", "master", cwd=repo_root + "main", "master", cwd=repo_root, scan=True ) head_names = {ref[len(remote) + 1 :] for remote, ref in heads.items()} - forge = fetch_forge_facts( - {r: run(["remote", "get-url", r], cwd=repo_root) for r in remotes} - ) + urls = {r: _git(["remote", "get-url", "--", r], repo_root) for r in remotes} + forge = (forge_source or fetch_forge_facts)(urls) if forge.unavailable: warnings.append(f"forge unavailable: {forge.unavailable}") lines, line_names = _lines(repo_root, merged_into, remotes, warnings) @@ -535,8 +648,11 @@ def plan_repo( *line_names, }, first_parent=set( - run(["rev-list", "--first-parent", default], cwd=repo_root).splitlines() + _git( + ["rev-list", "--first-parent", END_OF_OPTIONS, default], repo_root + ).splitlines() ), + urls=urls, forge=forge, containers=containers, host_ns=host_ns, @@ -637,6 +753,7 @@ def execute(repo: Repo, item: Item) -> None: [ "push", f"--force-with-lease=refs/heads/{item.branch}:{sha}", + "--", remote, f":refs/heads/{item.branch}", ], diff --git a/src/pkgmgr/actions/branch/gc_forge.py b/src/pkgmgr/actions/branch/gc_forge.py index 7a5730c..1daf34c 100644 --- a/src/pkgmgr/actions/branch/gc_forge.py +++ b/src/pkgmgr/actions/branch/gc_forge.py @@ -79,7 +79,21 @@ def fetch_forge_facts(remote_urls: dict[str, str]) -> ForgeFacts: except CredentialError as exc: facts.unavailable = f"no GitHub token ({exc})" return facts + return facts_with_token(slugs, token) + +def facts_with_token(slugs: list[tuple[str, str]], token: str) -> ForgeFacts: + """ + Fetch the pull-request and protection facts of GitHub repositories. + + Args: + slugs: (owner, name) of each repository. + token: the GitHub token sent as bearer. + + Returns: + The facts, or facts whose unavailable names the failing repository. + """ + facts = ForgeFacts() http = HttpClient(timeout_s=30) headers = { "Authorization": f"Bearer {token}", diff --git a/src/pkgmgr/actions/branch/tree.py b/src/pkgmgr/actions/branch/tree.py new file mode 100644 index 0000000..2240027 --- /dev/null +++ b/src/pkgmgr/actions/branch/tree.py @@ -0,0 +1,342 @@ +from __future__ import annotations + +import fcntl +import json +import os +import re +import sys +import tempfile +import time +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import asdict +from datetime import datetime, timezone + +import yaml + +from pkgmgr.core.credentials.store_keys import build_keyring_key +from pkgmgr.core.git.errors import GitBaseError + +from . import gc +from .gc_forge import GITHUB_HOST, ForgeFacts, facts_with_token, github_slugs + +WORKTREES_ROOT = os.path.join( + os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share"), + "worktrees", +) +WORKTREE_DEPTH = 6 +CACHE_TTL = 600 +SLUG = re.compile(r"^(?!\.\.?$)[\w.-]+$") + + +def now() -> float: + return time.time() + + +def _iso(stamp: float) -> str: + return datetime.fromtimestamp(stamp, timezone.utc).isoformat(timespec="seconds") + + +def tree_file() -> str: + state = os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state") + return os.environ.get("PKGMGR_TREE_FILE") or os.path.join( + state, "pkgmgr", "tree", "branch-tree.json" + ) + + +def _cache_path(owner: str, name: str) -> str: + base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache") + return os.path.join(base, "pkgmgr", "forge", owner, f"{name}.json") + + +def secret_service_token() -> tuple[str | None, str | None]: + """ + Read the GitHub token from the Secret Service without unlocking it. + + Returns: + (token, None), or (None, why no token is available). + """ + try: + import secretstorage + except ImportError: + return None, "secretstorage not installed" + key = build_keyring_key("github", GITHUB_HOST, None) + try: + connection = secretstorage.dbus_init() + try: + items = list( + secretstorage.search_items( + connection, {"service": key.service, "username": key.username} + ) + ) + unlocked = [item for item in items if not item.is_locked()] + if not items: + return None, "no GitHub token in the Secret Service" + if not unlocked: + return None, "Secret Service locked" + return unlocked[0].get_secret().decode(), None + finally: + connection.close() + except (secretstorage.exceptions.SecretStorageException, OSError) as exc: + return None, f"Secret Service unavailable: {exc}" + + +@contextmanager +def _locked(path: str) -> Iterator[None]: + folder, name = os.path.split(os.path.abspath(path)) + os.makedirs(folder, exist_ok=True) + with open(os.path.join(folder, f".{name}.lock"), "w", encoding="utf-8") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + yield + + +def _replace(path: str, text: str) -> None: + folder, name = os.path.split(os.path.abspath(path)) + fd, tmp = tempfile.mkstemp(dir=folder, prefix=f".{name}.") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + finally: + if os.path.exists(tmp): + os.unlink(tmp) + + +def _write(path: str, text: str) -> None: + with _locked(path): + _replace(path, text) + + +def _cached(path: str) -> ForgeFacts | None: + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + if now() - data["fetched_at"] >= CACHE_TTL: + return None + return ForgeFacts( + merged_heads=data["merged_heads"], + open_heads=data["open_heads"], + protected=set(data["protected"]), + ) + except (OSError, ValueError, KeyError, TypeError): + return None + + +def _store(path: str, facts: ForgeFacts) -> None: + data = { + "fetched_at": now(), + "merged_heads": facts.merged_heads, + "open_heads": facts.open_heads, + "protected": sorted(facts.protected), + } + _write(path, json.dumps(data)) + + +def _unavailable(reason: str | None) -> tuple[ForgeFacts, str]: + return ForgeFacts(unavailable=reason), f"unavailable: {reason}" + + +def forge_facts( + urls: dict[str, str], token: str | None, reason: str | None +) -> tuple[ForgeFacts, str]: + """ + Forge facts of one repository from the cache, or live once the cache is stale. + + Args: + urls: remote name -> URL. + token: GitHub token from the Secret Service, or None. + reason: why the token is None. + + Returns: + The facts and their source: none, cached, live or unavailable: . + """ + slugs = github_slugs(urls) + if not slugs: + return ForgeFacts(), "none" + invalid = [f"{o}/{n}" for o, n in slugs if not (SLUG.match(o) and SLUG.match(n))] + if invalid: + return _unavailable(f"invalid GitHub slug {invalid[0]}") + merged, label = ForgeFacts(), "cached" + for owner, name in slugs: + path = _cache_path(owner, name) + facts = _cached(path) + if facts is None: + if token is None: + return _unavailable(reason) + facts = facts_with_token([(owner, name)], token) + if facts.unavailable: + return _unavailable(facts.unavailable) + try: + _store(path, facts) + label = "live" if label == "cached" else label + except OSError as exc: + label = f"live: cache not written ({exc})" + merged.merged_heads.update(facts.merged_heads) + merged.open_heads.update(facts.open_heads) + merged.protected |= facts.protected + return merged, label + + +def _missing(root: str, known: bool, scanned_at: str) -> dict: + return { + "path": root, + "remote": None, + "default": None, + "missing": True, + "known": known, + "forge": None, + "scanned_at": scanned_at, + "warnings": [], + "items": [], + } + + +def _scan( + root: str, + known: bool, + token: str | None, + reason: str | None, + containers: list[gc.Container] | None, + host_ns: bool, +) -> dict: + scanned_at = _iso(now()) + entry = _missing(root, known, scanned_at) + if not os.path.exists(os.path.join(root, ".git")): + return entry + sources: list[str] = [] + + def source(urls: dict[str, str]) -> ForgeFacts: + facts, label = forge_facts(urls, token, reason) + sources.append(label) + return facts + + entry["missing"] = False + try: + repo = gc.plan_repo( + root, + containers, + host_ns, + merged_into=[], + sizes=False, + min_age=None, + fetch=False, + forge_source=source, + ) + except GitBaseError as exc: + entry["warnings"] = [f"scan failed: {str(exc).strip().splitlines()[0]}"] + return entry + owner = next((r for r in repo.remotes if repo.default.startswith(f"{r}/")), None) + entry.update( + remote=repo.urls[repo.remotes[0]] if repo.remotes else None, + default=repo.default.removeprefix(f"{owner}/") if owner else repo.default, + forge=sources[0], + warnings=repo.warnings, + items=[asdict(item) for item in repo.items], + ) + return entry + + +def _owner_root(checkout: str) -> str | None: + try: + with open(os.path.join(checkout, ".git"), encoding="utf-8") as fh: + gitdir = fh.read().strip().removeprefix("gitdir: ") + except OSError: + return None + parent = os.path.dirname(os.path.realpath(os.path.join(checkout, gitdir))) + if os.path.basename(parent) != "worktrees": + return None + common = os.path.dirname(parent) + return os.path.dirname(common) if os.path.basename(common) == ".git" else common + + +def _shared_owners(worktrees_root: str) -> list[str]: + owners: set[str] = set() + base = worktrees_root.rstrip("/").count("/") + for path, dirs, files in os.walk(worktrees_root): + dirs.sort() + owner = _owner_root(path) if ".git" in files else None + if owner: + owners.add(owner) + if ( + ".git" in files + or ".git" in dirs + or path.count("/") - base >= WORKTREE_DEPTH + ): + dirs.clear() + return sorted(owners) + + +def branch_tree( + repo_dirs: list[str], *, fmt: str, output: str | None, worktrees_root: str +) -> int: + """ + Print or write the branch and worktree state of every repository, without fetching. + + Args: + repo_dirs: directories of the repositories pkgmgr knows. + fmt: json or yaml. + output: file to write atomically; None prints to stdout. + worktrees_root: shared worktree root whose unknown owners are added. + + Returns: + 0 on success, 1 when the output could not be written. + """ + known = list(dict.fromkeys(os.path.realpath(d) for d in repo_dirs)) + orphans = [d for d in _shared_owners(worktrees_root) if d not in known] + token, reason = secret_service_token() + host_ns = gc.host_pid_namespace() + containers = gc.list_compose_containers() + data = { + "generated_at": _iso(now()), + "host_pid_namespace": host_ns, + "delete_ready": token is not None, + "delete_reason": reason, + "repos": [ + _scan(root, root in known, token, reason, containers, host_ns) + for root in known + orphans + ], + } + text = ( + yaml.safe_dump(data, sort_keys=False) + if fmt == "yaml" + else json.dumps(data, indent=2) + "\n" + ) + if output is None: + sys.stdout.write(text) + return 0 + try: + _write(output, text) + except OSError as exc: + print(f"Cannot write {output}: {exc}", file=sys.stderr) + return 1 + return 0 + + +def refresh_tree_entry(repo_root: str) -> None: + """ + Rescan one repository into the tree file, if the file exists and holds it. + + Args: + repo_root: main worktree root of the repository. + + Skipped outside the host PID namespace and when the file's entry is newer. + """ + path = tree_file() + if not os.path.isfile(path) or not gc.host_pid_namespace(): + return + root = os.path.realpath(repo_root) + token, reason = secret_service_token() + fresh = _scan(root, True, token, reason, gc.list_compose_containers(), True) + with _locked(path): + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + index = next( + (i for i, entry in enumerate(data["repos"]) if entry["path"] == root), None + ) + if index is None or data["repos"][index]["scanned_at"] > fresh["scanned_at"]: + return + fresh["known"] = data["repos"][index]["known"] + data["repos"][index] = fresh + _replace(path, json.dumps(data, indent=2) + "\n") diff --git a/src/pkgmgr/cli/commands/branch.py b/src/pkgmgr/cli/commands/branch.py index dafce80..cfee18b 100644 --- a/src/pkgmgr/cli/commands/branch.py +++ b/src/pkgmgr/cli/commands/branch.py @@ -9,6 +9,7 @@ from pkgmgr.actions.branch import ( close_branch, drop_branch, open_branch, + tree, ) from pkgmgr.actions.branch.gc import main_repo_root from pkgmgr.cli.context import CLIContext @@ -16,11 +17,27 @@ from pkgmgr.core.git.errors import GitBaseError from pkgmgr.core.repository.dir import get_repo_dir +def _known_dirs(ctx: CLIContext) -> list[str]: + return [ + os.path.expanduser(str(r["directory"])) + if r.get("directory") + else get_repo_dir(ctx.repositories_base_dir, r) + for r in ctx.all_repositories + ] + + def _gc_repo_roots(args, ctx: CLIContext) -> list[str]: if not args.all: return [main_repo_root(".")] - dirs = [get_repo_dir(ctx.repositories_base_dir, r) for r in ctx.all_repositories] - return [d for d in dirs if os.path.exists(os.path.join(d, ".git"))] + return [d for d in _known_dirs(ctx) if os.path.exists(os.path.join(d, ".git"))] + + +def _refresh(roots: list[str]) -> None: + for root in roots: + try: + tree.refresh_tree_entry(root) + except (OSError, ValueError, KeyError, GitBaseError) as exc: + print(f"WARN tree file not rewritten for {root}: {exc}", file=sys.stderr) def handle_branch(args, ctx: CLIContext) -> None: @@ -35,7 +52,18 @@ def handle_branch(args, ctx: CLIContext) -> None: [--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] """ + if args.subcommand == "tree": + sys.exit( + tree.branch_tree( + _known_dirs(ctx), + fmt=args.format, + output=args.output, + worktrees_root=tree.WORKTREES_ROOT, + ) + ) + if args.subcommand in ("gc", "purge"): try: roots = _gc_repo_roots(args, ctx) @@ -53,6 +81,8 @@ def handle_branch(args, ctx: CLIContext) -> None: code = branch_gc(roots, apply=args.apply, as_json=args.json, **options) else: code = branch_purge(roots, **options) + if args.subcommand == "purge" or args.apply: + _refresh(roots) sys.exit(code) if args.subcommand == "open": @@ -61,6 +91,7 @@ def handle_branch(args, ctx: CLIContext) -> None: base_branch=getattr(args, "base", "main"), cwd=".", ) + _refresh([main_repo_root(".")]) return if args.subcommand == "close": @@ -70,6 +101,7 @@ def handle_branch(args, ctx: CLIContext) -> None: cwd=".", force=getattr(args, "force", False), ) + _refresh([main_repo_root(".")]) return if args.subcommand == "drop": @@ -79,6 +111,7 @@ def handle_branch(args, ctx: CLIContext) -> None: cwd=".", force=getattr(args, "force", False), ) + _refresh([main_repo_root(".")]) return print(f"Unknown branch subcommand: {args.subcommand}") diff --git a/src/pkgmgr/cli/parser/branch_cmd.py b/src/pkgmgr/cli/parser/branch_cmd.py index 17b1f5c..dda094b 100644 --- a/src/pkgmgr/cli/parser/branch_cmd.py +++ b/src/pkgmgr/cli/parser/branch_cmd.py @@ -115,6 +115,25 @@ def add_branch_subparsers( help="Print the plan as JSON", ) + branch_tree = branch_subparsers.add_parser( + "tree", + help=( + "Print the branches and worktrees of every pkgmgr-managed repository " + "with their gc state, without fetching" + ), + ) + branch_tree.add_argument( + "--format", + choices=("json", "yaml"), + default="json", + help="Output format (default: json)", + ) + branch_tree.add_argument( + "--output", + metavar="FILE", + help="Write atomically to FILE instead of stdout", + ) + branch_purge = branch_subparsers.add_parser( "purge", help="Delete unmerged or dirty branches after a confirmation per item", diff --git a/src/pkgmgr/core/git/commands/delete_local_branch.py b/src/pkgmgr/core/git/commands/delete_local_branch.py index 6e07237..012bdef 100644 --- a/src/pkgmgr/core/git/commands/delete_local_branch.py +++ b/src/pkgmgr/core/git/commands/delete_local_branch.py @@ -11,7 +11,7 @@ class GitDeleteLocalBranchError(GitCommandError): def delete_local_branch(branch: str, cwd: str = ".", force: bool = False) -> None: flag = "-D" if force else "-d" try: - run(["branch", flag, branch], cwd=cwd) + run(["branch", flag, "--", branch], cwd=cwd) except GitRunError as exc: raise GitDeleteLocalBranchError( f"Failed to delete local branch {branch!r} (flag {flag}).", diff --git a/src/pkgmgr/core/git/queries/list_remotes.py b/src/pkgmgr/core/git/queries/list_remotes.py index 83ba7ec..6945fb5 100644 --- a/src/pkgmgr/core/git/queries/list_remotes.py +++ b/src/pkgmgr/core/git/queries/list_remotes.py @@ -3,13 +3,13 @@ from __future__ import annotations from ..run import run -def list_remotes(cwd: str = ".") -> list[str]: +def list_remotes(cwd: str = ".", scan: bool = False) -> list[str]: """ Return a list of configured git remotes (e.g. ['origin', 'upstream']). - Raises GitBaseError if the command fails. + scan=True runs git hardened. Raises GitBaseError if the command fails. """ - output = run(["remote"], cwd=cwd) + output = run(["remote"], cwd=cwd, scan=scan) if not output: return [] return [line.strip() for line in output.splitlines() if line.strip()] diff --git a/src/pkgmgr/core/git/queries/resolve_base_branch.py b/src/pkgmgr/core/git/queries/resolve_base_branch.py index 35d4070..24c88ec 100644 --- a/src/pkgmgr/core/git/queries/resolve_base_branch.py +++ b/src/pkgmgr/core/git/queries/resolve_base_branch.py @@ -37,12 +37,13 @@ def resolve_base_branch( preferred: str = "main", fallback: str = "master", cwd: str = ".", + scan: bool = False, ) -> str: """ Resolve the base branch to use. Try `preferred` first (default: main), - fall back to `fallback` (default: master). + fall back to `fallback` (default: master); scan=True runs git hardened. Raises GitBaseBranchNotFoundError if neither exists. Raises GitRunError for other git failures (e.g., not a git repository). @@ -51,7 +52,7 @@ def resolve_base_branch( for candidate in (preferred, fallback): try: - run(["rev-parse", "--verify", candidate], cwd=cwd) + run(["rev-parse", "--verify", candidate], cwd=cwd, scan=scan) return candidate except GitRunError as exc: if _is_branch_missing_error(exc): diff --git a/src/pkgmgr/core/git/run.py b/src/pkgmgr/core/git/run.py index a0d5d2c..9fef59d 100644 --- a/src/pkgmgr/core/git/run.py +++ b/src/pkgmgr/core/git/run.py @@ -1,35 +1,84 @@ from __future__ import annotations +import os import subprocess from .errors import GitNotRepositoryError, GitRunError +SCAN_FLAGS = ( + "-c", + "core.fsmonitor=false", + "-c", + "core.hooksPath=/dev/null", + "-c", + "core.pager=cat", + "-c", + "gc.auto=0", + "-c", + "maintenance.auto=false", + "-c", + "log.showSignature=false", +) +SCAN_ENV = { + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_NO_LAZY_FETCH": "1", + "GIT_OPTIONAL_LOCKS": "0", + "GIT_TERMINAL_PROMPT": "0", + "GIT_ASKPASS": "/bin/false", + "LC_ALL": "C", +} +SCAN_UNSET = ( + "GIT_DIR", + "GIT_WORK_TREE", + "GIT_INDEX_FILE", + "GIT_EXTERNAL_DIFF", + "GIT_PAGER", + "PAGER", +) +SCAN_TIMEOUT = 60 + def _is_not_repo_error(stderr: str) -> bool: msg = (stderr or "").lower() 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} + + def run( args: list[str], *, cwd: str = ".", preview: bool = False, + scan: bool = False, ) -> str: """ Run a Git command and return its stdout as a stripped string. - If preview=True, the command is printed but NOT executed. + Args: + args: the git arguments. + cwd: directory to run in. + preview: print the command instead of running it. + scan: read a repository an agent may have written: no fsmonitor, hook, + pager, lazy fetch, system config or prompt, stdin closed, a timeout. - Raises GitRunError (or a subclass) if execution fails. + Raises GitRunError (or a subclass) if execution fails or times out. """ - cmd = ["git"] + args + cmd = ["git", *SCAN_FLAGS, *args] if scan else ["git"] + args cmd_str = " ".join(cmd) if preview: print(f"[PREVIEW] Would run in {cwd!r}: {cmd_str}") return "" + hardened = ( + {"env": _scan_env(), "stdin": subprocess.DEVNULL, "timeout": SCAN_TIMEOUT} + if scan + else {} + ) try: result = subprocess.run( cmd, @@ -37,7 +86,13 @@ def run( check=True, capture_output=True, text=True, + **hardened, ) + except subprocess.TimeoutExpired as exc: + err = GitRunError(f"Git command timed out in {cwd!r}: {cmd_str}") + err.cwd, err.cmd, err.cmd_str, err.returncode = cwd, cmd, cmd_str, -1 + err.stdout, err.stderr = "", "" + raise err from exc except subprocess.CalledProcessError as exc: stderr = exc.stderr or "" stdout = exc.stdout or "" diff --git a/tests/integration/test_branch_gc.py b/tests/integration/test_branch_gc.py index 43ee9ca..13e1d5a 100644 --- a/tests/integration/test_branch_gc.py +++ b/tests/integration/test_branch_gc.py @@ -331,6 +331,54 @@ class TestBranchGc(unittest.TestCase): self.assertTrue(item.result.startswith("skipped: changed since plan")) self.assertTrue(os.path.isdir(self.merged_wt)) + def test_worktree_path_with_newlines_cannot_forge_a_record(self) -> None: + forged = os.path.join(self.root, "a\n\nworktree -evil") + git(self.repo, "worktree", "add", "-b", "forged", forged, "main") + + paths = {tree.path: tree.branch for tree in gc._worktrees(self.repo)} + + self.assertEqual(paths[forged], "forged") + self.assertNotIn("-evil", paths) + + def test_ref_named_like_an_option_writes_no_file(self) -> None: + target = os.path.join(self.root, "pwn") + os.makedirs(target) + remote = f"--output={target}" + main = git(self.repo, "rev-parse", "main") + git(self.repo, "remote", "remove", "origin") + git(self.repo, "config", f"remote.{remote}.url", self.origin) + git(self.repo, "update-ref", f"refs/remotes/{remote}/main", main) + git( + self.repo, + "symbolic-ref", + f"refs/remotes/{remote}/HEAD", + f"refs/remotes/{remote}/main", + ) + + plan = self._plan(host_ns=True, cwds={}) + + self.assertEqual(plan.default, f"{remote}/main") + self.assertEqual(os.listdir(target), []) + + def test_execute_deletes_branches_named_like_options(self) -> None: + sha = git(self.repo, "rev-parse", "merged") + other = os.path.join(self.root, "other.git") + git(self.root, "init", "--bare", other) + git(self.repo, "push", other, "merged:refs/heads/-evil") + git(self.repo, "config", "remote.-r.url", other) + git(self.repo, "update-ref", "refs/heads/-evil", sha) + git(self.repo, "update-ref", "refs/remotes/-r/-evil", sha) + plan = self._plan(host_ns=True, cwds={}) + item = self._items(plan)["-evil"] + self.assertEqual(item.remotes, {"-r": sha}) + + with patch.object(gc, "_process_cwds", return_value={}): + gc.execute(plan, item) + + self.assertEqual(item.result, "done") + self.assertEqual(git(self.repo, "for-each-ref", "refs/heads/-evil"), "") + self.assertEqual(git(other, "for-each-ref", "refs/heads/-evil"), "") + def test_branch_gc_json_and_refusal(self) -> None: out = io.StringIO() with ( diff --git a/tests/integration/test_branch_tree.py b/tests/integration/test_branch_tree.py new file mode 100644 index 0000000..9704235 --- /dev/null +++ b/tests/integration/test_branch_tree.py @@ -0,0 +1,440 @@ +from __future__ import annotations + +import io +import json +import os +import shutil +import stat +import subprocess +import tempfile +import time +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + +import yaml + +from pkgmgr.actions.branch import gc, tree +from pkgmgr.actions.branch.gc_forge import ForgeFacts +from pkgmgr.cli.commands.branch import handle_branch +from pkgmgr.cli.context import CLIContext +from pkgmgr.cli.parser import create_parser +from pkgmgr.core.git.run import run + +GIT_ENV = { + **os.environ, + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@example.org", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@example.org", +} +VOLATILE = ("generated_at", "scanned_at") + + +def git(cwd: str, *args: str) -> str: + return subprocess.run( + ["git", "-c", "commit.gpgsign=false", *args], + cwd=cwd, + env=GIT_ENV, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def stable(data): + if isinstance(data, dict): + return { + key: stable(value) for key, value in data.items() if key not in VOLATILE + } + if isinstance(data, list): + return [stable(value) for value in data] + return data + + +class TestBranchTree(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.root = os.path.realpath(self._tmp.name) + self.origin = os.path.join(self.root, "origin.git") + self.repo = os.path.join(self.root, "repos", "repo") + self.shared = os.path.join(self.root, "shared-worktrees") + os.makedirs(self.shared) + git(self.root, "init", "--bare", "-b", "main", self.origin) + git(self.root, "init", "-b", "main", self.repo) + git(self.repo, "remote", "add", "origin", self.origin) + self._commit("init") + git(self.repo, "push", "-u", "origin", "main") + git(self.repo, "remote", "set-head", "origin", "main") + self._write(os.path.join(self.repo, ".git", "info"), "exclude", ".worktrees/\n") + + self._pushed("merged") + git(self.repo, "merge", "--no-ff", "-m", "merge merged", "merged") + git(self.repo, "push", "origin", "main") + self._pushed("ahead") + git(self.repo, "checkout", "ahead") + self._commit("ahead-local") + git(self.repo, "checkout", "main") + self._pushed("behind", commits=2) + git(self.repo, "branch", "-f", "behind", "behind~1") + self._pushed("gone") + git(self.repo, "push", "origin", "--delete", "gone") + + self.changed = os.path.join(self.root, "repos", "repo", ".worktrees", "changed") + git(self.repo, "worktree", "add", "-b", "changed", self.changed, "main") + self._write(self.changed, "staged.txt", "s") + git(self.changed, "add", "staged.txt") + self._write(self.changed, "init.txt", "edited") + self._write(self.changed, "loose.txt", "u") + self.detached = os.path.join(self.shared, "repo", "detached") + git(self.repo, "worktree", "add", "--detach", self.detached, "main") + self.missing = os.path.join(self.root, "wt-missing") + git(self.repo, "worktree", "add", "-b", "missing", self.missing, "main") + shutil.rmtree(self.missing) + + self.facts = ForgeFacts(merged_heads={}, open_heads={}) + self.tree_file = os.path.join(self.root, "state", "tree", "branch-tree.json") + self.env = patch.dict( + os.environ, + { + "PKGMGR_TREE_FILE": self.tree_file, + "XDG_CACHE_HOME": os.path.join(self.root, "cache"), + }, + ) + self.env.start() + self.addCleanup(self.env.stop) + self.patches = [ + 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( + tree, + "secret_service_token", + return_value=(None, "Secret Service locked"), + ), + ] + for patcher in self.patches: + patcher.start() + self.addCleanup(patcher.stop) + + def _write(self, cwd: str, name: str, text: str) -> None: + with open(os.path.join(cwd, name), "w", encoding="utf-8") as fh: + fh.write(text) + + def _commit(self, name: str, cwd: str | None = None) -> None: + self._write(cwd or self.repo, f"{name}.txt", name) + git(cwd or self.repo, "add", f"{name}.txt") + git(cwd or self.repo, "commit", "-m", name) + + def _pushed(self, name: str, commits: int = 1) -> None: + git(self.repo, "checkout", "-b", name, "main") + for number in range(commits): + self._commit(f"{name}-{number}") + git(self.repo, "push", "-u", "origin", name) + git(self.repo, "checkout", "main") + + def _tree(self, fmt: str = "json", repos: list[str] | None = None) -> str: + out = io.StringIO() + with redirect_stdout(out): + code = tree.branch_tree( + repos or [self.repo], fmt=fmt, output=None, worktrees_root=self.shared + ) + self.assertEqual(code, 0) + return out.getvalue() + + def _entry(self) -> dict: + data = json.loads(self._tree()) + return next(repo for repo in data["repos"] if repo["path"] == self.repo) + + def _by_branch(self, entry: dict) -> dict[str, dict]: + return {item["branch"]: item for item in entry["items"] if item["branch"]} + + def test_json_carries_repository_branch_and_worktree_fields(self) -> None: + data = json.loads(self._tree()) + entry = next(repo for repo in data["repos"] if repo["path"] == self.repo) + branches = self._by_branch(entry) + + self.assertIn("generated_at", data) + self.assertEqual( + (entry["remote"], entry["default"], entry["missing"]), + (self.origin, "main", False), + ) + self.assertEqual( + ( + branches["ahead"]["upstream"], + branches["ahead"]["ahead"], + branches["ahead"]["behind"], + ), + ("origin/ahead", 1, 0), + ) + self.assertEqual( + (branches["behind"]["ahead"], branches["behind"]["behind"]), (0, 1) + ) + self.assertTrue(branches["gone"]["gone"]) + self.assertFalse(branches["merged"]["gone"]) + changed = branches["changed"] + self.assertEqual( + (changed["worktree"], changed["worktree_state"]), (self.changed, "changed") + ) + self.assertEqual( + (changed["staged"], changed["unstaged"], changed["untracked"]), (1, 1, 1) + ) + self.assertEqual(changed["head"], git(self.repo, "rev-parse", "main")) + self.assertEqual(changed["state"], "dirty") + self.assertEqual(branches["missing"]["worktree_state"], "missing") + self.assertEqual(branches["main"]["worktree_state"], "clean") + detached = [item for item in entry["items"] if item["detached"]] + self.assertEqual( + [(item["worktree"], item["worktree_state"]) for item in detached], + [(self.detached, "clean")], + ) + self.assertEqual(branches["merged"]["merged_by"], "ancestor") + self.assertEqual(entry["forge"], "none") + self.assertEqual( + (data["delete_ready"], data["delete_reason"]), + (False, "Secret Service locked"), + ) + + def test_yaml_prints_the_same_data(self) -> None: + self.assertEqual( + stable(yaml.safe_load(self._tree("yaml"))), + stable(json.loads(self._tree("json"))), + ) + + def test_tree_and_gc_share_every_state(self) -> None: + out = io.StringIO() + with ( + patch.object(gc, "fetch_forge_facts", return_value=ForgeFacts()), + redirect_stdout(out), + ): + gc.branch_gc( + [self.repo], + apply=False, + as_json=True, + include=[], + exclude=[], + merged_into=[], + sizes=False, + min_age=None, + ) + planned = { + (item["branch"], item["worktree"]): ( + item["state"], + item["merged_by"], + item["skip"], + ) + for item in json.loads(out.getvalue())["repos"][0]["items"] + } + listed = { + (item["branch"], item["worktree"]): ( + item["state"], + item["merged_by"], + item["skip"], + ) + for item in self._entry()["items"] + } + self.assertEqual(listed, planned) + + def test_the_scan_runs_no_hook_fsmonitor_or_local_filter(self) -> None: + marker = os.path.join(self.root, "ran") + script = os.path.join(self.root, "plant.sh") + self._write(self.root, "plant.sh", f"#!/bin/sh\ntouch {marker}\n") + os.chmod(script, stat.S_IRWXU) + hooks = os.path.join(self.root, "hooks") + os.makedirs(hooks) + for hook in ("post-index-change", "reference-transaction", "post-checkout"): + shutil.copy(script, os.path.join(hooks, hook)) + git(self.repo, "config", "core.fsmonitor", script) + git(self.repo, "config", "core.hooksPath", hooks) + git(self.repo, "config", "core.pager", script) + git(self.repo, "config", "filter.planted.clean", script) + self._write(self.changed, ".gitattributes", "* filter=planted\n") + + with patch("subprocess.run", wraps=subprocess.run) as spy: + changed = self._by_branch(self._entry())["changed"] + self.assertFalse(os.path.exists(marker)) + self.assertEqual( + (changed["worktree_state"], changed["staged"]), ("changed", None) + ) + self.assertIn("filter.planted.clean", changed["reason"]) + scans = [ + call + for call in spy.call_args_list + if call.args and call.args[0][:1] == ["git"] + ] + self.assertTrue(scans) + for call in scans: + with self.subTest(argv=call.args[0][:6]): + self.assertNotIn("fetch", call.args[0]) + self.assertEqual(call.kwargs["env"]["GIT_NO_LAZY_FETCH"], "1") + self.assertIn("core.fsmonitor=false", call.args[0]) + + def test_a_worktree_with_submodules_is_not_read(self) -> None: + head = git(self.changed, "rev-parse", "HEAD") + git(self.changed, "update-index", "--add", "--cacheinfo", f"160000,{head},sub") + changed = self._by_branch(self._entry())["changed"] + self.assertEqual((changed["state"], changed["staged"]), ("dirty", None)) + self.assertIn("submodules", changed["reason"]) + + def test_the_scan_never_runs_a_planted_gpg_program(self) -> None: + marker = os.path.join(self.root, "verified") + self._write(self.root, "gpg.sh", f"#!/bin/sh\ntouch {marker}\nexit 1\n") + os.chmod(os.path.join(self.root, "gpg.sh"), stat.S_IRWXU) + git(self.repo, "config", "log.showSignature", "true") + git(self.repo, "config", "gpg.program", os.path.join(self.root, "gpg.sh")) + body = git(self.repo, "cat-file", "commit", "HEAD").replace( + "\n\n", + "\ngpgsig -----BEGIN PGP SIGNATURE-----\n \n -----END PGP SIGNATURE-----\n\n", + 1, + ) + signed = subprocess.run( + ["git", "hash-object", "-t", "commit", "-w", "--stdin"], + cwd=self.repo, + input=body, + capture_output=True, + text=True, + check=True, + ).stdout.strip() + git(self.repo, "log", "-1", signed) + self.assertTrue(os.path.exists(marker)) + os.unlink(marker) + run(["log", "-1", "--format=%ct", signed], cwd=self.repo, scan=True) + self.assertFalse(os.path.exists(marker)) + + def test_output_is_written_atomically(self) -> None: + target = os.path.join(self.root, "out", "branch-tree.json") + code = tree.branch_tree( + [self.repo], fmt="json", output=target, worktrees_root=self.shared + ) + self.assertEqual(code, 0) + with open(target, encoding="utf-8") as fh: + self.assertEqual(json.load(fh)["repos"][0]["path"], self.repo) + self.assertEqual( + sorted(os.listdir(os.path.dirname(target))), + [".branch-tree.json.lock", "branch-tree.json"], + ) + + def test_lists_every_known_repository_also_plain_and_missing(self) -> None: + plain = os.path.join(self.root, "repos", "plain") + git(self.root, "init", "-b", "main", plain) + self._commit("plain", cwd=plain) + absent = os.path.join(self.root, "repos", "absent") + ctx = CLIContext( + config_merged={}, + repositories_base_dir=os.path.join(self.root, "base"), + all_repositories=[ + { + "provider": "x", + "account": "y", + "repository": "repo", + "directory": self.repo, + }, + { + "provider": "x", + "account": "y", + "repository": "plain", + "directory": plain, + }, + { + "provider": "x", + "account": "y", + "repository": "absent", + "directory": absent, + }, + ], + binaries_dir="", + user_config_path="", + ) + args = create_parser("test").parse_args(["branch", "tree", "--format", "json"]) + out = io.StringIO() + with ( + patch.object(tree, "WORKTREES_ROOT", self.shared), + redirect_stdout(out), + self.assertRaises(SystemExit) as done, + ): + handle_branch(args, ctx) + self.assertEqual(done.exception.code, 0) + repos = {repo["path"]: repo for repo in json.loads(out.getvalue())["repos"]} + self.assertEqual(set(repos), {self.repo, plain, absent}) + self.assertEqual([item["branch"] for item in repos[plain]["items"]], ["main"]) + self.assertEqual((repos[absent]["missing"], repos[absent]["items"]), (True, [])) + + def test_a_shared_worktree_of_an_unknown_repository_adds_it(self) -> None: + other = os.path.join(self.root, "elsewhere", "other") + git(self.root, "init", "-b", "main", other) + self._commit("other", cwd=other) + orphan = os.path.join(self.shared, "other", "feature") + git(other, "worktree", "add", "-b", "feature", orphan, "main") + repos = {repo["path"]: repo for repo in json.loads(self._tree())["repos"]} + self.assertEqual( + (repos[self.repo]["known"], repos[other]["known"]), (True, False) + ) + self.assertIn(orphan, [item["worktree"] for item in repos[other]["items"]]) + + def test_forge_facts_come_from_the_secret_service_at_most_every_ten_minutes( + self, + ) -> None: + git(self.repo, "remote", "set-url", "origin", "https://github.com/me/repo.git") + locked = self._entry() + self.assertEqual(locked["forge"], "unavailable: Secret Service locked") + self.assertEqual(self._by_branch(locked)["merged"]["merged_by"], "ancestor") + self.assertEqual(self._by_branch(locked)["ahead"]["skip"], "forge unavailable") + facts = ForgeFacts( + merged_heads={ + git(self.repo, "rev-parse", ref): 9 for ref in ("ahead", "origin/ahead") + }, + open_heads={}, + ) + with ( + patch.object(tree, "secret_service_token", return_value=("t0ken", None)), + patch.object(tree, "facts_with_token", return_value=facts) as live, + ): + first = self._by_branch(self._entry()) + second = self._by_branch(self._entry()) + self.assertEqual(live.call_count, 1) + with patch.object(tree, "now", return_value=time.time() + 601): + self._entry() + self.assertEqual(live.call_count, 2) + self.assertEqual( + (first["ahead"]["merged_by"], second["ahead"]["merged_by"]), + ("pr#9", "pr#9"), + ) + data = json.loads(self._tree()) + self.assertEqual( + (data["delete_ready"], data["delete_reason"]), + (False, "Secret Service locked"), + ) + self.assertEqual(data["repos"][0]["forge"], "cached") + with patch.object(tree, "secret_service_token", return_value=("t0ken", None)): + self.assertTrue(json.loads(self._tree())["delete_ready"]) + + def test_an_unwritable_cache_keeps_the_scan_and_names_it(self) -> None: + git(self.repo, "remote", "set-url", "origin", "https://github.com/me/repo.git") + blocker = os.path.join(self.root, "cache-is-a-file") + self._write(self.root, "cache-is-a-file", "") + with ( + patch.dict(os.environ, {"XDG_CACHE_HOME": blocker}), + patch.object(tree, "secret_service_token", return_value=("t0ken", None)), + patch.object(tree, "facts_with_token", return_value=ForgeFacts()), + ): + entry = self._entry() + self.assertTrue(entry["forge"].startswith("live: cache not written")) + self.assertIn("main", self._by_branch(entry)) + + def test_branch_actions_rewrite_the_tree_file_when_it_exists(self) -> None: + tree.branch_tree( + [self.repo], fmt="json", output=self.tree_file, worktrees_root=self.shared + ) + args = create_parser("test").parse_args(["branch", "open", "fresh"]) + cwd = os.getcwd() + os.chdir(self.repo) + self.addCleanup(os.chdir, cwd) + handle_branch(args, ctx=None) + with open(self.tree_file, encoding="utf-8") as fh: + entry = json.load(fh)["repos"][0] + self.assertIn("fresh", [item["branch"] for item in entry["items"]]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/actions/branch/test_tree_secret_service.py b/tests/unit/pkgmgr/actions/branch/test_tree_secret_service.py new file mode 100644 index 0000000..7d32d5c --- /dev/null +++ b/tests/unit/pkgmgr/actions/branch/test_tree_secret_service.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +from pkgmgr.actions.branch import tree + + +class SecretStorageError(Exception): + pass + + +def fake_secretstorage(items: list[MagicMock]) -> types.ModuleType: + module = types.ModuleType("secretstorage") + module.exceptions = types.SimpleNamespace(SecretStorageException=SecretStorageError) + module.connection = MagicMock() + module.dbus_init = MagicMock(return_value=module.connection) + module.search_items = MagicMock(return_value=iter(items)) + return module + + +def item(locked: bool, secret: bytes = b"t0ken") -> MagicMock: + found = MagicMock() + found.is_locked.return_value = locked + found.get_secret.return_value = secret + return found + + +class TestSecretServiceToken(unittest.TestCase): + def _token(self, module: types.ModuleType) -> tuple[str | None, str | None]: + with patch.dict(sys.modules, {"secretstorage": module}): + return tree.secret_service_token() + + def test_a_locked_item_is_never_unlocked(self) -> None: + locked = item(True) + module = fake_secretstorage([locked]) + self.assertEqual(self._token(module), (None, "Secret Service locked")) + locked.get_secret.assert_not_called() + locked.unlock.assert_not_called() + module.connection.close.assert_called_once() + + def test_an_unlocked_item_returns_its_token(self) -> None: + module = fake_secretstorage([item(True), item(False, b"live")]) + self.assertEqual(self._token(module), ("live", None)) + module.search_items.assert_called_once_with( + module.connection, {"service": "pkgmgr:github", "username": "github.com|-"} + ) + + def test_no_item_and_no_bus_name_their_reason(self) -> None: + self.assertEqual( + self._token(fake_secretstorage([])), + (None, "no GitHub token in the Secret Service"), + ) + module = fake_secretstorage([]) + module.dbus_init.side_effect = SecretStorageError("no bus") + self.assertEqual( + self._token(module), (None, "Secret Service unavailable: no bus") + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py b/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py index a1f6e0a..284ddd4 100644 --- a/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py +++ b/tests/unit/pkgmgr/core/git/queries/test_resolve_base_branch.py @@ -16,7 +16,9 @@ class TestResolveBaseBranch(unittest.TestCase): mock_run.return_value = "dummy" result = resolve_base_branch("main", "master", cwd=".") self.assertEqual(result, "main") - mock_run.assert_called_with(["rev-parse", "--verify", "main"], cwd=".") + mock_run.assert_called_with( + ["rev-parse", "--verify", "main"], cwd=".", scan=False + ) @patch("pkgmgr.core.git.queries.resolve_base_branch.run") def test_resolves_fallback(self, mock_run):