fix(branch): gc local-only merged branches without a forge token

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 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-12 14:11:31 +00:00
1 parent 9542f68f90
commit 814322f8f5
2 files changed
+99 -3

No files matched your search

+21 -1
View File
@@ -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
+78 -2
View File
@@ -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)