AIOSAI
f4b203a74e
- seedgo: new check_introspection_rich_formatting (delegation-aware) + introspection.md richness section + tests - 13 branches: wrap print_introspection in Rich markup -> all at seedgo 100% - S202 CLI polish: @hooks --help rewrite (hooksound on/off/status surfaced), spawn repair help clarity, drone Rich colour restore for --help/introspection/status - flow: playbook plan-type (PPLAN) self-serve templates (default.md, sunday_merge.md SOP) + register-overrides-auto fix + --help rewrite - hooks: setup.sh installs auto_process bridges + seedgo snapshot fixture learns them (TDPLAN-0005 followup, clears commit-gate blocker) - remove orphaned devpulse/SETUP.md (vectorized to @memory) - CHANGELOG [2026.W23]
842 lines
28 KiB
Python
842 lines
28 KiB
Python
# =================== AIPass ====================
|
|
# Name: test_cli_routing.py
|
|
# Description: CLI Routing Tests for Drone (adapted from universal template)
|
|
# Version: 1.0.0
|
|
# Created: 2026-03-27
|
|
# Modified: 2026-03-27
|
|
# =============================================
|
|
|
|
"""
|
|
CLI Routing Tests for Drone
|
|
|
|
Drone's CLI entry point is apps/drone.py (not cli_handler.py).
|
|
This file tests print_help, print_introspection, and short_help (-h)
|
|
using drone.py's functions directly.
|
|
|
|
Covers the 2 missing CLI routing items:
|
|
- short_help (CR-002)
|
|
- print_help (CR-007)
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
def test_print_help(capsys: pytest.CaptureFixture[str]) -> None: # CR-007
|
|
"""print_help() runs without error and produces stdout output."""
|
|
from aipass.drone.apps.drone import print_help
|
|
|
|
print_help()
|
|
captured = capsys.readouterr()
|
|
assert len(captured.out) > 0, "print_help() must produce output"
|
|
|
|
|
|
def test_print_introspection(capsys: pytest.CaptureFixture[str]) -> None: # CR-008
|
|
"""print_introspection() runs without error and produces stdout output."""
|
|
from aipass.drone.apps.drone import print_introspection
|
|
|
|
print_introspection()
|
|
captured = capsys.readouterr()
|
|
assert len(captured.out) > 0, "print_introspection() must produce output"
|
|
|
|
|
|
def test_short_help() -> None: # CR-002
|
|
"""drone -h flag triggers help and exits cleanly."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "-h"]):
|
|
result = main()
|
|
assert result == 0, "drone -h must return exit code 0"
|
|
|
|
|
|
# ===========================================================================
|
|
# main() dispatch — version, help, introspection
|
|
# ===========================================================================
|
|
|
|
_DRONE = "aipass.drone.apps.drone"
|
|
|
|
|
|
class TestMainVersion:
|
|
"""drone --version and -V flags."""
|
|
|
|
def test_version_long_flag(self) -> None:
|
|
"""--version prints version and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "--version"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_version_short_flag(self) -> None:
|
|
"""-V prints version and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "-V"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
|
|
class TestMainHelp:
|
|
"""drone --help, -h, and help command."""
|
|
|
|
def test_help_long_flag(self) -> None:
|
|
"""--help returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "--help"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_help_word(self) -> None:
|
|
"""bare 'help' returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "help"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
|
|
class TestMainNoArgs:
|
|
"""drone with no args shows introspection."""
|
|
|
|
def test_no_args_introspection(self) -> None:
|
|
"""No args calls print_introspection and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone"]),
|
|
patch(f"{_DRONE}.print_introspection"),
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_no_args_registry_error(self) -> None:
|
|
"""RegistryError during introspection returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
from aipass.drone.apps.modules import RegistryError
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone"]),
|
|
patch(
|
|
f"{_DRONE}.print_introspection",
|
|
side_effect=RegistryError("no registry"),
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
# ===========================================================================
|
|
# main() dispatch — built-in commands
|
|
# ===========================================================================
|
|
|
|
|
|
class TestMainSystems:
|
|
"""drone systems command."""
|
|
|
|
def test_systems_success(self) -> None:
|
|
"""systems delegates to _handle_systems and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "systems"]),
|
|
patch(f"{_DRONE}._handle_systems", return_value=0) as mock_sys,
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
mock_sys.assert_called_once()
|
|
|
|
def test_systems_registry_error(self) -> None:
|
|
"""RegistryError in systems returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
from aipass.drone.apps.modules import RegistryError
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "systems"]),
|
|
patch(
|
|
f"{_DRONE}._handle_systems",
|
|
side_effect=RegistryError("broken"),
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
def test_systems_unexpected_error(self) -> None:
|
|
"""Unexpected Exception in systems returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "systems"]),
|
|
patch(
|
|
f"{_DRONE}._handle_systems",
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
class TestMainScan:
|
|
"""drone scan command."""
|
|
|
|
def test_scan_no_target(self) -> None:
|
|
"""scan with no target returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "scan"]):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
def test_scan_success(self) -> None:
|
|
"""scan with target delegates to scan module."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "scan", "@seedgo"]),
|
|
patch(
|
|
"aipass.drone.apps.modules.scan.scan",
|
|
return_value=[{"name": "audit"}],
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_scan_failure(self) -> None:
|
|
"""scan returning None means failure -> exit 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "scan", "@seedgo"]),
|
|
patch("aipass.drone.apps.modules.scan.scan", return_value=None),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
class TestMainActivate:
|
|
"""drone activate command."""
|
|
|
|
def test_activate_no_target(self) -> None:
|
|
"""activate with no target shows help and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "activate"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_activate_help_flag(self) -> None:
|
|
"""activate --help shows help and returns 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "activate", "--help"]):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_activate_with_target(self) -> None:
|
|
"""activate with target delegates to _handle_activate."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "activate", "@seedgo"]),
|
|
patch(f"{_DRONE}._handle_activate", return_value=0) as mock_act,
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
mock_act.assert_called_once_with("@seedgo")
|
|
|
|
|
|
class TestMainList:
|
|
"""drone list command."""
|
|
|
|
def test_list_delegates(self) -> None:
|
|
"""list delegates to _handle_list."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "list"]),
|
|
patch(f"{_DRONE}._handle_list", return_value=0) as mock_list,
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
mock_list.assert_called_once()
|
|
|
|
|
|
class TestMainRemove:
|
|
"""drone remove command."""
|
|
|
|
def test_remove_no_name(self) -> None:
|
|
"""remove with no name returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with patch.object(sys, "argv", ["drone", "remove"]):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
def test_remove_with_name(self) -> None:
|
|
"""remove with name delegates to _handle_remove."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "remove", "audit"]),
|
|
patch(f"{_DRONE}._handle_remove", return_value=0) as mock_rm,
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
mock_rm.assert_called_once_with("audit")
|
|
|
|
|
|
class TestMainAtTarget:
|
|
"""drone @target routing."""
|
|
|
|
def test_at_target_delegates(self) -> None:
|
|
"""@target routes to _handle_target."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "@seedgo", "audit"]),
|
|
patch(f"{_DRONE}._handle_target", return_value=0) as mock_tgt,
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
mock_tgt.assert_called_once_with(["@seedgo", "audit"])
|
|
|
|
|
|
class TestMainModuleRouting:
|
|
"""drone bare module name routing."""
|
|
|
|
def test_discovered_module_bool_true(self) -> None:
|
|
"""Discovered module returning True yields exit 0."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
mock_mod = type(sys)("fake_mod")
|
|
mock_mod.handle_command = lambda cmd, args: True
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "config", "list"]),
|
|
patch(
|
|
f"{_DRONE}._discover_modules",
|
|
return_value=[("config", "Config module")],
|
|
),
|
|
patch(
|
|
f"{_DRONE}.importlib.import_module",
|
|
return_value=mock_mod,
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_discovered_module_bool_false(self) -> None:
|
|
"""Discovered module returning False yields exit 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
mock_mod = type(sys)("fake_mod")
|
|
mock_mod.handle_command = lambda cmd, args: False
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "config", "broken"]),
|
|
patch(
|
|
f"{_DRONE}._discover_modules",
|
|
return_value=[("config", "Config module")],
|
|
),
|
|
patch(
|
|
f"{_DRONE}.importlib.import_module",
|
|
return_value=mock_mod,
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
def test_discovered_module_dict_result(self) -> None:
|
|
"""Discovered module returning dict uses stdout/stderr/exit_code."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
mock_mod = type(sys)("fake_mod")
|
|
mock_mod.handle_command = lambda cmd, args: {
|
|
"stdout": "output",
|
|
"stderr": "",
|
|
"exit_code": 0,
|
|
}
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "config", "list"]),
|
|
patch(
|
|
f"{_DRONE}._discover_modules",
|
|
return_value=[("config", "Config module")],
|
|
),
|
|
patch(
|
|
f"{_DRONE}.importlib.import_module",
|
|
return_value=mock_mod,
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_discovered_module_exception(self) -> None:
|
|
"""Module raising exception returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "config", "list"]),
|
|
patch(
|
|
f"{_DRONE}._discover_modules",
|
|
return_value=[("config", "Config module")],
|
|
),
|
|
patch(
|
|
f"{_DRONE}.importlib.import_module",
|
|
side_effect=ImportError("nope"),
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
class TestMainCustomCommand:
|
|
"""drone custom command matching fallback."""
|
|
|
|
def test_custom_command_matched(self) -> None:
|
|
"""Custom command matched returns its result."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "audit", "aipass"]),
|
|
patch(f"{_DRONE}._discover_modules", return_value=[]),
|
|
patch(f"{_DRONE}._handle_custom_command", return_value=0),
|
|
):
|
|
result = main()
|
|
assert result == 0
|
|
|
|
def test_custom_command_not_matched(self) -> None:
|
|
"""Unmatched command falls through to unknown."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "nonexistent_cmd"]),
|
|
patch(f"{_DRONE}._discover_modules", return_value=[]),
|
|
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
|
|
patch(
|
|
"aipass.drone.apps.modules.resolver.branch_exists",
|
|
return_value=False,
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
class TestMainUnknownCommand:
|
|
"""drone unknown command handling with branch hint."""
|
|
|
|
def test_unknown_bare_branch_name(self) -> None:
|
|
"""Bare branch name shows @ prefix hint."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "seedgo"]),
|
|
patch(f"{_DRONE}._discover_modules", return_value=[]),
|
|
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
|
|
patch(
|
|
"aipass.drone.apps.modules.resolver.branch_exists",
|
|
return_value=True,
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
def test_unknown_branch_check_fails(self) -> None:
|
|
"""Exception in branch_exists doesn't crash -- still returns 1."""
|
|
from aipass.drone.apps.drone import main
|
|
|
|
with (
|
|
patch.object(sys, "argv", ["drone", "broken_cmd"]),
|
|
patch(f"{_DRONE}._discover_modules", return_value=[]),
|
|
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
|
|
patch(
|
|
"aipass.drone.apps.modules.resolver.branch_exists",
|
|
side_effect=Exception("boom"),
|
|
),
|
|
):
|
|
result = main()
|
|
assert result == 1
|
|
|
|
|
|
# ===========================================================================
|
|
# _handle_systems() paths
|
|
# ===========================================================================
|
|
|
|
|
|
class TestHandleSystems:
|
|
"""_handle_systems() logic paths."""
|
|
|
|
def test_no_registry(self) -> None:
|
|
"""No registry in CWD tree returns 0 with message."""
|
|
from aipass.drone.apps.drone import _handle_systems
|
|
|
|
with patch(f"{_DRONE}._cwd_has_registry", return_value=False):
|
|
result = _handle_systems()
|
|
assert result == 0
|
|
|
|
def test_with_branches_and_modules(self) -> None:
|
|
"""Normal case with modules and branches."""
|
|
from aipass.drone.apps.drone import _handle_systems
|
|
|
|
branches = [
|
|
{"name": "drone", "profile": "library", "description": "Router"},
|
|
{"name": "myapp", "profile": "agent"},
|
|
]
|
|
with (
|
|
patch(f"{_DRONE}._cwd_has_registry", return_value=True),
|
|
patch(f"{_DRONE}.get_all_branches", return_value=branches),
|
|
patch(f"{_DRONE}.list_modules", return_value=["git"]),
|
|
patch(
|
|
f"{_DRONE}.get_module_info",
|
|
return_value=type("I", (), {"description": "Git ops"})(),
|
|
),
|
|
):
|
|
result = _handle_systems()
|
|
assert result == 0
|
|
|
|
def test_aipass_home_hint(self) -> None:
|
|
"""Shows hint when AIPASS_HOME not set and drone not in branches."""
|
|
import os
|
|
|
|
from aipass.drone.apps.drone import _handle_systems
|
|
|
|
with (
|
|
patch(f"{_DRONE}._cwd_has_registry", return_value=True),
|
|
patch(f"{_DRONE}.get_all_branches", return_value=[]),
|
|
patch(f"{_DRONE}.list_modules", return_value=[]),
|
|
patch.dict(os.environ, {}, clear=False),
|
|
):
|
|
os.environ.pop("AIPASS_HOME", None)
|
|
result = _handle_systems()
|
|
assert result == 0
|
|
|
|
|
|
# ===========================================================================
|
|
# _handle_module() paths
|
|
# ===========================================================================
|
|
|
|
|
|
class TestHandleModule:
|
|
"""_handle_module() introspection, help, and command routing."""
|
|
|
|
def test_no_args_introspection(self) -> None:
|
|
"""No args shows introspection text."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(
|
|
f"{_DRONE}.get_module_introspective",
|
|
return_value="Module info",
|
|
):
|
|
result = _handle_module("git", [])
|
|
assert result == 0
|
|
|
|
def test_no_args_no_introspection(self) -> None:
|
|
"""No args with empty introspection text shows fallback."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(f"{_DRONE}.get_module_introspective", return_value=""):
|
|
result = _handle_module("git", [])
|
|
assert result == 0
|
|
|
|
def test_help_flag(self) -> None:
|
|
"""--help shows help text."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(f"{_DRONE}.get_module_help", return_value="Help text"):
|
|
result = _handle_module("git", ["--help"])
|
|
assert result == 0
|
|
|
|
def test_help_no_text(self) -> None:
|
|
"""--help with no text shows fallback."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(f"{_DRONE}.get_module_help", return_value=""):
|
|
result = _handle_module("git", ["--help"])
|
|
assert result == 0
|
|
|
|
def test_command_routing(self) -> None:
|
|
"""Command routes through route_module_command."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(
|
|
f"{_DRONE}.route_module_command",
|
|
return_value={"stdout": "ok", "stderr": "", "exit_code": 0},
|
|
):
|
|
result = _handle_module("git", ["status"])
|
|
assert result == 0
|
|
|
|
def test_command_import_error(self) -> None:
|
|
"""ImportError during module command returns 1."""
|
|
from aipass.drone.apps.drone import _handle_module
|
|
|
|
with patch(
|
|
f"{_DRONE}.route_module_command",
|
|
side_effect=ImportError("missing"),
|
|
):
|
|
result = _handle_module("git", ["status"])
|
|
assert result == 1
|
|
|
|
|
|
# ===========================================================================
|
|
# _handle_target() paths
|
|
# ===========================================================================
|
|
|
|
|
|
class TestHandleTarget:
|
|
"""_handle_target() routing for @branch commands."""
|
|
|
|
def test_module_route(self) -> None:
|
|
"""Module target (non-interactive) routes to _handle_module."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=True),
|
|
patch(f"{_DRONE}._handle_module", return_value=0) as mock_hm,
|
|
):
|
|
result = _handle_target(["@git", "diff"])
|
|
assert result == 0
|
|
mock_hm.assert_called_once_with("git", ["diff"])
|
|
|
|
def test_no_args_introspection(self) -> None:
|
|
"""@target with no args routes via route_command with interactive=True."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
mock_result = CommandResult(
|
|
stdout="",
|
|
stderr="",
|
|
exit_code=0,
|
|
branch="seedgo",
|
|
command="",
|
|
)
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=False),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result) as mock_route,
|
|
):
|
|
result = _handle_target(["@seedgo"])
|
|
assert result == 0
|
|
mock_route.assert_called_once_with("@seedgo", interactive=True)
|
|
|
|
def test_help_flag(self) -> None:
|
|
"""@target --help routes via route_command with interactive=True."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
mock_result = CommandResult(
|
|
stdout="",
|
|
stderr="",
|
|
exit_code=0,
|
|
branch="seedgo",
|
|
command="--help",
|
|
)
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=False),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result) as mock_route,
|
|
):
|
|
result = _handle_target(["@seedgo", "--help"])
|
|
assert result == 0
|
|
mock_route.assert_called_once_with("@seedgo", "--help", interactive=True)
|
|
|
|
def test_command_routing(self) -> None:
|
|
"""@target command routes via route_command."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
mock_result = CommandResult(
|
|
stdout="output",
|
|
stderr="",
|
|
exit_code=0,
|
|
branch="seedgo",
|
|
command="audit",
|
|
)
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=False),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result),
|
|
):
|
|
result = _handle_target(["@seedgo", "audit", "aipass"])
|
|
assert result == 0
|
|
|
|
def test_short_help_flag(self) -> None:
|
|
"""@target -h routes via route_command with interactive=True."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
mock_result = CommandResult(stdout="", stderr="", exit_code=0, branch="seedgo", command="-h")
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=False),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result) as mock_route,
|
|
):
|
|
result = _handle_target(["@seedgo", "-h"])
|
|
assert result == 0
|
|
mock_route.assert_called_once_with("@seedgo", "-h", interactive=True)
|
|
|
|
def test_status_routes_interactive(self) -> None:
|
|
"""status command routes with interactive=True for Rich color output."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
mock_result = CommandResult(stdout="", stderr="", exit_code=0, branch="hooks", command="status")
|
|
with (
|
|
patch(f"{_DRONE}.is_module", return_value=False),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result) as mock_route,
|
|
):
|
|
result = _handle_target(["@hooks", "status"])
|
|
assert result == 0
|
|
call_kwargs = mock_route.call_args.kwargs
|
|
assert call_kwargs["interactive"] is True
|
|
|
|
def test_help_flag_module_fallback(self) -> None:
|
|
"""--help BranchNotFoundError for a module falls back to _handle_module."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.modules import BranchNotFoundError
|
|
|
|
with (
|
|
patch(f"{_DRONE}.is_module", side_effect=[False, True]),
|
|
patch(f"{_DRONE}.route_command", side_effect=BranchNotFoundError("not found")),
|
|
patch(f"{_DRONE}._handle_module", return_value=0) as mock_hm,
|
|
):
|
|
result = _handle_target(["@seedgo", "--help"])
|
|
assert result == 0
|
|
mock_hm.assert_called_once_with("seedgo", ["--help"])
|
|
|
|
def test_branch_not_found_module_fallback(self) -> None:
|
|
"""BranchNotFoundError for a module falls back to _handle_module."""
|
|
from aipass.drone.apps.drone import _handle_target
|
|
from aipass.drone.apps.modules import BranchNotFoundError
|
|
|
|
with (
|
|
patch(f"{_DRONE}.is_module", side_effect=[False, True]),
|
|
patch(
|
|
f"{_DRONE}.route_command",
|
|
side_effect=BranchNotFoundError("not found"),
|
|
),
|
|
patch(f"{_DRONE}._handle_module", return_value=0) as mock_hm,
|
|
):
|
|
result = _handle_target(["@seedgo", "audit"])
|
|
assert result == 0
|
|
mock_hm.assert_called_once()
|
|
|
|
|
|
# ===========================================================================
|
|
# _handle_custom_command() paths
|
|
# ===========================================================================
|
|
|
|
|
|
class TestHandleCustomCommand:
|
|
"""_handle_custom_command() matching and routing."""
|
|
|
|
_MATCH = "aipass.drone.apps.modules.commands.match"
|
|
|
|
def test_no_match(self) -> None:
|
|
"""No match returns -1 sentinel."""
|
|
from aipass.drone.apps.drone import _handle_custom_command
|
|
|
|
with patch(self._MATCH, return_value=None):
|
|
result = _handle_custom_command(["unknown"])
|
|
assert result == -1
|
|
|
|
def test_matched_routes_success(self) -> None:
|
|
"""Matched command routes and returns exit code."""
|
|
from aipass.drone.apps.drone import _handle_custom_command
|
|
from aipass.drone.apps.handlers.executor import CommandResult
|
|
|
|
cmd_data = {
|
|
"target": "@seedgo",
|
|
"command": "audit",
|
|
"args": ["aipass"],
|
|
}
|
|
mock_result = CommandResult(
|
|
stdout="ok",
|
|
stderr="",
|
|
exit_code=0,
|
|
branch="seedgo",
|
|
command="audit",
|
|
)
|
|
with (
|
|
patch(self._MATCH, return_value=(cmd_data, [])),
|
|
patch(f"{_DRONE}.route_command", return_value=mock_result),
|
|
):
|
|
result = _handle_custom_command(["audit"])
|
|
assert result == 0
|
|
|
|
|
|
# ===========================================================================
|
|
# Helper functions
|
|
# ===========================================================================
|
|
|
|
|
|
class TestReadInboxMessageId:
|
|
"""_read_inbox_message_id() edge cases."""
|
|
|
|
def test_valid_index(self, tmp_path: Path) -> None:
|
|
"""Returns message ID for valid display-order index (reversed from array)."""
|
|
inbox = tmp_path / "inbox.json"
|
|
inbox.write_text(
|
|
json.dumps({"messages": [{"id": "newest"}, {"id": "oldest"}]}),
|
|
encoding="utf-8",
|
|
)
|
|
from aipass.drone.apps.drone import _read_inbox_message_id
|
|
|
|
assert _read_inbox_message_id(inbox, 1) == "oldest"
|
|
assert _read_inbox_message_id(inbox, 2) == "newest"
|
|
|
|
def test_out_of_range(self, tmp_path: Path) -> None:
|
|
"""Returns None for out-of-range index."""
|
|
inbox = tmp_path / "inbox.json"
|
|
inbox.write_text(
|
|
json.dumps({"messages": [{"id": "abc"}]}),
|
|
encoding="utf-8",
|
|
)
|
|
from aipass.drone.apps.drone import _read_inbox_message_id
|
|
|
|
assert _read_inbox_message_id(inbox, 5) is None
|
|
assert _read_inbox_message_id(inbox, 0) is None
|
|
|
|
def test_corrupt_file(self, tmp_path: Path) -> None:
|
|
"""Returns None for corrupt inbox."""
|
|
inbox = tmp_path / "inbox.json"
|
|
inbox.write_text("{bad json", encoding="utf-8")
|
|
from aipass.drone.apps.drone import _read_inbox_message_id
|
|
|
|
assert _read_inbox_message_id(inbox, 1) is None
|
|
|
|
|
|
class TestDiscoverModules:
|
|
"""_discover_modules() auto-discovery."""
|
|
|
|
def test_discovers_modules_with_handle_command(self) -> None:
|
|
"""Modules with handle_command are discovered."""
|
|
from aipass.drone.apps.drone import _discover_modules
|
|
|
|
result = _discover_modules()
|
|
names = [m[0] for m in result]
|
|
assert "git_module" in names
|
|
assert "resolver" in names
|
|
|
|
def test_skips_private_files(self) -> None:
|
|
"""Files starting with _ are skipped."""
|
|
from aipass.drone.apps.drone import _discover_modules
|
|
|
|
result = _discover_modules()
|
|
names = [m[0] for m in result]
|
|
assert "__init__" not in names
|
|
|
|
|
|
class TestCliEntryPoint:
|
|
"""cli.py entry point."""
|
|
|
|
def test_cli_main_calls_drone_main(self) -> None:
|
|
"""cli.main() calls drone main and exits."""
|
|
from aipass.drone.cli import main as cli_main
|
|
|
|
with (
|
|
patch("aipass.drone.cli._drone_main", return_value=0),
|
|
pytest.raises(SystemExit) as exc_info,
|
|
):
|
|
cli_main()
|
|
assert exc_info.value.code == 0
|