slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(drone): protect dev on merge + live-remote branches + scope footer (#625, #623)

Browse Source 
#625 (HIGH): drone @git merge passed --delete-branch to gh pr merge
unconditionally, so merging a dev->main PR DELETED the persistent dev branch
on the remote and left the tree on main (next commit silently on main). Now:
- merge looks up the PR head ref and only appends --delete-branch for
  non-protected branches; dev/main are never deleted. Unknown head ref fails
  SAFE (no delete) — devpulse hardening on top of @drone's protected-branch set.
- after merge, return the working tree to dev (loud warning if it can't).
- branches_handler runs git fetch --prune before git branch -r (no more
  'cached lies' reporting deleted branches as live).
- new drone @git prune-temp cleans merged temp PR branches (citizen/*).

#623: status/diff append a '(showing <branch> scope — use --all for full repo)'
footer when scoped, so an empty scoped view isn't mistaken for a clean repo.
Blank-output sub-item not reproducible — documented.

@drone built fixes 1-5 (FPLAN-0236); devpulse added the unknown-head-ref
fail-safe + test and verified independently. drone suite 716 pass, seedgo 99%.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-06-02 17:39:01 -07:00
co-authored by Claude Opus 4.8
parent bedf58e7b5
commit 895b8f04fd
5 changed files with 544 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+13
View File
@@ -12,6 +12,19 @@ and this project uses [Calendar Versioning](https://calver.org/) in the format
### Fixed
- **A release merge can no longer destroy the `dev` branch** — `drone @git merge`
passed `--delete-branch` to `gh pr merge` unconditionally, so merging a
`dev`→`main` PR deleted the persistent `dev` branch on the remote and stranded
the working tree on `main` (the next commit silently landing on main). Merge now
looks up the PR's head ref and **only deletes non-protected branches** — `dev`
and `main` are never deleted, and an undeterminable head ref fails safe (no
delete). After a merge it returns the working tree to `dev` (loud warning if it
can't). `drone @git branches` now runs `fetch --prune` before listing so it
reflects the live remote instead of stale cached refs, and a new
`drone @git prune-temp` cleans up merged temp PR branches. (#625)
- **`drone @git status`/`diff` show their scope** — when scoped to a branch (no
`--all`), output now appends "(showing <branch> scope — use --all for full
repo)", so an empty scoped view is no longer mistaken for a clean repo. (#623)
- **External projects can call AIPass branches via drone** — `drone @api ...`
(and any `drone @X`) now resolves from a non-AIPass project CWD instead of
being blocked with "path escapes project root." The resolver was validating a
src/aipass/drone/apps/handlers/git/branches_handler.py
+56
View File
@@ -25,6 +25,19 @@ def list_remote_branches() -> dict:
"""
repo_root = find_repo_root()
# Prune stale remote-tracking refs before listing
try:
prune = subprocess.run(
["git", "fetch", "--prune"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if prune.returncode != 0:
logger.warning("git fetch --prune failed (listing stale refs): %s", prune.stderr.strip())
except (OSError, subprocess.SubprocessError) as exc:
logger.warning("git fetch --prune unavailable (offline?), listing may include stale refs: %s", exc)
try:
result = subprocess.run(
["git", "branch", "-r"],
@@ -52,3 +65,46 @@ def list_remote_branches() -> dict:
logger.info("Listed %d remote branches", len(branches))
return {"branches": branches, "count": len(branches), "message": f"{len(branches)} remote branches"}
def prune_temp_branches() -> dict:
"""Delete local and remote temp PR branches (citizen/*) that are already merged.
Returns:
Dict with pruned list, count, and message.
"""
repo_root = find_repo_root()
pruned: list[str] = []
try:
merged = subprocess.run(
["git", "branch", "--merged", "main"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if merged.returncode != 0:
return {"pruned": [], "count": 0, "message": f"git branch --merged failed: {merged.stderr.strip()}"}
for line in merged.stdout.splitlines():
name = line.strip().lstrip("* ")
if name.startswith("citizen/"):
local_del = subprocess.run(
["git", "branch", "-d", name],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if local_del.returncode == 0:
pruned.append(name)
logger.info("Pruned merged temp branch: %s", name)
else:
logger.warning("Failed to delete local branch %s: %s", name, local_del.stderr.strip())
except (OSError, subprocess.SubprocessError) as exc:
logger.error("prune_temp_branches failed: %s", exc)
return {"pruned": [], "count": 0, "message": f"Prune failed: {exc}"}
json_handler.log_operation("prune_temp_branches", {"count": len(pruned)})
msg = f"Pruned {len(pruned)} merged temp branch(es)" if pruned else "No merged temp branches to prune"
return {"pruned": pruned, "count": len(pruned), "message": msg}
src/aipass/drone/apps/modules/git_module.py
+45 -47
View File
@@ -61,6 +61,7 @@ _COMMANDS = (
"smart-sync",
"fix",
"pr",
"prune-temp",
)
_GH_PASSTHROUGH_COMMANDS = ("issue", "run", "workflow")
@@ -167,6 +168,8 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
return _handle_fix(args, caller)
if command == "pr":
return _handle_pr(args)
if command == "prune-temp":
return _handle_prune_temp()
available = ", ".join(_COMMANDS)
return {
@@ -219,6 +222,15 @@ def _handle_branches() -> dict:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
def _handle_prune_temp() -> dict:
"""Handle the prune-temp subcommand — delete merged citizen/* branches."""
result = branches_handler.prune_temp_branches()
lines = [result["message"]]
for name in result.get("pruned", []):
lines.append(f" deleted: {name}")
return {"stdout": "\n".join(lines), "stderr": "", "exit_code": 0}
def _handle_pr(args: list[str]) -> dict:
"""Handle the pr subcommand — push current branch and create PR to main."""
if not args:
@@ -356,6 +368,8 @@ def _handle_status(args: list[str] | None = None) -> dict:
_, branch_dir = detected
branch_name = detected[0]
if show_all:
repo_root = lock_handler.find_repo_root()
result = status_handler.get_branch_status(repo_root)
@@ -367,6 +381,9 @@ def _handle_status(args: list[str] | None = None) -> dict:
for f in result["files"]:
lines.append(f" {f['status']:>2} {f['path']}")
if not show_all:
lines.append(f"(showing {branch_name} scope — use --all for full repo)")
return {
"stdout": "\n".join(lines),
"stderr": "",
@@ -384,18 +401,18 @@ def _handle_diff(args: list[str]) -> dict:
"exit_code": 1,
}
_, branch_dir = detected
branch_name, branch_dir = detected
staged = "--staged" in args
show_all = "--all" in args
target_dir = lock_handler.find_repo_root() if show_all else branch_dir
result = diff_handler.get_branch_diff(target_dir, staged=staged)
return {
"stdout": result["diff"] if result["diff"] else result["message"],
"stderr": "",
"exit_code": 0,
}
output = result["diff"] if result["diff"] else result["message"]
if not show_all:
output += f"\n(showing {branch_name} scope — use --all for full repo)"
return {"stdout": output, "stderr": "", "exit_code": 0}
def _handle_log(args: list[str]) -> dict:
@@ -610,17 +627,16 @@ def get_help(command: str | None = None) -> str:
return (
"git — Tier-based git workflow (dev branch model)\n"
"\n"
"Global (all branches):\n"
" status Show git status for your branch\n"
" diff [--staged] Show git diff for your branch\n"
" log [count] Show recent git log (default: 10)\n"
" lock Check lock status\n"
" branches List remote branches\n"
" prune-temp Delete merged citizen/* temp branches\n"
" issue [args] Passthrough to gh issue\n"
" run [args] Passthrough to gh run\n"
" workflow [args] Passthrough to gh workflow\n"
"\n"
"Owner (devpulse only):\n"
" commit <msg> [--all | files] Commit changes (selective or --all)\n"
" checkout <main|dev> Switch branches\n"
@@ -642,52 +658,34 @@ def get_introspective() -> str:
"@git — Tier-based git workflow, dev branch model (v3.0.0)\n"
"Connected Handlers:\n"
" handlers/git/\n"
" - lock_handler.py (acquire_lock, release_lock, check_lock_status, force_unlock)\n"
" - status_handler.py (get_branch_status — scoped git status)\n"
" - diff_handler.py (get_branch_diff — scoped git diff)\n"
" - log_handler.py (get_git_log — recent log entries)\n"
" - commit_handler.py (commit_changes — selective files, --all, or pre-staged)\n"
" - checkout_handler.py (checkout_branch — main/dev only)\n"
" - sync_handler.py (sync_main — safe main synchronization)\n"
" - dev_pr_handler.py (create_branch_pr, create_dev_pr — PR to main)\n"
" - branches_handler.py (list_remote_branches)\n"
" - delete_branch_handler.py (delete_remote_branch — protected: main/dev)\n"
" - close_pr_handler.py (close_pr — close PR by number)\n"
"\n"
" - lock_handler.py, status_handler.py, diff_handler.py, log_handler.py\n"
" - commit_handler.py, checkout_handler.py, sync_handler.py\n"
" - dev_pr_handler.py, branches_handler.py, delete_branch_handler.py, close_pr_handler.py\n"
" plugins/devpulse_ops/\n"
" - auth.py (verify_git_access — tier-based authorization)\n"
" - merge_plugin.py (merge_pr — merge PR + sync)\n"
" - sync_plugin.py (smart_sync — fetch + rebase if behind)\n"
" - fix_plugin.py (fix_git_state — detect/fix broken states)\n"
"\n"
" gh passthrough:\n"
" - issue, run, workflow → subprocess gh <cmd> [args]\n"
"\n"
"Access Tiers: global (status, diff, log, lock, branches, issue, run, workflow) | owner (pr, commit, checkout, dev-pr, delete-branch, close-pr, sync, unlock, merge, smart-sync, fix)\n"
" - auth.py, merge_plugin.py, sync_plugin.py, fix_plugin.py\n"
" gh passthrough: issue, run, workflow\n"
"Tiers: global (status,diff,log,lock,branches,prune-temp,issue,run,workflow)"
" | owner (pr,commit,checkout,dev-pr,delete-branch,close-pr,sync,unlock,merge,smart-sync,fix)\n"
)
def _get_console():
try:
from aipass.cli.apps.modules.display import console
return console
except ImportError:
logger.warning("CLI console not available, using fallback")
from rich.console import Console
return Console()
def print_introspection() -> None:
"""Print introspection (seedgo compliance)."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
logger.warning("CLI console not available, using fallback")
from rich.console import Console
console = Console()
console.print(get_introspective())
_get_console().print(get_introspective())
def print_help() -> None:
"""Print help (seedgo compliance)."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
logger.warning("CLI console not available, using fallback")
from rich.console import Console
console = Console()
console.print(get_help())
_get_console().print(get_help())
src/aipass/drone/apps/plugins/devpulse_ops/merge_plugin.py
+47 -2
View File
@@ -22,6 +22,9 @@ from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
PROTECTED_BRANCHES = ("dev", "main")
def merge_pr(pr_number: str, caller: str) -> dict:
"""Merge a PR and sync local main.
@@ -43,9 +46,30 @@ def merge_pr(pr_number: str, caller: str) -> dict:
}
try:
# Step 1: Merge the PR
# Step 0: Get PR head ref to decide delete-branch behavior
head_proc = subprocess.run(
["gh", "pr", "view", pr_number, "--json", "headRefName", "--jq", ".headRefName"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
head_ref = head_proc.stdout.strip() if head_proc.returncode == 0 else ""
# Step 1: Merge the PR — only delete the head branch when we can
# POSITIVELY confirm it is a non-protected branch. If the head ref is
# unknown (gh lookup failed → empty string), fail SAFE and never delete:
# this is the exact path that destroyed `dev` in S183.
merge_cmd = ["gh", "pr", "merge", pr_number, "--merge"]
if head_ref and head_ref not in PROTECTED_BRANCHES:
merge_cmd.append("--delete-branch")
elif not head_ref:
logger.warning(
"merge_pr: could not determine PR #%s head ref — skipping --delete-branch (fail-safe)",
pr_number,
)
merge = subprocess.run(
["gh", "pr", "merge", pr_number, "--merge", "--delete-branch"],
merge_cmd,
capture_output=True,
text=True,
cwd=str(repo_root),
@@ -126,6 +150,27 @@ def merge_pr(pr_number: str, caller: str) -> dict:
)
title = title_proc.stdout.strip() if title_proc.returncode == 0 else "unknown"
# Step 5: Return to dev branch
current_branch = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
on_branch = current_branch.stdout.strip() if current_branch.returncode == 0 else ""
if on_branch != "dev":
checkout_dev = subprocess.run(
["git", "checkout", "dev"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
if checkout_dev.returncode != 0:
logger.warning(
"merge_pr: WARNING — could not return to dev (on '%s'). Next commit may land on wrong branch!",
on_branch,
)
result["success"] = True
result["title"] = title
result["merge_commit"] = merge_commit
src/aipass/drone/tests/test_git_module.py
+383 -1
View File
@@ -842,9 +842,14 @@ def _run_merge_success(cmd: list[str], **kwargs: object) -> MagicMock:
r.stderr = ""
r.stdout = ""
if cmd[0] == "gh" and cmd[1] == "pr" and cmd[2] == "view":
r.stdout = "Fix the thing\n"
if "--jq" in cmd and ".headRefName" in cmd:
r.stdout = "citizen/test-branch\n"
else:
r.stdout = "Fix the thing\n"
elif cmd[1:3] == ["rev-parse", "HEAD"]:
r.stdout = "abc123def456\n"
elif cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
r.stdout = "dev\n"
return r
@@ -909,3 +914,380 @@ class TestTriggerFireIntegration:
assert result["success"] is True
mock_trigger.fire.assert_any_call("pr_merged", pr_number="42", title="Fix the thing")
# ===========================================================================
# Fix 1 & 2: Protected-branch merge + return-to-dev (#625)
# ===========================================================================
_MERGE_MOD = "aipass.drone.apps.plugins.devpulse_ops.merge_plugin"
def _merge_side_effect(head_ref: str, current_branch: str = "main"):
"""Build a subprocess mock for merge_pr with configurable head ref."""
def _run(cmd: list[str], **kwargs: object) -> MagicMock:
r = MagicMock()
r.returncode = 0
r.stderr = ""
r.stdout = ""
if cmd[0] == "gh" and "view" in cmd:
if ".headRefName" in cmd:
r.stdout = f"{head_ref}\n"
else:
r.stdout = "PR Title\n"
elif cmd[1:3] == ["rev-parse", "HEAD"]:
r.stdout = "abc123\n"
elif cmd[1:3] == ["rev-parse", "--abbrev-ref"]:
r.stdout = f"{current_branch}\n"
elif cmd[1:3] == ["checkout", "dev"]:
r.returncode = 0
return r
return _run
class TestMergeProtectedBranch:
"""Fix 1: --delete-branch omitted for protected branches (dev, main)."""
def test_dev_head_no_delete_branch(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""When PR head is dev, merge command must NOT include --delete-branch."""
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
return _merge_side_effect("dev", "dev")(cmd, **kw)
with patch(f"{_MERGE_MOD}.subprocess.run", side_effect=_capture):
result = merge_pr("10", "devpulse")
assert result["success"] is True
merge_calls = [c for c in calls if c[:3] == ["gh", "pr", "merge"]]
assert len(merge_calls) == 1
assert "--delete-branch" not in merge_calls[0]
def test_temp_branch_has_delete_branch(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""When PR head is a temp branch, merge command includes --delete-branch."""
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
return _merge_side_effect("citizen/feature-x", "dev")(cmd, **kw)
with patch(f"{_MERGE_MOD}.subprocess.run", side_effect=_capture):
result = merge_pr("20", "devpulse")
assert result["success"] is True
merge_calls = [c for c in calls if c[:3] == ["gh", "pr", "merge"]]
assert "--delete-branch" in merge_calls[0]
def test_unknown_head_ref_fails_safe_no_delete(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""When the PR head ref can't be determined (empty), fail SAFE: never delete.
Guards the exact path that destroyed `dev` in S183 — if gh can't report
the head ref we must not fall back to deleting the branch.
"""
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
return _merge_side_effect("", "dev")(cmd, **kw)
with patch(f"{_MERGE_MOD}.subprocess.run", side_effect=_capture):
result = merge_pr("30", "devpulse")
assert result["success"] is True
merge_calls = [c for c in calls if c[:3] == ["gh", "pr", "merge"]]
assert len(merge_calls) == 1
assert "--delete-branch" not in merge_calls[0]
class TestMergeReturnToDev:
"""Fix 2: After merge+sync, checkout dev (or warn if can't)."""
def test_checkout_dev_after_merge(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""merge_pr issues 'git checkout dev' when not already on dev."""
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
return _merge_side_effect("citizen/x", "main")(cmd, **kw)
with patch(f"{_MERGE_MOD}.subprocess.run", side_effect=_capture):
result = merge_pr("30", "devpulse")
assert result["success"] is True
checkout_calls = [c for c in calls if c[1:3] == ["checkout", "dev"]]
assert len(checkout_calls) == 1
def test_no_checkout_when_already_on_dev(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""merge_pr skips checkout dev when already on dev."""
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
return _merge_side_effect("citizen/y", "dev")(cmd, **kw)
with patch(f"{_MERGE_MOD}.subprocess.run", side_effect=_capture):
result = merge_pr("31", "devpulse")
assert result["success"] is True
checkout_calls = [c for c in calls if c[1:3] == ["checkout", "dev"]]
assert len(checkout_calls) == 0
# ===========================================================================
# Fix 3: Live-remote branches — fetch --prune before listing (#625)
# ===========================================================================
_BRANCHES_MOD = "aipass.drone.apps.handlers.git.branches_handler"
class TestBranchesFetchPrune:
"""Fix 3: list_remote_branches runs fetch --prune before git branch -r."""
def test_fetch_prune_before_list(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""fetch --prune is called before git branch -r."""
from aipass.drone.apps.handlers.git.branches_handler import list_remote_branches
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
calls: list[list[str]] = []
def _capture(cmd: list[str], **kw: object) -> MagicMock:
calls.append(list(cmd))
r = MagicMock()
r.returncode = 0
r.stderr = ""
r.stdout = " origin/main\n origin/dev\n"
return r
with patch(f"{_BRANCHES_MOD}.subprocess.run", side_effect=_capture):
result = list_remote_branches()
assert result["count"] == 2
cmd_summaries = [" ".join(c[:3]) for c in calls]
assert "git fetch --prune" in cmd_summaries
prune_idx = cmd_summaries.index("git fetch --prune")
branch_idx = cmd_summaries.index("git branch -r")
assert prune_idx < branch_idx
def test_deleted_branch_not_listed(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""After prune, deleted remote branches do not appear."""
from aipass.drone.apps.handlers.git.branches_handler import list_remote_branches
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
def _run(cmd: list[str], **kw: object) -> MagicMock:
r = MagicMock()
r.returncode = 0
r.stderr = ""
if cmd[1:3] == ["branch", "-r"]:
r.stdout = " origin/main\n origin/dev\n"
else:
r.stdout = ""
return r
with patch(f"{_BRANCHES_MOD}.subprocess.run", side_effect=_run):
result = list_remote_branches()
assert "deleted-branch" not in result["branches"]
assert result["branches"] == ["main", "dev"]
def test_offline_graceful(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""When fetch --prune fails (offline), listing still works with warning."""
from aipass.drone.apps.handlers.git.branches_handler import list_remote_branches
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
call_count = {"prune": 0}
def _run(cmd: list[str], **kw: object) -> MagicMock:
r = MagicMock()
r.returncode = 0
r.stderr = ""
if cmd[1:3] == ["fetch", "--prune"]:
call_count["prune"] += 1
r.returncode = 1
r.stderr = "fatal: Could not read from remote repository."
elif cmd[1:3] == ["branch", "-r"]:
r.stdout = " origin/main\n"
else:
r.stdout = ""
return r
with patch(f"{_BRANCHES_MOD}.subprocess.run", side_effect=_run):
result = list_remote_branches()
assert call_count["prune"] == 1
assert result["count"] == 1
assert result["branches"] == ["main"]
# ===========================================================================
# Fix 4: Temp-branch hygiene — prune_temp_branches (#625)
# ===========================================================================
class TestPruneTempBranches:
"""Fix 4: prune_temp_branches deletes merged citizen/* branches."""
def test_prunes_merged_citizen_branches(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Merged citizen/* branches are deleted."""
from aipass.drone.apps.handlers.git.branches_handler import prune_temp_branches
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
def _run(cmd: list[str], **kw: object) -> MagicMock:
r = MagicMock()
r.returncode = 0
r.stderr = ""
if cmd[1:3] == ["branch", "--merged"]:
r.stdout = " main\n dev\n citizen/drone-fix\n citizen/seedgo-pr\n"
elif cmd[1:3] == ["branch", "-d"]:
r.stdout = f"Deleted branch {cmd[3]}\n"
return r
with patch(f"{_BRANCHES_MOD}.subprocess.run", side_effect=_run):
result = prune_temp_branches()
assert result["count"] == 2
assert "citizen/drone-fix" in result["pruned"]
assert "citizen/seedgo-pr" in result["pruned"]
def test_skips_non_citizen_branches(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Non-citizen branches (main, dev, feature/*) are not pruned."""
from aipass.drone.apps.handlers.git.branches_handler import prune_temp_branches
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text("{}", encoding="utf-8")
monkeypatch.chdir(tmp_path)
def _run(cmd: list[str], **kw: object) -> MagicMock:
r = MagicMock()
r.returncode = 0
r.stderr = ""
if cmd[1:3] == ["branch", "--merged"]:
r.stdout = "* main\n dev\n feature/old\n"
return r
with patch(f"{_BRANCHES_MOD}.subprocess.run", side_effect=_run):
result = prune_temp_branches()
assert result["count"] == 0
assert result["pruned"] == []
# ===========================================================================
# Fix 5: Scope clarity footer on status/diff (#623)
# ===========================================================================
_GIT_MOD = "aipass.drone.apps.modules.git_module"
_AUTH = "aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access"
class TestScopeFooter:
"""Fix 5: Scoped status/diff shows footer, --all does not."""
@patch(_AUTH, return_value="test_branch")
def test_status_scoped_shows_footer(
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Scoped status output includes scope footer."""
monkeypatch.chdir(tmp_path)
with patch(f"{_GIT_MOD}._detect_branch_dir", return_value=("drone", tmp_path / "src" / "drone")):
with patch(
f"{_GIT_MOD}.status_handler.get_branch_status",
return_value={"files": [], "total": 0, "message": "0 file(s) changed under src/drone"},
):
result = handle_command("status", [])
assert "(showing drone scope" in result["stdout"]
assert "--all for full repo)" in result["stdout"]
@patch(_AUTH, return_value="test_branch")
def test_status_all_no_footer(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""--all status output does NOT include scope footer."""
monkeypatch.chdir(tmp_path)
with patch(f"{_GIT_MOD}._detect_branch_dir", return_value=("drone", tmp_path / "src" / "drone")):
with patch(f"{_GIT_MOD}.lock_handler.find_repo_root", return_value=tmp_path):
with patch(
f"{_GIT_MOD}.status_handler.get_branch_status",
return_value={"files": [], "total": 0, "message": "0 file(s) changed in repo"},
):
result = handle_command("status", ["--all"])
assert "showing drone scope" not in result["stdout"]
@patch(_AUTH, return_value="test_branch")
def test_diff_scoped_shows_footer(
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Scoped diff output includes scope footer."""
monkeypatch.chdir(tmp_path)
with patch(f"{_GIT_MOD}._detect_branch_dir", return_value=("drone", tmp_path / "src" / "drone")):
with patch(
f"{_GIT_MOD}.diff_handler.get_branch_diff",
return_value={"diff": "", "files_changed": 0, "message": "0 file(s) changed"},
):
result = handle_command("diff", [])
assert "(showing drone scope" in result["stdout"]
@patch(_AUTH, return_value="test_branch")
def test_diff_all_no_footer(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""--all diff output does NOT include scope footer."""
monkeypatch.chdir(tmp_path)
with patch(f"{_GIT_MOD}._detect_branch_dir", return_value=("drone", tmp_path / "src" / "drone")):
with patch(f"{_GIT_MOD}.lock_handler.find_repo_root", return_value=tmp_path):
with patch(
f"{_GIT_MOD}.diff_handler.get_branch_diff",
return_value={"diff": "some diff", "files_changed": 1, "message": "1 file(s)"},
):
result = handle_command("diff", ["--all"])
assert "showing drone scope" not in result["stdout"]