From 814322f8f5de94963540d65b0d8cf28b6ff1b800 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Thu, 12 Nov 2026 14:11:31 +0000 Subject: [PATCH] fix(branch): gc local-only merged branches without a forge token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without a forge token every gc item was skipped before any local check. _needs_forge now decides which items still need PR and protection facts: a branch no worktree holds, without upstream or remote branch, merged into the default by ancestry is decided locally and removed without the forge. Two integration tests fail on 2.4.0 and assert the HTTP client is never called. 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: 100 % 🛡️ Security: 100 % 🏆 Score: 88 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- src/pkgmgr/actions/branch/gc.py | 22 +++++++- tests/integration/test_branch_gc.py | 80 ++++++++++++++++++++++++++++- 2 files changed, 99 insertions(+), 3 deletions(-) diff --git a/src/pkgmgr/actions/branch/gc.py b/src/pkgmgr/actions/branch/gc.py index 1d51eff..c326bd4 100644 --- a/src/pkgmgr/actions/branch/gc.py +++ b/src/pkgmgr/actions/branch/gc.py @@ -415,6 +415,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,7 +445,7 @@ 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 diff --git a/tests/integration/test_branch_gc.py b/tests/integration/test_branch_gc.py index 1748d88..ad6309a 100644 --- a/tests/integration/test_branch_gc.py +++ b/tests/integration/test_branch_gc.py @@ -6,15 +6,17 @@ 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 +GITHUB_URL = "https://github.com/example/demo.git" GIT_ENV = { **os.environ, "GIT_AUTHOR_NAME": "t", @@ -241,6 +243,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)