feat(branch): add pkgmgr branch tree, a hardened meta-tree of all repos
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=<path> 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
c2f0a56699
commit
54f0cbb37b
14 files changed
+1226
-88
No files matched your search
@@ -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()
|
||||
@@ -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):
|
||||
|
||||
Reference in new issue
Block a user