diff --git a/src/aipass/drone/README.md b/src/aipass/drone/README.md index 3da4440c..7b43d6be 100644 --- a/src/aipass/drone/README.md +++ b/src/aipass/drone/README.md @@ -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 # 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 + diff --git a/src/aipass/drone/apps/drone.py b/src/aipass/drone/apps/drone.py index 957a090f..9c1743bc 100644 --- a/src/aipass/drone/apps/drone.py +++ b/src/aipass/drone/apps/drone.py @@ -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 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 ` -- 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.") diff --git a/src/aipass/drone/apps/handlers/command_registry/__init__.py b/src/aipass/drone/apps/handlers/command_registry/__init__.py new file mode 100644 index 00000000..ad555bfc --- /dev/null +++ b/src/aipass/drone/apps/handlers/command_registry/__init__.py @@ -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``). +""" diff --git a/src/aipass/drone/apps/handlers/command_registry/formatters.py b/src/aipass/drone/apps/handlers/command_registry/formatters.py new file mode 100644 index 00000000..0ee06a9e --- /dev/null +++ b/src/aipass/drone/apps/handlers/command_registry/formatters.py @@ -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() diff --git a/src/aipass/drone/apps/handlers/command_registry/lookup.py b/src/aipass/drone/apps/handlers/command_registry/lookup.py new file mode 100644 index 00000000..3f68c57c --- /dev/null +++ b/src/aipass/drone/apps/handlers/command_registry/lookup.py @@ -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 [] diff --git a/src/aipass/drone/apps/handlers/command_registry/ops.py b/src/aipass/drone/apps/handlers/command_registry/ops.py new file mode 100644 index 00000000..10c90f88 --- /dev/null +++ b/src/aipass/drone/apps/handlers/command_registry/ops.py @@ -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"] diff --git a/src/aipass/drone/apps/handlers/scanning/__init__.py b/src/aipass/drone/apps/handlers/scanning/__init__.py new file mode 100644 index 00000000..865174b2 --- /dev/null +++ b/src/aipass/drone/apps/handlers/scanning/__init__.py @@ -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", +] diff --git a/src/aipass/drone/apps/handlers/scanning/formatters.py b/src/aipass/drone/apps/handlers/scanning/formatters.py new file mode 100644 index 00000000..8d39477b --- /dev/null +++ b/src/aipass/drone/apps/handlers/scanning/formatters.py @@ -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() diff --git a/src/aipass/drone/apps/handlers/scanning/scanner.py b/src/aipass/drone/apps/handlers/scanning/scanner.py new file mode 100644 index 00000000..5e5ffa63 --- /dev/null +++ b/src/aipass/drone/apps/handlers/scanning/scanner.py @@ -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 diff --git a/src/aipass/drone/apps/modules/commands.py b/src/aipass/drone/apps/modules/commands.py new file mode 100644 index 00000000..31449c42 --- /dev/null +++ b/src/aipass/drone/apps/modules/commands.py @@ -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: [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 [args...] [--desc=...] [--branch=...]") + console.print(" remove Remove a custom command") + console.print(" list [branch] List commands (optionally by branch)") + console.print(" lookup 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) diff --git a/src/aipass/drone/apps/modules/scan.py b/src/aipass/drone/apps/modules/scan.py new file mode 100644 index 00000000..3b780b99 --- /dev/null +++ b/src/aipass/drone/apps/modules/scan.py @@ -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 diff --git a/src/aipass/drone/tests/test_activation.py b/src/aipass/drone/tests/test_activation.py new file mode 100644 index 00000000..7646a33e --- /dev/null +++ b/src/aipass/drone/tests/test_activation.py @@ -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 `` 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, + ) diff --git a/src/aipass/drone/tests/test_commands.py b/src/aipass/drone/tests/test_commands.py new file mode 100644 index 00000000..09c8f08e --- /dev/null +++ b/src/aipass/drone/tests/test_commands.py @@ -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 diff --git a/src/aipass/drone/tests/test_scan.py b/src/aipass/drone/tests/test_scan.py new file mode 100644 index 00000000..e87ee466 --- /dev/null +++ b/src/aipass/drone/tests/test_scan.py @@ -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