From 78ae5f63b60b3c30f450946e39843973adf960a1 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 1 Mar 2026 12:37:05 -0800 Subject: [PATCH 1/2] feat(routing): add Routing & Discovery module v1.0.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete implementation of symbolic @branch addressing for multi-agent systems. Phase 1 — Core Routing: - Registry management (BRANCH_REGISTRY.json) - Symbolic resolution (@branch → absolute path) - Branch lifecycle (register, list, exists, info) - Configuration (env vars, custom paths) Phase 2 — Command Routing: - route_command() — execute commands on remote branches - Safe subprocess execution (no shell=True) - Module discovery (--help parsing + filesystem scan) Phase 3 — Help System & Release Prep: - HelpResult dataclass (structured help output) - get_system_help() — aggregate help across all branches - route_all() — broadcast commands to all active branches - Complete API documentation (docs/api.md) - pyproject.toml configured for PyPI (v1.0.0) - trinity-pattern moved to optional dependency 106 tests, all passing. Ready for PyPI publication. Co-Authored-By: Claude Opus 4.6 --- README.md | 98 ++++++ pyproject.toml | 29 +- src/aipass/routing/__init__.py | 97 ++++++ src/aipass/routing/config.py | 55 +++ src/aipass/routing/discovery.py | 278 +++++++++++++++ src/aipass/routing/exceptions.py | 50 +++ src/aipass/routing/executor.py | 97 ++++++ src/aipass/routing/registry.py | 260 ++++++++++++++ src/aipass/routing/resolver.py | 117 +++++++ src/aipass/routing/router.py | 159 +++++++++ tests/test_routing_core.py | 391 +++++++++++++++++++++ tests/test_routing_integration.py | 504 +++++++++++++++++++++++++++ tests/test_routing_phase2.py | 546 ++++++++++++++++++++++++++++++ 13 files changed, 2678 insertions(+), 3 deletions(-) create mode 100644 src/aipass/routing/__init__.py create mode 100644 src/aipass/routing/config.py create mode 100644 src/aipass/routing/discovery.py create mode 100644 src/aipass/routing/exceptions.py create mode 100644 src/aipass/routing/executor.py create mode 100644 src/aipass/routing/registry.py create mode 100644 src/aipass/routing/resolver.py create mode 100644 src/aipass/routing/router.py create mode 100644 tests/test_routing_core.py create mode 100644 tests/test_routing_integration.py create mode 100644 tests/test_routing_phase2.py diff --git a/README.md b/README.md index 567845ce..26ebe759 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,104 @@ pip install aipass > **Note:** This package is not yet published to PyPI. This repo is in early development. +## Features + +### Routing & Discovery (v1.0) + +Symbolic addressing for multi-agent systems. Instead of hard-coded paths, agents use `@branch` symbolic names that resolve to actual locations at runtime. + +**Quick Start:** + +```python +from aipass.routing import initialize_registry, register_branch, resolve_branch + +# Initialize registry (first time only) +initialize_registry() + +# Register your agents +register_branch("my_agent", "/path/to/agents/my_agent", branch_type="agent") +register_branch("researcher", "/path/to/agents/researcher", branch_type="agent") +register_branch("monitor", "/path/to/services/monitor", branch_type="service") + +# Resolve symbolic names to paths +agent_path = resolve_branch("@my_agent") +# Returns: "/path/to/agents/my_agent" + +# Works with or without @ prefix +researcher_path = resolve_branch("researcher") +# Returns: "/path/to/agents/researcher" +``` + +**Discovery:** + +```python +from aipass.routing import list_branches, branch_exists, get_branch_info + +# Check if a branch exists +if branch_exists("@my_agent"): + print("Agent found!") + +# List all registered branches +all_branches = list_branches() +# Returns: ["@my_agent", "@researcher", "@monitor"] + +# List branches by type +agents_only = list_branches(branch_type="agent") +# Returns: ["@my_agent", "@researcher"] + +# Get full branch metadata +info = get_branch_info("@my_agent") +# Returns: { +# "name": "my_agent", +# "path": "/path/to/agents/my_agent", +# "type": "agent", +# "status": "active", +# "created": "2026-03-01T10:00:00Z" +# } +``` + +**Configuration:** + +By default, the registry is stored at `~/.aipass/BRANCH_REGISTRY.json`. You can customize this: + +```python +from aipass.routing import set_registry_path + +# Set custom registry location +set_registry_path("/custom/path/to/registry.json") +``` + +Or via environment variable: + +```bash +export AIPASS_REGISTRY_PATH=/custom/path/to/registry.json +``` + +**Integration with Trinity Pattern:** + +```python +from trinity_pattern import Agent +from aipass.routing import resolve_branch + +# Before: hard-coded paths +agent = Agent(directory="/home/user/agents/my_agent") + +# After: symbolic addressing +agent_dir = resolve_branch("@my_agent") +agent = Agent(directory=agent_dir) +``` + +**Error Handling:** + +```python +from aipass.routing import resolve_branch, BranchNotFoundError + +try: + path = resolve_branch("@nonexistent") +except BranchNotFoundError as e: + print(f"Branch not found: {e}") +``` + ## License MIT diff --git a/pyproject.toml b/pyproject.toml index fca81599..ffe6afec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "aipass" -version = "0.1.0-dev" +version = "1.0.0" description = "Orchestration framework for autonomous AI agent ecosystems" readme = "README.md" license = "MIT" @@ -12,15 +12,38 @@ requires-python = ">=3.10" authors = [ { name = "AIPass", email = "aipass.system@gmail.com" }, ] -dependencies = [ - "trinity-pattern>=1.0.0", +keywords = ["ai", "agents", "routing", "discovery", "multi-agent", "orchestration"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Software Development :: Libraries :: Python Modules", + "Topic :: Scientific/Engineering :: Artificial Intelligence", ] +dependencies = [] + +[project.urls] +Homepage = "https://github.com/AIOSAI/AIPass" +Repository = "https://github.com/AIOSAI/AIPass" +"Bug Tracker" = "https://github.com/AIOSAI/AIPass/issues" [project.optional-dependencies] +trinity = [ + "trinity-pattern>=1.0.0", +] dev = [ "pytest", + "pytest-cov", "ruff", ] +[tool.pytest.ini_options] +testpaths = ["tests"] + [tool.ruff] line-length = 120 diff --git a/src/aipass/routing/__init__.py b/src/aipass/routing/__init__.py new file mode 100644 index 00000000..7ae3ede0 --- /dev/null +++ b/src/aipass/routing/__init__.py @@ -0,0 +1,97 @@ +""" +AIPass Routing & Discovery Module. + +Provides symbolic addressing for multi-agent systems. Resolves @branch names +to absolute paths at runtime. + +Core API: + resolve_branch(name) -> str # @branch → absolute path + register_branch(name, path, type) # Add branch to registry + list_branches(type, status) # List all branches + branch_exists(name) -> bool # Check if branch exists + get_branch_info(name) -> dict # Get branch metadata + +Command Routing (Phase 2): + route_command(target, command, args, timeout) -> CommandResult + discover_modules(target) -> list # Available commands for a branch + get_help(target, command) -> HelpResult # Structured help for a branch + +@all Operations (Phase 3): + route_all(command, args, timeout) -> dict[str, CommandResult] + get_system_help() -> dict[str, HelpResult] # Help across all branches + +Registry Management: + initialize_registry() # Create empty registry + set_registry_path(path) # Set custom registry location + get_registry_path() -> Path # Get current registry path + +Example: + >>> from aipass.routing import resolve_branch, register_branch + >>> register_branch("my_agent", "/path/to/agent", "agent") + >>> path = resolve_branch("@my_agent") + >>> print(path) + /path/to/agent + + >>> from aipass.routing import route_command + >>> result = route_command("@my_agent", "status") + >>> print(result.stdout) + + >>> from aipass.routing import route_all + >>> results = route_all("status") + >>> for branch, r in results.items(): + ... print(f"{branch}: exit={r.exit_code}") +""" + +from .config import get_registry_path, reset_registry_path, set_registry_path +from .discovery import HelpResult, discover_modules, get_help, get_system_help +from .exceptions import ( + BranchAlreadyExistsError, + BranchNotFoundError, + CommandExecutionError, + InvalidPathError, + RegistryCorruptError, + RegistryError, + RegistryNotFoundError, + RegistryPermissionError, + RoutingError, +) +from .executor import CommandResult +from .registry import add_branch as register_branch +from .registry import initialize_registry +from .resolver import branch_exists, get_branch_info, list_branches, resolve_branch +from .router import route_all, route_command + +__version__ = "1.0.0" + +__all__ = [ + # Core API + "resolve_branch", + "register_branch", + "list_branches", + "branch_exists", + "get_branch_info", + # Command routing (Phase 2) + "route_command", + "discover_modules", + "get_help", + "CommandResult", + # Help & discovery (Phase 3) + "HelpResult", + "get_system_help", + "route_all", + # Registry management + "initialize_registry", + "set_registry_path", + "get_registry_path", + "reset_registry_path", + # Exceptions + "RoutingError", + "BranchNotFoundError", + "BranchAlreadyExistsError", + "InvalidPathError", + "RegistryError", + "RegistryNotFoundError", + "RegistryCorruptError", + "RegistryPermissionError", + "CommandExecutionError", +] diff --git a/src/aipass/routing/config.py b/src/aipass/routing/config.py new file mode 100644 index 00000000..36292a5b --- /dev/null +++ b/src/aipass/routing/config.py @@ -0,0 +1,55 @@ +""" +Registry configuration management. + +Handles registry path configuration with environment variable support. +""" + +import os +from pathlib import Path +from typing import Optional + + +_registry_path: Optional[Path] = None + + +def get_registry_path() -> Path: + """ + Get the current registry path. + + Priority: + 1. Explicitly set path via set_registry_path() + 2. AIPASS_REGISTRY_PATH environment variable + 3. Default: ~/.aipass/BRANCH_REGISTRY.json + + Returns: + Path to the registry file + """ + global _registry_path + + if _registry_path is not None: + return _registry_path + + env_path = os.environ.get("AIPASS_REGISTRY_PATH") + if env_path: + return Path(env_path) + + return Path.home() / ".aipass" / "BRANCH_REGISTRY.json" + + +def set_registry_path(path: str | Path) -> None: + """ + Set a custom registry path. + + Args: + path: Path to the registry file + """ + global _registry_path + _registry_path = Path(path) + + +def reset_registry_path() -> None: + """ + Reset registry path to default (useful for testing). + """ + global _registry_path + _registry_path = None diff --git a/src/aipass/routing/discovery.py b/src/aipass/routing/discovery.py new file mode 100644 index 00000000..ec592d0b --- /dev/null +++ b/src/aipass/routing/discovery.py @@ -0,0 +1,278 @@ +""" +Module and command discovery for AIPass branch introspection. + +Introspects branch capabilities by querying entry points for help text +and scanning module directories as a fallback. +""" + +import logging +import subprocess +from dataclasses import dataclass, field +from pathlib import Path +from typing import Dict, List, Optional + +from .exceptions import CommandExecutionError +from .resolver import list_branches, resolve_branch + +logger = logging.getLogger(__name__) + + +@dataclass +class HelpResult: + """ + Structured result from a help query. + + Attributes: + branch: The branch name the help was retrieved from. + command: The specific command queried, or None for top-level help. + text: Raw help text (stdout or stderr from the entry point). + commands_found: List of command names parsed from the help text. + """ + + branch: str + command: Optional[str] + text: str + commands_found: List[str] = field(default_factory=list) + + +def _get_entry_point(branch_path: str, branch_name: str) -> Optional[Path]: + """ + Return the apps/{branch_name}.py entry point path if it exists, else None. + + Args: + branch_path: Absolute path to the branch directory. + branch_name: Lowercase branch name. + + Returns: + Path to the entry point, or None if it does not exist. + """ + entry_point = Path(branch_path) / "apps" / f"{branch_name}.py" + return entry_point if entry_point.exists() else None + + +def _scan_modules_directory(branch_path: str) -> List[str]: + """ + Scan apps/modules/ for .py files and return their stems as command names. + + Used as a fallback when the entry point's --help output cannot be parsed. + + Args: + branch_path: Absolute path to the branch directory. + + Returns: + Sorted list of module stem names (without .py extension), excluding + __init__ and __main__. + """ + modules_dir = Path(branch_path) / "apps" / "modules" + if not modules_dir.is_dir(): + return [] + + excluded = {"__init__", "__main__"} + return sorted( + f.stem + for f in modules_dir.glob("*.py") + if f.stem not in excluded + ) + + +def _parse_help_for_commands(help_text: str) -> List[str]: + """ + Parse --help output to extract a list of available commands. + + Handles multiple CLI framework formats: + - Section-based (custom CLIs, argparse subparsers): looks for headers + labelled "commands", "subcommands", or "available commands". + - Click-style: looks for indented entries under "Commands:" or "Options:". + - Positional args listing: lines matching ``{command}`` or ``[command]`` + at the start of usage lines. + + Falls back to returning an empty list if no recognisable pattern is found. + + Args: + help_text: Raw stdout from ``python3 entry.py --help``. + + Returns: + List of command names parsed from the help text. + """ + commands: List[str] = [] + in_commands_section = False + + section_markers = {"commands", "subcommands", "available commands"} + + for line in help_text.splitlines(): + stripped = line.strip() + + # Detect section header lines (e.g. "Commands:", "Available commands:") + if any(marker in stripped.lower() for marker in section_markers): + in_commands_section = True + continue + + # A blank line ends the current section. + if in_commands_section and not stripped: + in_commands_section = False + continue + + if in_commands_section: + # Indented lines that start with a word are command entries. + if line.startswith((" ", "\t")) and stripped: + # Take the first token as the command name. + token = stripped.split()[0] + # Skip lines that look like option flags. + if not token.startswith("-"): + commands.append(token) + + return commands + + +def discover_modules(target: str) -> List[str]: + """ + Discover available commands for a branch. + + Resolution order: + 1. Resolve @target to its absolute path via the registry. + 2. Attempt to run ``python3 apps/{name}.py --help`` and parse the output + for a commands section. + 3. If parsing yields nothing, fall back to scanning ``apps/modules/`` for + .py file stems. + + Args: + target: Symbolic branch name with or without @ prefix (e.g., "@flow"). + + Returns: + List of discovered command/module names (may be empty). + + Raises: + BranchNotFoundError: If the target branch is not in the registry. + """ + # Step 1: resolve — BranchNotFoundError propagates to caller. + branch_path = resolve_branch(target) + branch_name = target.lstrip("@").lower() + + # Step 2: try --help on the entry point. + entry_point = _get_entry_point(branch_path, branch_name) + if entry_point is not None: + try: + result = subprocess.run( + ["python3", 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") + + commands = _parse_help_for_commands(help_text) + if commands: + return commands + except (subprocess.TimeoutExpired, OSError): + pass + + # Step 3: fallback — scan modules directory. + return _scan_modules_directory(branch_path) + + +def get_help(target: str, command: Optional[str] = None) -> HelpResult: + """ + Get structured help for a branch or a specific command. + + Executes: + python3 apps/{name}.py --help (when command is None) + python3 apps/{name}.py {command} --help (when command is given) + + Args: + target: Symbolic branch name with or without @ prefix (e.g., "@flow"). + command: Specific command to get help for (default: None). + + Returns: + HelpResult with branch, command, text, and commands_found populated. + + Raises: + BranchNotFoundError: If the target branch is not in the registry. + CommandExecutionError: If the entry point does not exist or execution + fails. + """ + branch_path = resolve_branch(target) + branch_name = target.lstrip("@").lower() + + entry_point = _get_entry_point(branch_path, branch_name) + if entry_point is None: + raise CommandExecutionError( + f"Entry point not found for branch '{branch_name}': " + f"{Path(branch_path) / 'apps' / (branch_name + '.py')}" + ) + + relative_entry = str(entry_point.relative_to(branch_path)) + if command is None: + cmd_args = [relative_entry, "--help"] + else: + cmd_args = [relative_entry, command, "--help"] + + try: + result = subprocess.run( + ["python3"] + cmd_args, + cwd=branch_path, + capture_output=True, + timeout=10, + shell=False, + ) + except subprocess.TimeoutExpired as e: + raise CommandExecutionError( + f"Help command timed out for branch '{branch_name}'" + ) from e + except OSError as e: + raise CommandExecutionError( + f"OS error getting help for branch '{branch_name}': {e}" + ) from e + + stdout = result.stdout.decode("utf-8", errors="replace") + stderr = result.stderr.decode("utf-8", errors="replace") + + # Many CLIs write help to stderr; return whichever is non-empty. + text = stdout if stdout.strip() else stderr + commands_found = _parse_help_for_commands(text) + + return HelpResult( + branch=branch_name, + command=command, + text=text, + commands_found=commands_found, + ) + + +def get_system_help() -> Dict[str, HelpResult]: + """ + Aggregate help across all active branches in the registry. + + Iterates over every active branch and attempts to retrieve top-level help + via ``get_help()``. Branches whose entry point is missing or whose help + command fails are silently skipped (the error is logged at DEBUG level). + + Returns: + Mapping of branch name (without @ prefix) to its HelpResult. + Branches that cannot be queried are omitted from the result. + + Example: + >>> results = get_system_help() + >>> for branch, help_result in results.items(): + ... print(f"{branch}: {help_result.commands_found}") + """ + results: Dict[str, HelpResult] = {} + + # list_branches returns names with @ prefix; iterate all active branches. + active_branches = list_branches(status="active") + + for symbolic_name in active_branches: + branch_name = symbolic_name.lstrip("@") + try: + help_result = get_help(symbolic_name) + results[branch_name] = help_result + except Exception as exc: + logger.debug( + "get_system_help: skipping branch '%s': %s", + branch_name, + exc, + ) + + return results diff --git a/src/aipass/routing/exceptions.py b/src/aipass/routing/exceptions.py new file mode 100644 index 00000000..619c4ff5 --- /dev/null +++ b/src/aipass/routing/exceptions.py @@ -0,0 +1,50 @@ +""" +Routing module custom exceptions. + +Defines the exception hierarchy for routing and branch resolution errors. +""" + + +class RoutingError(Exception): + """Base exception for all routing-related errors.""" + pass + + +class BranchNotFoundError(RoutingError): + """Raised when a branch cannot be found in the registry.""" + pass + + +class BranchAlreadyExistsError(RoutingError): + """Raised when attempting to register a branch that already exists.""" + pass + + +class InvalidPathError(RoutingError): + """Raised when a path is invalid or doesn't exist.""" + pass + + +class RegistryError(RoutingError): + """Base exception for registry-related errors.""" + pass + + +class RegistryNotFoundError(RegistryError): + """Raised when the registry file doesn't exist.""" + pass + + +class RegistryCorruptError(RegistryError): + """Raised when the registry file is corrupted or invalid JSON.""" + pass + + +class RegistryPermissionError(RegistryError): + """Raised when there are permission issues accessing the registry.""" + pass + + +class CommandExecutionError(RoutingError): + """Raised when command execution fails (Phase 2 feature).""" + pass diff --git a/src/aipass/routing/executor.py b/src/aipass/routing/executor.py new file mode 100644 index 00000000..10568e4f --- /dev/null +++ b/src/aipass/routing/executor.py @@ -0,0 +1,97 @@ +""" +Safe subprocess execution for branch command routing. + +Wraps subprocess.run with safety guards: timeout enforcement, no shell injection, +captured output, and consistent error wrapping via CommandExecutionError. + +Defines CommandResult, the shared return type for all routing operations. +""" + +import subprocess +from dataclasses import dataclass +from typing import List + +from .exceptions import CommandExecutionError + + +@dataclass +class CommandResult: + """ + Result of a routed command execution. + + Attributes: + stdout: Captured standard output from the command. + stderr: Captured standard error from the command. + exit_code: Process exit code (0 typically indicates success). + branch: The branch name the command was routed to. + command: The command string that was executed. + """ + + stdout: str + stderr: str + exit_code: int + branch: str + command: str + + +def execute_command( + executable: str, + args: List[str], + cwd: str, + timeout: int = 30, +) -> CommandResult: + """ + Execute a command via subprocess with safety guards. + + Args: + executable: The executable to run (e.g., "python3"). + args: List of arguments (e.g., ["apps/flow.py", "status"]). + cwd: Working directory for command execution (absolute path). + timeout: Maximum execution time in seconds (default: 30). + + Returns: + CommandResult with stdout, stderr, exit_code populated. + branch and command fields are empty strings; route_command populates them. + + Raises: + CommandExecutionError: If the process times out, the executable is not + found, or any other OS-level error prevents execution. + + Notes: + - Never uses shell=True to prevent shell injection attacks. + - Always captures stdout and stderr separately. + - Decodes output as UTF-8, replacing undecodable bytes. + """ + full_cmd = [executable] + list(args) + + try: + result = subprocess.run( + full_cmd, + cwd=cwd, + capture_output=True, + timeout=timeout, + shell=False, + ) + except subprocess.TimeoutExpired as e: + raise CommandExecutionError( + f"Command timed out after {timeout}s: {' '.join(full_cmd)}" + ) from e + except FileNotFoundError as e: + raise CommandExecutionError( + f"Executable not found: {executable!r}" + ) from e + except OSError as e: + raise CommandExecutionError( + f"OS error executing command: {e}" + ) from e + + stdout = result.stdout.decode("utf-8", errors="replace") + stderr = result.stderr.decode("utf-8", errors="replace") + + return CommandResult( + stdout=stdout, + stderr=stderr, + exit_code=result.returncode, + branch="", + command="", + ) diff --git a/src/aipass/routing/registry.py b/src/aipass/routing/registry.py new file mode 100644 index 00000000..ec1104d6 --- /dev/null +++ b/src/aipass/routing/registry.py @@ -0,0 +1,260 @@ +""" +Registry operations for branch management. + +Handles loading, saving, and managing the BRANCH_REGISTRY.json file. +""" + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .config import get_registry_path +from .exceptions import ( + BranchAlreadyExistsError, + InvalidPathError, + RegistryCorruptError, + RegistryNotFoundError, + RegistryPermissionError, +) + + +def load_registry() -> Dict[str, Any]: + """ + Load the branch registry from disk. + + Returns: + Registry dictionary with version, branches, and metadata + + Raises: + RegistryNotFoundError: If registry file doesn't exist + RegistryCorruptError: If registry file is invalid JSON + RegistryPermissionError: If registry file cannot be read + """ + registry_path = get_registry_path() + + if not registry_path.exists(): + raise RegistryNotFoundError( + f"Registry not found at {registry_path}. " + "Run initialize_registry() to create one." + ) + + try: + with open(registry_path, "r", encoding="utf-8") as f: + data = json.load(f) + except PermissionError as e: + raise RegistryPermissionError(f"Permission denied reading registry: {e}") + except json.JSONDecodeError as e: + raise RegistryCorruptError(f"Registry file is corrupted: {e}") + except Exception as e: + raise RegistryCorruptError(f"Failed to read registry: {e}") + + # Validate structure + if not isinstance(data, dict): + raise RegistryCorruptError("Registry must be a JSON object") + + if "branches" not in data: + raise RegistryCorruptError("Registry missing 'branches' field") + + return data + + +def save_registry(registry: Dict[str, Any]) -> None: + """ + Save the branch registry to disk. + + Args: + registry: Registry dictionary to save + + Raises: + RegistryPermissionError: If registry file cannot be written + """ + registry_path = get_registry_path() + + # Ensure directory exists + registry_path.parent.mkdir(parents=True, exist_ok=True) + + # Update metadata timestamp + if "metadata" not in registry: + registry["metadata"] = {} + + registry["metadata"]["last_updated"] = datetime.now(timezone.utc).isoformat() + registry["metadata"]["managed_by"] = "aipass.routing" + + try: + with open(registry_path, "w", encoding="utf-8") as f: + json.dump(registry, f, indent=2) + except PermissionError as e: + raise RegistryPermissionError(f"Permission denied writing registry: {e}") + except Exception as e: + raise RegistryPermissionError(f"Failed to write registry: {e}") + + +def initialize_registry() -> None: + """ + Create a new empty registry file. + + Creates the registry directory and file if they don't exist. + If registry already exists, does nothing. + """ + registry_path = get_registry_path() + + if registry_path.exists(): + return + + registry = { + "version": "1.0", + "branches": {}, + "metadata": { + "last_updated": datetime.now(timezone.utc).isoformat(), + "managed_by": "aipass.routing", + }, + } + + save_registry(registry) + + +def add_branch( + name: str, + path: str | Path, + branch_type: str = "agent", + status: str = "active", +) -> None: + """ + Add a branch to the registry. + + Args: + name: Branch name (without @ prefix) + path: Absolute path to branch directory + branch_type: Type of branch (agent, service, module, etc.) + status: Branch status (active, inactive, archived) + + Raises: + BranchAlreadyExistsError: If branch name already exists + InvalidPathError: If path doesn't exist or isn't a directory + RegistryNotFoundError: If registry doesn't exist + """ + # Validate path + path_obj = Path(path).resolve() + if not path_obj.exists(): + raise InvalidPathError(f"Path does not exist: {path}") + if not path_obj.is_dir(): + raise InvalidPathError(f"Path is not a directory: {path}") + + # Load registry + try: + registry = load_registry() + except RegistryNotFoundError: + # Auto-initialize if registry doesn't exist + initialize_registry() + registry = load_registry() + + # Check if branch already exists + if name in registry["branches"]: + raise BranchAlreadyExistsError( + f"Branch '{name}' already exists in registry" + ) + + # Add branch entry + registry["branches"][name] = { + "name": name, + "path": str(path_obj), + "type": branch_type, + "status": status, + "created": datetime.now(timezone.utc).isoformat(), + } + + save_registry(registry) + + +def remove_branch(name: str) -> None: + """ + Remove a branch from the registry. + + Args: + name: Branch name (without @ prefix) + + Raises: + RegistryNotFoundError: If registry doesn't exist + """ + registry = load_registry() + + if name in registry["branches"]: + del registry["branches"][name] + save_registry(registry) + + +def update_branch_status(name: str, status: str) -> None: + """ + Update the status of a branch. + + Args: + name: Branch name (without @ prefix) + status: New status (active, inactive, archived) + + Raises: + RegistryNotFoundError: If registry doesn't exist + """ + registry = load_registry() + + if name in registry["branches"]: + registry["branches"][name]["status"] = status + registry["branches"][name]["last_seen"] = datetime.now(timezone.utc).isoformat() + save_registry(registry) + + +def get_all_branches( + branch_type: Optional[str] = None, + status: str = "active", +) -> List[Dict[str, Any]]: + """ + Get all branches from the registry, optionally filtered. + + Args: + branch_type: Filter by branch type (optional) + status: Filter by status (default: "active") + + Returns: + List of branch dictionaries + + Raises: + RegistryNotFoundError: If registry doesn't exist + """ + try: + registry = load_registry() + except RegistryNotFoundError: + return [] + + branches = registry.get("branches", {}).values() + + # Apply filters + filtered = [] + for branch in branches: + if status and branch.get("status") != status: + continue + if branch_type and branch.get("type") != branch_type: + continue + filtered.append(branch) + + return filtered + + +def get_branch_by_name(name: str) -> Optional[Dict[str, Any]]: + """ + Get a single branch by name. + + Args: + name: Branch name (without @ prefix) + + Returns: + Branch dictionary or None if not found + + Raises: + RegistryNotFoundError: If registry doesn't exist + """ + try: + registry = load_registry() + except RegistryNotFoundError: + return None + + return registry.get("branches", {}).get(name) diff --git a/src/aipass/routing/resolver.py b/src/aipass/routing/resolver.py new file mode 100644 index 00000000..d99d0597 --- /dev/null +++ b/src/aipass/routing/resolver.py @@ -0,0 +1,117 @@ +""" +Branch resolution logic. + +Resolves symbolic @branch names to absolute paths and metadata. +""" + +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .exceptions import BranchNotFoundError, RegistryNotFoundError +from .registry import get_all_branches, get_branch_by_name, load_registry + + +def normalize_branch_name(symbolic_name: str) -> str: + """ + Normalize a symbolic branch name. + + Strips @ prefix if present and returns clean branch name. + + Args: + symbolic_name: Branch name with or without @ prefix + + Returns: + Clean branch name without @ prefix + """ + if symbolic_name.startswith("@"): + return symbolic_name[1:] + return symbolic_name + + +def resolve_branch(symbolic_name: str) -> str: + """ + Resolve a symbolic branch name to its absolute path. + + Args: + symbolic_name: Branch name with or without @ prefix (e.g., "@my_agent" or "my_agent") + + Returns: + Absolute path to branch directory as string + + Raises: + BranchNotFoundError: If branch not in registry + RegistryNotFoundError: If registry file missing or corrupt + """ + # Let RegistryNotFoundError propagate from load_registry + registry = load_registry() + + name = normalize_branch_name(symbolic_name) + branch = registry.get("branches", {}).get(name) + + if branch is None: + raise BranchNotFoundError( + f"Branch '{symbolic_name}' not found in registry" + ) + + return branch["path"] + + +def branch_exists(symbolic_name: str) -> bool: + """ + Check if a branch exists in the registry. + + Args: + symbolic_name: Branch name with or without @ prefix + + Returns: + True if branch exists, False otherwise + """ + name = normalize_branch_name(symbolic_name) + branch = get_branch_by_name(name) + return branch is not None + + +def get_branch_info(symbolic_name: str) -> Dict[str, Any]: + """ + Get full metadata for a branch. + + Args: + symbolic_name: Branch name with or without @ prefix + + Returns: + Dictionary with branch metadata (name, path, type, status, timestamps) + + Raises: + BranchNotFoundError: If branch not in registry + RegistryNotFoundError: If registry file missing or corrupt + """ + # Let RegistryNotFoundError propagate from load_registry + registry = load_registry() + + name = normalize_branch_name(symbolic_name) + branch = registry.get("branches", {}).get(name) + + if branch is None: + raise BranchNotFoundError( + f"Branch '{symbolic_name}' not found in registry" + ) + + return branch + + +def list_branches( + branch_type: Optional[str] = None, + status: str = "active", +) -> List[str]: + """ + List all registered branches, optionally filtered by type and status. + + Args: + branch_type: Filter by branch type (optional) + status: Filter by status (default: "active") + + Returns: + List of branch names with @ prefix + """ + branches = get_all_branches(branch_type=branch_type, status=status) + return [f"@{branch['name']}" for branch in branches] diff --git a/src/aipass/routing/router.py b/src/aipass/routing/router.py new file mode 100644 index 00000000..f1130e21 --- /dev/null +++ b/src/aipass/routing/router.py @@ -0,0 +1,159 @@ +""" +Command routing logic for the AIPass routing module. + +Routes commands to branch entry points by resolving symbolic @branch names, +locating the branch's apps/{name}.py entry point, and executing via subprocess. +""" + +import logging +from pathlib import Path +from typing import Dict, List, Optional + +from .exceptions import CommandExecutionError +from .executor import CommandResult, execute_command +from .resolver import list_branches, resolve_branch + +logger = logging.getLogger(__name__) + + +def _find_entry_point(branch_path: str, branch_name: str) -> Path: + """ + Locate the apps/{branch_name}.py entry point for a branch. + + Args: + branch_path: Absolute path to the branch directory. + branch_name: Lowercase branch name from the registry. + + Returns: + Path to the entry point file. + + Raises: + CommandExecutionError: If the entry point does not exist. + """ + entry_point = Path(branch_path) / "apps" / f"{branch_name}.py" + if not entry_point.exists(): + raise CommandExecutionError( + f"Entry point not found for branch '{branch_name}': {entry_point}" + ) + return entry_point + + +def route_command( + target: str, + command: str, + args: Optional[List[str]] = None, + timeout: int = 30, +) -> CommandResult: + """ + Route a command to a branch's entry point. + + Resolves @target to an absolute path, locates the branch entry point at + {path}/apps/{branch_name}.py, then executes: + python3 apps/{name}.py {command} [args...] + + The process runs with cwd set to the branch root so relative paths inside + the entry point (e.g. imports of sibling modules) resolve correctly. + + Args: + target: Symbolic branch name with or without @ prefix (e.g., "@flow"). + command: Command to pass to the branch entry point (e.g., "status"). + args: Optional list of additional arguments (default: None). + timeout: Maximum execution time in seconds (default: 30). + + Returns: + CommandResult with stdout, stderr, exit_code, branch, and command. + + Raises: + BranchNotFoundError: If the target branch is not in the registry. + CommandExecutionError: If the entry point is missing or execution fails. + """ + if args is None: + args = [] + + # Step 1: Resolve @target → absolute path. BranchNotFoundError propagates. + branch_path = resolve_branch(target) + + # Derive the canonical branch name (strip leading @). + branch_name = target.lstrip("@").lower() + + # Step 2: Locate entry point. + entry_point = _find_entry_point(branch_path, branch_name) + + # Step 3: Build argument list and execute. + # Use a relative path to apps/{name}.py so the subprocess cwd is the branch root. + relative_entry = str(entry_point.relative_to(branch_path)) + cmd_args = [relative_entry, command] + list(args) + + result = execute_command( + executable="python3", + args=cmd_args, + cwd=branch_path, + timeout=timeout, + ) + + # Annotate result with routing context. + return CommandResult( + stdout=result.stdout, + stderr=result.stderr, + exit_code=result.exit_code, + branch=branch_name, + command=command, + ) + + +def route_all( + command: str, + args: Optional[List[str]] = None, + timeout: int = 30, +) -> Dict[str, CommandResult]: + """ + Route the same command to ALL active branches in the registry. + + Iterates over every active branch and calls ``route_command`` on each. + Individual branch failures (missing entry point, execution error) are + logged at WARNING level and recorded as a CommandResult with exit_code=-1 + and the error message in stderr. Processing continues regardless of + per-branch failures. + + Args: + command: Command to pass to every branch entry point (e.g., "status"). + args: Optional list of additional arguments forwarded to each branch. + timeout: Maximum execution time per branch in seconds (default: 30). + + Returns: + Dictionary mapping branch name (without @ prefix) to its CommandResult. + Branches that fail are included with exit_code=-1 and error in stderr. + + Example: + >>> results = route_all("status") + >>> for branch, result in results.items(): + ... print(f"{branch}: exit={result.exit_code}") + """ + if args is None: + args = [] + + results: Dict[str, CommandResult] = {} + + active_branches = list_branches(status="active") + + for symbolic_name in active_branches: + branch_name = symbolic_name.lstrip("@") + try: + result = route_command(symbolic_name, command, args=list(args), timeout=timeout) + results[branch_name] = result + except Exception as exc: + logger.warning( + "route_all: branch '%s' failed for command '%s': %s", + branch_name, + command, + exc, + ) + results[branch_name] = CommandResult( + stdout="", + stderr=str(exc), + exit_code=-1, + branch=branch_name, + command=command, + ) + + return results diff --git a/tests/test_routing_core.py b/tests/test_routing_core.py new file mode 100644 index 00000000..a9514768 --- /dev/null +++ b/tests/test_routing_core.py @@ -0,0 +1,391 @@ +""" +Unit tests for routing core functionality. + +Tests all Phase 1 routing functions with >80% coverage target. +""" + +import json +import tempfile +from pathlib import Path + +import pytest + +from aipass.routing import ( + BranchAlreadyExistsError, + BranchNotFoundError, + InvalidPathError, + RegistryNotFoundError, + branch_exists, + get_branch_info, + get_registry_path, + initialize_registry, + list_branches, + register_branch, + reset_registry_path, + resolve_branch, + set_registry_path, +) + + +@pytest.fixture +def temp_registry(tmp_path): + """Create a temporary registry for testing.""" + registry_path = tmp_path / "test_registry.json" + set_registry_path(registry_path) + initialize_registry() + yield registry_path + reset_registry_path() + + +@pytest.fixture +def sample_branches(tmp_path): + """Create sample branch directories for testing.""" + branches = {} + for name in ["agent1", "agent2", "service1"]: + branch_dir = tmp_path / "branches" / name + branch_dir.mkdir(parents=True, exist_ok=True) + branches[name] = branch_dir + return branches + + +class TestConfiguration: + """Test registry configuration.""" + + def test_default_registry_path(self): + """Test default registry path is ~/.aipass/BRANCH_REGISTRY.json.""" + reset_registry_path() + path = get_registry_path() + assert path == Path.home() / ".aipass" / "BRANCH_REGISTRY.json" + + def test_set_custom_registry_path(self, tmp_path): + """Test setting custom registry path.""" + custom_path = tmp_path / "custom_registry.json" + set_registry_path(custom_path) + assert get_registry_path() == custom_path + reset_registry_path() + + def test_reset_registry_path(self, tmp_path): + """Test resetting registry path to default.""" + custom_path = tmp_path / "custom_registry.json" + set_registry_path(custom_path) + reset_registry_path() + path = get_registry_path() + assert path == Path.home() / ".aipass" / "BRANCH_REGISTRY.json" + + +class TestRegistryInitialization: + """Test registry initialization.""" + + def test_initialize_creates_registry(self, tmp_path): + """Test initialize_registry creates valid registry file.""" + registry_path = tmp_path / "new_registry.json" + set_registry_path(registry_path) + + initialize_registry() + + assert registry_path.exists() + with open(registry_path, encoding="utf-8") as f: + data = json.load(f) + + assert data["version"] == "1.0" + assert "branches" in data + assert isinstance(data["branches"], dict) + assert "metadata" in data + + reset_registry_path() + + def test_initialize_idempotent(self, temp_registry): + """Test initialize_registry is idempotent (safe to call multiple times).""" + # First initialization done by fixture + first_content = temp_registry.read_text() + + # Second initialization should not change anything + initialize_registry() + second_content = temp_registry.read_text() + + assert first_content == second_content + + +class TestBranchRegistration: + """Test branch registration functionality.""" + + def test_register_branch_basic(self, temp_registry, sample_branches): + """Test basic branch registration.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + # Verify branch was added + with open(temp_registry, encoding="utf-8") as f: + data = json.load(f) + + assert "agent1" in data["branches"] + branch = data["branches"]["agent1"] + assert branch["name"] == "agent1" + assert Path(branch["path"]) == branch_path + assert branch["type"] == "agent" + assert branch["status"] == "active" + assert "created" in branch + + def test_register_branch_with_path_object(self, temp_registry, sample_branches): + """Test branch registration with Path object.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", branch_path, "agent") + + assert branch_exists("agent1") + + def test_register_branch_duplicate_raises_error(self, temp_registry, sample_branches): + """Test registering duplicate branch raises error.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + with pytest.raises(BranchAlreadyExistsError): + register_branch("agent1", str(branch_path), "agent") + + def test_register_branch_nonexistent_path_raises_error(self, temp_registry): + """Test registering nonexistent path raises error.""" + fake_path = "/nonexistent/path/to/branch" + + with pytest.raises(InvalidPathError): + register_branch("fake_branch", fake_path, "agent") + + def test_register_branch_file_not_directory_raises_error(self, temp_registry, tmp_path): + """Test registering a file instead of directory raises error.""" + file_path = tmp_path / "file.txt" + file_path.write_text("test") + + with pytest.raises(InvalidPathError): + register_branch("bad_branch", str(file_path), "agent") + + def test_register_branch_auto_initializes_registry(self, tmp_path): + """Test registering branch auto-initializes registry if it doesn't exist.""" + registry_path = tmp_path / "new_registry.json" + set_registry_path(registry_path) + + # Registry doesn't exist yet + assert not registry_path.exists() + + # Register branch should auto-initialize + branch_path = tmp_path / "agent" + branch_path.mkdir() + register_branch("agent", str(branch_path), "agent") + + # Registry should now exist + assert registry_path.exists() + assert branch_exists("agent") + + reset_registry_path() + + +class TestBranchResolution: + """Test branch resolution functionality.""" + + def test_resolve_branch_with_at_prefix(self, temp_registry, sample_branches): + """Test resolving branch with @ prefix.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + resolved = resolve_branch("@agent1") + assert Path(resolved) == branch_path + + def test_resolve_branch_without_at_prefix(self, temp_registry, sample_branches): + """Test resolving branch without @ prefix.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + resolved = resolve_branch("agent1") + assert Path(resolved) == branch_path + + def test_resolve_branch_not_found_raises_error(self, temp_registry): + """Test resolving nonexistent branch raises error.""" + with pytest.raises(BranchNotFoundError): + resolve_branch("@nonexistent") + + def test_resolve_branch_no_registry_raises_error(self, tmp_path): + """Test resolving branch without registry raises error.""" + registry_path = tmp_path / "missing_registry.json" + set_registry_path(registry_path) + + with pytest.raises(RegistryNotFoundError): + resolve_branch("@agent1") + + reset_registry_path() + + +class TestBranchExists: + """Test branch existence checking.""" + + def test_branch_exists_true(self, temp_registry, sample_branches): + """Test branch_exists returns True for registered branch.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + assert branch_exists("@agent1") + assert branch_exists("agent1") + + def test_branch_exists_false(self, temp_registry): + """Test branch_exists returns False for unregistered branch.""" + assert not branch_exists("@nonexistent") + assert not branch_exists("nonexistent") + + def test_branch_exists_no_registry_returns_false(self, tmp_path): + """Test branch_exists returns False when registry doesn't exist.""" + registry_path = tmp_path / "missing_registry.json" + set_registry_path(registry_path) + + assert not branch_exists("@agent1") + + reset_registry_path() + + +class TestBranchInfo: + """Test branch metadata retrieval.""" + + def test_get_branch_info_basic(self, temp_registry, sample_branches): + """Test getting branch info.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + info = get_branch_info("@agent1") + + assert info["name"] == "agent1" + assert Path(info["path"]) == branch_path + assert info["type"] == "agent" + assert info["status"] == "active" + assert "created" in info + + def test_get_branch_info_without_at_prefix(self, temp_registry, sample_branches): + """Test getting branch info without @ prefix.""" + branch_path = sample_branches["agent1"] + register_branch("agent1", str(branch_path), "agent") + + info = get_branch_info("agent1") + assert info["name"] == "agent1" + + def test_get_branch_info_not_found_raises_error(self, temp_registry): + """Test getting info for nonexistent branch raises error.""" + with pytest.raises(BranchNotFoundError): + get_branch_info("@nonexistent") + + +class TestListBranches: + """Test branch listing functionality.""" + + def test_list_branches_all(self, temp_registry, sample_branches): + """Test listing all branches.""" + register_branch("agent1", str(sample_branches["agent1"]), "agent") + register_branch("agent2", str(sample_branches["agent2"]), "agent") + register_branch("service1", str(sample_branches["service1"]), "service") + + branches = list_branches() + + assert "@agent1" in branches + assert "@agent2" in branches + assert "@service1" in branches + assert len(branches) == 3 + + def test_list_branches_by_type(self, temp_registry, sample_branches): + """Test listing branches filtered by type.""" + register_branch("agent1", str(sample_branches["agent1"]), "agent") + register_branch("agent2", str(sample_branches["agent2"]), "agent") + register_branch("service1", str(sample_branches["service1"]), "service") + + agents = list_branches(branch_type="agent") + services = list_branches(branch_type="service") + + assert "@agent1" in agents + assert "@agent2" in agents + assert "@service1" not in agents + assert len(agents) == 2 + + assert "@service1" in services + assert "@agent1" not in services + assert len(services) == 1 + + def test_list_branches_empty_registry(self, temp_registry): + """Test listing branches with empty registry.""" + branches = list_branches() + assert branches == [] + + def test_list_branches_no_registry(self, tmp_path): + """Test listing branches when registry doesn't exist.""" + registry_path = tmp_path / "missing_registry.json" + set_registry_path(registry_path) + + branches = list_branches() + assert branches == [] + + reset_registry_path() + + +class TestErrorHandling: + """Test error handling scenarios.""" + + def test_corrupt_registry_raises_error(self, tmp_path): + """Test corrupted registry file raises error.""" + registry_path = tmp_path / "corrupt_registry.json" + registry_path.write_text("invalid json {{{") + + set_registry_path(registry_path) + + with pytest.raises(Exception): # RegistryCorruptError or similar + resolve_branch("@agent1") + + reset_registry_path() + + def test_missing_branches_field_raises_error(self, tmp_path): + """Test registry missing 'branches' field raises error.""" + registry_path = tmp_path / "bad_registry.json" + registry_path.write_text(json.dumps({"version": "1.0"})) + + set_registry_path(registry_path) + + with pytest.raises(Exception): # RegistryCorruptError + resolve_branch("@agent1") + + reset_registry_path() + + +class TestEndToEnd: + """End-to-end integration tests.""" + + def test_full_workflow(self, tmp_path): + """Test complete workflow: initialize → register → resolve → list → exists.""" + # Setup + registry_path = tmp_path / "registry.json" + set_registry_path(registry_path) + initialize_registry() + + # Create branch directories + agent1_path = tmp_path / "agent1" + agent1_path.mkdir() + agent2_path = tmp_path / "agent2" + agent2_path.mkdir() + + # Register branches + register_branch("agent1", str(agent1_path), "agent") + register_branch("agent2", str(agent2_path), "agent") + + # Verify existence + assert branch_exists("@agent1") + assert branch_exists("@agent2") + assert not branch_exists("@agent3") + + # Resolve paths + resolved1 = resolve_branch("@agent1") + resolved2 = resolve_branch("agent2") + assert Path(resolved1) == agent1_path + assert Path(resolved2) == agent2_path + + # Get info + info = get_branch_info("@agent1") + assert info["name"] == "agent1" + assert info["type"] == "agent" + + # List branches + branches = list_branches() + assert len(branches) == 2 + assert "@agent1" in branches + assert "@agent2" in branches + + # Cleanup + reset_registry_path() diff --git a/tests/test_routing_integration.py b/tests/test_routing_integration.py new file mode 100644 index 00000000..e2a72ad9 --- /dev/null +++ b/tests/test_routing_integration.py @@ -0,0 +1,504 @@ +""" +Integration tests for the AIPass routing module (Phase 3). + +End-to-end tests that exercise the full workflow using real temp directories +with proper branch file structures. All subprocess calls are mocked so no +real Python processes are spawned. +""" + +import json +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from aipass.routing import ( + BranchNotFoundError, + CommandExecutionError, + CommandResult, + HelpResult, + branch_exists, + discover_modules, + get_branch_info, + get_help, + initialize_registry, + list_branches, + register_branch, + reset_registry_path, + resolve_branch, + route_all, + route_command, + set_registry_path, +) +from aipass.routing.discovery import get_system_help + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_cp(stdout=b"", stderr=b"", returncode=0): + """Build a mock subprocess.CompletedProcess.""" + cp = MagicMock(spec=subprocess.CompletedProcess) + cp.stdout = stdout + cp.stderr = stderr + cp.returncode = returncode + return cp + + +def _make_branch(root: Path, name: str) -> Path: + """ + Create a minimal branch directory layout under root. + + Layout: + root/{name}/ + apps/ + {name}.py ← entry point + modules/ + status.py + info.py + """ + branch_dir = root / name + apps_dir = branch_dir / "apps" + modules_dir = apps_dir / "modules" + modules_dir.mkdir(parents=True) + (apps_dir / f"{name}.py").write_text(f"# {name} entry point\n") + (modules_dir / "status.py").write_text("") + (modules_dir / "info.py").write_text("") + return branch_dir + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def registry(tmp_path): + """Isolated registry backed by a temp directory.""" + registry_path = tmp_path / "registry.json" + set_registry_path(registry_path) + initialize_registry() + yield tmp_path + reset_registry_path() + + +@pytest.fixture +def multi_branch_registry(tmp_path): + """ + Registry pre-populated with three branches: alpha, beta, gamma. + + Each has a full apps/{name}.py + apps/modules/ structure. + """ + registry_path = tmp_path / "registry.json" + set_registry_path(registry_path) + initialize_registry() + + for name in ("alpha", "beta", "gamma"): + branch_dir = _make_branch(tmp_path, name) + register_branch(name, str(branch_dir), "agent") + + yield tmp_path + reset_registry_path() + + +# --------------------------------------------------------------------------- +# 1. Full workflow: initialize → register → resolve → discover → help +# --------------------------------------------------------------------------- + + +class TestFullWorkflow: + """End-to-end happy-path tests.""" + + def test_initialize_register_resolve(self, tmp_path): + """Complete init → register → resolve cycle.""" + registry_path = tmp_path / "reg.json" + set_registry_path(registry_path) + initialize_registry() + + branch_dir = _make_branch(tmp_path, "mybot") + register_branch("mybot", str(branch_dir), "agent") + + resolved = resolve_branch("@mybot") + assert Path(resolved) == branch_dir + + reset_registry_path() + + def test_register_multiple_branches_and_list(self, registry, tmp_path): + """Register several branches and verify list_branches returns all.""" + names = ["alice", "bob", "charlie"] + for name in names: + bd = _make_branch(tmp_path, name) + register_branch(name, str(bd), "agent") + + found = list_branches() + for name in names: + assert f"@{name}" in found + assert len(found) == 3 + + def test_resolve_then_discover_then_help(self, multi_branch_registry): + """Resolve a branch, discover its modules, then get help.""" + # Resolve + path = resolve_branch("@alpha") + assert Path(path).is_dir() + + # Discover — mock help output with a Commands section + help_bytes = b"Usage: alpha.py\n\nCommands:\n status Check status\n info Show info\n" + with patch("subprocess.run", return_value=_make_cp(stdout=help_bytes)): + modules = discover_modules("@alpha") + assert "status" in modules + assert "info" in modules + + # Help — returns structured HelpResult + with patch("subprocess.run", return_value=_make_cp(stdout=help_bytes)): + result = get_help("@alpha") + assert isinstance(result, HelpResult) + assert result.branch == "alpha" + assert result.command is None + assert "status" in result.commands_found + + def test_route_command_after_register(self, multi_branch_registry): + """route_command reaches a registered branch successfully.""" + cp = _make_cp(stdout=b"alpha is running\n", returncode=0) + with patch("subprocess.run", return_value=cp): + result = route_command("@alpha", "status") + + assert result.exit_code == 0 + assert "alpha is running" in result.stdout + assert result.branch == "alpha" + assert result.command == "status" + + def test_branch_exists_after_register(self, registry, tmp_path): + """branch_exists returns True immediately after registration.""" + bd = _make_branch(tmp_path, "sentinel") + register_branch("sentinel", str(bd), "service") + + assert branch_exists("@sentinel") + assert branch_exists("sentinel") + assert not branch_exists("@ghost") + + def test_get_branch_info_after_register(self, registry, tmp_path): + """get_branch_info returns correct metadata after registration.""" + bd = _make_branch(tmp_path, "worker") + register_branch("worker", str(bd), "service") + + info = get_branch_info("@worker") + assert info["name"] == "worker" + assert info["type"] == "service" + assert info["status"] == "active" + assert Path(info["path"]) == bd + assert "created" in info + + +# --------------------------------------------------------------------------- +# 2. HelpResult structure +# --------------------------------------------------------------------------- + + +class TestHelpResult: + """Verify HelpResult dataclass fields and behaviour.""" + + def test_help_result_fields(self, multi_branch_registry): + """HelpResult exposes branch, command, text, commands_found.""" + help_bytes = b"Usage: beta.py\n\nCommands:\n run Run a task\n stop Stop\n" + with patch("subprocess.run", return_value=_make_cp(stdout=help_bytes)): + result = get_help("@beta") + + assert result.branch == "beta" + assert result.command is None + assert "Usage" in result.text + assert isinstance(result.commands_found, list) + assert "run" in result.commands_found + assert "stop" in result.commands_found + + def test_help_result_command_level(self, multi_branch_registry): + """HelpResult.command is set when a specific command is queried.""" + help_bytes = b"Usage: beta.py run [options]\n --verbose Verbose output\n" + with patch("subprocess.run", return_value=_make_cp(stdout=help_bytes)): + result = get_help("@beta", command="run") + + assert result.command == "run" + assert "Usage" in result.text + + def test_help_result_stderr_fallback(self, multi_branch_registry): + """HelpResult.text comes from stderr when stdout is empty.""" + with patch("subprocess.run", return_value=_make_cp(stdout=b"", stderr=b"from stderr\n")): + result = get_help("@gamma") + + assert "from stderr" in result.text + + def test_help_result_empty_commands_found(self, multi_branch_registry): + """commands_found is empty when help text has no commands section.""" + with patch("subprocess.run", return_value=_make_cp(stdout=b"No commands here.\n")): + result = get_help("@gamma") + + assert result.commands_found == [] + + def test_help_result_importable_from_package(self): + """HelpResult is importable directly from aipass.routing.""" + from aipass.routing import HelpResult as HR + assert HR is HelpResult + + +# --------------------------------------------------------------------------- +# 3. @all routing: route_all +# --------------------------------------------------------------------------- + + +class TestRouteAll: + """Tests for route_all — fan-out command to all active branches.""" + + def test_route_all_reaches_all_branches(self, multi_branch_registry): + """route_all returns a result for every registered branch.""" + cp = _make_cp(stdout=b"ok\n", returncode=0) + with patch("subprocess.run", return_value=cp): + results = route_all("status") + + assert set(results.keys()) == {"alpha", "beta", "gamma"} + + def test_route_all_returns_command_results(self, multi_branch_registry): + """Each value in route_all output is a CommandResult.""" + cp = _make_cp(stdout=b"running\n", returncode=0) + with patch("subprocess.run", return_value=cp): + results = route_all("status") + + for branch_name, result in results.items(): + assert isinstance(result, CommandResult) + assert result.branch == branch_name + assert result.command == "status" + + def test_route_all_with_args(self, multi_branch_registry): + """route_all forwards extra args to every branch.""" + cp = _make_cp(stdout=b"verbose\n", returncode=0) + with patch("subprocess.run", return_value=cp) as mock_run: + route_all("run", args=["--verbose"]) + + # Every call should include --verbose + for call in mock_run.call_args_list: + cmd_list = call[0][0] + assert "--verbose" in cmd_list + + def test_route_all_continues_on_branch_failure(self, multi_branch_registry): + """route_all continues even when one branch raises CommandExecutionError.""" + call_count = 0 + + def side_effect(*a, **kw): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise CommandExecutionError("branch exploded") + return _make_cp(stdout=b"ok\n", returncode=0) + + with patch("subprocess.run", side_effect=side_effect): + results = route_all("status") + + # All three branches present in results + assert len(results) == 3 + + def test_route_all_failed_branch_has_negative_exit_code(self, multi_branch_registry): + """A branch that raises an error gets exit_code=-1 in route_all result.""" + def side_effect(*a, **kw): + raise CommandExecutionError("no entry point") + + with patch("subprocess.run", side_effect=side_effect): + results = route_all("status") + + for result in results.values(): + assert result.exit_code == -1 + assert result.stderr != "" + + def test_route_all_empty_registry(self, registry): + """route_all returns empty dict when no branches are registered.""" + results = route_all("status") + assert results == {} + + def test_route_all_timeout_per_branch(self, multi_branch_registry): + """route_all forwards the timeout to each branch call.""" + cp = _make_cp(stdout=b"ok\n", returncode=0) + with patch("subprocess.run", return_value=cp) as mock_run: + route_all("status", timeout=5) + + for call in mock_run.call_args_list: + _, kwargs = call + assert kwargs.get("timeout") == 5 + + def test_route_all_importable_from_package(self): + """route_all is importable from aipass.routing.""" + from aipass.routing import route_all as ra + assert callable(ra) + + +# --------------------------------------------------------------------------- +# 4. get_system_help — aggregated help across all branches +# --------------------------------------------------------------------------- + + +class TestGetSystemHelp: + """Tests for get_system_help — fan-out help query across all branches.""" + + def test_system_help_returns_dict(self, multi_branch_registry): + """get_system_help returns a dict keyed by branch name.""" + help_bytes = b"Usage: x.py\n\nCommands:\n status Check\n" + with patch("subprocess.run", return_value=_make_cp(stdout=help_bytes)): + results = get_system_help() + + assert isinstance(results, dict) + assert set(results.keys()) == {"alpha", "beta", "gamma"} + + def test_system_help_values_are_help_results(self, multi_branch_registry): + """Each value in get_system_help output is a HelpResult.""" + with patch("subprocess.run", return_value=_make_cp(stdout=b"help\n")): + results = get_system_help() + + for result in results.values(): + assert isinstance(result, HelpResult) + + def test_system_help_skips_failing_branches(self, multi_branch_registry): + """get_system_help omits branches whose help command fails.""" + call_count = 0 + + def side_effect(*a, **kw): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise CommandExecutionError("boom") + return _make_cp(stdout=b"help\n") + + with patch("subprocess.run", side_effect=side_effect): + results = get_system_help() + + # At least two of the three branches should succeed + assert len(results) >= 2 + + def test_system_help_empty_when_no_branches(self, registry): + """get_system_help returns empty dict with no registered branches.""" + results = get_system_help() + assert results == {} + + +# --------------------------------------------------------------------------- +# 5. Error recovery — branches disappear mid-operation +# --------------------------------------------------------------------------- + + +class TestErrorRecovery: + """Tests for graceful handling of unexpected runtime conditions.""" + + def test_resolve_nonexistent_branch_raises(self, registry): + """Resolving an unregistered branch raises BranchNotFoundError.""" + with pytest.raises(BranchNotFoundError): + resolve_branch("@ghost") + + def test_route_command_nonexistent_branch_raises(self, registry): + """Routing to an unregistered branch raises BranchNotFoundError.""" + with pytest.raises(BranchNotFoundError): + route_command("@ghost", "status") + + def test_discover_modules_nonexistent_branch_raises(self, registry): + """discover_modules on unregistered branch raises BranchNotFoundError.""" + with pytest.raises(BranchNotFoundError): + discover_modules("@ghost") + + def test_get_help_nonexistent_branch_raises(self, registry): + """get_help on unregistered branch raises BranchNotFoundError.""" + with pytest.raises(BranchNotFoundError): + get_help("@ghost") + + def test_branch_without_entry_point_raises_on_route(self, tmp_path): + """route_command raises CommandExecutionError when entry point missing.""" + registry_path = tmp_path / "reg.json" + set_registry_path(registry_path) + initialize_registry() + + # Register a branch that has NO apps/{name}.py + no_entry = tmp_path / "bare" + no_entry.mkdir() + register_branch("bare", str(no_entry), "agent") + + with pytest.raises(CommandExecutionError, match="Entry point not found"): + route_command("@bare", "status") + + reset_registry_path() + + def test_branch_without_entry_point_raises_on_help(self, tmp_path): + """get_help raises CommandExecutionError when entry point missing.""" + registry_path = tmp_path / "reg.json" + set_registry_path(registry_path) + initialize_registry() + + no_entry = tmp_path / "bare2" + no_entry.mkdir() + register_branch("bare2", str(no_entry), "agent") + + with pytest.raises(CommandExecutionError, match="Entry point not found"): + get_help("@bare2") + + reset_registry_path() + + +# --------------------------------------------------------------------------- +# 6. Registry persistence across multiple operations +# --------------------------------------------------------------------------- + + +class TestRegistryPersistence: + """Verify the registry file survives multiple read/write cycles.""" + + def test_registry_persists_across_register_calls(self, tmp_path): + """Each register_branch call is durable in the JSON file.""" + registry_path = tmp_path / "persist.json" + set_registry_path(registry_path) + initialize_registry() + + for name in ("p1", "p2", "p3"): + bd = _make_branch(tmp_path, name) + register_branch(name, str(bd), "agent") + + # Read the raw JSON and verify all branches are there + with open(registry_path, encoding="utf-8") as fh: + data = json.load(fh) + + assert "p1" in data["branches"] + assert "p2" in data["branches"] + assert "p3" in data["branches"] + + reset_registry_path() + + def test_registry_metadata_updated_on_write(self, tmp_path): + """Registry metadata.last_updated is refreshed on each write.""" + registry_path = tmp_path / "meta.json" + set_registry_path(registry_path) + initialize_registry() + + bd = _make_branch(tmp_path, "meta_agent") + register_branch("meta_agent", str(bd), "agent") + + with open(registry_path, encoding="utf-8") as fh: + data = json.load(fh) + + assert "last_updated" in data["metadata"] + assert data["metadata"]["managed_by"] == "aipass.routing" + + reset_registry_path() + + def test_list_branches_reflects_all_registrations(self, tmp_path): + """list_branches always reflects the on-disk state.""" + registry_path = tmp_path / "list_test.json" + set_registry_path(registry_path) + initialize_registry() + + bd1 = _make_branch(tmp_path, "lx1") + bd2 = _make_branch(tmp_path, "lx2") + register_branch("lx1", str(bd1), "agent") + + assert len(list_branches()) == 1 + + register_branch("lx2", str(bd2), "agent") + + assert len(list_branches()) == 2 + assert "@lx1" in list_branches() + assert "@lx2" in list_branches() + + reset_registry_path() diff --git a/tests/test_routing_phase2.py b/tests/test_routing_phase2.py new file mode 100644 index 00000000..f28f3965 --- /dev/null +++ b/tests/test_routing_phase2.py @@ -0,0 +1,546 @@ +""" +Unit tests for Phase 2 routing functionality. + +Tests command routing, subprocess execution, and module discovery with >80% +coverage on new code. All subprocess calls are mocked — no real processes run. +""" + +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from aipass.routing import ( + BranchNotFoundError, + CommandExecutionError, + CommandResult, + discover_modules, + get_help, + initialize_registry, + register_branch, + reset_registry_path, + route_command, + set_registry_path, +) +from aipass.routing.executor import execute_command +from aipass.routing.router import _find_entry_point + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def temp_registry(tmp_path): + """Temporary registry backed by a tmp directory.""" + registry_path = tmp_path / "test_registry.json" + set_registry_path(registry_path) + initialize_registry() + yield tmp_path + reset_registry_path() + + +@pytest.fixture +def branch_with_entry(tmp_path): + """ + A registered branch that has a valid apps/{name}.py entry point. + + Layout: + tmp_path/ + myagent/ + apps/ + myagent.py + modules/ + status.py + info.py + """ + branch_dir = tmp_path / "myagent" + apps_dir = branch_dir / "apps" + modules_dir = apps_dir / "modules" + modules_dir.mkdir(parents=True) + + entry = apps_dir / "myagent.py" + entry.write_text("# stub entry point\n") + + (modules_dir / "status.py").write_text("") + (modules_dir / "info.py").write_text("") + (modules_dir / "__init__.py").write_text("") + + registry_path = tmp_path / "registry.json" + set_registry_path(registry_path) + initialize_registry() + register_branch("myagent", str(branch_dir), "agent") + + yield branch_dir + + reset_registry_path() + + +@pytest.fixture +def branch_without_entry(tmp_path): + """A registered branch that has NO apps/{name}.py entry point.""" + branch_dir = tmp_path / "noentry" + apps_dir = branch_dir / "apps" + modules_dir = apps_dir / "modules" + modules_dir.mkdir(parents=True) + (modules_dir / "do_thing.py").write_text("") + + registry_path = tmp_path / "registry.json" + set_registry_path(registry_path) + initialize_registry() + register_branch("noentry", str(branch_dir), "agent") + + yield branch_dir + + reset_registry_path() + + +def _make_completed_process(stdout=b"", stderr=b"", returncode=0): + """Helper to build a subprocess.CompletedProcess mock.""" + cp = MagicMock(spec=subprocess.CompletedProcess) + cp.stdout = stdout + cp.stderr = stderr + cp.returncode = returncode + return cp + + +# --------------------------------------------------------------------------- +# CommandResult structure +# --------------------------------------------------------------------------- + + +class TestCommandResult: + """Verify CommandResult is a proper dataclass with expected fields.""" + + def test_fields_present(self): + """CommandResult exposes stdout, stderr, exit_code, branch, command.""" + result = CommandResult( + stdout="hello", + stderr="", + exit_code=0, + branch="myagent", + command="status", + ) + assert result.stdout == "hello" + assert result.stderr == "" + assert result.exit_code == 0 + assert result.branch == "myagent" + assert result.command == "status" + + def test_non_zero_exit_code(self): + """CommandResult stores non-zero exit codes without raising.""" + result = CommandResult( + stdout="", stderr="error", exit_code=1, branch="b", command="c" + ) + assert result.exit_code == 1 + + def test_dataclass_equality(self): + """Two CommandResult instances with identical values compare equal.""" + a = CommandResult("out", "err", 0, "b", "c") + b = CommandResult("out", "err", 0, "b", "c") + assert a == b + + +# --------------------------------------------------------------------------- +# executor.execute_command +# --------------------------------------------------------------------------- + + +class TestExecuteCommand: + """Tests for the low-level subprocess wrapper.""" + + def test_successful_execution(self, tmp_path): + """execute_command returns CommandResult on success.""" + cp = _make_completed_process(stdout=b"ok\n", returncode=0) + with patch("subprocess.run", return_value=cp) as mock_run: + result = execute_command("python3", ["script.py"], cwd=str(tmp_path)) + + assert result.stdout == "ok\n" + assert result.exit_code == 0 + mock_run.assert_called_once() + + def test_no_shell_true(self, tmp_path): + """execute_command never passes shell=True to subprocess.run.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + execute_command("python3", ["x.py"], cwd=str(tmp_path)) + + _, kwargs = mock_run.call_args + assert kwargs.get("shell") is False + + def test_timeout_raises_command_execution_error(self, tmp_path): + """TimeoutExpired is wrapped in CommandExecutionError.""" + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="python3", timeout=5), + ): + with pytest.raises(CommandExecutionError, match="timed out"): + execute_command("python3", ["x.py"], cwd=str(tmp_path), timeout=5) + + def test_file_not_found_raises_command_execution_error(self, tmp_path): + """FileNotFoundError is wrapped in CommandExecutionError.""" + with patch("subprocess.run", side_effect=FileNotFoundError): + with pytest.raises(CommandExecutionError, match="not found"): + execute_command("nonexistent_bin", [], cwd=str(tmp_path)) + + def test_os_error_raises_command_execution_error(self, tmp_path): + """Generic OSError is wrapped in CommandExecutionError.""" + with patch("subprocess.run", side_effect=OSError("permission denied")): + with pytest.raises(CommandExecutionError, match="OS error"): + execute_command("python3", [], cwd=str(tmp_path)) + + def test_stderr_captured(self, tmp_path): + """execute_command captures stderr separately from stdout.""" + cp = _make_completed_process(stdout=b"", stderr=b"warn\n", returncode=1) + with patch("subprocess.run", return_value=cp): + result = execute_command("python3", ["x.py"], cwd=str(tmp_path)) + + assert result.stderr == "warn\n" + assert result.exit_code == 1 + + def test_branch_and_command_empty_strings(self, tmp_path): + """execute_command sets branch and command to empty strings.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp): + result = execute_command("python3", [], cwd=str(tmp_path)) + + assert result.branch == "" + assert result.command == "" + + def test_timeout_forwarded(self, tmp_path): + """Custom timeout value is passed through to subprocess.run.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + execute_command("python3", [], cwd=str(tmp_path), timeout=99) + + _, kwargs = mock_run.call_args + assert kwargs["timeout"] == 99 + + +# --------------------------------------------------------------------------- +# router.route_command +# --------------------------------------------------------------------------- + + +class TestRouteCommand: + """Tests for the high-level command routing function.""" + + def test_route_command_success(self, branch_with_entry): + """route_command returns populated CommandResult on success.""" + cp = _make_completed_process(stdout=b"running\n", returncode=0) + with patch("subprocess.run", return_value=cp): + result = route_command("@myagent", "status") + + assert result.stdout == "running\n" + assert result.exit_code == 0 + assert result.branch == "myagent" + assert result.command == "status" + + def test_route_command_without_at_prefix(self, branch_with_entry): + """route_command accepts branch names without the @ prefix.""" + cp = _make_completed_process(stdout=b"ok\n", returncode=0) + with patch("subprocess.run", return_value=cp): + result = route_command("myagent", "status") + + assert result.branch == "myagent" + + def test_route_command_with_args(self, branch_with_entry): + """route_command forwards extra args to subprocess.""" + cp = _make_completed_process(stdout=b"done\n", returncode=0) + with patch("subprocess.run", return_value=cp) as mock_run: + route_command("@myagent", "run", args=["--verbose", "--dry-run"]) + + call_args = mock_run.call_args[0][0] # positional list + assert "--verbose" in call_args + assert "--dry-run" in call_args + + def test_route_command_missing_branch_raises_error(self, temp_registry): + """route_command raises BranchNotFoundError for unknown branch.""" + with pytest.raises(BranchNotFoundError): + route_command("@ghost", "status") + + def test_route_command_missing_entry_point_raises_error(self, branch_without_entry): + """route_command raises CommandExecutionError when entry point absent.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + route_command("@noentry", "status") + + def test_route_command_timeout(self, branch_with_entry): + """route_command propagates timeout as CommandExecutionError.""" + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="python3", timeout=1), + ): + with pytest.raises(CommandExecutionError, match="timed out"): + route_command("@myagent", "status", timeout=1) + + def test_route_command_nonzero_exit_does_not_raise(self, branch_with_entry): + """Non-zero exit code is returned, not raised.""" + cp = _make_completed_process(stderr=b"fail\n", returncode=2) + with patch("subprocess.run", return_value=cp): + result = route_command("@myagent", "status") + + assert result.exit_code == 2 + assert result.stderr == "fail\n" + + def test_route_command_default_timeout_is_30(self, branch_with_entry): + """Default timeout forwarded to subprocess is 30 seconds.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + route_command("@myagent", "status") + + _, kwargs = mock_run.call_args + assert kwargs["timeout"] == 30 + + +# --------------------------------------------------------------------------- +# router._find_entry_point +# --------------------------------------------------------------------------- + + +class TestFindEntryPoint: + """Tests for the internal entry-point locator.""" + + def test_finds_existing_entry_point(self, branch_with_entry): + """_find_entry_point returns path when entry point exists.""" + ep = _find_entry_point(str(branch_with_entry), "myagent") + assert ep.exists() + assert ep.name == "myagent.py" + + def test_raises_when_entry_point_missing(self, branch_without_entry): + """_find_entry_point raises CommandExecutionError when file absent.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + _find_entry_point(str(branch_without_entry), "noentry") + + +# --------------------------------------------------------------------------- +# discovery.discover_modules +# --------------------------------------------------------------------------- + + +class TestDiscoverModules: + """Tests for branch capability discovery.""" + + def test_discover_from_help_output(self, branch_with_entry): + """discover_modules parses commands from --help output.""" + help_output = ( + b"Usage: myagent.py [command]\n\n" + b"Commands:\n" + b" status Show current status\n" + b" deploy Deploy the agent\n" + ) + cp = _make_completed_process(stdout=help_output, returncode=0) + with patch("subprocess.run", return_value=cp): + modules = discover_modules("@myagent") + + assert "status" in modules + assert "deploy" in modules + + def test_discover_falls_back_to_modules_dir(self, branch_with_entry): + """discover_modules falls back to scanning modules/ when help unparseable.""" + cp = _make_completed_process(stdout=b"No commands here.\n", returncode=0) + with patch("subprocess.run", return_value=cp): + modules = discover_modules("@myagent") + + # Falls back to scanning apps/modules/ which contains status.py, info.py + assert "status" in modules + assert "info" in modules + # __init__ is excluded + assert "__init__" not in modules + + def test_discover_falls_back_when_subprocess_fails(self, branch_with_entry): + """discover_modules falls back to modules/ on subprocess OSError.""" + with patch("subprocess.run", side_effect=OSError("boom")): + modules = discover_modules("@myagent") + + assert "status" in modules + assert "info" in modules + + def test_discover_falls_back_on_timeout(self, branch_with_entry): + """discover_modules falls back to modules/ on help timeout.""" + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="python3", timeout=10), + ): + modules = discover_modules("@myagent") + + assert isinstance(modules, list) + + def test_discover_missing_branch_raises_error(self, temp_registry): + """discover_modules raises BranchNotFoundError for unknown branch.""" + with pytest.raises(BranchNotFoundError): + discover_modules("@phantom") + + def test_discover_no_entry_no_modules(self, branch_without_entry): + """discover_modules returns empty list when no entry and no modules match help.""" + cp = _make_completed_process(stdout=b"nothing useful\n", returncode=0) + # branch_without_entry has no entry point, so subprocess is never called for + # --help (discovery skips it), but it does have a modules/ dir. + with patch("subprocess.run", return_value=cp): + modules = discover_modules("@noentry") + + # Should find do_thing.py from the modules dir. + assert "do_thing" in modules + + def test_discover_returns_list(self, branch_with_entry): + """discover_modules always returns a list.""" + cp = _make_completed_process(stdout=b"", returncode=0) + with patch("subprocess.run", return_value=cp): + result = discover_modules("@myagent") + + assert isinstance(result, list) + + +# --------------------------------------------------------------------------- +# discovery.get_help +# --------------------------------------------------------------------------- + + +class TestGetHelp: + """Tests for branch help text retrieval (get_help returns HelpResult).""" + + def test_get_help_branch_level(self, branch_with_entry): + """get_help returns HelpResult with text when command is None.""" + help_text = b"Usage: myagent.py [command]\n\nCommands:\n status\n" + cp = _make_completed_process(stdout=help_text, returncode=0) + with patch("subprocess.run", return_value=cp): + result = get_help("@myagent") + + assert "Usage" in result.text + assert result.branch == "myagent" + assert result.command is None + + def test_get_help_command_level(self, branch_with_entry): + """get_help passes the command name when command is specified.""" + help_text = b"Usage: myagent.py status [options]\n" + cp = _make_completed_process(stdout=help_text, returncode=0) + with patch("subprocess.run", return_value=cp) as mock_run: + result = get_help("@myagent", command="status") + + call_args = mock_run.call_args[0][0] + assert "status" in call_args + assert "--help" in call_args + assert "Usage" in result.text + assert result.command == "status" + + def test_get_help_falls_back_to_stderr(self, branch_with_entry): + """get_help uses stderr text when stdout is empty.""" + cp = _make_completed_process(stdout=b"", stderr=b"help via stderr\n", returncode=0) + with patch("subprocess.run", return_value=cp): + result = get_help("@myagent") + + assert "help via stderr" in result.text + + def test_get_help_missing_branch_raises_error(self, temp_registry): + """get_help raises BranchNotFoundError for unknown branch.""" + with pytest.raises(BranchNotFoundError): + get_help("@phantom") + + def test_get_help_missing_entry_point_raises_error(self, branch_without_entry): + """get_help raises CommandExecutionError when entry point absent.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + get_help("@noentry") + + def test_get_help_timeout_raises_error(self, branch_with_entry): + """get_help raises CommandExecutionError on timeout.""" + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="python3", timeout=10), + ): + with pytest.raises(CommandExecutionError, match="timed out"): + get_help("@myagent") + + def test_get_help_os_error_raises_error(self, branch_with_entry): + """get_help raises CommandExecutionError on OS error.""" + with patch("subprocess.run", side_effect=OSError("no such file")): + with pytest.raises(CommandExecutionError, match="OS error"): + get_help("@myagent") + + def test_get_help_without_at_prefix(self, branch_with_entry): + """get_help accepts branch name without @ prefix.""" + cp = _make_completed_process(stdout=b"help text\n", returncode=0) + with patch("subprocess.run", return_value=cp): + result = get_help("myagent") + + assert "help text" in result.text + + def test_get_help_commands_found_populated(self, branch_with_entry): + """get_help populates commands_found from parsed help text.""" + help_text = b"Usage: myagent.py\n\nCommands:\n status Show status\n deploy Deploy\n" + cp = _make_completed_process(stdout=help_text, returncode=0) + with patch("subprocess.run", return_value=cp): + result = get_help("@myagent") + + assert "status" in result.commands_found + assert "deploy" in result.commands_found + + +# --------------------------------------------------------------------------- +# Executor safety: no shell injection +# --------------------------------------------------------------------------- + + +class TestExecutorSafety: + """Verify the executor never enables shell features.""" + + def test_shell_is_false(self, tmp_path): + """subprocess.run is always called with shell=False.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + execute_command("python3", ["script.py", "arg with spaces"], cwd=str(tmp_path)) + + _, kwargs = mock_run.call_args + assert kwargs.get("shell") is False + + def test_args_passed_as_list(self, tmp_path): + """Command is passed as a list, never as a shell string.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + execute_command("python3", ["a", "b", "c"], cwd=str(tmp_path)) + + positional_cmd = mock_run.call_args[0][0] + assert isinstance(positional_cmd, list) + + def test_shell_metacharacters_not_interpreted(self, tmp_path): + """Shell metacharacters in args are passed verbatim, not interpreted.""" + cp = _make_completed_process() + with patch("subprocess.run", return_value=cp) as mock_run: + execute_command( + "python3", + ["script.py", "; rm -rf /", "$(evil)"], + cwd=str(tmp_path), + ) + + positional_cmd = mock_run.call_args[0][0] + # Metacharacter strings survive intact as list elements. + assert "; rm -rf /" in positional_cmd + assert "$(evil)" in positional_cmd + + +# --------------------------------------------------------------------------- +# Public API import smoke test +# --------------------------------------------------------------------------- + + +class TestPublicAPIImports: + """Verify all Phase 2 symbols are importable from the top-level package.""" + + def test_route_command_importable(self): + from aipass.routing import route_command + assert callable(route_command) + + def test_discover_modules_importable(self): + from aipass.routing import discover_modules + assert callable(discover_modules) + + def test_get_help_importable(self): + from aipass.routing import get_help + assert callable(get_help) + + def test_command_result_importable(self): + from aipass.routing import CommandResult + assert CommandResult is not None + + def test_version_updated(self): + from aipass.routing import __version__ + assert __version__ == "1.0.0" From 002bcb672f0b62967fff739e1fed2fea739dc7c8 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 1 Mar 2026 20:21:03 -0800 Subject: [PATCH 2/2] fix(routing): remove unused imports to pass CI lint Remove unused `pathlib.Path` and `RegistryNotFoundError` from resolver.py and unused `tempfile` from test_routing_core.py, resolving ruff F401 errors. Co-Authored-By: Claude Sonnet 4.6 --- src/aipass/routing/resolver.py | 3 +-- tests/test_routing_core.py | 1 - 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/src/aipass/routing/resolver.py b/src/aipass/routing/resolver.py index d99d0597..34767cee 100644 --- a/src/aipass/routing/resolver.py +++ b/src/aipass/routing/resolver.py @@ -4,10 +4,9 @@ Branch resolution logic. Resolves symbolic @branch names to absolute paths and metadata. """ -from pathlib import Path from typing import Any, Dict, List, Optional -from .exceptions import BranchNotFoundError, RegistryNotFoundError +from .exceptions import BranchNotFoundError from .registry import get_all_branches, get_branch_by_name, load_registry diff --git a/tests/test_routing_core.py b/tests/test_routing_core.py index a9514768..ca73bb16 100644 --- a/tests/test_routing_core.py +++ b/tests/test_routing_core.py @@ -5,7 +5,6 @@ Tests all Phase 1 routing functions with >80% coverage target. """ import json -import tempfile from pathlib import Path import pytest