HAL9000
14f134a463
- Implement plugin CLI subcommand group with list, show, enable, disable, install, remove commands - Add JSON/YAML output format support for all plugin commands - Create Behave BDD tests for plugin CLI functionality - Full type annotations and pyright compliance - Supports plugin state management (ACTIVATED, DEACTIVATED, DISCOVERED, ERRORED) Closes #5756
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""Step definitions for plugin CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
|
|
@when('I run "{command}"')
|
|
def step_run_command(context: Any, command: str) -> None:
|
|
"""Run a CLI command and capture output."""
|
|
try:
|
|
result = subprocess.run(
|
|
command.split(),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
context.command_output = result.stdout + result.stderr
|
|
context.command_returncode = result.returncode
|
|
except subprocess.TimeoutExpired:
|
|
context.command_output = "Command timed out"
|
|
context.command_returncode = 124
|
|
|
|
|
|
@then('the output should contain "{text}"')
|
|
def step_output_contains(context: Any, text: str) -> None:
|
|
"""Check if output contains text."""
|
|
assert text in context.command_output, (
|
|
f"Expected '{text}' in output, got: {context.command_output}"
|
|
)
|
|
|
|
|
|
@then("the output should be valid JSON")
|
|
def step_output_is_json(context: Any) -> None:
|
|
"""Check if output is valid JSON."""
|
|
try:
|
|
context.json_output = json.loads(context.command_output)
|
|
except json.JSONDecodeError as e:
|
|
raise AssertionError(f"Output is not valid JSON: {e}")
|
|
|
|
|
|
@then('the JSON should contain a plugin named "{name}"')
|
|
def step_json_contains_plugin(context: Any, name: str) -> None:
|
|
"""Check if JSON output contains a plugin with given name."""
|
|
if isinstance(context.json_output, list):
|
|
plugin_names = [p.get("name") for p in context.json_output]
|
|
assert name in plugin_names, (
|
|
f"Plugin '{name}' not found in JSON. Found: {plugin_names}"
|
|
)
|
|
else:
|
|
assert context.json_output.get("name") == name, (
|
|
f"Expected plugin name '{name}', got: {context.json_output}"
|
|
)
|
|
|
|
|
|
@given('a plugin named "{name}" is registered')
|
|
def step_register_plugin(context: Any, name: str) -> None:
|
|
"""Register a test plugin."""
|
|
from cleveragents.infrastructure.plugins.manager import PluginManager
|
|
from cleveragents.infrastructure.plugins.types import PluginDescriptor
|
|
|
|
manager = PluginManager()
|
|
descriptor = PluginDescriptor(
|
|
name=name,
|
|
version="1.0.0",
|
|
description=f"Test plugin {name}",
|
|
module_path="test.module",
|
|
class_name="TestPlugin",
|
|
)
|
|
manager.register_plugin(descriptor)
|
|
context.plugin_manager = manager
|
|
|
|
|
|
@given('a plugin named "{name}" is registered but disabled')
|
|
def step_register_disabled_plugin(context: Any, name: str) -> None:
|
|
"""Register a disabled test plugin."""
|
|
step_register_plugin(context, name)
|
|
|
|
|
|
@given('a plugin named "{name}" is registered and enabled')
|
|
def step_register_enabled_plugin(context: Any, name: str) -> None:
|
|
"""Register and enable a test plugin."""
|
|
from cleveragents.infrastructure.plugins.manager import PluginManager
|
|
from cleveragents.infrastructure.plugins.types import PluginDescriptor, PluginState
|
|
|
|
manager = PluginManager()
|
|
descriptor = PluginDescriptor(
|
|
name=name,
|
|
version="1.0.0",
|
|
description=f"Test plugin {name}",
|
|
module_path="test.module",
|
|
class_name="TestPlugin",
|
|
)
|
|
descriptor.state = PluginState.ACTIVATED
|
|
manager.register_plugin(descriptor)
|
|
context.plugin_manager = manager
|