AIOSAIandClaude Opus 4.6
78ae5f63b6
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 <noreply@anthropic.com>
547 lines
21 KiB
Python
547 lines
21 KiB
Python
"""
|
|
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"
|