feat(system): test
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
@@ -81,12 +81,11 @@ drone @memory archive # Archive memories to vector store
|
||||
drone @memory search <query> # Search archived memories
|
||||
```
|
||||
|
||||
### Git Workflow
|
||||
### Git
|
||||
```
|
||||
drone @git pr 'description' # Create a pull request
|
||||
drone @git status # Git status (branch-scoped)
|
||||
drone @git sync # Sync with main
|
||||
drone @git lock / unlock # Lock/unlock the repo
|
||||
drone @git diff # See changes in your branch
|
||||
drone @git log # Recent commits
|
||||
```
|
||||
|
||||
### Infrastructure
|
||||
|
||||
@@ -81,12 +81,11 @@ drone @memory archive # Archive memories to vector store
|
||||
drone @memory search <query> # Search archived memories
|
||||
```
|
||||
|
||||
### Git Workflow
|
||||
### Git
|
||||
```
|
||||
drone @git pr 'description' # Create a pull request
|
||||
drone @git status # Git status (branch-scoped)
|
||||
drone @git sync # Sync with main
|
||||
drone @git lock / unlock # Lock/unlock the repo
|
||||
drone @git diff # See changes in your branch
|
||||
drone @git log # Recent commits
|
||||
```
|
||||
|
||||
### Infrastructure
|
||||
|
||||
@@ -15,7 +15,7 @@ You are DEVPULSE — Patrick's primary AI collaborator and orchestration hub for
|
||||
- **Delegate heavy code to sub-agents** (`run_in_background: true`). Fire and forget, move on immediately. Launch → continue → get notified → report results. Never block waiting on agents.
|
||||
- Use `drone @branch --help` for command syntax. Use `drone systems` for branch list.
|
||||
- **Always wake after sending dispatch emails.** Send email → wake. Every time. No asking.
|
||||
- **Start watchdog after any dispatch.** Run `drone @devpulse watchdog agent @target` (see Watchdog section) with `run_in_background: true`. Don't wait for the user to ask.
|
||||
- **Start watchdog after any dispatch.** Use Monitor tool: `drone @devpulse watchdog agent @target` (timeout_ms=600000, persistent=false). This streams state changes live into the conversation. Never use run_in_background for watchdog — notifications get buried in task files.
|
||||
|
||||
## Branch Experts — Ask Before Rebuilding
|
||||
|
||||
@@ -32,36 +32,35 @@ When a task belongs to a specialist's DOMAIN, ask them. You can still investigat
|
||||
| Command routing | @drone | @branch resolution, subprocess |
|
||||
| Memory, vectors | @memory | ChromaDB, search, archival |
|
||||
|
||||
## Git Workflow — Always on Main, Drone Only, Never Merge
|
||||
## Git Workflow — Dev Branch, Drone Only, You Are the Gatekeeper
|
||||
|
||||
**Three rules, in order:**
|
||||
**You are the only branch with git write access.** All git/gh commands are blocked at the project level (`Bash(git *)`, `Bash(gh *)`). Drone bypasses this via subprocess — and drone's tier system only grants write access to devpulse.
|
||||
|
||||
1. **Always on main. No exceptions.** You don't create branches. You don't tell other agents to create branches. Branches exist only inside the atomic `drone @git system-pr` window which commits → creates branch → pushes → PRs → returns HEAD to main. Every other moment: you're on main.
|
||||
**Three rules:**
|
||||
|
||||
2. **Never merge PRs.** That's the user's role. You fix, you PR, you stop. The user says "merge X" or merges themselves. Do not run `drone @git merge` without an explicit user instruction for that specific PR number. Past PRs, closed PRs, your own PRs — none of them auto-qualify. User-merges-only is the rule.
|
||||
1. **Work on dev, merge to main when satisfied.** All work happens on the `dev` branch. Stack changes until a feature is complete. Test in Docker against dev. When satisfied: `drone @git merge dev` squash-merges to main.
|
||||
|
||||
3. **Local files are source of truth.** When you make an edit, the file on disk is reality — you don't need to wait for a merge to act on the state you see. But that also means: if the truth is wrong, fix it locally first, then PR. Don't assume remote state matches.
|
||||
2. **You commit, agents don't.** Dispatched agents build code and run tests. They report results. You review the diff and commit via `drone @git commit`. No agent PRs.
|
||||
|
||||
Why main-only: AIPass repo has one shared HEAD. Linger on a non-main HEAD and every agent's next edit lands on the wrong branch. Work gets stranded. Dispatch briefs must never say "create a branch as step 1" — that's what caused the S101 merge mess.
|
||||
|
||||
Never use raw git commands (git commit, git push, git checkout anything, gh pr create). `Bash(git checkout*)` and `Bash(git add -f*)` are denied system-wide in `.claude/settings.json`. Drone handles everything correctly.
|
||||
3. **Local files are source of truth.** When you edit a file, the state on disk IS reality. If the truth is wrong, fix it locally first, then commit.
|
||||
|
||||
```
|
||||
drone @git system-pr "description" # System-wide PR (devpulse only) — commit, branch, push, PR, back to main
|
||||
drone @git pr "description" # Branch-scoped PR (any branch) — dispatched agents use this
|
||||
drone @git status # What changed? (all branches can use)
|
||||
drone @git diff # See the diff (all branches can use)
|
||||
drone @git log # Recent commits (all branches can use)
|
||||
drone @git commit "description" # Commit changes (devpulse only)
|
||||
drone @git checkout dev # Switch to dev branch (devpulse only)
|
||||
drone @git checkout main # Switch to main (devpulse only)
|
||||
drone @git system-pr "description" # System-wide PR (devpulse only)
|
||||
drone @git merge <PR#> # Merge a PR (devpulse only, user must request)
|
||||
drone @git smart-sync # Fetch + rebase if behind (devpulse only)
|
||||
drone @git sync # Pull latest (devpulse only)
|
||||
drone @git smart-sync # Fetch + rebase (devpulse only)
|
||||
drone @git fix # Fix broken git states (devpulse only)
|
||||
drone @git status # What changed?
|
||||
drone @git sync # Pull latest main
|
||||
drone @git lock # Check PR lock status
|
||||
```
|
||||
|
||||
**Dispatch briefs must say `drone @git pr`, not `drone @git system-pr`.** system-pr is devpulse-only. Agents dispatched to branches use `drone @git pr` for their own branch-scoped PRs.
|
||||
**Dispatch briefs must NOT reference any git commands.** Agents have zero git access. They build, test, report. You handle git.
|
||||
|
||||
Read-only git commands are fine: `git status`, `git diff`, `git log`.
|
||||
|
||||
**Never cd to repo root.** `drone @git system-pr` requires `.trinity/passport.json` in the CWD hierarchy. If you cd to the repo root, it fails. Stage files with relative paths from devpulse: `git add ../../../HERALD.md`. Always run drone commands from this directory.
|
||||
**Never cd to repo root.** Drone git commands require `.trinity/passport.json` in the CWD hierarchy. Always run drone commands from this directory.
|
||||
|
||||
## Dispatch — Fresh vs Continue
|
||||
|
||||
@@ -107,7 +106,7 @@ Watchdog is a real devpulse module now (not a bash one-liner). After dispatching
|
||||
**Pattern:**
|
||||
```bash
|
||||
drone @ai_mail dispatch @target "Subject" "Body"
|
||||
drone @devpulse watchdog agent @target # run_in_background: true
|
||||
drone @devpulse watchdog agent @target # use Monitor tool (not run_in_background)
|
||||
```
|
||||
|
||||
The handler resolves `@target` → branch path → `.ai_mail.local/.dispatch.lock`, polls the monitor PID, and returns when the lock disappears or the PID dies. Crash vs success is distinguished by `last_bounce.json`. Default timeout 1800s — override with `--timeout SECONDS`.
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
# Session Wrap-Up
|
||||
|
||||
Purpose: Button up everything at the end of a session — or before a /compact. Memories, plans, git — all tidy. Works for both closing out a chat and preparing for compaction.
|
||||
|
||||
**Workflow:** `/prep` → review output → close chat or `/compact`
|
||||
|
||||
## Execution
|
||||
|
||||
1. Read `.trinity/passport.json` first — re-absorb your identity before writing anything
|
||||
2. Do ALL of the following, then confirm what was updated
|
||||
|
||||
## 1. Memories
|
||||
|
||||
Each memory file plays a distinct role. Update based on what actually changed this session.
|
||||
|
||||
- **`.trinity/passport.json`** — IDENTITY. Who you are: role, capabilities, principles. Only update if identity genuinely evolved this session.
|
||||
- **`.trinity/local.json`** — YOUR MEMORY. Add/update session entry with a summary of work done. Add key_learnings for anything learned. Trim oldest sessions if over 20.
|
||||
- **`.trinity/observations.json`** — YOUR MEMORY OF THE USER. Collaboration insights, preferences, friction points. Skip if nothing new about the user this session.
|
||||
- **`STATUS.local.md`** — PUBLIC STATUS BEACON. Current work, known issues, todos, notepad. Auto-synced to central STATUS.md on PR events — this is how other branches see you. Keep Current Work accurate.
|
||||
|
||||
## 2. Active Plans
|
||||
|
||||
- Check any DPLANs or FPLANs referenced in this session
|
||||
- Update their execution logs, status, decision logs with current state
|
||||
- If a plan was completed, note it (but don't close — the user does that)
|
||||
|
||||
## 3. Git State
|
||||
|
||||
- Run `git status` — report uncommitted changes
|
||||
- If there's a logical commit waiting, suggest it (don't commit without asking)
|
||||
- Note the current branch and any open PRs
|
||||
|
||||
## 4. Inbox
|
||||
|
||||
- Run `drone @ai_mail inbox 2>/dev/null` — report any unread emails
|
||||
- Close any that were already processed but not formally closed
|
||||
|
||||
## 5. Loose Ends
|
||||
|
||||
- Flag anything in-flight: running background agents, dispatched branches waiting for replies, pending decisions
|
||||
- If anything can't survive compaction (e.g., agent IDs needed for resume), write it to STATUS.local.md Notepad
|
||||
|
||||
## Confirm
|
||||
|
||||
List everything updated. Format:
|
||||
```
|
||||
Prep complete:
|
||||
- local.json: [what was added]
|
||||
- observations.json: [updated / skipped]
|
||||
- STATUS.local.md: [updated / skipped]
|
||||
- Plans: [which ones updated]
|
||||
- Git: [branch, uncommitted count, suggestion]
|
||||
- Inbox: [count, action taken]
|
||||
- Loose ends: [any flagged]
|
||||
|
||||
Ready to close out or /compact.
|
||||
```
|
||||
@@ -299,9 +299,30 @@ def is_same_file_as_last(file_path: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _project_has_own_posttooluse_hooks() -> bool:
|
||||
"""Check if CWD is inside a project with its own PostToolUse hooks."""
|
||||
search = Path.cwd()
|
||||
home = Path.home()
|
||||
while search != home and search.parent != search:
|
||||
settings = search / ".claude" / "settings.json"
|
||||
if settings.exists():
|
||||
try:
|
||||
data = json.loads(settings.read_text(encoding="utf-8"))
|
||||
ptu = data.get("hooks", {}).get("PostToolUse", [])
|
||||
if ptu:
|
||||
return True
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
search = search.parent
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Main hook entry point."""
|
||||
try:
|
||||
if _project_has_own_posttooluse_hooks():
|
||||
return
|
||||
|
||||
input_data = json.load(sys.stdin)
|
||||
tool_name = input_data.get("tool_name", "")
|
||||
tool_input = input_data.get("tool_input", {})
|
||||
@@ -363,4 +384,7 @@ Fix these errors in {Path(file_path).name} now. Do not skip or defer."""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("PostToolUse", "provider", __file__, main)
|
||||
|
||||
@@ -6,12 +6,35 @@ Injects branch-specific prompts based on CWD. When working in a branch
|
||||
directory, loads .aipass/aipass_local_prompt.md and outputs it so the
|
||||
AI sees branch-specific context.
|
||||
|
||||
Version: 1.0.0
|
||||
When CWD is inside a project that has its own UserPromptSubmit hooks
|
||||
(e.g. a standalone aipass-init project), this provider-level hook exits
|
||||
silently to avoid double-firing.
|
||||
|
||||
Version: 1.1.0
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _project_has_own_hooks() -> bool:
|
||||
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
|
||||
search = Path.cwd()
|
||||
home = Path.home()
|
||||
while search != home and search.parent != search:
|
||||
settings = search / ".claude" / "settings.json"
|
||||
if settings.exists():
|
||||
try:
|
||||
data = json.loads(settings.read_text(encoding="utf-8"))
|
||||
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
|
||||
if ups:
|
||||
return True
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
search = search.parent
|
||||
return False
|
||||
|
||||
|
||||
def find_branch_root() -> Path | None:
|
||||
"""
|
||||
Find the branch root directory.
|
||||
@@ -39,6 +62,9 @@ def find_branch_root() -> Path | None:
|
||||
|
||||
|
||||
def main():
|
||||
if _project_has_own_hooks():
|
||||
return
|
||||
|
||||
branch_root = find_branch_root()
|
||||
|
||||
if branch_root:
|
||||
@@ -50,4 +76,9 @@ def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("UserPromptSubmit", "provider", __file__, main)
|
||||
|
||||
@@ -5,13 +5,34 @@ Email Notification Hook - Notifies of new emails on prompt submit.
|
||||
Checks the current branch's inbox for unread emails and displays
|
||||
a notification if any exist.
|
||||
|
||||
Version: 1.0.0
|
||||
When CWD is inside a project that has its own UserPromptSubmit hooks,
|
||||
this provider-level hook exits silently to avoid double-firing.
|
||||
|
||||
Version: 1.1.0
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _project_has_own_hooks() -> bool:
|
||||
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
|
||||
search = Path.cwd()
|
||||
home = Path.home()
|
||||
while search != home and search.parent != search:
|
||||
settings = search / ".claude" / "settings.json"
|
||||
if settings.exists():
|
||||
try:
|
||||
data = json.loads(settings.read_text(encoding="utf-8"))
|
||||
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
|
||||
if ups:
|
||||
return True
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
search = search.parent
|
||||
return False
|
||||
|
||||
|
||||
def find_repo_root() -> Path | None:
|
||||
"""Find the repo root (contains pyproject.toml or .git)."""
|
||||
search = Path.cwd()
|
||||
@@ -80,6 +101,9 @@ def count_new_emails(branch_root: Path) -> int:
|
||||
|
||||
|
||||
def main():
|
||||
if _project_has_own_hooks():
|
||||
return
|
||||
|
||||
branch_root = find_branch_root()
|
||||
if not branch_root:
|
||||
return
|
||||
@@ -93,4 +117,9 @@ def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("UserPromptSubmit", "provider", __file__, main)
|
||||
|
||||
@@ -5,13 +5,34 @@ Identity Injector - Injects branch identity on every prompt.
|
||||
Reads from [BRANCH].id.json and outputs core identity fields.
|
||||
Finds the branch root by walking up from CWD looking for apps/ or *.id.json.
|
||||
|
||||
Version: 1.0.0
|
||||
When CWD is inside a project that has its own UserPromptSubmit hooks,
|
||||
this provider-level hook exits silently to avoid double-firing.
|
||||
|
||||
Version: 1.1.0
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _project_has_own_hooks() -> bool:
|
||||
"""Check if CWD is inside a project with its own UserPromptSubmit hooks."""
|
||||
search = Path.cwd()
|
||||
home = Path.home()
|
||||
while search != home and search.parent != search:
|
||||
settings = search / ".claude" / "settings.json"
|
||||
if settings.exists():
|
||||
try:
|
||||
data = json.loads(settings.read_text(encoding="utf-8"))
|
||||
ups = data.get("hooks", {}).get("UserPromptSubmit", [])
|
||||
if ups:
|
||||
return True
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
search = search.parent
|
||||
return False
|
||||
|
||||
|
||||
def find_repo_root() -> Path | None:
|
||||
"""Find the repo root (contains pyproject.toml or .git)."""
|
||||
search = Path.cwd()
|
||||
@@ -97,6 +118,9 @@ def format_identity(data: dict) -> str:
|
||||
|
||||
|
||||
def main():
|
||||
if _project_has_own_hooks():
|
||||
return
|
||||
|
||||
branch_root = find_branch_root()
|
||||
if not branch_root:
|
||||
return
|
||||
@@ -115,4 +139,9 @@ def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("UserPromptSubmit", "provider", __file__, main)
|
||||
|
||||
@@ -85,7 +85,7 @@ def _get_git_info():
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
subprocess.run(
|
||||
status = subprocess.run(
|
||||
["git", "diff", "--stat", "--cached", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
@@ -165,4 +165,7 @@ Context just compacted. Below is your live state. Use it to continue seamlessly.
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("PreCompact", "provider", __file__, main)
|
||||
|
||||
@@ -29,18 +29,41 @@ def _find_repo_root() -> Path | None:
|
||||
AIPASS_ROOT = _find_repo_root()
|
||||
|
||||
|
||||
def _get_cwd_branch() -> str | None:
|
||||
"""Detect which branch directory (src/aipass/<name>) the CWD is in."""
|
||||
cwd = Path.cwd().resolve()
|
||||
if AIPASS_ROOT is None:
|
||||
return None
|
||||
src = AIPASS_ROOT / "src" / "aipass"
|
||||
try:
|
||||
rel = cwd.relative_to(src)
|
||||
return rel.parts[0] if rel.parts else None
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def get_modified_py_files() -> list[str]:
|
||||
"""Get Python files modified in the working tree (unstaged + staged)."""
|
||||
"""Get Python files modified in the working tree, scoped to the CWD branch.
|
||||
|
||||
Only returns files inside the current branch's directory (or repo-root files).
|
||||
This prevents dispatched agents' changes from triggering violations on the
|
||||
orchestrator or other agents sharing the worktree.
|
||||
"""
|
||||
if AIPASS_ROOT is None:
|
||||
return []
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
|
||||
)
|
||||
cwd_branch = _get_cwd_branch()
|
||||
files = []
|
||||
for line in result.stdout.strip().split("\n"):
|
||||
line = line.strip()
|
||||
if line.endswith(".py") and not line.startswith(".claude/"):
|
||||
if cwd_branch and line.startswith("src/aipass/"):
|
||||
file_branch = line.split("/")[2] if len(line.split("/")) > 2 else None
|
||||
if file_branch and file_branch != cwd_branch:
|
||||
continue
|
||||
full = AIPASS_ROOT / line
|
||||
if full.exists():
|
||||
files.append(str(full))
|
||||
@@ -77,6 +100,29 @@ def run_seedgo_checklist(file_path: str) -> list[str]:
|
||||
return []
|
||||
|
||||
|
||||
def check_hook_readme_accountability() -> str | None:
|
||||
"""Check if hook files changed but README wasn't updated. Returns reminder or None."""
|
||||
if AIPASS_ROOT is None:
|
||||
return None
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
|
||||
)
|
||||
changed = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
|
||||
|
||||
hook_files_changed = any(f.startswith(".claude/hooks/") and f.endswith(".py") for f in changed)
|
||||
readme_changed = ".claude/hooks/README.md" in changed
|
||||
|
||||
if hook_files_changed and not readme_changed:
|
||||
return (
|
||||
"Hook files were modified but .claude/hooks/README.md was not updated. "
|
||||
"Consider updating the README to reflect your changes."
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
json.load(sys.stdin)
|
||||
@@ -85,6 +131,8 @@ def main():
|
||||
if not modified:
|
||||
return # Nothing to check
|
||||
|
||||
readme_reminder = check_hook_readme_accountability()
|
||||
|
||||
all_violations = {}
|
||||
for f in modified:
|
||||
vs = run_seedgo_checklist(f)
|
||||
@@ -92,23 +140,30 @@ def main():
|
||||
name = Path(f).name
|
||||
all_violations[name] = vs
|
||||
|
||||
if not all_violations:
|
||||
return # All clear
|
||||
if all_violations:
|
||||
# Build the block reason
|
||||
lines = ["Standards violations found in files you modified:\n"]
|
||||
for fname, vs in all_violations.items():
|
||||
lines.append(f" {fname}:")
|
||||
for v in vs:
|
||||
lines.append(f" - {v}")
|
||||
lines.append("\nFix these violations before finishing.")
|
||||
|
||||
# Build the block reason
|
||||
lines = ["Standards violations found in files you modified:\n"]
|
||||
for fname, vs in all_violations.items():
|
||||
lines.append(f" {fname}:")
|
||||
for v in vs:
|
||||
lines.append(f" - {v}")
|
||||
lines.append("\nFix these violations before finishing.")
|
||||
if readme_reminder:
|
||||
lines.append(f"\n⚠️ {readme_reminder}")
|
||||
|
||||
output = {"decision": "block", "reason": "\n".join(lines)}
|
||||
print(json.dumps(output))
|
||||
output = {"decision": "block", "reason": "\n".join(lines)}
|
||||
print(json.dumps(output))
|
||||
elif readme_reminder:
|
||||
output = {"decision": "allow", "reason": f"⚠️ {readme_reminder}"}
|
||||
print(json.dumps(output))
|
||||
|
||||
except Exception:
|
||||
pass # Silent fail — don't block on errors
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from hook_log import run_and_log
|
||||
|
||||
run_and_log("SubagentStop", "provider", __file__, main)
|
||||
|
||||
@@ -2,38 +2,8 @@
|
||||
"permissions": {
|
||||
"allow": [],
|
||||
"deny": [
|
||||
"Bash(git reset*)",
|
||||
"Bash(git rebase*)",
|
||||
"Bash(git merge*)",
|
||||
"Bash(git config*)",
|
||||
"Bash(git push --force*)",
|
||||
"Bash(git push -f *)",
|
||||
"Bash(git checkout -- *)",
|
||||
"Bash(git checkout .*)",
|
||||
"Bash(git restore --staged*)",
|
||||
"Bash(git restore .*)",
|
||||
"Bash(git clean*)",
|
||||
"Bash(git branch -D*)",
|
||||
"Bash(git stash drop*)",
|
||||
"Bash(git stash clear*)",
|
||||
"Bash(rm -rf*)",
|
||||
"Bash(rm -r *)",
|
||||
"Bash(git checkout -b*)",
|
||||
"Bash(git commit*)",
|
||||
"Bash(git push*)",
|
||||
"Bash(git add -f*)",
|
||||
"Bash(gh pr close*)",
|
||||
"Bash(gh pr create*)",
|
||||
"Bash(*&& git commit*)",
|
||||
"Bash(*&& git push*)",
|
||||
"Bash(*&& git checkout -b*)",
|
||||
"Bash(*&& git add -f*)",
|
||||
"Bash(*&& git reset*)",
|
||||
"Bash(*&& gh pr close*)",
|
||||
"Bash(*&& gh pr create*)",
|
||||
"Bash(*; git commit*)",
|
||||
"Bash(*; git push*)",
|
||||
"Bash(*; git checkout -b*)"
|
||||
"Bash(rm -r *)"
|
||||
],
|
||||
"ask": []
|
||||
},
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"metadata": {
|
||||
"id": "6e3e877e-56ed-4ec5-892f-81073257dc90",
|
||||
"name": "DEVPULSE",
|
||||
"version": "1.0.0",
|
||||
"created": "2026-05-12",
|
||||
"last_updated": "2026-05-12",
|
||||
"total_branches": 0
|
||||
},
|
||||
"branches": []
|
||||
}
|
||||
@@ -154,6 +154,21 @@
|
||||
"lines": [20],
|
||||
"reason": "Test file imports lock_handler directly to test its public interface. Unit tests require direct access to implementation components."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_git_access.py",
|
||||
"standard": "architecture",
|
||||
"reason": "Test file — intentionally outside 3-layer structure. tests/ is a peer of apps/, not part of it."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_git_access.py",
|
||||
"standard": "encapsulation",
|
||||
"reason": "Test file imports handlers directly to test their public interface. Unit tests require direct access to implementation components."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_git_access.py",
|
||||
"standard": "documentation",
|
||||
"reason": "Test class docstrings describe intent; per-method docstrings would be noise for assertion-named test methods."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_system_pr.py",
|
||||
"standard": "architecture",
|
||||
@@ -179,6 +194,18 @@
|
||||
"standard": "unused_function",
|
||||
"lines": [232, 263],
|
||||
"reason": "Public CRUD API surface (update_command, command_exists) — tested, available for programmatic use. No CLI subcommand wired yet."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/git/pr_handler.py",
|
||||
"standard": "unused_function",
|
||||
"lines": [108],
|
||||
"reason": "create_pr() is deprecated per FPLAN-0210 — pr command blocked at auth tier. Handler kept for backwards compatibility; still tested in test_git_module.py."
|
||||
},
|
||||
{
|
||||
"file": "apps/plugins/devpulse_ops/auth.py",
|
||||
"standard": "unused_function",
|
||||
"lines": [84],
|
||||
"reason": "verify_caller() replaced by verify_git_access() for centralized auth. Still exported as public API and tested directly in test_system_pr.py."
|
||||
}
|
||||
],
|
||||
"notes": {
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
"""Git workflow handlers — lock, status, sync, PR."""
|
||||
"""Git workflow handlers — lock, status, sync, PR, diff, log, commit, checkout."""
|
||||
|
||||
from . import lock_handler as lock_handler # explicit re-export for type checkers
|
||||
from . import status_handler as status_handler
|
||||
from . import sync_handler as sync_handler
|
||||
from . import pr_handler as pr_handler
|
||||
from . import diff_handler as diff_handler
|
||||
from . import log_handler as log_handler
|
||||
from . import commit_handler as commit_handler
|
||||
from . import checkout_handler as checkout_handler
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: checkout_handler.py
|
||||
# Description: Branch checkout handler with hard guard
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-12
|
||||
# Modified: 2026-05-12
|
||||
# =============================================
|
||||
|
||||
"""Branch checkout handler with hard guard."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
|
||||
|
||||
_ALLOWED_TARGETS = ("main", "dev")
|
||||
|
||||
|
||||
def checkout_branch(target: str) -> dict:
|
||||
"""Switch to target branch (main or dev only)."""
|
||||
if target not in _ALLOWED_TARGETS:
|
||||
allowed = ", ".join(_ALLOWED_TARGETS)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": f"Checkout denied: only {allowed} branches are allowed. Got '{target}'.",
|
||||
"exit_code": 1,
|
||||
"current_branch": "",
|
||||
}
|
||||
|
||||
repo_root = find_repo_root()
|
||||
|
||||
try:
|
||||
status = subprocess.run(
|
||||
["git", "status", "--porcelain"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
if status.stdout.strip():
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Cannot checkout: uncommitted changes in working tree. Commit or stash first.",
|
||||
"exit_code": 1,
|
||||
"current_branch": "",
|
||||
}
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
logger.error("git status check failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": f"Failed to check working tree status: {exc}",
|
||||
"exit_code": 1,
|
||||
"current_branch": "",
|
||||
}
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "checkout", target],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
logger.error("git checkout failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": f"git checkout failed: {exc}",
|
||||
"exit_code": 1,
|
||||
"current_branch": "",
|
||||
}
|
||||
|
||||
json_handler.log_operation(
|
||||
"checkout_branch",
|
||||
{"target": target, "exit_code": result.returncode},
|
||||
)
|
||||
|
||||
return {
|
||||
"stdout": result.stdout.strip(),
|
||||
"stderr": result.stderr.strip(),
|
||||
"exit_code": result.returncode,
|
||||
"current_branch": target if result.returncode == 0 else "",
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: commit_handler.py
|
||||
# Description: Commit handler with scoped staging
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-12
|
||||
# Modified: 2026-05-12
|
||||
# =============================================
|
||||
|
||||
"""Commit handler with scoped staging."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
|
||||
|
||||
|
||||
def stage_branch_dir(branch_dir: Path, repo_root: Path | None = None) -> dict:
|
||||
"""Stage all changes under branch_dir.
|
||||
|
||||
Shared utility used by both commit_handler and pr_handler.
|
||||
"""
|
||||
if repo_root is None:
|
||||
repo_root = find_repo_root()
|
||||
|
||||
try:
|
||||
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"stage_branch_dir: branch_dir %s not relative to repo root %s",
|
||||
branch_dir,
|
||||
repo_root,
|
||||
)
|
||||
rel_dir = branch_dir
|
||||
|
||||
add_result = subprocess.run(
|
||||
["git", "add", str(rel_dir) + "/"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
|
||||
if add_result.returncode != 0:
|
||||
return {
|
||||
"success": False,
|
||||
"rel_dir": rel_dir,
|
||||
"message": f"Failed to stage files: {add_result.stderr.strip()}",
|
||||
}
|
||||
|
||||
return {"success": True, "rel_dir": rel_dir, "message": f"Staged files under {rel_dir}"}
|
||||
|
||||
|
||||
def commit_changes(
|
||||
message: str,
|
||||
branch_dir: Path | None = None,
|
||||
all_files: bool = False,
|
||||
) -> dict:
|
||||
"""Commit staged changes or all changes under branch_dir."""
|
||||
repo_root = find_repo_root()
|
||||
|
||||
if all_files and branch_dir:
|
||||
stage_result = stage_branch_dir(branch_dir, repo_root)
|
||||
if not stage_result["success"]:
|
||||
return {"stdout": "", "stderr": stage_result["message"], "exit_code": 1}
|
||||
|
||||
diff_check = subprocess.run(
|
||||
["git", "diff", "--cached", "--quiet"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
if diff_check.returncode == 0:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Nothing to commit: no changes staged",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
cmd = ["git", "commit", "-m", message]
|
||||
if branch_dir:
|
||||
try:
|
||||
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
|
||||
except ValueError as exc:
|
||||
logger.warning("commit_changes: branch_dir not relative to repo root: %s", exc)
|
||||
rel_dir = branch_dir
|
||||
cmd.extend(["--", str(rel_dir) + "/"])
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
logger.error("git commit failed: %s", exc)
|
||||
return {"stdout": "", "stderr": f"git commit failed: {exc}", "exit_code": 1}
|
||||
|
||||
json_handler.log_operation(
|
||||
"commit_changes",
|
||||
{"message": message, "all_files": all_files, "exit_code": result.returncode},
|
||||
)
|
||||
|
||||
return {
|
||||
"stdout": result.stdout.strip(),
|
||||
"stderr": result.stderr.strip(),
|
||||
"exit_code": result.returncode,
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: diff_handler.py
|
||||
# Description: Scoped git diff for branch directories
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-12
|
||||
# Modified: 2026-05-12
|
||||
# =============================================
|
||||
|
||||
"""Scoped git diff for branch directories."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
|
||||
|
||||
|
||||
def get_branch_diff(branch_dir: Path, staged: bool = False) -> dict:
|
||||
"""Get git diff filtered to files under branch_dir."""
|
||||
repo_root = find_repo_root()
|
||||
|
||||
cmd = ["git", "diff"]
|
||||
if staged:
|
||||
cmd.append("--staged")
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
logger.error("git diff failed: %s", exc)
|
||||
return {"diff": "", "files_changed": 0, "message": f"git diff failed: {exc}"}
|
||||
|
||||
if result.returncode != 0:
|
||||
return {
|
||||
"diff": "",
|
||||
"files_changed": 0,
|
||||
"message": f"git diff error: {result.stderr.strip()}",
|
||||
}
|
||||
|
||||
try:
|
||||
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"get_branch_diff: branch_dir %s not relative to repo root %s",
|
||||
branch_dir,
|
||||
repo_root,
|
||||
)
|
||||
rel_dir = branch_dir
|
||||
|
||||
rel_prefix = rel_dir.as_posix() + "/"
|
||||
|
||||
filtered_lines: list[str] = []
|
||||
include_block = False
|
||||
files_changed = 0
|
||||
for line in result.stdout.splitlines():
|
||||
if line.startswith("diff --git"):
|
||||
include_block = rel_prefix in line
|
||||
if include_block:
|
||||
files_changed += 1
|
||||
if include_block:
|
||||
filtered_lines.append(line)
|
||||
|
||||
filtered_diff = "\n".join(filtered_lines)
|
||||
message = f"{files_changed} file(s) changed under {rel_dir}"
|
||||
|
||||
json_handler.log_operation(
|
||||
"get_branch_diff",
|
||||
{"branch_dir": str(branch_dir), "files_changed": files_changed, "staged": staged},
|
||||
)
|
||||
logger.info(message)
|
||||
|
||||
return {"diff": filtered_diff, "files_changed": files_changed, "message": message}
|
||||
@@ -0,0 +1,47 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: log_handler.py
|
||||
# Description: Git log handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-12
|
||||
# Modified: 2026-05-12
|
||||
# =============================================
|
||||
|
||||
"""Git log handler."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
|
||||
|
||||
|
||||
def get_git_log(count: int = 10) -> dict:
|
||||
"""Get recent git log entries."""
|
||||
repo_root = find_repo_root()
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "log", "--oneline", f"-{count}"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
logger.error("git log failed: %s", exc)
|
||||
return {"entries": [], "count": 0, "message": f"git log failed: {exc}"}
|
||||
|
||||
if result.returncode != 0:
|
||||
return {
|
||||
"entries": [],
|
||||
"count": 0,
|
||||
"message": f"git log error: {result.stderr.strip()}",
|
||||
}
|
||||
|
||||
entries = [line for line in result.stdout.splitlines() if line.strip()]
|
||||
|
||||
json_handler.log_operation("get_git_log", {"count": len(entries)})
|
||||
logger.info("Retrieved %d log entries", len(entries))
|
||||
|
||||
return {"entries": entries, "count": len(entries), "message": f"{len(entries)} log entries"}
|
||||
@@ -28,6 +28,7 @@ from aipass.drone.apps.handlers.git.lock_handler import (
|
||||
find_repo_root,
|
||||
release_lock,
|
||||
)
|
||||
from aipass.drone.apps.handlers.git.commit_handler import stage_branch_dir
|
||||
|
||||
|
||||
def _has_credential_helper() -> bool:
|
||||
@@ -166,24 +167,12 @@ def create_pr(branch_name: str, description: str, branch_dir: Path) -> dict:
|
||||
lock_acquired = True
|
||||
|
||||
# Step 3: Stage only files under branch_dir (on main)
|
||||
try:
|
||||
rel_dir = branch_dir.resolve().relative_to(repo_root.resolve())
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"create_pr: branch_dir %s not relative to repo root %s, using absolute", branch_dir, repo_root
|
||||
)
|
||||
rel_dir = branch_dir
|
||||
|
||||
add_result = subprocess.run(
|
||||
["git", "add", str(rel_dir) + "/"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
if add_result.returncode != 0:
|
||||
result["message"] = f"Failed to stage files: {add_result.stderr.strip()}"
|
||||
stage_result = stage_branch_dir(branch_dir, repo_root)
|
||||
if not stage_result["success"]:
|
||||
result["message"] = stage_result["message"]
|
||||
logger.error(result["message"])
|
||||
return result
|
||||
rel_dir = stage_result["rel_dir"]
|
||||
|
||||
# Step 4: Check if anything was staged
|
||||
diff_check = subprocess.run(
|
||||
|
||||
@@ -20,15 +20,37 @@ from pathlib import Path
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.drone.apps.handlers.git import lock_handler, status_handler, sync_handler, pr_handler
|
||||
from aipass.drone.apps.handlers.git import (
|
||||
lock_handler,
|
||||
status_handler,
|
||||
sync_handler,
|
||||
diff_handler,
|
||||
log_handler,
|
||||
commit_handler,
|
||||
checkout_handler,
|
||||
)
|
||||
|
||||
DRONE_MODULE = {
|
||||
"name": "git",
|
||||
"version": "1.0.0",
|
||||
"description": "Git workflow — PR, status, sync, lock management",
|
||||
"version": "2.0.0",
|
||||
"description": "Git workflow — tier-based access, status, diff, log, commit, checkout, sync, lock",
|
||||
}
|
||||
|
||||
_COMMANDS = ("pr", "status", "sync", "lock", "unlock", "system-pr", "merge", "smart-sync", "fix")
|
||||
_COMMANDS = (
|
||||
"status",
|
||||
"diff",
|
||||
"log",
|
||||
"lock",
|
||||
"commit",
|
||||
"checkout",
|
||||
"sync",
|
||||
"unlock",
|
||||
"system-pr",
|
||||
"merge",
|
||||
"smart-sync",
|
||||
"fix",
|
||||
"pr",
|
||||
)
|
||||
|
||||
|
||||
def _detect_branch_dir() -> tuple[str, Path] | None:
|
||||
@@ -63,42 +85,64 @@ def _detect_branch_dir() -> tuple[str, Path] | None:
|
||||
def handle_command(command: str | None = None, args: list[str] | None = None) -> dict:
|
||||
"""Route a git command to the appropriate handler.
|
||||
|
||||
Auth is centralized: verify_git_access() is called once at the top,
|
||||
before any routing. Global-tier commands pass for all callers;
|
||||
owner-tier commands require devpulse.
|
||||
|
||||
Args:
|
||||
command: The subcommand (pr, status, sync, lock, unlock).
|
||||
command: The subcommand (status, diff, log, commit, checkout, etc.).
|
||||
args: Optional list of arguments.
|
||||
|
||||
Returns:
|
||||
Dict with stdout, stderr, and exit_code.
|
||||
"""
|
||||
if not args:
|
||||
if command is None:
|
||||
print_introspection()
|
||||
return {"stdout": "", "stderr": "", "exit_code": 0}
|
||||
args = []
|
||||
if command in ("--help", "-h") or (args and args[0] in ("--help", "-h")):
|
||||
print_help()
|
||||
return {"stdout": "", "stderr": "", "exit_code": 0}
|
||||
|
||||
json_handler.log_operation("git_handle_command", {"command": command, "args": args})
|
||||
if command is None:
|
||||
print_introspection()
|
||||
return {"stdout": "", "stderr": "", "exit_code": 0}
|
||||
|
||||
cmd: str = command
|
||||
try:
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_git_access
|
||||
|
||||
caller = verify_git_access(cmd)
|
||||
except PermissionError as exc:
|
||||
logger.error("git access denied: %s", exc)
|
||||
return {"stdout": "", "stderr": str(exc), "exit_code": 1}
|
||||
|
||||
json_handler.log_operation("git_handle_command", {"command": command, "args": args, "caller": caller})
|
||||
|
||||
if command == "system-pr":
|
||||
return _handle_system_pr(args)
|
||||
if command == "merge":
|
||||
return _handle_merge(args)
|
||||
if command == "smart-sync":
|
||||
return _handle_smart_sync(args)
|
||||
if command == "fix":
|
||||
return _handle_fix(args)
|
||||
if command == "pr":
|
||||
return _handle_pr(args)
|
||||
if command == "status":
|
||||
return _handle_status()
|
||||
if command == "sync":
|
||||
return _handle_sync(args)
|
||||
if command == "diff":
|
||||
return _handle_diff(args)
|
||||
if command == "log":
|
||||
return _handle_log(args)
|
||||
if command == "lock":
|
||||
return _handle_lock()
|
||||
if command == "commit":
|
||||
return _handle_commit(args)
|
||||
if command == "checkout":
|
||||
return _handle_checkout(args)
|
||||
if command == "sync":
|
||||
return _handle_sync(args)
|
||||
if command == "unlock":
|
||||
return _handle_unlock(args)
|
||||
if command == "system-pr":
|
||||
return _handle_system_pr(args, caller)
|
||||
if command == "merge":
|
||||
return _handle_merge(args, caller)
|
||||
if command == "smart-sync":
|
||||
return _handle_smart_sync(caller)
|
||||
if command == "fix":
|
||||
return _handle_fix(args, caller)
|
||||
if command == "pr":
|
||||
return {"stdout": "", "stderr": "Agent PRs are deprecated.", "exit_code": 1}
|
||||
|
||||
available = ", ".join(_COMMANDS)
|
||||
return {
|
||||
@@ -108,8 +152,8 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
|
||||
}
|
||||
|
||||
|
||||
def _handle_system_pr(args: list[str]) -> dict:
|
||||
"""Handle the system-pr subcommand (devpulse-only)."""
|
||||
def _handle_system_pr(args: list[str], caller: str) -> dict:
|
||||
"""Handle the system-pr subcommand (owner-tier, auth pre-checked)."""
|
||||
if not args:
|
||||
return {
|
||||
"stdout": "",
|
||||
@@ -120,7 +164,6 @@ def _handle_system_pr(args: list[str]) -> dict:
|
||||
description = " ".join(args)
|
||||
|
||||
try:
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
|
||||
from aipass.drone.apps.plugins.devpulse_ops.pr_plugin import create_system_pr
|
||||
except ImportError as exc:
|
||||
logger.error("Failed to import devpulse_ops plugin: %s", exc)
|
||||
@@ -130,16 +173,6 @@ def _handle_system_pr(args: list[str]) -> dict:
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
caller = verify_caller()
|
||||
except PermissionError as exc:
|
||||
logger.error("system-pr authorization failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": str(exc),
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
result = create_system_pr(description, caller)
|
||||
|
||||
if result["success"]:
|
||||
@@ -155,8 +188,8 @@ def _handle_system_pr(args: list[str]) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _handle_merge(args: list[str]) -> dict:
|
||||
"""Handle the merge subcommand (devpulse-only)."""
|
||||
def _handle_merge(args: list[str], caller: str) -> dict:
|
||||
"""Handle the merge subcommand (owner-tier, auth pre-checked)."""
|
||||
if not args:
|
||||
return {
|
||||
"stdout": "",
|
||||
@@ -167,7 +200,6 @@ def _handle_merge(args: list[str]) -> dict:
|
||||
pr_number = args[0]
|
||||
|
||||
try:
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
|
||||
from aipass.drone.apps.plugins.devpulse_ops.merge_plugin import merge_pr
|
||||
except ImportError as exc:
|
||||
logger.error("Failed to import devpulse_ops merge plugin: %s", exc)
|
||||
@@ -177,16 +209,6 @@ def _handle_merge(args: list[str]) -> dict:
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
caller = verify_caller()
|
||||
except PermissionError as exc:
|
||||
logger.error("merge authorization failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": str(exc),
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
result = merge_pr(pr_number, caller)
|
||||
|
||||
if result["success"]:
|
||||
@@ -202,10 +224,9 @@ def _handle_merge(args: list[str]) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _handle_smart_sync(args: list[str]) -> dict:
|
||||
"""Handle the smart-sync subcommand (devpulse-only)."""
|
||||
def _handle_smart_sync(caller: str) -> dict:
|
||||
"""Handle the smart-sync subcommand (owner-tier, auth pre-checked)."""
|
||||
try:
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
|
||||
from aipass.drone.apps.plugins.devpulse_ops.sync_plugin import smart_sync
|
||||
except ImportError as exc:
|
||||
logger.error("Failed to import devpulse_ops sync plugin: %s", exc)
|
||||
@@ -215,16 +236,6 @@ def _handle_smart_sync(args: list[str]) -> dict:
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
caller = verify_caller()
|
||||
except PermissionError as exc:
|
||||
logger.error("smart-sync authorization failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": str(exc),
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
result = smart_sync(caller)
|
||||
|
||||
if result["success"]:
|
||||
@@ -240,10 +251,9 @@ def _handle_smart_sync(args: list[str]) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _handle_fix(args: list[str]) -> dict:
|
||||
"""Handle the fix subcommand (devpulse-only)."""
|
||||
def _handle_fix(args: list[str], caller: str) -> dict:
|
||||
"""Handle the fix subcommand (owner-tier, auth pre-checked)."""
|
||||
try:
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import verify_caller
|
||||
from aipass.drone.apps.plugins.devpulse_ops.fix_plugin import fix_git_state
|
||||
except ImportError as exc:
|
||||
logger.error("Failed to import devpulse_ops fix plugin: %s", exc)
|
||||
@@ -253,16 +263,6 @@ def _handle_fix(args: list[str]) -> dict:
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
caller = verify_caller()
|
||||
except PermissionError as exc:
|
||||
logger.error("fix authorization failed: %s", exc)
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": str(exc),
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
dry_run = "--dry-run" in (args or [])
|
||||
result = fix_git_state(caller, dry_run=dry_run)
|
||||
|
||||
@@ -279,42 +279,8 @@ def _handle_fix(args: list[str]) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _handle_pr(args: list[str]) -> dict:
|
||||
"""Handle the PR subcommand."""
|
||||
if not args:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Usage: drone @git pr <description>",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
description = " ".join(args)
|
||||
detected = _detect_branch_dir()
|
||||
if detected is None:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Cannot detect branch directory from CWD. Run from within src/aipass/<branch>/",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
branch_name, branch_dir = detected
|
||||
result = pr_handler.create_pr(branch_name, description, branch_dir)
|
||||
|
||||
if result["success"]:
|
||||
return {
|
||||
"stdout": f"PR created: {result['pr_url']}\nBranch: {result['feature_branch']}",
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
}
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": result["message"],
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
|
||||
def _handle_status() -> dict:
|
||||
"""Handle the status subcommand."""
|
||||
"""Handle the status subcommand (global tier)."""
|
||||
detected = _detect_branch_dir()
|
||||
if detected is None:
|
||||
return {
|
||||
@@ -337,8 +303,96 @@ def _handle_status() -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _handle_diff(args: list[str]) -> dict:
|
||||
"""Handle the diff subcommand (global tier)."""
|
||||
detected = _detect_branch_dir()
|
||||
if detected is None:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Cannot detect branch directory from CWD. Run from within src/aipass/<branch>/",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
_, branch_dir = detected
|
||||
staged = "--staged" in args
|
||||
result = diff_handler.get_branch_diff(branch_dir, staged=staged)
|
||||
|
||||
return {
|
||||
"stdout": result["diff"] if result["diff"] else result["message"],
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
}
|
||||
|
||||
|
||||
def _handle_log(args: list[str]) -> dict:
|
||||
"""Handle the log subcommand (global tier)."""
|
||||
count = 10
|
||||
for arg in args:
|
||||
try:
|
||||
count = int(arg)
|
||||
break
|
||||
except ValueError as exc:
|
||||
logger.warning("Invalid log count argument '%s': %s", arg, exc)
|
||||
continue
|
||||
|
||||
result = log_handler.get_git_log(count=count)
|
||||
|
||||
if result["entries"]:
|
||||
return {
|
||||
"stdout": "\n".join(result["entries"]),
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
}
|
||||
return {
|
||||
"stdout": result["message"],
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
}
|
||||
|
||||
|
||||
def _handle_commit(args: list[str]) -> dict:
|
||||
"""Handle the commit subcommand (owner tier)."""
|
||||
if not args:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Usage: drone @git commit <message> [--all]",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
all_files = "--all" in args
|
||||
msg_parts = [a for a in args if a != "--all"]
|
||||
message = " ".join(msg_parts)
|
||||
|
||||
if not message:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Commit message cannot be empty",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
branch_dir = None
|
||||
if all_files:
|
||||
detected = _detect_branch_dir()
|
||||
if detected:
|
||||
_, branch_dir = detected
|
||||
|
||||
return commit_handler.commit_changes(message, branch_dir=branch_dir, all_files=all_files)
|
||||
|
||||
|
||||
def _handle_checkout(args: list[str]) -> dict:
|
||||
"""Handle the checkout subcommand (owner tier)."""
|
||||
if not args:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "Usage: drone @git checkout <main|dev>",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
return checkout_handler.checkout_branch(args[0])
|
||||
|
||||
|
||||
def _handle_sync(args: list[str]) -> dict:
|
||||
"""Handle the sync subcommand."""
|
||||
"""Handle the sync subcommand (owner tier)."""
|
||||
autostash = "--autostash" in args
|
||||
result = sync_handler.sync_main(autostash=autostash)
|
||||
|
||||
@@ -356,7 +410,7 @@ def _handle_sync(args: list[str]) -> dict:
|
||||
|
||||
|
||||
def _handle_lock() -> dict:
|
||||
"""Handle the lock subcommand (check status)."""
|
||||
"""Handle the lock subcommand (global tier)."""
|
||||
result = lock_handler.check_lock_status()
|
||||
return {
|
||||
"stdout": json.dumps(result, indent=2),
|
||||
@@ -366,7 +420,7 @@ def _handle_lock() -> dict:
|
||||
|
||||
|
||||
def _handle_unlock(args: list[str]) -> dict:
|
||||
"""Handle the unlock subcommand (force only)."""
|
||||
"""Handle the unlock subcommand (owner tier)."""
|
||||
if "--force" not in args:
|
||||
return {
|
||||
"stdout": "",
|
||||
@@ -398,100 +452,109 @@ def get_help(command: str | None = None) -> str:
|
||||
Help text string.
|
||||
"""
|
||||
if command == "pr":
|
||||
return (
|
||||
"git pr <description> — Create a PR with scoped changes\n"
|
||||
" Stages only files under the caller's branch directory,\n"
|
||||
" creates a feature branch, commits, pushes, and opens a PR.\n"
|
||||
)
|
||||
return "git pr — DEPRECATED. Agent PRs are no longer supported. Devpulse handles git.\n"
|
||||
if command == "status":
|
||||
return "git status — Show git status filtered to your branch directory\n"
|
||||
return "git status — Show git status filtered to your branch directory [global]\n"
|
||||
if command == "diff":
|
||||
return (
|
||||
"git diff [--staged] — Show git diff filtered to your branch directory [global]\n"
|
||||
" Options:\n"
|
||||
" --staged Show staged changes only.\n"
|
||||
)
|
||||
if command == "log":
|
||||
return "git log [count] — Show recent git log entries (default: 10) [global]\n"
|
||||
if command == "lock":
|
||||
return "git lock — Check current lock status [global]\n Shows lock holder, age, stale/orphan detection.\n"
|
||||
if command == "commit":
|
||||
return (
|
||||
"git commit <message> [--all] — Commit staged changes [owner]\n"
|
||||
" Options:\n"
|
||||
" --all Stage all changes under your branch directory first.\n"
|
||||
)
|
||||
if command == "checkout":
|
||||
return "git checkout <main|dev> — Switch branches (main or dev only) [owner]\n"
|
||||
if command == "sync":
|
||||
return (
|
||||
"git sync [--autostash] — Checkout main and pull latest changes\n"
|
||||
"git sync [--autostash] — Checkout main and pull latest changes [owner]\n"
|
||||
" Options:\n"
|
||||
" --autostash Stash local changes before pull and restore after.\n"
|
||||
" Use when sync fails with 'unstaged changes' error.\n"
|
||||
)
|
||||
if command == "lock":
|
||||
return "git lock — Check current lock status\n Shows lock holder, age, stale/orphan detection.\n"
|
||||
if command == "unlock":
|
||||
return "git unlock --force — Force-release the PR lock\n Removes .git_pr.lock regardless of holder.\n"
|
||||
return "git unlock --force — Force-release the PR lock [owner]\n"
|
||||
if command == "system-pr":
|
||||
return (
|
||||
"git system-pr <description> — Create a system-wide PR (devpulse only)\n"
|
||||
" Stages all tracked changes, creates a disposable feature branch,\n"
|
||||
" and opens a PR. Requires devpulse passport authorization.\n"
|
||||
"git system-pr <description> — Create a system-wide PR [owner]\n"
|
||||
" Stages all tracked changes, creates a feature branch, and opens a PR.\n"
|
||||
)
|
||||
if command == "merge":
|
||||
return (
|
||||
"git merge <PR#> — Merge a PR and sync local main (devpulse only)\n"
|
||||
"git merge <PR#> — Merge a PR and sync local main [owner]\n"
|
||||
" Runs gh pr merge --merge --delete-branch, then git pull --rebase.\n"
|
||||
)
|
||||
if command == "smart-sync":
|
||||
return (
|
||||
"git smart-sync — Fetch origin and rebase if behind (devpulse only)\n"
|
||||
" Detects divergence and rebases safely; aborts on conflict.\n"
|
||||
)
|
||||
return "git smart-sync — Fetch origin and rebase if behind [owner]\n"
|
||||
if command == "fix":
|
||||
return (
|
||||
"git fix [--dry-run] — Detect and fix common broken git states (devpulse only)\n"
|
||||
"git fix [--dry-run] — Detect and fix broken git states [owner]\n"
|
||||
"\n"
|
||||
"Detected states and actions:\n"
|
||||
" Stuck rebase .git/rebase-merge or rebase-apply exists\n"
|
||||
" → git rebase --abort\n"
|
||||
" Detached HEAD HEAD is not on a named branch\n"
|
||||
" → git checkout main\n"
|
||||
" Diverged local main and origin/main have diverged\n"
|
||||
" → git fetch + git merge origin/main --no-edit\n"
|
||||
" Dirty index files are staged but not committed\n"
|
||||
" → git reset HEAD (unstages all)\n"
|
||||
" Stuck rebase → git rebase --abort\n"
|
||||
" Detached HEAD → git checkout main\n"
|
||||
" Diverged → git fetch + git merge origin/main\n"
|
||||
" Dirty index → git reset HEAD\n"
|
||||
"\n"
|
||||
"Options:\n"
|
||||
" --dry-run Report detected states and proposed actions without\n"
|
||||
" executing any fixes. Safe to run anytime.\n"
|
||||
" --dry-run Report without executing fixes.\n"
|
||||
)
|
||||
|
||||
return (
|
||||
"git — Git workflow: PR, status, sync, lock management\n"
|
||||
"git — Tier-based git workflow\n"
|
||||
"\n"
|
||||
"Commands:\n"
|
||||
" pr <description> Create a PR with scoped changes\n"
|
||||
" system-pr <desc> Create a system-wide PR (devpulse only)\n"
|
||||
" merge <PR#> Merge a PR (devpulse only)\n"
|
||||
" smart-sync Fetch + rebase if behind (devpulse only)\n"
|
||||
" fix Fix broken git states (devpulse only)\n"
|
||||
"Global (all branches):\n"
|
||||
" status Show git status for your branch\n"
|
||||
" sync Checkout main and pull\n"
|
||||
" diff [--staged] Show git diff for your branch\n"
|
||||
" log [count] Show recent git log (default: 10)\n"
|
||||
" lock Check lock status\n"
|
||||
" unlock --force Force-release the PR lock\n"
|
||||
"\n"
|
||||
"Policy: raw git commands are blocked for agents via .claude/settings.json:\n"
|
||||
" - git checkout* (any form) — use `drone @git sync` instead\n"
|
||||
" - git add -f* / --force* — use scoped `drone @git pr` instead\n"
|
||||
"Culturally also avoid: raw git commit/push, gh pr create. Go through drone.\n"
|
||||
"Owner (devpulse only):\n"
|
||||
" commit <msg> [--all] Commit staged changes\n"
|
||||
" checkout <main|dev> Switch branches\n"
|
||||
" sync [--autostash] Checkout main and pull\n"
|
||||
" unlock --force Force-release the PR lock\n"
|
||||
" system-pr <desc> Create a system-wide PR\n"
|
||||
" merge <PR#> Merge a PR\n"
|
||||
" smart-sync Fetch + rebase if behind\n"
|
||||
" fix [--dry-run] Fix broken git states\n"
|
||||
"\n"
|
||||
"Deprecated:\n"
|
||||
" pr Agent PRs removed — devpulse handles git\n"
|
||||
)
|
||||
|
||||
|
||||
def get_introspective() -> str:
|
||||
"""Return introspection text showing connected handlers."""
|
||||
return (
|
||||
"@git — Git workflow: PR, status, sync, lock management\n"
|
||||
"@git — Tier-based git workflow (v2.0.0)\n"
|
||||
"\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, stage_branch_dir)\n"
|
||||
" - checkout_handler.py (checkout_branch — main/dev only)\n"
|
||||
" - sync_handler.py (sync_main — safe main synchronization)\n"
|
||||
" - pr_handler.py (create_pr — full PR workflow with lockfile)\n"
|
||||
" - pr_handler.py (create_pr — DEPRECATED)\n"
|
||||
"\n"
|
||||
" plugins/devpulse_ops/\n"
|
||||
" - auth.py (verify_caller — passport-based authorization)\n"
|
||||
" - auth.py (verify_git_access — tier-based authorization)\n"
|
||||
" - pr_plugin.py (create_system_pr — system-wide PR workflow)\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"
|
||||
"Commands: pr, system-pr, merge, smart-sync, fix, status, sync, lock, unlock\n"
|
||||
"Access Tiers: global (status, diff, log, lock) | owner (commit, checkout, sync, unlock, system-pr, merge, smart-sync, fix)\n"
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -24,6 +24,27 @@ from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
|
||||
|
||||
ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
|
||||
|
||||
GIT_ACCESS_TIERS: dict[str, dict] = {
|
||||
"global": {
|
||||
"commands": ["status", "diff", "log", "lock"],
|
||||
"description": "Read-only — available to all branches",
|
||||
},
|
||||
"owner": {
|
||||
"commands": [
|
||||
"commit",
|
||||
"checkout",
|
||||
"sync",
|
||||
"unlock",
|
||||
"system-pr",
|
||||
"merge",
|
||||
"smart-sync",
|
||||
"fix",
|
||||
],
|
||||
"allowed_callers": ["devpulse"],
|
||||
"description": "Write operations — project owner only",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _find_caller() -> str:
|
||||
"""Walk up from CWD to find passport.json and return branch name.
|
||||
@@ -83,3 +104,46 @@ def verify_caller() -> str:
|
||||
)
|
||||
logger.info("Caller '%s' authorized for devpulse operations", name)
|
||||
return name
|
||||
|
||||
|
||||
def verify_git_access(command: str) -> str:
|
||||
"""Check if the calling branch is authorized for this git command.
|
||||
|
||||
Uses GIT_ACCESS_TIERS to determine access level. Global-tier commands
|
||||
are available to all branches; owner-tier commands require the caller
|
||||
to be in the allowed_callers list.
|
||||
|
||||
Returns:
|
||||
The caller's branch name if authorized.
|
||||
|
||||
Raises:
|
||||
PermissionError: If the caller is not authorized for this command.
|
||||
"""
|
||||
if command == "pr":
|
||||
raise PermissionError("Agent PRs are deprecated. Build code, run tests, report results. Devpulse handles git.")
|
||||
|
||||
global_cmds = GIT_ACCESS_TIERS["global"]["commands"]
|
||||
owner_tier = GIT_ACCESS_TIERS["owner"]
|
||||
|
||||
if command in global_cmds:
|
||||
caller = _find_caller()
|
||||
json_handler.log_operation(
|
||||
"git_access_verify",
|
||||
{"caller": caller, "command": command, "tier": "global"},
|
||||
)
|
||||
return caller
|
||||
|
||||
if command in owner_tier["commands"]:
|
||||
caller = _find_caller()
|
||||
allowed = owner_tier["allowed_callers"]
|
||||
if caller not in allowed:
|
||||
msg = f"Branch '{caller}' is not authorized for '{command}'. Only {allowed} can use owner-tier commands."
|
||||
logger.error(msg)
|
||||
raise PermissionError(msg)
|
||||
json_handler.log_operation(
|
||||
"git_access_verify",
|
||||
{"caller": caller, "command": command, "tier": "owner"},
|
||||
)
|
||||
return caller
|
||||
|
||||
raise PermissionError(f"Unknown git command: '{command}'")
|
||||
|
||||
@@ -518,26 +518,32 @@ class TestGitModuleRouting:
|
||||
assert "sync_plugin" in intro
|
||||
assert "fix_plugin" in intro
|
||||
|
||||
def test_handle_merge_no_args(self) -> None:
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
return_value="devpulse",
|
||||
)
|
||||
def test_handle_merge_no_args(self, _mock_access: MagicMock) -> None:
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
result = handle_command("merge", [])
|
||||
assert result["exit_code"] == 1
|
||||
assert "Usage" in result["stderr"]
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
return_value="devpulse",
|
||||
)
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.merge_plugin.find_repo_root")
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.merge_plugin.subprocess.run")
|
||||
def test_handle_merge_routes_correctly(
|
||||
self,
|
||||
mock_run: MagicMock,
|
||||
mock_root: MagicMock,
|
||||
mock_verify: MagicMock,
|
||||
_mock_access: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
mock_verify.return_value = "devpulse"
|
||||
mock_root.return_value = tmp_path
|
||||
|
||||
proc = MagicMock()
|
||||
@@ -549,19 +555,21 @@ class TestGitModuleRouting:
|
||||
result = handle_command("merge", ["42"])
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
return_value="devpulse",
|
||||
)
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.sync_plugin.find_repo_root")
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.sync_plugin.subprocess.run")
|
||||
def test_handle_smart_sync_routes_correctly(
|
||||
self,
|
||||
mock_run: MagicMock,
|
||||
mock_root: MagicMock,
|
||||
mock_verify: MagicMock,
|
||||
_mock_access: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
mock_verify.return_value = "devpulse"
|
||||
mock_root.return_value = tmp_path
|
||||
|
||||
def side_effect(cmd: list[str], **kwargs: object) -> MagicMock:
|
||||
@@ -579,19 +587,21 @@ class TestGitModuleRouting:
|
||||
result = handle_command("smart-sync", [])
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
return_value="devpulse",
|
||||
)
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.fix_plugin.find_repo_root")
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.fix_plugin.subprocess.run")
|
||||
def test_handle_fix_routes_correctly(
|
||||
self,
|
||||
mock_run: MagicMock,
|
||||
mock_root: MagicMock,
|
||||
mock_verify: MagicMock,
|
||||
_mock_access: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
mock_verify.return_value = "devpulse"
|
||||
mock_root.return_value = tmp_path
|
||||
git_dir = tmp_path / ".git"
|
||||
git_dir.mkdir()
|
||||
|
||||
@@ -0,0 +1,577 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: test_git_access.py
|
||||
# Description: Tests for tier-based git access, new handlers, and PR deprecation
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-12
|
||||
# Modified: 2026-05-12
|
||||
# =============================================
|
||||
|
||||
"""Tests for tier-based git access, new handlers (diff, log, commit, checkout), and PR deprecation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import (
|
||||
GIT_ACCESS_TIERS,
|
||||
verify_git_access,
|
||||
)
|
||||
from aipass.drone.apps.handlers.git.diff_handler import get_branch_diff
|
||||
from aipass.drone.apps.handlers.git.log_handler import get_git_log
|
||||
from aipass.drone.apps.handlers.git.commit_handler import commit_changes, stage_branch_dir
|
||||
from aipass.drone.apps.handlers.git.checkout_handler import checkout_branch
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Fixtures
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def devpulse_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Create a temp directory with a devpulse passport."""
|
||||
trinity = tmp_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
passport = trinity / "passport.json"
|
||||
passport.write_text(
|
||||
json.dumps({"branch_info": {"branch_name": "devpulse"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def seedgo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Create a temp directory with a non-owner passport."""
|
||||
trinity = tmp_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
passport = trinity / "passport.json"
|
||||
passport.write_text(
|
||||
json.dumps({"branch_info": {"branch_name": "seedgo"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.chdir(tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def repo_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Create a temp repo root with AIPASS_REGISTRY.json."""
|
||||
registry = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry.write_text("{}", encoding="utf-8")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 1. GIT_ACCESS_TIERS config structure
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestGitAccessTiers:
|
||||
"""Verify the tier config is correctly structured."""
|
||||
|
||||
def test_tiers_has_global_and_owner(self) -> None:
|
||||
assert "global" in GIT_ACCESS_TIERS
|
||||
assert "owner" in GIT_ACCESS_TIERS
|
||||
|
||||
def test_global_commands(self) -> None:
|
||||
cmds = GIT_ACCESS_TIERS["global"]["commands"]
|
||||
assert "status" in cmds
|
||||
assert "diff" in cmds
|
||||
assert "log" in cmds
|
||||
assert "lock" in cmds
|
||||
|
||||
def test_owner_commands(self) -> None:
|
||||
cmds = GIT_ACCESS_TIERS["owner"]["commands"]
|
||||
assert "commit" in cmds
|
||||
assert "checkout" in cmds
|
||||
assert "sync" in cmds
|
||||
assert "unlock" in cmds
|
||||
assert "system-pr" in cmds
|
||||
assert "merge" in cmds
|
||||
assert "smart-sync" in cmds
|
||||
assert "fix" in cmds
|
||||
|
||||
def test_owner_allowed_callers(self) -> None:
|
||||
allowed = GIT_ACCESS_TIERS["owner"]["allowed_callers"]
|
||||
assert allowed == ["devpulse"]
|
||||
|
||||
def test_pr_not_in_any_tier(self) -> None:
|
||||
all_cmds = GIT_ACCESS_TIERS["global"]["commands"] + GIT_ACCESS_TIERS["owner"]["commands"]
|
||||
assert "pr" not in all_cmds
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 2. verify_git_access — tier enforcement
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestVerifyGitAccessGlobal:
|
||||
"""Global-tier commands should pass for any caller."""
|
||||
|
||||
def test_status_allowed_for_any_branch(self, devpulse_dir: Path) -> None:
|
||||
assert verify_git_access("status") == "devpulse"
|
||||
|
||||
def test_diff_allowed_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
assert verify_git_access("diff") == "seedgo"
|
||||
|
||||
def test_log_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
|
||||
assert verify_git_access("log") == "seedgo"
|
||||
|
||||
def test_lock_allowed_for_any_branch(self, seedgo_dir: Path) -> None:
|
||||
assert verify_git_access("lock") == "seedgo"
|
||||
|
||||
|
||||
class TestVerifyGitAccessOwner:
|
||||
"""Owner-tier commands should only pass for devpulse."""
|
||||
|
||||
def test_commit_allowed_for_devpulse(self, devpulse_dir: Path) -> None:
|
||||
assert verify_git_access("commit") == "devpulse"
|
||||
|
||||
def test_commit_denied_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="not authorized"):
|
||||
verify_git_access("commit")
|
||||
|
||||
def test_checkout_denied_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="not authorized"):
|
||||
verify_git_access("checkout")
|
||||
|
||||
def test_sync_denied_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="not authorized"):
|
||||
verify_git_access("sync")
|
||||
|
||||
def test_unlock_denied_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="not authorized"):
|
||||
verify_git_access("unlock")
|
||||
|
||||
def test_system_pr_denied_for_seedgo(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="not authorized"):
|
||||
verify_git_access("system-pr")
|
||||
|
||||
|
||||
class TestVerifyGitAccessPrDeprecated:
|
||||
"""PR command should be denied with deprecation message."""
|
||||
|
||||
def test_pr_deprecated_for_devpulse(self, devpulse_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="deprecated"):
|
||||
verify_git_access("pr")
|
||||
|
||||
def test_pr_deprecated_for_any_branch(self, seedgo_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="deprecated"):
|
||||
verify_git_access("pr")
|
||||
|
||||
|
||||
class TestVerifyGitAccessUnknown:
|
||||
"""Unknown commands should be denied."""
|
||||
|
||||
def test_unknown_command_denied(self, devpulse_dir: Path) -> None:
|
||||
with pytest.raises(PermissionError, match="Unknown git command"):
|
||||
verify_git_access("nonexistent")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 3. diff_handler
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestDiffHandler:
|
||||
"""Scoped git diff tests."""
|
||||
|
||||
def test_basic_diff(self, repo_dir: Path) -> None:
|
||||
diff_output = (
|
||||
"diff --git a/src/aipass/api/foo.py b/src/aipass/api/foo.py\n"
|
||||
"--- a/src/aipass/api/foo.py\n"
|
||||
"+++ b/src/aipass/api/foo.py\n"
|
||||
"@@ -1,3 +1,4 @@\n"
|
||||
"+new line\n"
|
||||
"diff --git a/src/aipass/drone/bar.py b/src/aipass/drone/bar.py\n"
|
||||
"--- a/src/aipass/drone/bar.py\n"
|
||||
"+++ b/src/aipass/drone/bar.py\n"
|
||||
)
|
||||
mock_result = MagicMock(returncode=0, stdout=diff_output, stderr="")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
|
||||
result = get_branch_diff(branch_dir)
|
||||
|
||||
assert result["files_changed"] == 1
|
||||
assert "src/aipass/api/foo.py" in result["diff"]
|
||||
assert "src/aipass/drone/bar.py" not in result["diff"]
|
||||
|
||||
def test_staged_diff(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result) as mock_run:
|
||||
get_branch_diff(branch_dir, staged=True)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert "--staged" in cmd
|
||||
|
||||
def test_empty_diff(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
|
||||
result = get_branch_diff(branch_dir)
|
||||
|
||||
assert result["files_changed"] == 0
|
||||
assert result["diff"] == ""
|
||||
|
||||
def test_git_failure(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=128, stderr="fatal: not a git repo", stdout="")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
|
||||
result = get_branch_diff(branch_dir)
|
||||
|
||||
assert result["files_changed"] == 0
|
||||
assert "error" in result["message"].lower()
|
||||
|
||||
def test_os_error(self, repo_dir: Path) -> None:
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", side_effect=OSError("git not found")):
|
||||
result = get_branch_diff(branch_dir)
|
||||
|
||||
assert result["files_changed"] == 0
|
||||
assert "failed" in result["message"].lower()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 4. log_handler
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestLogHandler:
|
||||
"""Git log tests."""
|
||||
|
||||
def test_basic_log(self, repo_dir: Path) -> None:
|
||||
log_output = "abc1234 feat: first\ndef5678 fix: second\n"
|
||||
mock_result = MagicMock(returncode=0, stdout=log_output, stderr="")
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
|
||||
result = get_git_log(count=5)
|
||||
|
||||
assert result["count"] == 2
|
||||
assert len(result["entries"]) == 2
|
||||
|
||||
def test_custom_count_passed_to_git(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result) as mock_run:
|
||||
get_git_log(count=25)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert "-25" in cmd
|
||||
|
||||
def test_git_failure(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=128, stderr="fatal: bad default", stdout="")
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
|
||||
result = get_git_log()
|
||||
|
||||
assert result["count"] == 0
|
||||
assert "error" in result["message"].lower()
|
||||
|
||||
def test_os_error(self, repo_dir: Path) -> None:
|
||||
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", side_effect=OSError("git not found")):
|
||||
result = get_git_log()
|
||||
|
||||
assert result["count"] == 0
|
||||
assert "failed" in result["message"].lower()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 5. commit_handler
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestStageBranchDir:
|
||||
"""Shared staging utility tests."""
|
||||
|
||||
def test_stage_success(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=0, stderr="")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_result):
|
||||
result = stage_branch_dir(branch_dir, repo_dir)
|
||||
|
||||
assert result["success"] is True
|
||||
|
||||
def test_stage_failure(self, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=1, stderr="fatal: pathspec error")
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_result):
|
||||
result = stage_branch_dir(branch_dir, repo_dir)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "failed" in result["message"].lower()
|
||||
|
||||
|
||||
class TestCommitChanges:
|
||||
"""Commit handler tests."""
|
||||
|
||||
def test_commit_staged(self, repo_dir: Path) -> None:
|
||||
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
|
||||
mock_commit = MagicMock(returncode=0, stdout="[main abc123] test commit", stderr="")
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
|
||||
side_effect=[mock_diff, mock_commit],
|
||||
):
|
||||
result = commit_changes("test commit")
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert "abc123" in result["stdout"]
|
||||
|
||||
def test_commit_nothing_staged(self, repo_dir: Path) -> None:
|
||||
mock_diff = MagicMock(returncode=0, stdout="", stderr="")
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.commit_handler.subprocess.run", return_value=mock_diff):
|
||||
result = commit_changes("test commit")
|
||||
|
||||
assert result["exit_code"] == 1
|
||||
assert "nothing to commit" in result["stderr"].lower()
|
||||
|
||||
def test_commit_all_stages_first(self, repo_dir: Path) -> None:
|
||||
mock_add = MagicMock(returncode=0, stderr="")
|
||||
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
|
||||
mock_commit = MagicMock(returncode=0, stdout="[main def456] all commit", stderr="")
|
||||
|
||||
branch_dir = repo_dir / "src" / "aipass" / "api"
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
|
||||
side_effect=[mock_add, mock_diff, mock_commit],
|
||||
):
|
||||
result = commit_changes("all commit", branch_dir=branch_dir, all_files=True)
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
def test_commit_os_error(self, repo_dir: Path) -> None:
|
||||
mock_diff = MagicMock(returncode=1, stdout="", stderr="")
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.commit_handler.subprocess.run",
|
||||
side_effect=[mock_diff, OSError("git not found")],
|
||||
):
|
||||
result = commit_changes("test commit")
|
||||
|
||||
assert result["exit_code"] == 1
|
||||
assert "failed" in result["stderr"].lower()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 6. checkout_handler
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestCheckoutHandler:
|
||||
"""Branch checkout with hard guard tests."""
|
||||
|
||||
def test_checkout_main_allowed(self, repo_dir: Path) -> None:
|
||||
mock_status = MagicMock(returncode=0, stdout="", stderr="")
|
||||
mock_checkout = MagicMock(returncode=0, stdout="", stderr="Switched to branch 'main'")
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
|
||||
side_effect=[mock_status, mock_checkout],
|
||||
):
|
||||
result = checkout_branch("main")
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert result["current_branch"] == "main"
|
||||
|
||||
def test_checkout_dev_allowed(self, repo_dir: Path) -> None:
|
||||
mock_status = MagicMock(returncode=0, stdout="", stderr="")
|
||||
mock_checkout = MagicMock(returncode=0, stdout="", stderr="Switched to branch 'dev'")
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
|
||||
side_effect=[mock_status, mock_checkout],
|
||||
):
|
||||
result = checkout_branch("dev")
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert result["current_branch"] == "dev"
|
||||
|
||||
def test_checkout_feature_branch_denied(self) -> None:
|
||||
result = checkout_branch("feat/my-feature")
|
||||
assert result["exit_code"] == 1
|
||||
assert "denied" in result["stderr"].lower()
|
||||
assert result["current_branch"] == ""
|
||||
|
||||
def test_checkout_arbitrary_branch_denied(self) -> None:
|
||||
result = checkout_branch("release/v2")
|
||||
assert result["exit_code"] == 1
|
||||
assert "denied" in result["stderr"].lower()
|
||||
|
||||
def test_checkout_dirty_tree_aborts(self, repo_dir: Path) -> None:
|
||||
mock_status = MagicMock(returncode=0, stdout=" M some/file.py\n", stderr="")
|
||||
|
||||
with patch("aipass.drone.apps.handlers.git.checkout_handler.subprocess.run", return_value=mock_status):
|
||||
result = checkout_branch("main")
|
||||
|
||||
assert result["exit_code"] == 1
|
||||
assert "uncommitted" in result["stderr"].lower()
|
||||
|
||||
def test_checkout_git_failure(self, repo_dir: Path) -> None:
|
||||
mock_status = MagicMock(returncode=0, stdout="", stderr="")
|
||||
mock_checkout = MagicMock(returncode=1, stdout="", stderr="error: pathspec 'main' did not match")
|
||||
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
|
||||
side_effect=[mock_status, mock_checkout],
|
||||
):
|
||||
result = checkout_branch("main")
|
||||
|
||||
assert result["exit_code"] == 1
|
||||
assert result["current_branch"] == ""
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 7. PR deprecation through handle_command
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestPrDeprecation:
|
||||
"""PR command returns deprecation message via centralized auth."""
|
||||
|
||||
def test_pr_returns_deprecation(self, devpulse_dir: Path) -> None:
|
||||
result = handle_command("pr", ["some description"])
|
||||
assert result["exit_code"] == 1
|
||||
assert "deprecated" in result["stderr"].lower()
|
||||
|
||||
def test_pr_no_args_also_deprecated(self, devpulse_dir: Path) -> None:
|
||||
result = handle_command("pr")
|
||||
assert result["exit_code"] == 1
|
||||
assert "deprecated" in result["stderr"].lower()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 8. New commands via handle_command routing
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestNewCommandRouting:
|
||||
"""Verify new commands route through handle_command correctly."""
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
|
||||
def test_diff_routes(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
|
||||
trinity = repo_dir / ".trinity"
|
||||
trinity.mkdir()
|
||||
passport = trinity / "passport.json"
|
||||
passport.write_text(json.dumps({"branch_info": {"branch_name": "test_branch"}}))
|
||||
|
||||
mock_result = MagicMock(returncode=0, stdout="", stderr="")
|
||||
with patch("aipass.drone.apps.handlers.git.diff_handler.subprocess.run", return_value=mock_result):
|
||||
result = handle_command("diff")
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
|
||||
def test_log_routes(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
|
||||
mock_result = MagicMock(returncode=0, stdout="abc123 test\n", stderr="")
|
||||
with patch("aipass.drone.apps.handlers.git.log_handler.subprocess.run", return_value=mock_result):
|
||||
result = handle_command("log")
|
||||
assert result["exit_code"] == 0
|
||||
assert "abc123" in result["stdout"]
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_commit_no_args_error(self, _mock_auth: MagicMock) -> None:
|
||||
result = handle_command("commit")
|
||||
assert result["exit_code"] == 1
|
||||
assert "usage" in result["stderr"].lower()
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_checkout_no_args_error(self, _mock_auth: MagicMock) -> None:
|
||||
result = handle_command("checkout")
|
||||
assert result["exit_code"] == 1
|
||||
assert "usage" in result["stderr"].lower()
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_checkout_routes_to_handler(self, _mock_auth: MagicMock, repo_dir: Path) -> None:
|
||||
mock_status = MagicMock(returncode=0, stdout="", stderr="")
|
||||
mock_checkout = MagicMock(returncode=0, stdout="", stderr="")
|
||||
with patch(
|
||||
"aipass.drone.apps.handlers.git.checkout_handler.subprocess.run",
|
||||
side_effect=[mock_status, mock_checkout],
|
||||
):
|
||||
result = handle_command("checkout", ["main"])
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_checkout_guard_rejects_feature(self, _mock_auth: MagicMock) -> None:
|
||||
result = handle_command("checkout", ["feat/bad"])
|
||||
assert result["exit_code"] == 1
|
||||
assert "denied" in result["stderr"].lower()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 9. Help text includes new commands and tiers
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestUpdatedHelp:
|
||||
"""Help and introspection reflect new commands and tiers."""
|
||||
|
||||
def test_help_includes_diff(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "diff" in text
|
||||
|
||||
def test_help_includes_log(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "log" in text
|
||||
|
||||
def test_help_includes_commit(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "commit" in text
|
||||
|
||||
def test_help_includes_checkout(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "checkout" in text
|
||||
|
||||
def test_help_shows_tier_sections(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "global" in text.lower()
|
||||
assert "owner" in text.lower()
|
||||
|
||||
def test_help_marks_pr_deprecated(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
text = get_help()
|
||||
assert "deprecated" in text.lower()
|
||||
|
||||
def test_introspection_includes_new_handlers(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_introspective
|
||||
|
||||
text = get_introspective()
|
||||
assert "diff_handler" in text
|
||||
assert "log_handler" in text
|
||||
assert "commit_handler" in text
|
||||
assert "checkout_handler" in text
|
||||
|
||||
def test_introspection_shows_tiers(self) -> None:
|
||||
from aipass.drone.apps.modules.git_module import get_introspective
|
||||
|
||||
text = get_introspective()
|
||||
assert "global" in text.lower()
|
||||
assert "owner" in text.lower()
|
||||
@@ -614,9 +614,14 @@ class TestGitModuleRouting:
|
||||
result = handle_command("bogus")
|
||||
assert result["exit_code"] == 1
|
||||
assert "unknown" in result["stderr"].lower()
|
||||
assert "pr" in result["stderr"]
|
||||
|
||||
def test_lock_routes_to_handler(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
return_value="test_branch",
|
||||
)
|
||||
def test_lock_routes_to_handler(
|
||||
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""lock command routes to check_lock_status."""
|
||||
registry = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry.write_text("{}", encoding="utf-8")
|
||||
@@ -627,13 +632,15 @@ class TestGitModuleRouting:
|
||||
data = json.loads(result["stdout"])
|
||||
assert data["locked"] is False
|
||||
|
||||
def test_unlock_requires_force(self) -> None:
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_unlock_requires_force(self, _mock_auth: MagicMock) -> None:
|
||||
"""unlock without --force returns error."""
|
||||
result = handle_command("unlock")
|
||||
assert result["exit_code"] == 1
|
||||
assert "--force" in result["stderr"]
|
||||
|
||||
def test_unlock_with_force(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_unlock_with_force(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""unlock --force routes to force_unlock."""
|
||||
registry = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry.write_text("{}", encoding="utf-8")
|
||||
@@ -642,7 +649,10 @@ class TestGitModuleRouting:
|
||||
result = handle_command("unlock", ["--force"])
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
def test_sync_routes_to_handler(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_sync_routes_to_handler(
|
||||
self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""sync command routes to sync_main."""
|
||||
registry = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry.write_text("{}", encoding="utf-8")
|
||||
@@ -661,7 +671,8 @@ class TestGitModuleRouting:
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
|
||||
def test_status_no_branch_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="test_branch")
|
||||
def test_status_no_branch_dir(self, _mock_auth: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""status outside a branch directory returns error."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
result = handle_command("status")
|
||||
@@ -669,17 +680,16 @@ class TestGitModuleRouting:
|
||||
assert "cannot detect" in result["stderr"].lower()
|
||||
|
||||
def test_pr_no_args(self) -> None:
|
||||
"""pr with no arguments returns usage error."""
|
||||
"""pr command is deprecated."""
|
||||
result = handle_command("pr")
|
||||
assert result["exit_code"] == 1
|
||||
assert "usage" in result["stderr"].lower()
|
||||
assert "deprecated" in result["stderr"].lower()
|
||||
|
||||
def test_pr_no_branch_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""pr outside a branch directory returns error."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
def test_pr_no_branch_dir(self) -> None:
|
||||
"""pr command is deprecated regardless of context."""
|
||||
result = handle_command("pr", ["some description"])
|
||||
assert result["exit_code"] == 1
|
||||
assert "cannot detect" in result["stderr"].lower()
|
||||
assert "deprecated" in result["stderr"].lower()
|
||||
|
||||
|
||||
class TestDetectBranchDir:
|
||||
@@ -756,7 +766,7 @@ class TestGitModuleHelp:
|
||||
"""Command-specific help returns relevant text."""
|
||||
text = get_help("pr")
|
||||
assert "pr" in text.lower()
|
||||
assert "description" in text.lower()
|
||||
assert "deprecated" in text.lower()
|
||||
|
||||
def test_introspective(self) -> None:
|
||||
"""Introspection lists connected handlers."""
|
||||
|
||||
@@ -319,11 +319,11 @@ class TestGitModuleSystemPrRouting:
|
||||
assert "system-pr" in help_text
|
||||
|
||||
def test_get_help_system_pr_specific(self) -> None:
|
||||
"""get_help('system-pr') output mentions devpulse as the authorized caller."""
|
||||
"""get_help('system-pr') output mentions owner as the tier label."""
|
||||
from aipass.drone.apps.modules.git_module import get_help
|
||||
|
||||
help_text = get_help("system-pr")
|
||||
assert "devpulse" in help_text
|
||||
assert "owner" in help_text.lower()
|
||||
|
||||
def test_get_introspective_includes_plugin(self) -> None:
|
||||
"""get_introspective() output mentions the devpulse_ops plugin."""
|
||||
@@ -332,7 +332,7 @@ class TestGitModuleSystemPrRouting:
|
||||
intro = get_introspective()
|
||||
assert "devpulse_ops" in intro
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access", return_value="devpulse")
|
||||
def test_handle_system_pr_no_args(self, mock_verify: MagicMock) -> None:
|
||||
"""handle_command('system-pr', []) exits with code 1 and a Usage message."""
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
@@ -341,12 +341,14 @@ class TestGitModuleSystemPrRouting:
|
||||
assert result["exit_code"] == 1
|
||||
assert "Usage" in result["stderr"]
|
||||
|
||||
@patch("aipass.drone.apps.plugins.devpulse_ops.auth.verify_caller")
|
||||
@patch(
|
||||
"aipass.drone.apps.plugins.devpulse_ops.auth.verify_git_access",
|
||||
side_effect=PermissionError("not authorized"),
|
||||
)
|
||||
def test_handle_system_pr_unauthorized(self, mock_verify: MagicMock) -> None:
|
||||
"""handle_command propagates PermissionError as exit_code 1 with the message."""
|
||||
from aipass.drone.apps.modules.git_module import handle_command
|
||||
|
||||
mock_verify.side_effect = PermissionError("not authorized")
|
||||
result = handle_command("system-pr", ["test"])
|
||||
assert result["exit_code"] == 1
|
||||
assert "not authorized" in result["stderr"]
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"metadata": {
|
||||
"version": "1.0.0",
|
||||
"last_updated": "2026-05-10",
|
||||
"last_updated": "2026-05-12",
|
||||
"description": "Template file tracking registry for ID-based updates"
|
||||
},
|
||||
"files": {
|
||||
@@ -155,7 +155,7 @@
|
||||
"content_hash": "a4cf0a8e3b4f",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f026": {
|
||||
"f015": {
|
||||
"path": "apps/modules/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
@@ -263,7 +263,7 @@
|
||||
"content_hash": "28e9ae373563",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f015": {
|
||||
"f026": {
|
||||
"path": "apps/plugins/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
|
||||
Reference in New Issue
Block a user