✨ feat(drone): add internal module routing system for drone commands
- introduce drone module registry with metadata and command routing - add seedgo as a registered internal drone module with adapter - update drone CLI to route commands to modules or branches by @target - enhance `drone systems` to list registered modules with descriptions - provide module help integration alongside branch help ✨ feat(seedgo): create drone adapter for seedgo module commands - implement handle_command forwarding to seedgo CLI subprocess - provide get_help function with command-specific help text - support commands: check, audit, list, init with options and examples ✨ feat(seedgo): add __main__.py to support `python -m seedgo` execution ✨ feat(seedgo plugin): add drone-compliance plugin to verify drone adapter compliance - checks existence of drone_adapter.py and required interface elements - validates presence of DRONE_MODULE dict, handle_command(), and get_help() - reports compliance results with pass/fail and score with detailed checks - skips non-target packages and handles syntax errors gracefully ✅ test(drone modules): add comprehensive tests for drone module registry and routing - test listing, info retrieval, command routing, and help retrieval - test CLI integration for `drone systems` and `drone @seedgo` commands - verify module priority over branch routing and error handling ✅ test(seedgo drone compliance): add tests for drone-compliance plugin - test passing cases with proper adapter and interface - test failures for missing adapter, metadata, or functions - test warnings for missing get_help function - test skipping non-target packages and file-not-found cases - test handling of syntax errors in adapter files
This commit is contained in:
AIOSAI
+76
-14
@@ -7,9 +7,10 @@ Usage:
|
||||
drone Show available commands
|
||||
drone --help Show help
|
||||
drone --version Show version
|
||||
drone systems List registered branches
|
||||
drone systems List registered branches and modules
|
||||
drone @branch command [args] Route command to branch
|
||||
drone @branch --help Show help for branch
|
||||
drone @module command [args] Route command to internal module
|
||||
drone @branch --help Show help for branch or module
|
||||
|
||||
Zero external dependencies — stdlib only (sys).
|
||||
"""
|
||||
@@ -26,22 +27,30 @@ from aipass.drone import (
|
||||
list_branches,
|
||||
route_command,
|
||||
)
|
||||
from aipass.drone.modules import (
|
||||
is_module,
|
||||
list_modules,
|
||||
get_module_info,
|
||||
route_module_command,
|
||||
get_module_help,
|
||||
)
|
||||
|
||||
_HELP_TEXT = """\
|
||||
drone — AIPass branch command router
|
||||
drone — AIPass command router
|
||||
|
||||
Usage:
|
||||
drone Show this help
|
||||
drone --help Show this help
|
||||
drone --version Show version
|
||||
drone systems List registered branches
|
||||
drone @branch command [args] Route command to branch
|
||||
drone @branch --help Show help for branch
|
||||
drone systems List registered branches and modules
|
||||
drone @target command [args] Route command to branch or module
|
||||
drone @target --help Show help for branch or module
|
||||
|
||||
Examples:
|
||||
drone systems
|
||||
drone @seedgo audit src/
|
||||
drone @seedgo check src/myfile.py
|
||||
drone @flow status
|
||||
drone @flow run --verbose
|
||||
drone @flow --help
|
||||
"""
|
||||
|
||||
@@ -82,23 +91,49 @@ def main() -> None:
|
||||
|
||||
|
||||
def _cmd_systems() -> None:
|
||||
"""Handle `drone systems` — list registered branches."""
|
||||
"""Handle `drone systems` — list registered branches and modules."""
|
||||
branches = list_branches()
|
||||
if not branches:
|
||||
print("No branches registered.")
|
||||
modules = list_modules()
|
||||
|
||||
if not branches and not modules:
|
||||
print("No branches or modules registered.")
|
||||
sys.exit(0)
|
||||
|
||||
print(f"Registered branches ({len(branches)}):")
|
||||
for name in sorted(branches):
|
||||
print(f" {name}")
|
||||
if modules:
|
||||
print(f"Modules ({len(modules)}):")
|
||||
for name in modules:
|
||||
info = get_module_info(name)
|
||||
if info:
|
||||
print(f" @{name} {info.description}")
|
||||
else:
|
||||
print(f" @{name} (not available)")
|
||||
if branches:
|
||||
print()
|
||||
|
||||
if branches:
|
||||
print(f"Branches ({len(branches)}):")
|
||||
for name in sorted(branches):
|
||||
print(f" {name}")
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def _cmd_branch(args: list[str]) -> None:
|
||||
"""Handle `drone @branch command [args]` or `drone @branch --help`."""
|
||||
"""Handle `drone @target command [args]` or `drone @target --help`.
|
||||
|
||||
Checks internal modules first (e.g. @seedgo), then falls back
|
||||
to branch registry for external branches.
|
||||
"""
|
||||
target = args[0]
|
||||
rest = args[1:]
|
||||
module_name = target.lstrip("@")
|
||||
|
||||
# Check if this is a registered internal module
|
||||
if is_module(module_name):
|
||||
_cmd_module(module_name, rest)
|
||||
return
|
||||
|
||||
# Fall through to branch routing
|
||||
# drone @branch --help
|
||||
if not rest or rest == ["--help"]:
|
||||
try:
|
||||
@@ -133,3 +168,30 @@ def _cmd_branch(args: list[str]) -> None:
|
||||
if result.stderr:
|
||||
print(result.stderr, end="", file=sys.stderr)
|
||||
sys.exit(result.exit_code)
|
||||
|
||||
|
||||
def _cmd_module(name: str, args: list[str]) -> None:
|
||||
"""Handle routing to an internal module (e.g. @seedgo)."""
|
||||
# drone @module --help (or no args)
|
||||
if not args or args == ["--help"]:
|
||||
help_text = get_module_help(name)
|
||||
if help_text:
|
||||
print(help_text, end="")
|
||||
else:
|
||||
print(f"No help available for @{name}.")
|
||||
sys.exit(0)
|
||||
|
||||
command = args[0]
|
||||
cmd_args = args[1:] if len(args) > 1 else None
|
||||
|
||||
try:
|
||||
result = route_module_command(name, command, cmd_args)
|
||||
except (ImportError, AttributeError) as exc:
|
||||
print(f"drone: module @{name} is registered but not available: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if result.get("stdout"):
|
||||
print(result["stdout"], end="")
|
||||
if result.get("stderr"):
|
||||
print(result["stderr"], end="", file=sys.stderr)
|
||||
sys.exit(result.get("exit_code", 0))
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Internal module registry for drone.
|
||||
|
||||
Routes @module commands to Python packages installed alongside drone,
|
||||
as opposed to external branches in BRANCH_REGISTRY.json.
|
||||
|
||||
Modules register by providing a drone_adapter module with:
|
||||
- DRONE_MODULE dict (name, version, description)
|
||||
- handle_command(command, args) -> dict with stdout/stderr/exit_code
|
||||
- get_help(command=None) -> str
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from dataclasses import dataclass
|
||||
|
||||
# Maps module name -> import path for its drone_adapter
|
||||
_MODULE_REGISTRY: dict[str, str] = {
|
||||
"seedgo": "seedgo.drone_adapter",
|
||||
# Future modules register here:
|
||||
# "prax": "aipass.prax.drone_adapter",
|
||||
# "cortex": "aipass.cortex.drone_adapter",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModuleInfo:
|
||||
"""Metadata about a registered module."""
|
||||
|
||||
name: str
|
||||
version: str
|
||||
description: str
|
||||
adapter_path: str
|
||||
|
||||
|
||||
def list_modules() -> list[str]:
|
||||
"""Return sorted list of registered module names."""
|
||||
return sorted(_MODULE_REGISTRY.keys())
|
||||
|
||||
|
||||
def get_module_info(name: str) -> ModuleInfo | None:
|
||||
"""Get module metadata without executing anything.
|
||||
|
||||
Returns None if the module is not registered or not importable.
|
||||
"""
|
||||
adapter_path = _MODULE_REGISTRY.get(name)
|
||||
if adapter_path is None:
|
||||
return None
|
||||
try:
|
||||
mod = importlib.import_module(adapter_path)
|
||||
meta = getattr(mod, "DRONE_MODULE", {})
|
||||
return ModuleInfo(
|
||||
name=meta.get("name", name),
|
||||
version=meta.get("version", "unknown"),
|
||||
description=meta.get("description", ""),
|
||||
adapter_path=adapter_path,
|
||||
)
|
||||
except ImportError:
|
||||
return None
|
||||
|
||||
|
||||
def is_module(name: str) -> bool:
|
||||
"""Check if name is a registered module (doesn't verify importability)."""
|
||||
return name in _MODULE_REGISTRY
|
||||
|
||||
|
||||
def route_module_command(name: str, command: str, args: list[str] | None = None) -> dict:
|
||||
"""Route a command to a module's drone adapter.
|
||||
|
||||
Returns dict with keys: stdout, stderr, exit_code.
|
||||
Raises: KeyError if module not registered, ImportError if adapter missing,
|
||||
AttributeError if adapter lacks handle_command.
|
||||
"""
|
||||
adapter_path = _MODULE_REGISTRY[name]
|
||||
mod = importlib.import_module(adapter_path)
|
||||
handler = getattr(mod, "handle_command")
|
||||
return handler(command, args)
|
||||
|
||||
|
||||
def get_module_help(name: str, command: str | None = None) -> str:
|
||||
"""Get help text from a module's drone adapter.
|
||||
|
||||
Returns help string, or empty string if unavailable.
|
||||
"""
|
||||
adapter_path = _MODULE_REGISTRY.get(name)
|
||||
if adapter_path is None:
|
||||
return ""
|
||||
try:
|
||||
mod = importlib.import_module(adapter_path)
|
||||
help_fn = getattr(mod, "get_help", None)
|
||||
if help_fn is None:
|
||||
return ""
|
||||
return help_fn(command)
|
||||
except (ImportError, AttributeError):
|
||||
return ""
|
||||
|
||||
|
||||
def register_module(name: str, adapter_path: str) -> None:
|
||||
"""Register a new module. Used for dynamic registration (e.g., plugins)."""
|
||||
_MODULE_REGISTRY[name] = adapter_path
|
||||
@@ -0,0 +1,6 @@
|
||||
"""Allow running seedgo as a module: python -m seedgo."""
|
||||
|
||||
from seedgo.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Drone adapter for seedgo — code standards framework.
|
||||
|
||||
Provides the drone module interface so `drone @seedgo` commands work.
|
||||
Delegates to the seedgo CLI via subprocess for clean stdout/stderr capture.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
DRONE_MODULE = {
|
||||
"name": "seedgo",
|
||||
"version": "1.0.0",
|
||||
"description": "Code standards framework — check and audit code against configurable standards",
|
||||
}
|
||||
|
||||
_HELP_TEXT = """\
|
||||
seedgo — code standards framework
|
||||
|
||||
Commands:
|
||||
check [FILE ...] Run checks on files (or project if no files given)
|
||||
audit [PATH ...] Alias for check (run audit on paths)
|
||||
list Show all discovered plugins
|
||||
init Initialize .seedgo/ config in current directory
|
||||
|
||||
Options (for check/audit):
|
||||
--format FORMAT Output format: human, json, github (default: human)
|
||||
--threshold N Pass threshold 0-100 (default: 75)
|
||||
--plugin NAME Run only this plugin
|
||||
|
||||
Examples:
|
||||
drone @seedgo check src/myfile.py
|
||||
drone @seedgo audit src/
|
||||
drone @seedgo list
|
||||
drone @seedgo init --profile strict
|
||||
"""
|
||||
|
||||
|
||||
def handle_command(command: str, args: list[str] | None = None) -> dict:
|
||||
"""Handle a command routed by drone.
|
||||
|
||||
Returns dict with stdout, stderr, exit_code.
|
||||
"""
|
||||
cmd_args = args or []
|
||||
|
||||
# Map commands to seedgo CLI arguments
|
||||
if command == "audit":
|
||||
# audit is an alias for check
|
||||
cli_args = ["check"] + cmd_args
|
||||
elif command in ("check", "list", "init"):
|
||||
cli_args = [command] + cmd_args
|
||||
else:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": f"seedgo: unknown command '{command}'\nRun 'drone @seedgo --help' for usage.\n",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "seedgo"] + cli_args,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
return {
|
||||
"stdout": result.stdout,
|
||||
"stderr": result.stderr,
|
||||
"exit_code": result.returncode,
|
||||
}
|
||||
except subprocess.TimeoutExpired:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "seedgo: command timed out after 120 seconds\n",
|
||||
"exit_code": 1,
|
||||
}
|
||||
except FileNotFoundError:
|
||||
return {
|
||||
"stdout": "",
|
||||
"stderr": "seedgo: Python executable not found\n",
|
||||
"exit_code": 1,
|
||||
}
|
||||
|
||||
|
||||
def get_help(command: str | None = None) -> str:
|
||||
"""Return help text for drone @seedgo."""
|
||||
if command is None:
|
||||
return _HELP_TEXT
|
||||
|
||||
# For specific command help, delegate to seedgo
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "seedgo", command, "--help"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
return result.stdout if result.stdout else _HELP_TEXT
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError):
|
||||
return _HELP_TEXT
|
||||
@@ -0,0 +1,212 @@
|
||||
"""Drone compliance check — verifies modules have proper drone integration.
|
||||
|
||||
Checks that Python packages in the aipass ecosystem provide:
|
||||
1. A drone_adapter module
|
||||
2. DRONE_MODULE metadata dict (name, version, description)
|
||||
3. handle_command() function
|
||||
4. get_help() function
|
||||
|
||||
This check runs on __init__.py files to identify packages, then looks
|
||||
for their drone_adapter.py sibling.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from seedgo.models import CheckItem, CheckResult, Severity
|
||||
|
||||
PLUGIN_NAME = "drone-compliance"
|
||||
PLUGIN_DESCRIPTION = "Verify modules provide drone adapter interface"
|
||||
PLUGIN_VERSION = "1.0.0"
|
||||
FILE_TYPES = ["__init__.py"]
|
||||
|
||||
|
||||
def check(file_path: str, config: dict | None = None) -> CheckResult:
|
||||
"""Check if the package containing this __init__.py is drone-compliant.
|
||||
|
||||
Args:
|
||||
file_path: Absolute path to an __init__.py file.
|
||||
config: Optional plugin config dict. Supports:
|
||||
target_packages (list[str]): Package names to check.
|
||||
Defaults to ["aipass", "seedgo"].
|
||||
|
||||
Returns:
|
||||
CheckResult with compliance check items.
|
||||
"""
|
||||
path = Path(file_path)
|
||||
|
||||
if not path.exists():
|
||||
return CheckResult(
|
||||
plugin=PLUGIN_NAME,
|
||||
passed=True,
|
||||
checks=[],
|
||||
file_path=file_path,
|
||||
metadata={"skipped": True, "reason": "file_not_found"},
|
||||
)
|
||||
|
||||
package_dir = path.parent
|
||||
package_name = package_dir.name
|
||||
|
||||
# Determine if this package should be checked
|
||||
cfg = config or {}
|
||||
target_packages = cfg.get("target_packages", ["aipass", "seedgo"])
|
||||
|
||||
should_check = package_name in target_packages
|
||||
|
||||
if not should_check:
|
||||
return CheckResult(
|
||||
plugin=PLUGIN_NAME,
|
||||
passed=True,
|
||||
checks=[
|
||||
CheckItem(
|
||||
name="scope",
|
||||
passed=True,
|
||||
message=f"Package '{package_name}' is not a drone-target module, skipped.",
|
||||
severity=Severity.INFO,
|
||||
)
|
||||
],
|
||||
score=100,
|
||||
file_path=file_path,
|
||||
metadata={"skipped": True},
|
||||
)
|
||||
|
||||
checks: list[CheckItem] = []
|
||||
adapter_path = package_dir / "drone_adapter.py"
|
||||
|
||||
# Check 1: drone_adapter.py exists
|
||||
has_adapter = adapter_path.exists()
|
||||
checks.append(
|
||||
CheckItem(
|
||||
name="adapter-exists",
|
||||
passed=has_adapter,
|
||||
message=(
|
||||
f"drone_adapter.py found in {package_name}/"
|
||||
if has_adapter
|
||||
else f"Missing drone_adapter.py in {package_name}/ — module is not drone-routable"
|
||||
),
|
||||
severity=Severity.ERROR if not has_adapter else Severity.INFO,
|
||||
fix_hint=(
|
||||
"Create drone_adapter.py with DRONE_MODULE, handle_command(), and get_help()"
|
||||
if not has_adapter
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
if not has_adapter:
|
||||
return CheckResult(
|
||||
plugin=PLUGIN_NAME,
|
||||
passed=False,
|
||||
checks=checks,
|
||||
score=0,
|
||||
file_path=file_path,
|
||||
metadata={},
|
||||
)
|
||||
|
||||
# Parse the adapter file with AST
|
||||
try:
|
||||
source = adapter_path.read_text()
|
||||
tree = ast.parse(source)
|
||||
except (SyntaxError, OSError) as exc:
|
||||
checks.append(
|
||||
CheckItem(
|
||||
name="adapter-parseable",
|
||||
passed=False,
|
||||
message=f"drone_adapter.py has syntax error: {exc}",
|
||||
severity=Severity.ERROR,
|
||||
)
|
||||
)
|
||||
return CheckResult(
|
||||
plugin=PLUGIN_NAME,
|
||||
passed=False,
|
||||
checks=checks,
|
||||
score=0,
|
||||
file_path=file_path,
|
||||
metadata={},
|
||||
)
|
||||
|
||||
# Check 2: DRONE_MODULE dict exists
|
||||
has_meta = any(
|
||||
isinstance(node, ast.Assign)
|
||||
and any(isinstance(t, ast.Name) and t.id == "DRONE_MODULE" for t in node.targets)
|
||||
for node in ast.walk(tree)
|
||||
)
|
||||
checks.append(
|
||||
CheckItem(
|
||||
name="drone-module-meta",
|
||||
passed=has_meta,
|
||||
message=(
|
||||
"DRONE_MODULE metadata dict found"
|
||||
if has_meta
|
||||
else "Missing DRONE_MODULE dict — drone can't read module metadata"
|
||||
),
|
||||
severity=Severity.ERROR if not has_meta else Severity.INFO,
|
||||
fix_hint=(
|
||||
'Add: DRONE_MODULE = {"name": "...", "version": "...", "description": "..."}'
|
||||
if not has_meta
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# Check 3: handle_command() function exists
|
||||
functions = [
|
||||
node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
]
|
||||
|
||||
has_handle = "handle_command" in functions
|
||||
checks.append(
|
||||
CheckItem(
|
||||
name="handle-command",
|
||||
passed=has_handle,
|
||||
message=(
|
||||
"handle_command() function found"
|
||||
if has_handle
|
||||
else "Missing handle_command() — drone can't route commands to this module"
|
||||
),
|
||||
severity=Severity.ERROR if not has_handle else Severity.INFO,
|
||||
fix_hint=(
|
||||
"Add: def handle_command(command: str, args: list[str] | None = None) -> dict:"
|
||||
if not has_handle
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# Check 4: get_help() function exists
|
||||
has_help = "get_help" in functions
|
||||
checks.append(
|
||||
CheckItem(
|
||||
name="get-help",
|
||||
passed=has_help,
|
||||
message=(
|
||||
"get_help() function found"
|
||||
if has_help
|
||||
else "Missing get_help() — drone can't show help for this module"
|
||||
),
|
||||
severity=Severity.WARNING if not has_help else Severity.INFO,
|
||||
fix_hint="Add: def get_help(command: str | None = None) -> str:" if not has_help else None,
|
||||
)
|
||||
)
|
||||
|
||||
# Score: errors block pass, warnings degrade
|
||||
error_checks = [c for c in checks if not c.passed and c.severity == Severity.ERROR]
|
||||
all_passed = len(error_checks) == 0
|
||||
|
||||
# Simple scoring: deduct for failures
|
||||
failed_errors = len(error_checks)
|
||||
failed_warnings = len([c for c in checks if not c.passed and c.severity == Severity.WARNING])
|
||||
total_weight = len(checks)
|
||||
deductions = failed_errors * 1.0 + failed_warnings * 0.5
|
||||
score = int(((total_weight - deductions) / max(total_weight, 1)) * 100)
|
||||
|
||||
return CheckResult(
|
||||
plugin=PLUGIN_NAME,
|
||||
passed=all_passed,
|
||||
checks=checks,
|
||||
score=score,
|
||||
file_path=file_path,
|
||||
metadata={"adapter_path": str(adapter_path)},
|
||||
)
|
||||
@@ -0,0 +1,176 @@
|
||||
"""
|
||||
Tests for the drone-compliance seedgo plugin.
|
||||
|
||||
Covers:
|
||||
- Passes on packages with proper drone_adapter.py
|
||||
- Fails on packages missing drone_adapter.py
|
||||
- Fails on adapter missing DRONE_MODULE / handle_command / get_help
|
||||
- Skips non-target packages
|
||||
- Handles syntax errors in adapter files
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
from seedgo.plugins.drone_compliance import PLUGIN_NAME, check
|
||||
|
||||
|
||||
class TestDroneCompliancePass:
|
||||
"""Packages with proper drone adapters should pass."""
|
||||
|
||||
def test_seedgo_adapter_passes(self, tmp_path: Path):
|
||||
"""seedgo's own drone_adapter.py passes all checks."""
|
||||
# Create a fake "seedgo" package with a proper adapter
|
||||
pkg = tmp_path / "seedgo"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text('"""Seedgo package."""\n')
|
||||
(pkg / "drone_adapter.py").write_text(
|
||||
textwrap.dedent("""\
|
||||
DRONE_MODULE = {"name": "seedgo", "version": "1.0.0", "description": "test"}
|
||||
|
||||
def handle_command(command, args=None):
|
||||
return {"stdout": "", "stderr": "", "exit_code": 0}
|
||||
|
||||
def get_help(command=None):
|
||||
return "help text"
|
||||
""")
|
||||
)
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["seedgo"]})
|
||||
|
||||
assert result.plugin == PLUGIN_NAME
|
||||
assert result.passed is True
|
||||
assert result.score == 100
|
||||
|
||||
def test_all_four_checks_present(self, tmp_path: Path):
|
||||
"""All 4 checks pass: adapter exists, DRONE_MODULE, handle_command, get_help."""
|
||||
pkg = tmp_path / "mymod"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
(pkg / "drone_adapter.py").write_text(
|
||||
textwrap.dedent("""\
|
||||
DRONE_MODULE = {"name": "mymod", "version": "0.1.0", "description": "test"}
|
||||
|
||||
def handle_command(command, args=None):
|
||||
return {}
|
||||
|
||||
def get_help(command=None):
|
||||
return ""
|
||||
""")
|
||||
)
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["mymod"]})
|
||||
assert result.passed is True
|
||||
|
||||
|
||||
class TestDroneComplianceFail:
|
||||
"""Packages missing drone adapter components should fail."""
|
||||
|
||||
def test_missing_adapter_fails(self, tmp_path: Path):
|
||||
"""Package without drone_adapter.py fails."""
|
||||
pkg = tmp_path / "seedgo"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["seedgo"]})
|
||||
|
||||
assert result.passed is False
|
||||
assert result.score == 0
|
||||
error_names = [c.name for c in result.checks if not c.passed]
|
||||
assert "adapter-exists" in error_names
|
||||
|
||||
def test_missing_drone_module_dict(self, tmp_path: Path):
|
||||
"""Adapter without DRONE_MODULE fails."""
|
||||
pkg = tmp_path / "mymod"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
(pkg / "drone_adapter.py").write_text(
|
||||
textwrap.dedent("""\
|
||||
def handle_command(command, args=None):
|
||||
return {}
|
||||
|
||||
def get_help(command=None):
|
||||
return ""
|
||||
""")
|
||||
)
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["mymod"]})
|
||||
assert result.passed is False
|
||||
failed = [c.name for c in result.checks if not c.passed]
|
||||
assert "drone-module-meta" in failed
|
||||
|
||||
def test_missing_handle_command(self, tmp_path: Path):
|
||||
"""Adapter without handle_command fails."""
|
||||
pkg = tmp_path / "mymod"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
(pkg / "drone_adapter.py").write_text(
|
||||
textwrap.dedent("""\
|
||||
DRONE_MODULE = {"name": "mymod", "version": "1.0.0", "description": "test"}
|
||||
|
||||
def get_help(command=None):
|
||||
return ""
|
||||
""")
|
||||
)
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["mymod"]})
|
||||
assert result.passed is False
|
||||
failed = [c.name for c in result.checks if not c.passed]
|
||||
assert "handle-command" in failed
|
||||
|
||||
def test_missing_get_help_is_warning(self, tmp_path: Path):
|
||||
"""Adapter without get_help is a warning, not an error — still passes."""
|
||||
pkg = tmp_path / "mymod"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
(pkg / "drone_adapter.py").write_text(
|
||||
textwrap.dedent("""\
|
||||
DRONE_MODULE = {"name": "mymod", "version": "1.0.0", "description": "test"}
|
||||
|
||||
def handle_command(command, args=None):
|
||||
return {}
|
||||
""")
|
||||
)
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["mymod"]})
|
||||
# Missing get_help is WARNING severity, not ERROR — should still pass
|
||||
assert result.passed is True
|
||||
warnings = [c for c in result.checks if not c.passed]
|
||||
assert any(c.name == "get-help" for c in warnings)
|
||||
|
||||
|
||||
class TestDroneComplianceSkip:
|
||||
"""Non-target packages should be skipped."""
|
||||
|
||||
def test_non_target_package_skipped(self, tmp_path: Path):
|
||||
"""Packages not in target_packages are skipped with pass."""
|
||||
pkg = tmp_path / "unrelated"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["seedgo"]})
|
||||
assert result.passed is True
|
||||
assert result.metadata.get("skipped") is True
|
||||
|
||||
def test_file_not_found_passes(self, tmp_path: Path):
|
||||
"""Non-existent file returns pass (skipped)."""
|
||||
result = check(str(tmp_path / "nonexistent" / "__init__.py"))
|
||||
assert result.passed is True
|
||||
|
||||
|
||||
class TestDroneComplianceEdgeCases:
|
||||
"""Edge cases for the compliance check."""
|
||||
|
||||
def test_syntax_error_in_adapter(self, tmp_path: Path):
|
||||
"""Adapter with syntax error fails gracefully."""
|
||||
pkg = tmp_path / "mymod"
|
||||
pkg.mkdir()
|
||||
(pkg / "__init__.py").write_text("")
|
||||
(pkg / "drone_adapter.py").write_text("def broken(:\n")
|
||||
|
||||
result = check(str(pkg / "__init__.py"), config={"target_packages": ["mymod"]})
|
||||
assert result.passed is False
|
||||
failed = [c.name for c in result.checks if not c.passed]
|
||||
assert "adapter-parseable" in failed
|
||||
@@ -0,0 +1,262 @@
|
||||
"""
|
||||
Tests for the drone module routing system (aipass.drone.modules).
|
||||
|
||||
Covers:
|
||||
- Module registry: list, lookup, register
|
||||
- Module info retrieval
|
||||
- Module command routing
|
||||
- Module help retrieval
|
||||
- CLI integration: drone systems shows modules, drone @seedgo routes correctly
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from aipass.drone.cli import main
|
||||
from aipass.drone.modules import (
|
||||
ModuleInfo,
|
||||
get_module_help,
|
||||
get_module_info,
|
||||
is_module,
|
||||
list_modules,
|
||||
register_module,
|
||||
route_module_command,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run_cli(*args: str) -> tuple[int, str, str]:
|
||||
"""Call main() with the given argv, capturing stdout/stderr and exit code."""
|
||||
stdout_lines: list[str] = []
|
||||
stderr_lines: list[str] = []
|
||||
|
||||
class _Cap:
|
||||
def __init__(self, buf: list[str]) -> None:
|
||||
self._buf = buf
|
||||
|
||||
def write(self, s: str) -> int:
|
||||
self._buf.append(s)
|
||||
return len(s)
|
||||
|
||||
def flush(self) -> None:
|
||||
pass
|
||||
|
||||
exit_code = 0
|
||||
|
||||
def _exit(code: int = 0) -> None:
|
||||
nonlocal exit_code
|
||||
exit_code = code
|
||||
raise SystemExit(code)
|
||||
|
||||
with (
|
||||
patch("sys.argv", ["drone", *args]),
|
||||
patch("sys.stdout", _Cap(stdout_lines)),
|
||||
patch("sys.stderr", _Cap(stderr_lines)),
|
||||
patch("sys.exit", side_effect=_exit),
|
||||
):
|
||||
try:
|
||||
main()
|
||||
except SystemExit:
|
||||
pass
|
||||
|
||||
return exit_code, "".join(stdout_lines), "".join(stderr_lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module registry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestModuleRegistry:
|
||||
"""Test the internal module registry."""
|
||||
|
||||
def test_seedgo_is_registered(self):
|
||||
"""seedgo is in the default module registry."""
|
||||
assert is_module("seedgo")
|
||||
|
||||
def test_unknown_module_not_registered(self):
|
||||
"""Unknown names return False."""
|
||||
assert not is_module("nonexistent")
|
||||
|
||||
def test_list_modules_includes_seedgo(self):
|
||||
"""list_modules includes seedgo."""
|
||||
modules = list_modules()
|
||||
assert "seedgo" in modules
|
||||
|
||||
def test_list_modules_returns_sorted(self):
|
||||
"""list_modules returns sorted names."""
|
||||
modules = list_modules()
|
||||
assert modules == sorted(modules)
|
||||
|
||||
def test_register_module_adds_new(self):
|
||||
"""register_module adds a new module."""
|
||||
register_module("test_mod", "test.path.adapter")
|
||||
assert is_module("test_mod")
|
||||
# Clean up
|
||||
from aipass.drone.modules import _MODULE_REGISTRY
|
||||
del _MODULE_REGISTRY["test_mod"]
|
||||
|
||||
|
||||
class TestModuleInfo:
|
||||
"""Test module info retrieval."""
|
||||
|
||||
def test_seedgo_info_returns_module_info(self):
|
||||
"""get_module_info for seedgo returns valid ModuleInfo."""
|
||||
info = get_module_info("seedgo")
|
||||
assert info is not None
|
||||
assert isinstance(info, ModuleInfo)
|
||||
assert info.name == "seedgo"
|
||||
assert info.version == "1.0.0"
|
||||
assert info.description != ""
|
||||
|
||||
def test_unknown_module_returns_none(self):
|
||||
"""get_module_info for unknown module returns None."""
|
||||
assert get_module_info("nonexistent") is None
|
||||
|
||||
def test_unimportable_module_returns_none(self):
|
||||
"""get_module_info returns None if adapter can't be imported."""
|
||||
register_module("broken", "nonexistent.module.path")
|
||||
info = get_module_info("broken")
|
||||
assert info is None
|
||||
# Clean up
|
||||
from aipass.drone.modules import _MODULE_REGISTRY
|
||||
del _MODULE_REGISTRY["broken"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module command routing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestModuleRouting:
|
||||
"""Test routing commands to modules."""
|
||||
|
||||
def test_route_seedgo_list(self):
|
||||
"""route_module_command to seedgo 'list' returns output."""
|
||||
result = route_module_command("seedgo", "list")
|
||||
assert isinstance(result, dict)
|
||||
assert "stdout" in result
|
||||
assert "exit_code" in result
|
||||
# seedgo list should succeed and show plugins
|
||||
assert result["exit_code"] == 0
|
||||
assert "plugin" in result["stdout"].lower() or "PLUGIN" in result["stdout"]
|
||||
|
||||
def test_route_unknown_command(self):
|
||||
"""Unknown seedgo command returns error."""
|
||||
result = route_module_command("seedgo", "nonexistent")
|
||||
assert result["exit_code"] == 1
|
||||
assert "unknown command" in result["stderr"]
|
||||
|
||||
def test_route_unknown_module_raises(self):
|
||||
"""Routing to unregistered module raises KeyError."""
|
||||
import pytest
|
||||
with pytest.raises(KeyError):
|
||||
route_module_command("nonexistent", "list")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module help
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestModuleHelp:
|
||||
"""Test module help retrieval."""
|
||||
|
||||
def test_seedgo_help_returns_text(self):
|
||||
"""get_module_help for seedgo returns non-empty help text."""
|
||||
help_text = get_module_help("seedgo")
|
||||
assert help_text != ""
|
||||
assert "seedgo" in help_text
|
||||
assert "check" in help_text or "audit" in help_text
|
||||
|
||||
def test_unknown_module_help_returns_empty(self):
|
||||
"""get_module_help for unknown module returns empty string."""
|
||||
assert get_module_help("nonexistent") == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI integration — drone systems shows modules
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCLISystemsModules:
|
||||
"""drone systems includes modules."""
|
||||
|
||||
def test_systems_shows_modules(self):
|
||||
"""drone systems output includes modules section."""
|
||||
with patch("aipass.drone.cli.list_branches", return_value=[]):
|
||||
code, out, _ = _run_cli("systems")
|
||||
assert code == 0
|
||||
assert "Modules" in out
|
||||
assert "@seedgo" in out
|
||||
|
||||
def test_systems_shows_module_description(self):
|
||||
"""drone systems shows module descriptions."""
|
||||
with patch("aipass.drone.cli.list_branches", return_value=[]):
|
||||
code, out, _ = _run_cli("systems")
|
||||
assert "standards" in out.lower() or "Standards" in out
|
||||
|
||||
def test_systems_shows_both_modules_and_branches(self):
|
||||
"""drone systems shows both modules and branches."""
|
||||
with patch("aipass.drone.cli.list_branches", return_value=["@flow", "@prax"]):
|
||||
code, out, _ = _run_cli("systems")
|
||||
assert code == 0
|
||||
assert "Modules" in out
|
||||
assert "Branches" in out
|
||||
assert "@seedgo" in out
|
||||
assert "@flow" in out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI integration — drone @seedgo routes to module
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCLIModuleRouting:
|
||||
"""drone @seedgo commands route through module system."""
|
||||
|
||||
def test_seedgo_help(self):
|
||||
"""drone @seedgo --help shows seedgo help text."""
|
||||
code, out, _ = _run_cli("@seedgo", "--help")
|
||||
assert code == 0
|
||||
assert "seedgo" in out
|
||||
assert "check" in out
|
||||
|
||||
def test_seedgo_no_args_shows_help(self):
|
||||
"""drone @seedgo with no command shows help."""
|
||||
code, out, _ = _run_cli("@seedgo")
|
||||
assert code == 0
|
||||
assert "seedgo" in out
|
||||
|
||||
def test_seedgo_list(self):
|
||||
"""drone @seedgo list shows plugins."""
|
||||
code, out, _ = _run_cli("@seedgo", "list")
|
||||
assert code == 0
|
||||
assert "plugin" in out.lower() or "PLUGIN" in out
|
||||
|
||||
def test_seedgo_unknown_command(self):
|
||||
"""drone @seedgo nonexistent exits with error."""
|
||||
code, _, err = _run_cli("@seedgo", "nonexistent")
|
||||
assert code == 1
|
||||
assert "unknown command" in err
|
||||
|
||||
def test_seedgo_check_specific_file(self):
|
||||
"""drone @seedgo check on a file runs checks."""
|
||||
code, out, _ = _run_cli("@seedgo", "check", "src/seedgo/__init__.py")
|
||||
assert code == 0 or code == 1 # May pass or fail, but should run
|
||||
# Should produce output from seedgo
|
||||
assert out != "" or _ != ""
|
||||
|
||||
def test_module_takes_priority_over_branch(self):
|
||||
"""Module routing takes priority over branch registry."""
|
||||
# Even if a branch named 'seedgo' existed, the module should handle it
|
||||
# We verify by NOT mocking route_command — if branch routing were tried,
|
||||
# it would fail because no registry exists
|
||||
code, out, _ = _run_cli("@seedgo", "--help")
|
||||
assert code == 0
|
||||
assert "seedgo" in out
|
||||
Reference in New Issue
Block a user