Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 30 additions & 6 deletions acp.py
Original file line number Diff line number Diff line change
Expand Up @@ -445,25 +445,32 @@ def is_github_user(username: str) -> bool:
return result.returncode == 0


def list_branches(show_all: bool = False) -> None:
def list_branches(show_all: bool = False, verbose: bool = False) -> None:
"""List ACP branches with linked PR titles.

By default, only shows branches with linked PRs.
With show_all=True, shows all ACP branches on upstream remote.
"""
gh_user = run(["gh", "api", "user", "--jq", ".login"], quiet=True)
if verbose:
print(f"GitHub user: '{gh_user}'")

if show_all:
remote = (
"upstream"
if run_check(["git", "remote", "get-url", "upstream"])
else "origin"
)
if verbose:
print(f"Fetching remote: '{remote}'")
run_check(["git", "fetch", "--prune", remote])
patterns = [f"{remote}/acp/*", f"{remote}/{gh_user}/acp/*"]
else:
patterns = ["*/acp/*", f"*/{gh_user}/acp/*"]

if verbose:
print(f"Searching patterns: {patterns}")

branches: list[str] = []
for pattern in patterns:
result = subprocess.run(
Expand All @@ -475,13 +482,20 @@ def list_branches(show_all: bool = False) -> None:
if result.returncode != 0:
print(f"Error: {result.stderr}", file=sys.stderr)
sys.exit(1)
for line in result.stdout.strip().splitlines():
branch = line.strip()
if " -> " in branch:
continue
found = [
line.strip()
for line in result.stdout.strip().splitlines()
if " -> " not in line and line.strip()
]
if verbose:
print(f" pattern '{pattern}': {found or '(none)'}")
for branch in found:
if branch not in branches:
branches.append(branch)

if verbose:
print(f"Total local ACP branches found: {len(branches)}")

pr_result = subprocess.run(
[
"gh",
Expand All @@ -504,10 +518,17 @@ def list_branches(show_all: bool = False) -> None:
for pr in json.loads(pr_result.stdout):
pr_map[pr["headRefName"]] = pr

if verbose:
print(
f"Open PRs from 'gh pr list': {list(pr_map.keys()) if pr_map else '(none)'}"
)

if not show_all:
branches = [
b for b in branches if (b.split("/", 1)[1] if "/" in b else b) in pr_map
]
if verbose:
print(f"Branches with linked PRs: {branches or '(none)'}")

if not branches:
if show_all:
Expand Down Expand Up @@ -921,6 +942,9 @@ def main() -> None:
action="store_true",
help="Show all ACP branches on upstream remote",
)
branches_parser.add_argument(
"-v", "--verbose", action="store_true", help="Show detailed output"
)

argcomplete.autocomplete(parser, default_completer=_no_files) # type: ignore[arg-type]
args = parser.parse_args()
Expand All @@ -935,7 +959,7 @@ def main() -> None:
elif args.command == "sync":
sync_fork(branch=args.branch, verbose=args.verbose)
elif args.command == "branches":
list_branches(show_all=args.all)
list_branches(show_all=args.all, verbose=args.verbose)
elif args.command == "checkout":
if not args.branch:
print("Error: Branch name required for checkout", file=sys.stderr)
Expand Down
57 changes: 54 additions & 3 deletions test_acp.py
Original file line number Diff line number Diff line change
Expand Up @@ -1502,6 +1502,45 @@ def test_skips_tracking_refs(self, mock_acp_run, mock_run_check, mock_run, capsy
assert "HEAD" not in output


class TestListBranchesVerbose:
@mock.patch("subprocess.run")
@mock.patch("acp.run", return_value="testuser")
def test_verbose_prints_user_and_patterns(self, mock_acp_run, mock_run, capsys):
mock_run.side_effect = [
subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr=""),
subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr=""),
subprocess.CompletedProcess(args=[], returncode=0, stdout="[]", stderr=""),
]
acp.list_branches(verbose=True)
out = capsys.readouterr().out
assert "GitHub user: 'testuser'" in out
assert "Searching patterns:" in out

@mock.patch("subprocess.run")
@mock.patch("acp.run", return_value="testuser")
def test_verbose_prints_open_prs(self, mock_acp_run, mock_run, capsys):
prs = [
{
"headRefName": "acp/testuser/1234",
"title": "feat",
"number": 1,
"url": "u",
}
]
mock_run.side_effect = [
subprocess.CompletedProcess(
args=[], returncode=0, stdout=" origin/acp/testuser/1234\n", stderr=""
),
subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr=""),
subprocess.CompletedProcess(
args=[], returncode=0, stdout=json.dumps(prs), stderr=""
),
]
acp.list_branches(verbose=True)
out = capsys.readouterr().out
assert "Open PRs from 'gh pr list'" in out


class TestSyncFork:
@mock.patch("subprocess.run")
@mock.patch("acp.run")
Expand Down Expand Up @@ -1722,19 +1761,31 @@ class TestBranchesCommand:
def test_branches_command(self, mock_list):
with mock.patch.object(sys, "argv", ["acp", "branches"]):
acp.main()
mock_list.assert_called_once_with(show_all=False)
mock_list.assert_called_once_with(show_all=False, verbose=False)

@mock.patch("acp.list_branches")
def test_branches_command_with_all(self, mock_list):
with mock.patch.object(sys, "argv", ["acp", "branches", "--all"]):
acp.main()
mock_list.assert_called_once_with(show_all=True)
mock_list.assert_called_once_with(show_all=True, verbose=False)

@mock.patch("acp.list_branches")
def test_branches_command_with_all_short(self, mock_list):
with mock.patch.object(sys, "argv", ["acp", "branches", "-a"]):
acp.main()
mock_list.assert_called_once_with(show_all=True)
mock_list.assert_called_once_with(show_all=True, verbose=False)

@mock.patch("acp.list_branches")
def test_branches_command_verbose(self, mock_list):
with mock.patch.object(sys, "argv", ["acp", "branches", "-v"]):
acp.main()
mock_list.assert_called_once_with(show_all=False, verbose=True)

@mock.patch("acp.list_branches")
def test_branches_command_verbose_long(self, mock_list):
with mock.patch.object(sys, "argv", ["acp", "branches", "--verbose"]):
acp.main()
mock_list.assert_called_once_with(show_all=False, verbose=True)


class TestPull:
Expand Down
Loading