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

Merge pull request #74 from AIOSAI/feat/drone-activated-commands

Browse Source 
feat(drone): restore activated commands — scan, activate, list, remove, execute
This commit is contained in:
AIPass
2026-03-17 19:08:05 -07:00
committed by GitHub
parent d731a3b180 26f120ab07
commit 49f96f2620
14 changed files with 2961 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/drone/README.md
+12 -2
View File
@@ -25,6 +25,10 @@ drone systems # List all registered modules and branches
drone @seedgo verify # Route "verify" to the seedgo module
drone @seedgo audit aipass # Route "audit aipass" to seedgo
drone @module --help # Show help for any module
drone scan @branch # Discover available commands in a branch
drone activate @branch # Scan + register all commands from a branch
drone list # List registered custom command shortcuts
drone remove <name> # Remove a custom command shortcut
drone --version # Show version
drone --help # Show usage information
```
@@ -91,10 +95,15 @@ drone/
│ │ ├── resolver.py # Branch resolution (@name -> path)
│ │ ├── router.py # Command routing via subprocess
│ │ ├── discovery.py # Module and command discovery
│ │ └── module_registry.py # Internal module routing
│ │ ├── module_registry.py # Internal module routing
│ │ ├── commands.py # Custom command shortcut orchestrator
│ │ └── scan.py # Branch command scanning
│ └── handlers/ # Implementation
│ ├── executor.py # Safe subprocess execution
│ └── exceptions.py # Exception hierarchy
│ ├── exceptions.py # Exception hierarchy
│ ├── json/ # Three-JSON Pattern handler
│ ├── scanning/ # Scan result formatting + discovery
│ └── command_registry/ # Command shortcut CRUD + lookup
├── docs/ # Documentation
└── tests/
```
@@ -140,3 +149,4 @@ To add: edit `interactive_commands` or `interactive_branches` in `_handle_target
---
**Last Updated:** 2026-03-17
src/aipass/drone/apps/drone.py
+139 -3
View File
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: drone.py
# Description: Drone - Command Router & Discovery
# Version: 1.0.0
# Version: 1.1.0
# Created: 2026-03-05
# Modified: 2026-03-08
# Modified: 2026-03-17
# =============================================
"""
@@ -31,7 +31,7 @@ from aipass.drone.apps.modules.module_registry import (
get_module_help,
)
VERSION = "1.0.0"
VERSION = "1.1.0"
# =============================================================================
@@ -49,6 +49,10 @@ def show_help() -> None:
console.print(" drone @target command \\[args] Route command to branch or module")
console.print(" drone @target --help Show help for branch or module")
console.print(" drone systems List registered branches and modules")
console.print(" drone scan @target Discover available commands in a branch")
console.print(" drone activate @target Register all commands from a branch")
console.print(" drone list List registered custom commands")
console.print(" drone remove <name> Remove a custom command")
console.print(" drone --help Show this help")
console.print(" drone --version Show version")
console.print()
@@ -57,6 +61,8 @@ def show_help() -> None:
console.print(" drone @seedgo list")
console.print(" drone @flow status")
console.print(" drone systems")
console.print(" drone activate @seedgo")
console.print(" drone audit (custom command shortcut)")
console.print()
@@ -169,6 +175,104 @@ def _handle_module(name: str, args: List[str]) -> int:
return result.get("exit_code", 0)
def _handle_activate(target: str) -> int:
"""Handle `drone activate @branch` -- scan and register all discovered commands."""
from aipass.drone.apps.modules.scan import scan
from aipass.drone.apps.modules.commands import add
from aipass.drone.apps.modules.commands import format_activation_results
branch_name = target.lstrip("@").lower()
results = scan(target)
if results is None:
err_console.print(f"drone: could not resolve '{target}'")
return 1
if not results:
return 0
added: list[str] = []
skipped: list[str] = []
for cmd in results:
name = cmd["name"]
description = cmd.get("description", "")
success = add(
name=name,
target=f"@{branch_name}",
command=name,
description=description,
source_branch=branch_name,
)
if success:
added.append(name)
else:
skipped.append(name)
format_activation_results(branch_name, added, skipped)
return 0
def _handle_list() -> int:
"""Handle `drone list` -- show all registered custom commands."""
from aipass.drone.apps.modules.commands import list_all
from aipass.drone.apps.modules.commands import format_command_list
commands = list_all()
format_command_list(commands)
return 0
def _handle_remove(name: str) -> int:
"""Handle `drone remove <name>` -- remove a custom command."""
from aipass.drone.apps.modules.commands import remove
from aipass.drone.apps.modules.commands import format_removal
success = remove(name)
format_removal(name, success)
return 0 if success else 1
def _handle_custom_command(args: list[str]) -> int:
"""Handle a custom command shortcut by matching and routing.
Uses greedy multi-word matching to resolve user input to a registered
custom command, then routes through the same path as ``@target`` commands.
"""
from aipass.drone.apps.modules.commands import match
matched = match(args)
if matched is None:
return -1 # Signal: not a custom command
cmd_data, remaining_args = matched
target = cmd_data["target"]
command = cmd_data["command"]
cmd_args = list(cmd_data.get("args", [])) + remaining_args
module_name = target.lstrip("@").lower()
# Interactive detection -- same logic as _handle_target
interactive_commands = ("monitor", "snapshot", "versioned")
interactive_branches = ("cli",)
interactive = command in interactive_commands or module_name in interactive_branches
try:
result = route_command(
target, command,
args=cmd_args if cmd_args else None,
interactive=interactive,
)
except (BranchNotFoundError, CommandExecutionError, RegistryError) as exc:
err_console.print(f"drone: {exc}")
return 1
if result.stdout:
console.print(result.stdout, end="", highlight=False)
if result.stderr:
err_console.print(result.stderr, end="", highlight=False)
return result.exit_code
def _handle_target(args: List[str]) -> int:
"""Handle `drone @target command [args]` or `drone @target --help`."""
target = args[0]
@@ -270,10 +374,42 @@ def main() -> int:
err_console.print(f"drone: {exc}")
return 1
# scan — discover available commands in a branch
if command == "scan":
if len(args) < 2:
err_console.print("drone: scan requires a target (e.g., drone scan @seedgo)")
return 1
from aipass.drone.apps.modules.scan import scan
results = scan(args[1])
return 0 if results is not None else 1
# activate — scan + register all discovered commands from a branch
if command == "activate":
if len(args) < 2:
err_console.print("drone: activate requires a target (e.g., drone activate @seedgo)")
return 1
return _handle_activate(args[1])
# list — show registered custom commands
if command == "list":
return _handle_list()
# remove — remove a custom command by name
if command == "remove":
if len(args) < 2:
err_console.print("drone: remove requires a command name (e.g., drone remove audit)")
return 1
return _handle_remove(args[1])
# @target — route to branch or module
if command.startswith("@"):
return _handle_target(args)
# Custom command matching (greedy multi-word, before unknown error)
custom_result = _handle_custom_command(args)
if custom_result != -1:
return custom_result
# Unknown command
err_console.print(f"drone: unknown command '{command}'")
err_console.print("Run 'drone --help' for usage.")
src/aipass/drone/apps/handlers/command_registry/__init__.py
+14
View File
@@ -0,0 +1,14 @@
# =================== AIPass ====================
# Name: command_registry/__init__.py
# Description: Custom command registry handler package
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Custom command registry handler package.
Provides persistent storage and lookup for user-defined command shortcuts
that map short names (e.g. ``audit``) to full drone commands
(e.g. ``drone @seedgo audit aipass``).
"""
src/aipass/drone/apps/handlers/command_registry/formatters.py
+117
View File
@@ -0,0 +1,117 @@
# =================== AIPass ====================
# Name: formatters.py
# Description: Rich output formatting for custom command registry
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Rich output formatting for custom command registry operations.
Renders command lists, activation results, and removal confirmations
using Rich tables and styled console output.
"""
from __future__ import annotations
try:
from aipass.cli.apps.modules import console
except ImportError:
from rich.console import Console
console = Console()
from rich.table import Table
from aipass.drone.apps.handlers.json import json_handler
def format_command_list(commands: list[dict]) -> None:
"""Display registered custom commands as a Rich table.
Args:
commands: List of command dicts with ``name``, ``target``,
``command``, ``args``, ``description`` keys.
"""
json_handler.log_operation(
"format_command_list",
{"command_count": len(commands)},
)
if not commands:
console.print()
console.print("No custom commands registered.")
console.print("Use 'drone activate @branch' to register commands from a branch.")
console.print()
return
console.print()
table = Table(show_header=True, header_style="bold")
table.add_column("Name", min_width=14)
table.add_column("Target", min_width=10)
table.add_column("Command", min_width=12)
table.add_column("Args", min_width=12)
table.add_column("Description", min_width=24)
for cmd in commands:
table.add_row(
cmd.get("name", ""),
cmd.get("target", ""),
cmd.get("command", ""),
" ".join(cmd.get("args", [])),
cmd.get("description", ""),
)
console.print(table)
console.print()
console.print(f"{len(commands)} custom command(s) registered.")
console.print()
def format_activation_results(
branch: str,
added: list[str],
skipped: list[str],
) -> None:
"""Display what was registered during activation.
Args:
branch: Branch name (with or without ``@`` prefix).
added: List of command names that were successfully registered.
skipped: List of command names that were skipped (already exist).
"""
display_name = branch if branch.startswith("@") else f"@{branch}"
console.print()
if added:
console.print(f"Activated {len(added)} command(s) from [bold]{display_name}[/bold]:")
for name in added:
console.print(f" + {name}")
if skipped:
if added:
console.print()
console.print(f"Skipped {len(skipped)} command(s) (already registered):")
for name in skipped:
console.print(f" - {name}")
if not added and not skipped:
console.print(f"No commands discovered in {display_name}.")
console.print()
def format_removal(name: str, success: bool) -> None:
"""Display the result of removing a custom command.
Args:
name: The command name that was targeted for removal.
success: Whether the removal succeeded.
"""
console.print()
if success:
console.print(f"Removed custom command '{name}'.")
else:
console.print(f"Command '{name}' not found in registry.")
console.print()
src/aipass/drone/apps/handlers/command_registry/lookup.py
+136
View File
@@ -0,0 +1,136 @@
# =================== AIPass ====================
# Name: lookup.py
# Description: Command lookup and matching for custom command shortcuts
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Command Lookup and Matching for Custom Commands.
Provides exact-match lookup and greedy multi-word matching so that
user input like ``plan create flow`` resolves to the longest registered
shortcut name (trying 3 words, then 2, then 1).
Usage:
from aipass.drone.apps.handlers.command_registry.lookup import (
lookup_command, match_command, list_commands,
)
cmd = lookup_command("audit")
result = match_command(["plan", "create", "flow", "--verbose"])
"""
from __future__ import annotations
from typing import Any
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from .ops import load_registry
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
MODULE_NAME = "command_lookup"
# ---------------------------------------------------------------------------
# Lookup functions
# ---------------------------------------------------------------------------
def lookup_command(name: str) -> dict[str, Any] | None:
"""Look up a custom command by exact name.
Args:
name: Shortcut name to look up.
Returns:
Command dict if found, None otherwise.
"""
try:
registry = load_registry()
return registry.get("commands", {}).get(name)
except Exception as exc:
logger.error("[%s] Failed to look up command '%s': %s", MODULE_NAME, name, exc)
return None
def match_command(args: list[str]) -> tuple[dict[str, Any], list[str]] | None:
"""Match user input against registered commands using greedy multi-word matching.
Tries the longest candidate first (up to 4 words), then progressively
shorter candidates until a match is found.
For example, given args ``["plan", "create", "flow", "--verbose"]``
it tries:
1. ``"plan create flow --verbose"`` (4 words)
2. ``"plan create flow"`` (3 words)
3. ``"plan create"`` (2 words)
4. ``"plan"`` (1 word)
Args:
args: List of whitespace-split user input tokens.
Returns:
Tuple of (command_dict, remaining_args) on match, None otherwise.
"""
if not args:
return None
try:
registry = load_registry()
commands = registry.get("commands", {})
except Exception as exc:
logger.error("[%s] Failed to load registry for matching: %s", MODULE_NAME, exc)
return None
for i in range(min(len(args), 4), 0, -1):
candidate = " ".join(args[:i])
if candidate in commands:
remaining = args[i:]
json_handler.log_operation(
"match_command",
{"matched": candidate, "remaining_args": remaining},
)
return (commands[candidate], remaining)
return None
def list_commands() -> list[dict[str, Any]]:
"""List all registered custom commands, sorted by name.
Returns:
List of command dicts sorted alphabetically by name.
"""
try:
registry = load_registry()
commands = registry.get("commands", {})
return sorted(commands.values(), key=lambda c: c.get("name", ""))
except Exception as exc:
logger.error("[%s] Failed to list commands: %s", MODULE_NAME, exc)
return []
def list_commands_by_branch(branch_name: str) -> list[dict[str, Any]]:
"""List custom commands filtered by source branch.
Args:
branch_name: Branch name to filter by (e.g. ``"seedgo"``).
Returns:
List of command dicts whose source_branch matches, sorted by name.
"""
try:
registry = load_registry()
commands = registry.get("commands", {})
filtered = [
cmd for cmd in commands.values()
if cmd.get("source_branch") == branch_name
]
return sorted(filtered, key=lambda c: c.get("name", ""))
except Exception as exc:
logger.error("[%s] Failed to list commands for branch '%s': %s", MODULE_NAME, branch_name, exc)
return []
src/aipass/drone/apps/handlers/command_registry/ops.py
+271
View File
@@ -0,0 +1,271 @@
# =================== AIPass ====================
# Name: ops.py
# Description: Registry CRUD operations for custom command shortcuts
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Registry CRUD Operations for Custom Commands.
Core operations for loading, saving, and managing the drone command registry.
The registry maps shortcut names to full drone commands so users can type
``audit`` instead of ``drone @seedgo audit aipass``.
Usage:
from aipass.drone.apps.handlers.command_registry.ops import (
load_registry, save_registry, add_command, remove_command,
)
registry = load_registry()
add_command("audit", "@seedgo", "audit", ["aipass"], "Run audit", "seedgo")
"""
from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
MODULE_NAME = "command_registry"
# ops.py -> command_registry/ -> handlers/ -> apps/ -> drone/
_BRANCH_ROOT: Path = Path(__file__).resolve().parents[3]
REGISTRY_FILE: Path = _BRANCH_ROOT / "drone_command_registry.json"
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _today() -> str:
"""Return today's date as ISO string."""
return datetime.now().date().isoformat()
def _empty_registry() -> dict[str, Any]:
"""Return a new empty registry structure."""
today = _today()
return {
"commands": {},
"metadata": {
"version": "1.0.0",
"last_updated": today,
"command_count": 0,
},
}
def _registry_path() -> Path:
"""Return the current registry file path.
Wrapped in a function so tests can monkeypatch ``ops.REGISTRY_FILE``.
"""
return REGISTRY_FILE
# ---------------------------------------------------------------------------
# Core CRUD
# ---------------------------------------------------------------------------
def load_registry() -> dict[str, Any]:
"""Load the command registry from disk, auto-creating if missing.
Returns:
Registry dict with ``commands`` and ``metadata`` keys.
"""
path = _registry_path()
if not path.exists():
logger.info("[%s] Registry not found, creating at %s", MODULE_NAME, path)
registry = _empty_registry()
save_registry(registry)
return registry
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("[%s] Corrupt registry, recreating: %s", MODULE_NAME, exc)
registry = _empty_registry()
save_registry(registry)
return registry
# Auto-heal missing keys
if not isinstance(data, dict):
logger.warning("[%s] Invalid registry structure, recreating", MODULE_NAME)
registry = _empty_registry()
save_registry(registry)
return registry
if "commands" not in data:
data["commands"] = {}
if "metadata" not in data:
data["metadata"] = {
"version": "1.0.0",
"last_updated": _today(),
"command_count": len(data["commands"]),
}
return data
def save_registry(data: dict[str, Any]) -> bool:
"""Save the command registry to disk with validation.
Args:
data: Registry dict to persist.
Returns:
True if saved successfully, False on error.
"""
path = _registry_path()
if not isinstance(data, dict) or "commands" not in data:
logger.error("[%s] Cannot save invalid registry structure", MODULE_NAME)
return False
# Refresh metadata
data.setdefault("metadata", {})
data["metadata"]["last_updated"] = _today()
data["metadata"]["command_count"] = len(data["commands"])
try:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
json.dump(data, fh, indent=2, ensure_ascii=False)
json_handler.log_operation(
"save_registry",
{"command_count": data["metadata"]["command_count"]},
module_name=MODULE_NAME,
)
return True
except OSError as exc:
logger.error("[%s] Failed to save registry: %s", MODULE_NAME, exc)
return False
def add_command(
name: str,
target: str,
command: str,
args: list[str] | None = None,
description: str = "",
source_branch: str = "",
) -> bool:
"""Add a new custom command to the registry.
Args:
name: Shortcut name (e.g. ``"audit"``).
target: Target branch (e.g. ``"@seedgo"``).
command: Actual command to run on the target.
args: Extra arguments for the command.
description: Human-readable description.
source_branch: Branch the command originated from.
Returns:
True if added, False if name already exists or save fails.
"""
registry = load_registry()
if name in registry["commands"]:
logger.warning("[%s] Command '%s' already exists", MODULE_NAME, name)
return False
registry["commands"][name] = {
"name": name,
"target": target,
"command": command,
"args": args if args is not None else [],
"description": description,
"created": _today(),
"source_branch": source_branch,
}
result = save_registry(registry)
if result:
json_handler.log_operation(
"add_command",
{"name": name, "target": target, "command": command},
module_name=MODULE_NAME,
)
return result
def remove_command(name: str) -> bool:
"""Remove a custom command from the registry.
Args:
name: Shortcut name to remove.
Returns:
True if removed, False if not found or save fails.
"""
registry = load_registry()
if name not in registry["commands"]:
logger.warning("[%s] Command '%s' not found", MODULE_NAME, name)
return False
del registry["commands"][name]
result = save_registry(registry)
if result:
json_handler.log_operation(
"remove_command",
{"name": name},
module_name=MODULE_NAME,
)
return result
def update_command(name: str, **kwargs: Any) -> bool:
"""Update fields of an existing custom command.
Args:
name: Shortcut name to update.
**kwargs: Fields to update (e.g. ``target="@prax"``, ``description="..."``).
Returns:
True if updated, False if not found or save fails.
"""
registry = load_registry()
if name not in registry["commands"]:
logger.warning("[%s] Command '%s' not found for update", MODULE_NAME, name)
return False
allowed_keys = {"target", "command", "args", "description", "source_branch"}
for key, value in kwargs.items():
if key in allowed_keys:
registry["commands"][name][key] = value
result = save_registry(registry)
if result:
json_handler.log_operation(
"update_command",
{"name": name, "updated_fields": list(kwargs.keys())},
module_name=MODULE_NAME,
)
return result
def command_exists(name: str) -> bool:
"""Check whether a custom command exists in the registry.
Args:
name: Shortcut name to check.
Returns:
True if the command exists, False otherwise.
"""
registry = load_registry()
return name in registry["commands"]
src/aipass/drone/apps/handlers/scanning/__init__.py
+27
View File
@@ -0,0 +1,27 @@
# =================== AIPass ====================
# Name: __init__.py
# Description: Scanning handler package for module command discovery
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Scanning handler package -- discovers available commands in branches."""
from aipass.drone.apps.handlers.scanning.scanner import (
scan_branch,
scan_help_output,
scan_module_files,
)
from aipass.drone.apps.handlers.scanning.formatters import (
format_no_commands,
format_scan_results,
)
__all__ = [
"format_no_commands",
"format_scan_results",
"scan_branch",
"scan_help_output",
"scan_module_files",
]
src/aipass/drone/apps/handlers/scanning/formatters.py
+77
View File
@@ -0,0 +1,77 @@
# =================== AIPass ====================
# Name: formatters.py
# Description: Rich output formatting for scan results
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Rich output formatting for scan results.
Renders discovered-command lists as clean Rich tables or informational
messages when no commands are found.
"""
from __future__ import annotations
try:
from aipass.cli.apps.modules import console
except ImportError:
from rich.console import Console
console = Console()
from rich.table import Table
from aipass.drone.apps.handlers.json import json_handler
def format_scan_results(branch_name: str, commands: list[dict]) -> None:
"""Display scan results as a Rich table.
Args:
branch_name: Branch name (with or without ``@`` prefix) for the title.
commands: List of command dicts with ``name``, ``description``, ``source``.
"""
display_name = branch_name if branch_name.startswith("@") else f"@{branch_name}"
json_handler.log_operation(
"format_scan_results",
{"branch": display_name, "command_count": len(commands)},
)
console.print()
console.print(f"Scan results for [bold]{display_name}[/bold]")
console.print()
table = Table(show_header=True, header_style="bold")
table.add_column("#", justify="right", style="dim", width=4)
table.add_column("Command", min_width=18)
table.add_column("Description", min_width=30)
table.add_column("Source", justify="center", width=8)
for idx, cmd in enumerate(commands, start=1):
table.add_row(
str(idx),
cmd.get("name", ""),
cmd.get("description", ""),
cmd.get("source", ""),
)
console.print(table)
console.print()
console.print(f"{len(commands)} command(s) discovered for {display_name}")
console.print()
def format_no_commands(branch_name: str) -> None:
"""Display a message when no commands are found for a branch.
Args:
branch_name: Branch name (with or without ``@`` prefix).
"""
display_name = branch_name if branch_name.startswith("@") else f"@{branch_name}"
console.print()
console.print(f"No commands discovered for [bold]{display_name}[/bold].")
console.print("Ensure the branch has an entry point with --help or modules with handle_command().")
console.print()
src/aipass/drone/apps/handlers/scanning/scanner.py
+207
View File
@@ -0,0 +1,207 @@
# =================== AIPass ====================
# Name: scanner.py
# Description: Module scanning for command discovery
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Core scanning logic for discovering available commands in a branch.
Combines two discovery strategies:
1. Run the branch entry point with ``--help`` and parse the output.
2. Scan ``apps/modules/*.py`` for files containing ``handle_command()``.
Results are merged and deduplicated, then returned as a flat list of dicts
suitable for display or downstream activation.
Reuses existing discovery-handler functions where possible.
"""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.discovery_handler import (
get_entry_point,
parse_help_for_commands,
)
# ---------------------------------------------------------------------------
# Help-output scanning
# ---------------------------------------------------------------------------
def scan_help_output(branch_path: str, branch_name: str) -> list[dict]:
"""Run the branch entry point with ``--help`` and parse discovered commands.
Args:
branch_path: Absolute path to the branch directory.
branch_name: Branch name (without ``@`` prefix).
Returns:
List of command dicts ``{"name": ..., "description": ..., "source": "help"}``.
"""
entry_point = get_entry_point(branch_path, branch_name)
if entry_point is None:
return []
try:
result = subprocess.run(
[sys.executable, str(entry_point.relative_to(branch_path)), "--help"],
cwd=branch_path,
capture_output=True,
timeout=10,
shell=False,
)
help_text = result.stdout.decode("utf-8", errors="replace")
if not help_text:
help_text = result.stderr.decode("utf-8", errors="replace")
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("scan_help_output: entry point execution failed for %s: %s", branch_name, exc)
return []
command_names = parse_help_for_commands(help_text)
# Build richer dicts by re-scanning the help text for descriptions
description_map = _extract_descriptions(help_text, command_names)
return [
{
"name": name,
"description": description_map.get(name, ""),
"source": "help",
}
for name in command_names
]
def _extract_descriptions(help_text: str, command_names: list[str]) -> dict[str, str]:
"""Best-effort extraction of one-line descriptions from help text.
Looks for lines like `` command_name Description text`` and pairs
the first token with the rest of the line.
"""
descriptions: dict[str, str] = {}
name_set = set(command_names)
for line in help_text.splitlines():
stripped = line.strip()
if not stripped:
continue
parts = stripped.split(None, 1)
if len(parts) == 2 and parts[0] in name_set:
descriptions[parts[0]] = parts[1].strip()
return descriptions
# ---------------------------------------------------------------------------
# Module-file scanning
# ---------------------------------------------------------------------------
def scan_module_files(branch_path: str) -> list[dict]:
"""Scan ``apps/modules/*.py`` for files that define ``handle_command()``.
Args:
branch_path: Absolute path to the branch directory.
Returns:
List of command dicts ``{"name": ..., "description": ..., "source": "module"}``.
"""
modules_dir = Path(branch_path) / "apps" / "modules"
if not modules_dir.is_dir():
return []
excluded = {"__init__", "__main__"}
results: list[dict] = []
for py_file in sorted(modules_dir.glob("*.py")):
if py_file.stem in excluded:
continue
try:
source = py_file.read_text(encoding="utf-8")
except OSError:
continue
if "def handle_command" not in source:
continue
# Extract a description from the module docstring (first line).
description = _extract_module_description(source)
results.append(
{
"name": py_file.stem,
"description": description,
"source": "module",
}
)
return results
def _extract_module_description(source: str) -> str:
"""Extract the first line of a module-level docstring, if present."""
for delimiter in ('"""', "'''"):
idx = source.find(delimiter)
if idx == -1:
continue
start = idx + len(delimiter)
end = source.find(delimiter, start)
if end == -1:
continue
docstring = source[start:end].strip()
first_line = docstring.split("\n", 1)[0].strip()
if first_line:
return first_line
return ""
# ---------------------------------------------------------------------------
# Full branch scan (merge + deduplicate)
# ---------------------------------------------------------------------------
def scan_branch(branch_path: str, branch_name: str) -> list[dict]:
"""Perform a full scan of a branch to discover available commands.
Combines both ``--help`` parsing and ``apps/modules/*.py`` file scanning,
deduplicates by command name (``--help`` wins on conflict), and returns
the merged result.
Args:
branch_path: Absolute path to the branch directory.
branch_name: Branch name (without ``@`` prefix).
Returns:
List of command dicts with keys ``name``, ``description``, ``source``.
"""
help_commands = scan_help_output(branch_path, branch_name)
module_commands = scan_module_files(branch_path)
# Merge: help results take priority for duplicates.
seen: dict[str, dict] = {}
for cmd in help_commands:
seen[cmd["name"]] = cmd
for cmd in module_commands:
if cmd["name"] not in seen:
seen[cmd["name"]] = cmd
merged = sorted(seen.values(), key=lambda c: c["name"])
json_handler.log_operation(
"scan_branch",
{
"branch": branch_name,
"help_count": len(help_commands),
"module_count": len(module_commands),
"total": len(merged),
},
)
return merged
src/aipass/drone/apps/modules/commands.py
+245
View File
@@ -0,0 +1,245 @@
# =================== AIPass ====================
# Name: commands.py
# Description: Module orchestrator for custom command shortcuts
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Module orchestrator for custom command shortcuts.
Thin orchestrator that delegates to the command_registry handler package
for all CRUD and lookup operations on user-defined command aliases.
"""
from __future__ import annotations
from typing import Any
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.command_registry.ops import (
add_command as _add_command,
command_exists as _command_exists,
remove_command as _remove_command,
update_command as _update_command,
)
from aipass.drone.apps.handlers.command_registry.lookup import (
list_commands as _list_commands,
list_commands_by_branch as _list_commands_by_branch,
lookup_command as _lookup_command,
match_command as _match_command,
)
from aipass.drone.apps.handlers.command_registry.formatters import (
format_activation_results,
format_command_list,
format_removal,
)
__all__ = [
"add",
"format_activation_results",
"format_command_list",
"format_removal",
"handle_command",
"list_all",
"lookup",
"match",
"print_help",
"print_introspection",
"remove",
]
# ---------------------------------------------------------------------------
# Standard module interface
# ---------------------------------------------------------------------------
def handle_command(command: str | None = None, args: list[str] | None = None) -> bool:
"""Route commands subcommands to handler functions.
Args:
command: The subcommand string (e.g. ``"add"``, ``"remove"``, ``"list"``).
args: List of arguments for the subcommand.
Returns:
True if the command succeeded, False otherwise.
"""
if not args:
if command is None:
print_introspection()
return True
args = []
json_handler.log_operation("handle_command", {"module": "commands", "command": command})
if command == "add":
if len(args) < 3:
logger.warning("commands add requires: <name> <target> <command> [args...] [--desc=...] [--branch=...]")
return False
name = args[0]
target = args[1]
cmd = args[2]
extra = args[3:]
# Extract optional --desc= and --branch= flags
description = ""
source_branch = ""
cmd_args: list[str] = []
for arg in extra:
if arg.startswith("--desc="):
description = arg[len("--desc="):]
elif arg.startswith("--branch="):
source_branch = arg[len("--branch="):]
else:
cmd_args.append(arg)
return add(name, target, cmd, cmd_args, description, source_branch)
if command == "remove":
if not args:
logger.warning("commands remove requires a command name")
return False
return remove(args[0])
if command == "list":
branch_filter = args[0] if args else None
if branch_filter:
cmds = _list_commands_by_branch(branch_filter)
else:
cmds = list_all()
for cmd_entry in cmds:
logger.info(
" %s -> %s %s %s",
cmd_entry.get("name", "?"),
cmd_entry.get("target", "?"),
cmd_entry.get("command", "?"),
" ".join(cmd_entry.get("args", [])),
)
if not cmds:
logger.info(" (no commands registered)")
return True
if command == "lookup":
if not args:
logger.warning("commands lookup requires a command name")
return False
result = lookup(args[0])
if result:
logger.info(" %s -> %s %s %s", result["name"], result["target"], result["command"], " ".join(result.get("args", [])))
else:
logger.warning(" Command '%s' not found", args[0])
return False
return True
logger.warning("commands: unknown subcommand '%s'", command)
return False
def print_introspection() -> None:
"""Display module introspection info."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
from rich.console import Console
console = Console()
console.print()
console.print("commands Module")
console.print("Custom command shortcuts — map short names to full drone commands.")
console.print()
console.print("Connected Handlers:")
console.print(" handlers/command_registry/")
console.print(" - ops.py (add_command, remove_command, update_command, command_exists)")
console.print(" - lookup.py (lookup_command, match_command, list_commands, list_commands_by_branch)")
console.print()
def print_help() -> None:
"""Print help for the commands module."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
from rich.console import Console
console = Console()
console.print("commands -- Custom command shortcuts")
console.print()
console.print("Commands:")
console.print(" add <name> <target> <cmd> [args...] [--desc=...] [--branch=...]")
console.print(" remove <name> Remove a custom command")
console.print(" list [branch] List commands (optionally by branch)")
console.print(" lookup <name> Look up a command by name")
# ---------------------------------------------------------------------------
# Delegated operations
# ---------------------------------------------------------------------------
def add(
name: str,
target: str,
command: str,
args: list[str] | None = None,
description: str = "",
source_branch: str = "",
) -> bool:
"""Add a custom command shortcut.
Args:
name: Shortcut name.
target: Target branch (e.g. ``"@seedgo"``).
command: Command to run on the target.
args: Extra arguments.
description: Human-readable description.
source_branch: Originating branch name.
Returns:
True if added successfully.
"""
return _add_command(name, target, command, args, description, source_branch)
def remove(name: str) -> bool:
"""Remove a custom command shortcut.
Args:
name: Shortcut name to remove.
Returns:
True if removed successfully.
"""
return _remove_command(name)
def list_all() -> list[dict[str, Any]]:
"""List all registered custom commands sorted by name.
Returns:
List of command dicts.
"""
return _list_commands()
def lookup(name: str) -> dict[str, Any] | None:
"""Look up a custom command by exact name.
Args:
name: Shortcut name.
Returns:
Command dict if found, None otherwise.
"""
return _lookup_command(name)
def match(args: list[str]) -> tuple[dict[str, Any], list[str]] | None:
"""Match user input to a registered command using greedy multi-word matching.
Args:
args: Whitespace-split user input tokens.
Returns:
Tuple of (command_dict, remaining_args) on match, None otherwise.
"""
return _match_command(args)
src/aipass/drone/apps/modules/scan.py
+138
View File
@@ -0,0 +1,138 @@
# =================== AIPass ====================
# Name: scan.py
# Description: Module orchestrator for branch command scanning
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Module orchestrator for branch command scanning.
Thin orchestrator that resolves ``@target`` to a branch path and delegates
scanning to the handler layer. Displays results via Rich formatters.
"""
from __future__ import annotations
from aipass.prax import logger
from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.scanning.scanner import scan_branch
from aipass.drone.apps.handlers.scanning.formatters import (
format_no_commands,
format_scan_results,
)
from aipass.drone.apps.modules.resolver import resolve_branch
__all__ = [
"handle_command",
"print_help",
"print_introspection",
"scan",
]
# ---------------------------------------------------------------------------
# Standard module interface
# ---------------------------------------------------------------------------
def handle_command(command: str | None = None, args: list[str] | None = None) -> bool:
"""Route scan subcommands to handler functions.
Args:
command: The subcommand string (currently unused -- scans are positional).
args: List of arguments; first element is the ``@target``.
Returns:
True if the command succeeded, False otherwise.
"""
if not args:
if command is None:
print_introspection()
return True
args = []
json_handler.log_operation("handle_command", {"module": "scan", "command": command})
if not args:
logger.warning("scan requires a target argument (e.g. scan @seedgo)")
return False
results = scan(args[0])
return results is not None
def print_introspection() -> None:
"""Display module introspection info."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
from rich.console import Console
console = Console()
console.print()
console.print("scan Module")
console.print("Branch command scanning -- discover available commands in a branch.")
console.print()
console.print("Connected Handlers:")
console.print(" handlers/scanning/")
console.print(" - scanner.py (scan_branch, scan_help_output, scan_module_files)")
console.print(" - formatters.py (format_scan_results, format_no_commands)")
console.print()
console.print("Connected Modules:")
console.print(" modules/")
console.print(" - resolver.py (resolve_branch -- branch name resolution)")
console.print()
def print_help() -> None:
"""Print help for the scan module."""
try:
from aipass.cli.apps.modules.display import console
except ImportError:
from rich.console import Console
console = Console()
console.print("scan -- Branch command scanning")
console.print()
console.print("Usage:")
console.print(" drone scan @target Scan a branch for available commands")
console.print()
console.print("Examples:")
console.print(" drone scan @seedgo Scan seedgo for commands")
console.print(" drone scan @flow Scan flow for commands")
# ---------------------------------------------------------------------------
# Core operation
# ---------------------------------------------------------------------------
def scan(target: str) -> list[dict] | None:
"""Resolve ``@target``, scan for commands, display and return results.
Args:
target: Symbolic branch name (e.g. ``@seedgo``).
Returns:
List of discovered command dicts, or None on resolution failure.
"""
branch_name = target.lstrip("@").lower()
try:
branch_path = resolve_branch(target)
except Exception as exc:
logger.warning("scan: could not resolve '%s': %s", target, exc)
try:
from aipass.cli.apps.modules import err_console
err_console.print(f"scan: could not resolve '{target}': {exc}")
except ImportError:
pass
return None
commands = scan_branch(branch_path, branch_name)
if commands:
format_scan_results(branch_name, commands)
else:
format_no_commands(branch_name)
return commands
src/aipass/drone/tests/test_activation.py
+654
View File
@@ -0,0 +1,654 @@
# =================== AIPass ====================
# Name: test_activation.py
# Description: Tests for command activation, listing, removal, and custom execution
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Tests for command activation, listing, removal, and custom execution.
Covers:
- ``drone activate @branch`` registers discovered commands
- ``drone list`` displays custom commands
- ``drone remove <name>`` removes a custom command
- Custom command execution via ``main()`` flow
- ``match_command`` integration with ``route_command``
- Formatter output for activation, listing, removal
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from aipass.drone.apps.handlers.command_registry import ops, lookup
from aipass.drone.apps.handlers.command_registry.formatters import (
format_activation_results,
format_command_list,
format_removal,
)
from aipass.drone.apps.handlers.executor import CommandResult
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def isolated_registry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Point the registry at a temp file so tests never touch the real one."""
registry_file = tmp_path / "drone_command_registry.json"
monkeypatch.setattr(ops, "REGISTRY_FILE", registry_file)
return registry_file
def _seed_commands(**commands: dict[str, Any]) -> None:
"""Register commands via ops.add_command for test setup."""
for name, data in commands.items():
ops.add_command(
name=name,
target=data.get("target", "@test"),
command=data.get("command", name),
args=data.get("args"),
description=data.get("description", ""),
source_branch=data.get("source_branch", "test"),
)
# ===================================================================
# 1. Formatters
# ===================================================================
class TestFormatCommandList:
"""Tests for format_command_list()."""
def test_displays_commands(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should display a table with command details."""
commands = [
{
"name": "audit",
"target": "@seedgo",
"command": "audit",
"args": ["aipass"],
"description": "Run audit",
},
{
"name": "check",
"target": "@seedgo",
"command": "check",
"args": [],
"description": "Run checks",
},
]
format_command_list(commands)
captured = capsys.readouterr()
assert "audit" in captured.out
assert "check" in captured.out
assert "@seedgo" in captured.out
assert "registered" in captured.out
assert "custom" in captured.out
def test_empty_list(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should show informational message when no commands exist."""
format_command_list([])
captured = capsys.readouterr()
assert "No custom commands registered" in captured.out
assert "drone activate" in captured.out
class TestFormatActivationResults:
"""Tests for format_activation_results()."""
def test_shows_added(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should display added commands."""
format_activation_results("seedgo", ["audit", "list"], [])
captured = capsys.readouterr()
assert "Activated" in captured.out
assert "@seedgo" in captured.out
assert "+ audit" in captured.out
assert "+ list" in captured.out
def test_shows_skipped(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should display skipped commands."""
format_activation_results("seedgo", [], ["audit"])
captured = capsys.readouterr()
assert "Skipped" in captured.out
assert "already registered" in captured.out
assert "- audit" in captured.out
def test_shows_both_added_and_skipped(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should display both added and skipped when both exist."""
format_activation_results("seedgo", ["list"], ["audit"])
captured = capsys.readouterr()
assert "Activated" in captured.out
assert "Skipped" in captured.out
def test_shows_no_commands(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should show a message when nothing was added or skipped."""
format_activation_results("seedgo", [], [])
captured = capsys.readouterr()
assert "No commands discovered" in captured.out
def test_adds_at_prefix(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should add @ prefix to branch name if missing."""
format_activation_results("seedgo", ["cmd"], [])
captured = capsys.readouterr()
assert "@seedgo" in captured.out
class TestFormatRemoval:
"""Tests for format_removal()."""
def test_success(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should show removal confirmation on success."""
format_removal("audit", True)
captured = capsys.readouterr()
assert "Removed custom command" in captured.out
assert "audit" in captured.out
def test_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should show not-found message on failure."""
format_removal("ghost", False)
captured = capsys.readouterr()
assert "not found" in captured.out
assert "ghost" in captured.out
# ===================================================================
# 2. _handle_activate
# ===================================================================
class TestHandleActivate:
"""Tests for _handle_activate() in drone.py."""
@patch("aipass.drone.apps.drone._handle_activate.__wrapped__", create=True)
def _get_handler(self):
"""Import the handler to test."""
from aipass.drone.apps.drone import _handle_activate
return _handle_activate
@patch("aipass.drone.apps.modules.commands.format_activation_results")
@patch("aipass.drone.apps.modules.commands.add")
@patch("aipass.drone.apps.modules.scan.scan")
def test_registers_discovered_commands(
self,
mock_scan: MagicMock,
mock_add: MagicMock,
mock_format: MagicMock,
) -> None:
"""Should register all commands discovered by scan."""
from aipass.drone.apps.drone import _handle_activate
mock_scan.return_value = [
{"name": "audit", "description": "Run audit", "source": "help"},
{"name": "list", "description": "List items", "source": "module"},
]
mock_add.return_value = True
result = _handle_activate("@seedgo")
assert result == 0
assert mock_add.call_count == 2
mock_format.assert_called_once()
# Check the added list in format call
call_args = mock_format.call_args
assert "audit" in call_args[0][1] # added list
assert "list" in call_args[0][1]
@patch("aipass.drone.apps.modules.commands.format_activation_results")
@patch("aipass.drone.apps.modules.commands.add")
@patch("aipass.drone.apps.modules.scan.scan")
def test_skips_existing_commands(
self,
mock_scan: MagicMock,
mock_add: MagicMock,
mock_format: MagicMock,
) -> None:
"""Should skip commands that already exist in registry."""
from aipass.drone.apps.drone import _handle_activate
mock_scan.return_value = [
{"name": "audit", "description": "Run audit", "source": "help"},
]
mock_add.return_value = False # Already exists
result = _handle_activate("@seedgo")
assert result == 0
call_args = mock_format.call_args
assert call_args[0][1] == [] # added = empty
assert "audit" in call_args[0][2] # skipped list
@patch("aipass.drone.apps.modules.scan.scan")
def test_returns_1_on_resolution_failure(self, mock_scan: MagicMock) -> None:
"""Should return 1 when scan cannot resolve the target."""
from aipass.drone.apps.drone import _handle_activate
mock_scan.return_value = None
result = _handle_activate("@nonexistent")
assert result == 1
@patch("aipass.drone.apps.modules.scan.scan")
def test_returns_0_on_empty_scan(self, mock_scan: MagicMock) -> None:
"""Should return 0 when scan finds no commands."""
from aipass.drone.apps.drone import _handle_activate
mock_scan.return_value = []
result = _handle_activate("@emptybranch")
assert result == 0
# ===================================================================
# 3. _handle_list
# ===================================================================
class TestHandleList:
"""Tests for _handle_list() in drone.py."""
@patch("aipass.drone.apps.modules.commands.format_command_list")
def test_calls_formatter(self, mock_format: MagicMock) -> None:
"""Should load commands and pass to formatter."""
from aipass.drone.apps.drone import _handle_list
ops.add_command("audit", "@seedgo", "audit")
result = _handle_list()
assert result == 0
mock_format.assert_called_once()
commands_arg = mock_format.call_args[0][0]
assert len(commands_arg) == 1
assert commands_arg[0]["name"] == "audit"
@patch("aipass.drone.apps.modules.commands.format_command_list")
def test_empty_registry(self, mock_format: MagicMock) -> None:
"""Should pass empty list to formatter when no commands exist."""
from aipass.drone.apps.drone import _handle_list
result = _handle_list()
assert result == 0
mock_format.assert_called_once_with([])
# ===================================================================
# 4. _handle_remove
# ===================================================================
class TestHandleRemove:
"""Tests for _handle_remove() in drone.py."""
@patch("aipass.drone.apps.modules.commands.format_removal")
def test_removes_existing(self, mock_format: MagicMock) -> None:
"""Should remove an existing command and return 0."""
from aipass.drone.apps.drone import _handle_remove
ops.add_command("audit", "@seedgo", "audit")
result = _handle_remove("audit")
assert result == 0
mock_format.assert_called_once_with("audit", True)
assert not ops.command_exists("audit")
@patch("aipass.drone.apps.modules.commands.format_removal")
def test_nonexistent_returns_1(self, mock_format: MagicMock) -> None:
"""Should return 1 when trying to remove a nonexistent command."""
from aipass.drone.apps.drone import _handle_remove
result = _handle_remove("ghost")
assert result == 1
mock_format.assert_called_once_with("ghost", False)
# ===================================================================
# 5. _handle_custom_command
# ===================================================================
class TestHandleCustomCommand:
"""Tests for _handle_custom_command() in drone.py."""
@patch("aipass.drone.apps.drone.route_command")
def test_routes_matched_command(self, mock_route: MagicMock) -> None:
"""Should route a matched custom command through route_command."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("audit", "@seedgo", "audit", args=["aipass"])
mock_route.return_value = CommandResult(
stdout="ok\n", stderr="", exit_code=0, branch="seedgo", command="audit",
)
result = _handle_custom_command(["audit"])
assert result == 0
mock_route.assert_called_once_with(
"@seedgo", "audit",
args=["aipass"],
interactive=False,
)
@patch("aipass.drone.apps.drone.route_command")
def test_appends_remaining_args(self, mock_route: MagicMock) -> None:
"""Should append remaining args to configured args."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("audit", "@seedgo", "audit", args=["aipass"])
mock_route.return_value = CommandResult(
stdout="", stderr="", exit_code=0, branch="seedgo", command="audit",
)
result = _handle_custom_command(["audit", "@drone"])
assert result == 0
mock_route.assert_called_once_with(
"@seedgo", "audit",
args=["aipass", "@drone"],
interactive=False,
)
def test_returns_negative_1_on_no_match(self) -> None:
"""Should return -1 when no custom command matches."""
from aipass.drone.apps.drone import _handle_custom_command
result = _handle_custom_command(["nonexistent"])
assert result == -1
@patch("aipass.drone.apps.drone.route_command")
def test_interactive_detection_for_command(self, mock_route: MagicMock) -> None:
"""Should set interactive=True for interactive commands."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("mon", "@prax", "monitor")
mock_route.return_value = CommandResult(
stdout="", stderr="", exit_code=0, branch="prax", command="monitor",
)
_handle_custom_command(["mon"])
call_kwargs = mock_route.call_args.kwargs
assert call_kwargs["interactive"] is True
@patch("aipass.drone.apps.drone.route_command")
def test_interactive_detection_for_branch(self, mock_route: MagicMock) -> None:
"""Should set interactive=True for CLI branch commands."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("status", "@cli", "status")
mock_route.return_value = CommandResult(
stdout="", stderr="", exit_code=0, branch="cli", command="status",
)
_handle_custom_command(["status"])
call_kwargs = mock_route.call_args.kwargs
assert call_kwargs["interactive"] is True
@patch("aipass.drone.apps.drone.route_command")
def test_propagates_exit_code(self, mock_route: MagicMock) -> None:
"""Should return the route_command exit code."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("failing", "@test", "fail")
mock_route.return_value = CommandResult(
stdout="", stderr="error\n", exit_code=2, branch="test", command="fail",
)
result = _handle_custom_command(["failing"])
assert result == 2
@patch("aipass.drone.apps.drone.route_command")
def test_handles_route_exception(self, mock_route: MagicMock) -> None:
"""Should return 1 when route_command raises."""
from aipass.drone.apps.drone import _handle_custom_command
from aipass.drone.apps.modules import BranchNotFoundError
ops.add_command("bad", "@ghost", "cmd")
mock_route.side_effect = BranchNotFoundError("not found")
result = _handle_custom_command(["bad"])
assert result == 1
@patch("aipass.drone.apps.drone.route_command")
def test_no_args_passes_none(self, mock_route: MagicMock) -> None:
"""Should pass args=None when configured args and remaining args are both empty."""
from aipass.drone.apps.drone import _handle_custom_command
ops.add_command("simple", "@test", "simple")
mock_route.return_value = CommandResult(
stdout="", stderr="", exit_code=0, branch="test", command="simple",
)
_handle_custom_command(["simple"])
call_kwargs = mock_route.call_args.kwargs
assert call_kwargs["args"] is None
# ===================================================================
# 6. main() integration
# ===================================================================
class TestMainIntegration:
"""Tests for main() routing of new commands."""
@patch("aipass.drone.apps.drone._handle_activate")
def test_activate_route(self, mock_activate: MagicMock) -> None:
"""main() routes 'activate @branch' to _handle_activate."""
from aipass.drone.apps.drone import main
mock_activate.return_value = 0
with patch("sys.argv", ["drone", "activate", "@seedgo"]):
result = main()
assert result == 0
mock_activate.assert_called_once_with("@seedgo")
def test_activate_no_target(self) -> None:
"""main() returns 1 when activate is called without a target."""
from aipass.drone.apps.drone import main
with patch("sys.argv", ["drone", "activate"]):
result = main()
assert result == 1
@patch("aipass.drone.apps.drone._handle_list")
def test_list_route(self, mock_list: MagicMock) -> None:
"""main() routes 'list' to _handle_list."""
from aipass.drone.apps.drone import main
mock_list.return_value = 0
with patch("sys.argv", ["drone", "list"]):
result = main()
assert result == 0
mock_list.assert_called_once()
@patch("aipass.drone.apps.drone._handle_remove")
def test_remove_route(self, mock_remove: MagicMock) -> None:
"""main() routes 'remove name' to _handle_remove."""
from aipass.drone.apps.drone import main
mock_remove.return_value = 0
with patch("sys.argv", ["drone", "remove", "audit"]):
result = main()
assert result == 0
mock_remove.assert_called_once_with("audit")
def test_remove_no_name(self) -> None:
"""main() returns 1 when remove is called without a name."""
from aipass.drone.apps.drone import main
with patch("sys.argv", ["drone", "remove"]):
result = main()
assert result == 1
@patch("aipass.drone.apps.drone._handle_custom_command")
def test_custom_command_route(self, mock_custom: MagicMock) -> None:
"""main() routes unrecognized commands to custom command matching."""
from aipass.drone.apps.drone import main
mock_custom.return_value = 0
with patch("sys.argv", ["drone", "audit"]):
result = main()
assert result == 0
mock_custom.assert_called_once_with(["audit"])
@patch("aipass.drone.apps.drone._handle_custom_command")
def test_unknown_command_when_no_custom_match(self, mock_custom: MagicMock) -> None:
"""main() shows unknown command when custom matching returns -1."""
from aipass.drone.apps.drone import main
mock_custom.return_value = -1
with patch("sys.argv", ["drone", "nonexistent"]):
result = main()
assert result == 1
@patch("aipass.drone.apps.drone.route_command")
def test_custom_command_end_to_end(self, mock_route: MagicMock) -> None:
"""Full integration: registered command routes through route_command."""
from aipass.drone.apps.drone import main
ops.add_command("audit", "@seedgo", "audit", args=["aipass"])
mock_route.return_value = CommandResult(
stdout="audit output\n", stderr="", exit_code=0,
branch="seedgo", command="audit",
)
with patch("sys.argv", ["drone", "audit"]):
result = main()
assert result == 0
mock_route.assert_called_once_with(
"@seedgo", "audit",
args=["aipass"],
interactive=False,
)
@patch("aipass.drone.apps.drone.route_command")
def test_custom_command_with_extra_args_end_to_end(self, mock_route: MagicMock) -> None:
"""Full integration: remaining args appended to configured args."""
from aipass.drone.apps.drone import main
ops.add_command("audit", "@seedgo", "audit", args=["aipass"])
mock_route.return_value = CommandResult(
stdout="", stderr="", exit_code=0,
branch="seedgo", command="audit",
)
with patch("sys.argv", ["drone", "audit", "@drone"]):
result = main()
assert result == 0
mock_route.assert_called_once_with(
"@seedgo", "audit",
args=["aipass", "@drone"],
interactive=False,
)
def test_builtin_commands_take_priority(self) -> None:
"""Built-in commands like 'systems' should NOT be overridden by custom commands."""
from aipass.drone.apps.drone import main
# Register a custom command named 'systems' (should be shadowed)
ops.add_command("systems", "@test", "systems")
with patch("aipass.drone.apps.drone._handle_systems", return_value=0) as mock_sys:
with patch("sys.argv", ["drone", "systems"]):
result = main()
assert result == 0
mock_sys.assert_called_once()
def test_at_target_takes_priority_over_custom(self) -> None:
"""@target routing should take priority over custom command matching."""
from aipass.drone.apps.drone import main
with patch("aipass.drone.apps.drone._handle_target", return_value=0) as mock_target:
with patch("sys.argv", ["drone", "@seedgo", "audit"]):
result = main()
assert result == 0
mock_target.assert_called_once()
# ===================================================================
# 7. match_command integration
# ===================================================================
class TestMatchCommandIntegration:
"""Tests verifying match_command works correctly with registered commands."""
def test_multi_word_custom_command(self) -> None:
"""Multi-word commands should match and leave remaining args."""
ops.add_command("plan create", "@flow", "create", args=["--type=plan"])
result = lookup.match_command(["plan", "create", "my-plan"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "plan create"
assert cmd["target"] == "@flow"
assert remaining == ["my-plan"]
@patch("aipass.drone.apps.drone.route_command")
def test_multi_word_end_to_end(self, mock_route: MagicMock) -> None:
"""Multi-word custom command routes correctly through main()."""
from aipass.drone.apps.drone import main
ops.add_command("plan create", "@flow", "create", args=["--type=plan"])
mock_route.return_value = CommandResult(
stdout="created\n", stderr="", exit_code=0,
branch="flow", command="create",
)
with patch("sys.argv", ["drone", "plan", "create", "my-plan"]):
result = main()
assert result == 0
mock_route.assert_called_once_with(
"@flow", "create",
args=["--type=plan", "my-plan"],
interactive=False,
)
src/aipass/drone/tests/test_commands.py
+431
View File
@@ -0,0 +1,431 @@
# =================== AIPass ====================
# Name: test_commands.py
# Description: Tests for the custom command registry
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Tests for the custom command registry.
Covers:
- CRUD operations (add, remove, update, exists)
- Exact-match lookup
- Multi-word greedy matching
- Registry auto-creation
- List filtering by branch
- Module orchestrator (handle_command)
"""
import json
from pathlib import Path
from typing import Any
import pytest
from aipass.drone.apps.handlers.command_registry import ops, lookup
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def isolated_registry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Point the registry at a temp file so tests never touch the real one."""
registry_file = tmp_path / "drone_command_registry.json"
monkeypatch.setattr(ops, "REGISTRY_FILE", registry_file)
return registry_file
def _seed_registry(registry_file: Path, commands: dict[str, Any] | None = None) -> None:
"""Write a registry file with pre-populated commands."""
if commands is None:
commands = {}
data = {
"commands": commands,
"metadata": {
"version": "1.0.0",
"last_updated": "2026-03-17",
"command_count": len(commands),
},
}
registry_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
# ===================================================================
# 1. Registry auto-creation
# ===================================================================
class TestRegistryAutoCreation:
def test_load_creates_file_when_missing(self, isolated_registry: Path) -> None:
"""load_registry() creates a new file when none exists."""
assert not isolated_registry.exists()
registry = ops.load_registry()
assert isolated_registry.exists()
assert isinstance(registry, dict)
assert "commands" in registry
assert "metadata" in registry
assert registry["metadata"]["command_count"] == 0
def test_load_recreates_on_corrupt_json(self, isolated_registry: Path) -> None:
"""Corrupt JSON triggers auto-recreation."""
isolated_registry.write_text("{{{bad json", encoding="utf-8")
registry = ops.load_registry()
assert registry["commands"] == {}
assert registry["metadata"]["command_count"] == 0
def test_load_recreates_on_non_dict(self, isolated_registry: Path) -> None:
"""Non-dict top-level JSON triggers auto-recreation."""
isolated_registry.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
registry = ops.load_registry()
assert isinstance(registry, dict)
assert "commands" in registry
def test_auto_heals_missing_keys(self, isolated_registry: Path) -> None:
"""Missing 'commands' or 'metadata' keys are auto-healed."""
isolated_registry.write_text(json.dumps({"other": "data"}), encoding="utf-8")
registry = ops.load_registry()
assert "commands" in registry
assert "metadata" in registry
# ===================================================================
# 2. CRUD operations
# ===================================================================
class TestCRUDOperations:
def test_add_command(self, isolated_registry: Path) -> None:
"""add_command creates a new entry in the registry."""
result = ops.add_command(
name="audit",
target="@seedgo",
command="audit",
args=["aipass"],
description="Run audit",
source_branch="seedgo",
)
assert result is True
registry = ops.load_registry()
assert "audit" in registry["commands"]
cmd = registry["commands"]["audit"]
assert cmd["name"] == "audit"
assert cmd["target"] == "@seedgo"
assert cmd["command"] == "audit"
assert cmd["args"] == ["aipass"]
assert cmd["description"] == "Run audit"
assert cmd["source_branch"] == "seedgo"
def test_add_duplicate_returns_false(self, isolated_registry: Path) -> None:
"""Adding a command with an existing name returns False."""
ops.add_command("audit", "@seedgo", "audit")
result = ops.add_command("audit", "@prax", "something_else")
assert result is False
def test_add_with_default_args(self, isolated_registry: Path) -> None:
"""add_command with no args defaults to empty list."""
ops.add_command("monitor", "@prax", "monitor")
registry = ops.load_registry()
assert registry["commands"]["monitor"]["args"] == []
def test_remove_command(self, isolated_registry: Path) -> None:
"""remove_command deletes an existing entry."""
ops.add_command("audit", "@seedgo", "audit")
result = ops.remove_command("audit")
assert result is True
registry = ops.load_registry()
assert "audit" not in registry["commands"]
def test_remove_nonexistent_returns_false(self, isolated_registry: Path) -> None:
"""Removing a nonexistent command returns False."""
result = ops.remove_command("nonexistent")
assert result is False
def test_update_command(self, isolated_registry: Path) -> None:
"""update_command modifies fields of an existing entry."""
ops.add_command("audit", "@seedgo", "audit", description="Old desc")
result = ops.update_command("audit", description="New desc", target="@prax")
assert result is True
registry = ops.load_registry()
cmd = registry["commands"]["audit"]
assert cmd["description"] == "New desc"
assert cmd["target"] == "@prax"
def test_update_nonexistent_returns_false(self, isolated_registry: Path) -> None:
"""Updating a nonexistent command returns False."""
result = ops.update_command("ghost", description="nope")
assert result is False
def test_update_ignores_disallowed_keys(self, isolated_registry: Path) -> None:
"""update_command ignores keys outside the allowed set."""
ops.add_command("audit", "@seedgo", "audit")
ops.update_command("audit", created="1970-01-01")
registry = ops.load_registry()
cmd = registry["commands"]["audit"]
# 'created' is not in the allowed set, so it should NOT be overwritten
assert cmd["created"] != "1970-01-01"
def test_command_exists_true(self, isolated_registry: Path) -> None:
"""command_exists returns True for registered commands."""
ops.add_command("audit", "@seedgo", "audit")
assert ops.command_exists("audit") is True
def test_command_exists_false(self, isolated_registry: Path) -> None:
"""command_exists returns False for unregistered commands."""
assert ops.command_exists("nonexistent") is False
def test_metadata_command_count_updates(self, isolated_registry: Path) -> None:
"""metadata.command_count tracks the number of commands after each save."""
ops.add_command("a", "@t", "c1")
ops.add_command("b", "@t", "c2")
registry = ops.load_registry()
assert registry["metadata"]["command_count"] == 2
ops.remove_command("a")
registry = ops.load_registry()
assert registry["metadata"]["command_count"] == 1
# ===================================================================
# 3. Lookup (exact match)
# ===================================================================
class TestLookup:
def test_lookup_found(self, isolated_registry: Path) -> None:
"""lookup_command returns the command dict for an exact match."""
ops.add_command("audit", "@seedgo", "audit", ["aipass"], "Run audit", "seedgo")
result = lookup.lookup_command("audit")
assert result is not None
assert result["name"] == "audit"
assert result["target"] == "@seedgo"
def test_lookup_not_found(self, isolated_registry: Path) -> None:
"""lookup_command returns None when no match exists."""
result = lookup.lookup_command("nonexistent")
assert result is None
# ===================================================================
# 4. Multi-word greedy matching
# ===================================================================
class TestMultiWordMatching:
def test_single_word_match(self, isolated_registry: Path) -> None:
"""match_command matches a single-word command."""
ops.add_command("audit", "@seedgo", "audit")
result = lookup.match_command(["audit", "--verbose"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "audit"
assert remaining == ["--verbose"]
def test_multi_word_match(self, isolated_registry: Path) -> None:
"""match_command matches a multi-word command name."""
ops.add_command("plan create", "@flow", "create")
result = lookup.match_command(["plan", "create", "my-plan"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "plan create"
assert remaining == ["my-plan"]
def test_greedy_longest_match(self, isolated_registry: Path) -> None:
"""match_command prefers the longest matching candidate."""
ops.add_command("plan", "@flow", "plan_list")
ops.add_command("plan create", "@flow", "plan_create")
result = lookup.match_command(["plan", "create", "my-plan"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "plan create"
assert remaining == ["my-plan"]
def test_four_word_match(self, isolated_registry: Path) -> None:
"""match_command handles up to 4-word candidates."""
ops.add_command("a b c d", "@test", "test_cmd")
result = lookup.match_command(["a", "b", "c", "d", "extra"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "a b c d"
assert remaining == ["extra"]
def test_no_match_returns_none(self, isolated_registry: Path) -> None:
"""match_command returns None when no candidate matches."""
ops.add_command("audit", "@seedgo", "audit")
result = lookup.match_command(["deploy", "staging"])
assert result is None
def test_empty_args_returns_none(self, isolated_registry: Path) -> None:
"""match_command returns None for empty args."""
result = lookup.match_command([])
assert result is None
def test_exact_match_no_remaining(self, isolated_registry: Path) -> None:
"""match_command with no extra args returns empty remaining list."""
ops.add_command("audit", "@seedgo", "audit")
result = lookup.match_command(["audit"])
assert result is not None
cmd, remaining = result
assert cmd["name"] == "audit"
assert remaining == []
# ===================================================================
# 5. List and filter by branch
# ===================================================================
class TestListAndFilter:
def test_list_commands_sorted(self, isolated_registry: Path) -> None:
"""list_commands returns all commands sorted by name."""
ops.add_command("zebra", "@t", "c1")
ops.add_command("alpha", "@t", "c2")
ops.add_command("middle", "@t", "c3")
result = lookup.list_commands()
assert len(result) == 3
names = [c["name"] for c in result]
assert names == ["alpha", "middle", "zebra"]
def test_list_commands_empty(self, isolated_registry: Path) -> None:
"""list_commands returns empty list when no commands exist."""
result = lookup.list_commands()
assert result == []
def test_list_by_branch(self, isolated_registry: Path) -> None:
"""list_commands_by_branch filters by source_branch."""
ops.add_command("audit", "@seedgo", "audit", source_branch="seedgo")
ops.add_command("monitor", "@prax", "monitor", source_branch="prax")
ops.add_command("check", "@seedgo", "check", source_branch="seedgo")
seedgo_cmds = lookup.list_commands_by_branch("seedgo")
assert len(seedgo_cmds) == 2
names = [c["name"] for c in seedgo_cmds]
assert "audit" in names
assert "check" in names
def test_list_by_branch_no_match(self, isolated_registry: Path) -> None:
"""list_commands_by_branch returns empty for nonexistent branch."""
ops.add_command("audit", "@seedgo", "audit", source_branch="seedgo")
result = lookup.list_commands_by_branch("nonexistent")
assert result == []
# ===================================================================
# 6. Save validation
# ===================================================================
class TestSaveValidation:
def test_save_rejects_non_dict(self, isolated_registry: Path) -> None:
"""save_registry rejects data that is not a dict."""
result = ops.save_registry([1, 2, 3]) # type: ignore[arg-type]
assert result is False
def test_save_rejects_missing_commands(self, isolated_registry: Path) -> None:
"""save_registry rejects a dict without 'commands' key."""
result = ops.save_registry({"metadata": {}})
assert result is False
# ===================================================================
# 7. Module orchestrator
# ===================================================================
class TestModuleOrchestrator:
def test_handle_command_introspection(self, isolated_registry: Path) -> None:
"""handle_command with no args triggers introspection."""
from aipass.drone.apps.modules.commands import handle_command
result = handle_command()
assert result is True
def test_handle_command_add(self, isolated_registry: Path) -> None:
"""handle_command 'add' creates a command."""
from aipass.drone.apps.modules.commands import handle_command
result = handle_command("add", ["test_cmd", "@branch", "do_thing", "--desc=A test"])
assert result is True
assert ops.command_exists("test_cmd") is True
registry = ops.load_registry()
assert registry["commands"]["test_cmd"]["description"] == "A test"
def test_handle_command_remove(self, isolated_registry: Path) -> None:
"""handle_command 'remove' deletes a command."""
from aipass.drone.apps.modules.commands import handle_command
ops.add_command("doomed", "@t", "c")
result = handle_command("remove", ["doomed"])
assert result is True
assert ops.command_exists("doomed") is False
def test_handle_command_list(self, isolated_registry: Path) -> None:
"""handle_command 'list' succeeds."""
from aipass.drone.apps.modules.commands import handle_command
ops.add_command("a", "@t", "c1")
result = handle_command("list", [])
assert result is True
def test_handle_command_unknown(self, isolated_registry: Path) -> None:
"""handle_command with unknown subcommand returns False."""
from aipass.drone.apps.modules.commands import handle_command
result = handle_command("unknown_sub", ["arg"])
assert result is False
src/aipass/drone/tests/test_scan.py
+493
View File
@@ -0,0 +1,493 @@
# =================== AIPass ====================
# Name: test_scan.py
# Description: Tests for branch command scanning
# Version: 1.0.0
# Created: 2026-03-17
# Modified: 2026-03-17
# =============================================
"""Tests for branch command scanning.
Covers handler-layer functions (scan_help_output, scan_module_files,
scan_branch) and the orchestration module (scan.handle_command, scan.scan).
"""
import subprocess
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from aipass.drone.apps.handlers.scanning.scanner import (
scan_branch,
scan_help_output,
scan_module_files,
)
from aipass.drone.apps.handlers.scanning.formatters import (
format_no_commands,
format_scan_results,
)
# =============================================================================
# scan_help_output tests
# =============================================================================
class TestScanHelpOutput:
"""Tests for scan_help_output()."""
def test_returns_commands_from_help(self, temp_test_dir: Path) -> None:
"""Should parse commands from --help output."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
mock_result = MagicMock()
mock_result.stdout = (
b"Usage: mybranch\n\n"
b"Commands:\n"
b" audit Run an audit\n"
b" list List items\n\n"
)
mock_result.stderr = b""
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
return_value=mock_result,
):
result = scan_help_output(str(temp_test_dir), "mybranch")
assert len(result) == 2
names = [c["name"] for c in result]
assert "audit" in names
assert "list" in names
assert all(c["source"] == "help" for c in result)
def test_extracts_descriptions(self, temp_test_dir: Path) -> None:
"""Should extract descriptions alongside command names."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
mock_result = MagicMock()
mock_result.stdout = (
b"Commands:\n"
b" deploy Deploy to production\n\n"
)
mock_result.stderr = b""
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
return_value=mock_result,
):
result = scan_help_output(str(temp_test_dir), "mybranch")
assert result[0]["description"] == "Deploy to production"
def test_returns_empty_when_no_entry_point(self, temp_test_dir: Path) -> None:
"""Should return empty list when no entry point exists."""
result = scan_help_output(str(temp_test_dir), "nonexistent")
assert result == []
def test_returns_empty_on_timeout(self, temp_test_dir: Path) -> None:
"""Should return empty list when subprocess times out."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "slow.py").write_text("# entry", encoding="utf-8")
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
side_effect=subprocess.TimeoutExpired(cmd="test", timeout=10),
):
result = scan_help_output(str(temp_test_dir), "slow")
assert result == []
def test_returns_empty_on_oserror(self, temp_test_dir: Path) -> None:
"""Should return empty list on OSError."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "broken.py").write_text("# entry", encoding="utf-8")
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
side_effect=OSError("No such file"),
):
result = scan_help_output(str(temp_test_dir), "broken")
assert result == []
def test_falls_back_to_stderr(self, temp_test_dir: Path) -> None:
"""Should parse stderr when stdout is empty."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
mock_result = MagicMock()
mock_result.stdout = b""
mock_result.stderr = b"Commands:\n check Run checks\n\n"
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
return_value=mock_result,
):
result = scan_help_output(str(temp_test_dir), "mybranch")
assert len(result) == 1
assert result[0]["name"] == "check"
# =============================================================================
# scan_module_files tests
# =============================================================================
class TestScanModuleFiles:
"""Tests for scan_module_files()."""
def test_finds_modules_with_handle_command(self, temp_test_dir: Path) -> None:
"""Should find .py files that define handle_command()."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
(modules_dir / "alpha.py").write_text(
'"""Alpha module."""\ndef handle_command(command=None, args=None): pass\n',
encoding="utf-8",
)
(modules_dir / "beta.py").write_text(
'"""Beta module."""\ndef handle_command(command=None, args=None): pass\n',
encoding="utf-8",
)
result = scan_module_files(str(temp_test_dir))
assert len(result) == 2
names = [c["name"] for c in result]
assert "alpha" in names
assert "beta" in names
assert all(c["source"] == "module" for c in result)
def test_skips_files_without_handle_command(self, temp_test_dir: Path) -> None:
"""Should skip modules that do not define handle_command()."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
(modules_dir / "utility.py").write_text(
"def helper(): pass\n",
encoding="utf-8",
)
result = scan_module_files(str(temp_test_dir))
assert result == []
def test_skips_init_and_main(self, temp_test_dir: Path) -> None:
"""Should exclude __init__.py and __main__.py."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
(modules_dir / "__init__.py").write_text(
"def handle_command(): pass\n",
encoding="utf-8",
)
(modules_dir / "__main__.py").write_text(
"def handle_command(): pass\n",
encoding="utf-8",
)
(modules_dir / "real.py").write_text(
'"""Real module."""\ndef handle_command(command=None, args=None): pass\n',
encoding="utf-8",
)
result = scan_module_files(str(temp_test_dir))
names = [c["name"] for c in result]
assert "__init__" not in names
assert "__main__" not in names
assert "real" in names
def test_returns_empty_when_no_modules_dir(self, temp_test_dir: Path) -> None:
"""Should return empty list when apps/modules/ does not exist."""
result = scan_module_files(str(temp_test_dir))
assert result == []
def test_extracts_module_description(self, temp_test_dir: Path) -> None:
"""Should extract the first line of the module docstring."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
(modules_dir / "config.py").write_text(
'"""Configuration management for the branch."""\n'
"def handle_command(command=None, args=None): pass\n",
encoding="utf-8",
)
result = scan_module_files(str(temp_test_dir))
assert len(result) == 1
assert result[0]["description"] == "Configuration management for the branch."
def test_results_are_sorted(self, temp_test_dir: Path) -> None:
"""Results should be sorted alphabetically by module name."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
for name in ["zebra", "apple", "mango"]:
(modules_dir / f"{name}.py").write_text(
f'"""{name} module."""\ndef handle_command(): pass\n',
encoding="utf-8",
)
result = scan_module_files(str(temp_test_dir))
names = [c["name"] for c in result]
assert names == sorted(names)
# =============================================================================
# scan_branch tests
# =============================================================================
class TestScanBranch:
"""Tests for scan_branch()."""
def test_merges_help_and_module_results(self, temp_test_dir: Path) -> None:
"""Should combine help and module results, deduplicating by name."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
modules_dir = apps_dir / "modules"
modules_dir.mkdir()
(modules_dir / "extra.py").write_text(
'"""Extra module."""\ndef handle_command(): pass\n',
encoding="utf-8",
)
mock_result = MagicMock()
mock_result.stdout = b"Commands:\n audit Run audit\n\n"
mock_result.stderr = b""
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
return_value=mock_result,
):
result = scan_branch(str(temp_test_dir), "mybranch")
names = [c["name"] for c in result]
assert "audit" in names
assert "extra" in names
def test_help_wins_on_duplicate(self, temp_test_dir: Path) -> None:
"""When a command appears in both help and module, help source should win."""
apps_dir = temp_test_dir / "apps"
apps_dir.mkdir(parents=True)
(apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
modules_dir = apps_dir / "modules"
modules_dir.mkdir()
# Module named "audit" -- same as a help command
(modules_dir / "audit.py").write_text(
'"""Audit module."""\ndef handle_command(): pass\n',
encoding="utf-8",
)
mock_result = MagicMock()
mock_result.stdout = b"Commands:\n audit Run audit from help\n\n"
mock_result.stderr = b""
with patch(
"aipass.drone.apps.handlers.scanning.scanner.subprocess.run",
return_value=mock_result,
):
result = scan_branch(str(temp_test_dir), "mybranch")
audit_cmds = [c for c in result if c["name"] == "audit"]
assert len(audit_cmds) == 1
assert audit_cmds[0]["source"] == "help"
def test_returns_empty_when_nothing_found(self, temp_test_dir: Path) -> None:
"""Should return empty list when no commands found anywhere."""
result = scan_branch(str(temp_test_dir), "empty")
assert result == []
def test_results_sorted_by_name(self, temp_test_dir: Path) -> None:
"""Results should be sorted alphabetically."""
modules_dir = temp_test_dir / "apps" / "modules"
modules_dir.mkdir(parents=True)
for name in ["zebra", "apple"]:
(modules_dir / f"{name}.py").write_text(
f'"""{name}."""\ndef handle_command(): pass\n',
encoding="utf-8",
)
result = scan_branch(str(temp_test_dir), "test")
names = [c["name"] for c in result]
assert names == sorted(names)
# =============================================================================
# format_scan_results tests
# =============================================================================
class TestFormatScanResults:
"""Tests for format_scan_results()."""
def test_produces_output(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should produce console output with command names."""
commands = [
{"name": "audit", "description": "Run audit", "source": "help"},
{"name": "list", "description": "List items", "source": "module"},
]
format_scan_results("testbranch", commands)
captured = capsys.readouterr()
assert "audit" in captured.out
assert "list" in captured.out
assert "@testbranch" in captured.out
assert "command" in captured.out and "discovered" in captured.out
def test_adds_at_prefix_if_missing(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should add @ prefix for display when missing."""
format_scan_results("mybranch", [{"name": "x", "description": "", "source": "help"}])
captured = capsys.readouterr()
assert "@mybranch" in captured.out
def test_preserves_at_prefix(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should not double the @ prefix."""
format_scan_results("@mybranch", [{"name": "x", "description": "", "source": "help"}])
captured = capsys.readouterr()
assert "@mybranch" in captured.out
assert "@@mybranch" not in captured.out
# =============================================================================
# format_no_commands tests
# =============================================================================
class TestFormatNoCommands:
"""Tests for format_no_commands()."""
def test_produces_output(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should produce informational console output."""
format_no_commands("emptybranch")
captured = capsys.readouterr()
assert "No commands discovered" in captured.out
assert "@emptybranch" in captured.out
def test_adds_at_prefix_if_missing(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Should add @ prefix for display."""
format_no_commands("mybranch")
captured = capsys.readouterr()
assert "@mybranch" in captured.out
# =============================================================================
# scan module handle_command tests
# =============================================================================
class TestScanHandleCommand:
"""Tests for scan.handle_command() routing."""
@patch("aipass.drone.apps.modules.scan.scan")
def test_routes_to_scan(self, mock_scan: MagicMock) -> None:
"""Should call scan() with the target argument."""
from aipass.drone.apps.modules.scan import handle_command
mock_scan.return_value = [{"name": "x", "description": "", "source": "help"}]
result = handle_command(command=None, args=["@testbranch"])
assert result is True
mock_scan.assert_called_once_with("@testbranch")
@patch("aipass.drone.apps.modules.scan.scan")
def test_returns_false_when_scan_fails(self, mock_scan: MagicMock) -> None:
"""Should return False when scan returns None (resolution failure)."""
from aipass.drone.apps.modules.scan import handle_command
mock_scan.return_value = None
result = handle_command(command=None, args=["@nonexistent"])
assert result is False
def test_no_args_shows_introspection(self) -> None:
"""Should call print_introspection when command is None and no args."""
from aipass.drone.apps.modules.scan import handle_command
with patch("aipass.drone.apps.modules.scan.print_introspection") as mock_intro:
result = handle_command(command=None, args=None)
assert result is True
mock_intro.assert_called_once()
def test_empty_args_returns_false(self) -> None:
"""Should return False when command is given but no args."""
from aipass.drone.apps.modules.scan import handle_command
result = handle_command(command="scan", args=[])
assert result is False
# =============================================================================
# scan module scan() tests
# =============================================================================
class TestScanFunction:
"""Tests for scan.scan() orchestration."""
@patch("aipass.drone.apps.modules.scan.resolve_branch")
@patch("aipass.drone.apps.modules.scan.scan_branch")
@patch("aipass.drone.apps.modules.scan.format_scan_results")
def test_resolves_and_scans(
self,
mock_format: MagicMock,
mock_scan_branch: MagicMock,
mock_resolve: MagicMock,
) -> None:
"""Should resolve target, scan, format, and return results."""
from aipass.drone.apps.modules.scan import scan
mock_resolve.return_value = "/fake/path"
mock_scan_branch.return_value = [
{"name": "audit", "description": "Run audit", "source": "help"},
]
result = scan("@testbranch")
mock_resolve.assert_called_once_with("@testbranch")
mock_scan_branch.assert_called_once_with("/fake/path", "testbranch")
mock_format.assert_called_once()
assert result is not None
assert len(result) == 1
@patch("aipass.drone.apps.modules.scan.resolve_branch")
@patch("aipass.drone.apps.modules.scan.scan_branch")
@patch("aipass.drone.apps.modules.scan.format_no_commands")
def test_shows_no_commands_message(
self,
mock_format_none: MagicMock,
mock_scan_branch: MagicMock,
mock_resolve: MagicMock,
) -> None:
"""Should display no-commands message when scan finds nothing."""
from aipass.drone.apps.modules.scan import scan
mock_resolve.return_value = "/fake/path"
mock_scan_branch.return_value = []
result = scan("@emptybranch")
mock_format_none.assert_called_once()
assert result is not None # Empty list, not None
assert len(result) == 0
@patch("aipass.drone.apps.modules.scan.resolve_branch", side_effect=Exception("not found"))
def test_returns_none_on_resolution_failure(self, mock_resolve: MagicMock) -> None:
"""Should return None when branch resolution fails."""
from aipass.drone.apps.modules.scan import scan
result = scan("@nonexistent")
assert result is None