forked from HAL9000/cleveragents-core
273 lines
9.9 KiB
Python
273 lines
9.9 KiB
Python
"""Step definitions for tool CLI uncovered branches."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import OrderedDict
|
|
from io import StringIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from rich.console import Console
|
|
|
|
from cleveragents.cli.commands.tool import _print_tool
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeToolWithCliDict:
|
|
"""A tool object that returns a custom OrderedDict from as_cli_dict."""
|
|
|
|
def __init__(self, data: dict):
|
|
self._data = data
|
|
|
|
def as_cli_dict(self) -> OrderedDict:
|
|
return OrderedDict(self._data)
|
|
|
|
|
|
def _capture_rich_output(tool, title="Tool") -> str:
|
|
"""Call _print_tool in rich format and capture console output."""
|
|
buf = StringIO()
|
|
test_console = Console(file=buf, force_terminal=False, width=120)
|
|
with patch("cleveragents.cli.commands.tool.console", test_console):
|
|
_print_tool(tool, title=title, fmt="rich")
|
|
return buf.getvalue()
|
|
|
|
|
|
# ===========================================================================
|
|
# Scenario: _get_tool_registry_service constructs service from container
|
|
# ===========================================================================
|
|
|
|
|
|
@given("tool cli branch the container returns a mock database url")
|
|
def step_tool_cli_branch_container_mock_db(context):
|
|
mock_container = MagicMock()
|
|
mock_container.database_url.return_value = "sqlite://"
|
|
context.mock_container = mock_container
|
|
|
|
|
|
@when("tool cli branch I call _get_tool_registry_service")
|
|
def step_tool_cli_branch_call_get_service(context):
|
|
mock_engine = MagicMock()
|
|
mock_session_factory = MagicMock()
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.tool.get_container",
|
|
return_value=context.mock_container,
|
|
)
|
|
if hasattr(context, "_unused")
|
|
else patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=context.mock_container,
|
|
),
|
|
patch(
|
|
"sqlalchemy.create_engine",
|
|
return_value=mock_engine,
|
|
) as _mock_create_engine,
|
|
patch(
|
|
"sqlalchemy.orm.sessionmaker",
|
|
return_value=mock_session_factory,
|
|
) as _mock_sessionmaker,
|
|
# The function does a local import of get_container, so we patch
|
|
# it at the location where it's imported.
|
|
patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=context.mock_container,
|
|
),
|
|
):
|
|
from cleveragents.cli.commands.tool import _get_tool_registry_service
|
|
|
|
result = _get_tool_registry_service()
|
|
context.service_result = result
|
|
|
|
|
|
@then("tool cli branch it should return a ToolRegistryService instance")
|
|
def step_tool_cli_branch_service_is_instance(context):
|
|
from cleveragents.application.services.tool_registry_service import (
|
|
ToolRegistryService,
|
|
)
|
|
|
|
assert isinstance(context.service_result, ToolRegistryService), (
|
|
f"Expected ToolRegistryService, got {type(context.service_result)}"
|
|
)
|
|
|
|
|
|
# ===========================================================================
|
|
# Scenario: _print_tool displays lifecycle dict with values
|
|
# ===========================================================================
|
|
|
|
|
|
@given("tool cli branch a tool with lifecycle hooks")
|
|
def step_tool_cli_branch_tool_with_lifecycle(context):
|
|
context.tool_obj = _FakeToolWithCliDict(
|
|
{
|
|
"name": "local/lifecycle-tool",
|
|
"description": "A tool with lifecycle",
|
|
"source": "custom",
|
|
"tool_type": "tool",
|
|
"namespace": "local",
|
|
"short_name": "lifecycle-tool",
|
|
"timeout": 60,
|
|
"capability": {"read_only": True},
|
|
"lifecycle": {
|
|
"on_start": "echo starting",
|
|
"on_stop": "echo stopping",
|
|
"on_error": None,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@when("tool cli branch I print the tool in rich format")
|
|
def step_tool_cli_branch_print_lifecycle_tool(context):
|
|
context.captured_output = _capture_rich_output(context.tool_obj)
|
|
|
|
|
|
@then("tool cli branch the output should contain lifecycle key values")
|
|
def step_tool_cli_branch_output_has_lifecycle(context):
|
|
output = context.captured_output
|
|
assert "on_start" in output, f"Expected 'on_start' in output:\n{output}"
|
|
assert "echo starting" in output, f"Expected 'echo starting' in output:\n{output}"
|
|
assert "on_stop" in output, f"Expected 'on_stop' in output:\n{output}"
|
|
assert "echo stopping" in output, f"Expected 'echo stopping' in output:\n{output}"
|
|
|
|
|
|
# ===========================================================================
|
|
# Scenario: _print_tool displays resource slots list with dict items
|
|
# ===========================================================================
|
|
|
|
|
|
@given("tool cli branch a tool with resource slots")
|
|
def step_tool_cli_branch_tool_with_slots(context):
|
|
context.tool_obj = _FakeToolWithCliDict(
|
|
{
|
|
"name": "local/slots-tool",
|
|
"description": "A tool with resource slots",
|
|
"source": "custom",
|
|
"tool_type": "tool",
|
|
"namespace": "local",
|
|
"short_name": "slots-tool",
|
|
"timeout": 120,
|
|
"capability": {},
|
|
"resource_slots": [
|
|
{
|
|
"name": "database",
|
|
"resource_type": "postgres",
|
|
"access": "read",
|
|
},
|
|
{
|
|
"name": "cache",
|
|
"resource_type": "redis",
|
|
"access": "write",
|
|
},
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
@when("tool cli branch I print the tool with slots in rich format")
|
|
def step_tool_cli_branch_print_slots_tool(context):
|
|
context.captured_output = _capture_rich_output(context.tool_obj)
|
|
|
|
|
|
@then("tool cli branch the output should contain resource slot details")
|
|
def step_tool_cli_branch_output_has_slots(context):
|
|
output = context.captured_output
|
|
assert "database" in output, f"Expected 'database' in output:\n{output}"
|
|
assert "postgres" in output, f"Expected 'postgres' in output:\n{output}"
|
|
assert "read" in output, f"Expected 'read' in output:\n{output}"
|
|
assert "cache" in output, f"Expected 'cache' in output:\n{output}"
|
|
assert "redis" in output, f"Expected 'redis' in output:\n{output}"
|
|
assert "write" in output, f"Expected 'write' in output:\n{output}"
|
|
|
|
|
|
# ===========================================================================
|
|
# Scenario: _print_tool displays both lifecycle and resource slots
|
|
# ===========================================================================
|
|
|
|
|
|
@given("tool cli branch a tool with lifecycle and resource slots")
|
|
def step_tool_cli_branch_tool_with_both(context):
|
|
context.tool_obj = _FakeToolWithCliDict(
|
|
{
|
|
"name": "local/full-tool",
|
|
"description": "A tool with both lifecycle and slots",
|
|
"source": "custom",
|
|
"tool_type": "tool",
|
|
"namespace": "local",
|
|
"short_name": "full-tool",
|
|
"timeout": 90,
|
|
"capability": {"writes": True, "checkpointable": False},
|
|
"lifecycle": {
|
|
"on_init": "setup.sh",
|
|
"on_teardown": "cleanup.sh",
|
|
},
|
|
"resource_slots": [
|
|
{
|
|
"name": "storage",
|
|
"resource_type": "s3",
|
|
"access": "readwrite",
|
|
},
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
@when("tool cli branch I print the full tool in rich format")
|
|
def step_tool_cli_branch_print_full_tool(context):
|
|
context.captured_output = _capture_rich_output(context.tool_obj)
|
|
|
|
|
|
@then("tool cli branch the output should contain both lifecycle and slot info")
|
|
def step_tool_cli_branch_output_has_both(context):
|
|
output = context.captured_output
|
|
# Lifecycle
|
|
assert "on_init" in output, f"Expected 'on_init' in output:\n{output}"
|
|
assert "setup.sh" in output, f"Expected 'setup.sh' in output:\n{output}"
|
|
assert "on_teardown" in output, f"Expected 'on_teardown' in output:\n{output}"
|
|
assert "cleanup.sh" in output, f"Expected 'cleanup.sh' in output:\n{output}"
|
|
# Resource slots
|
|
assert "storage" in output, f"Expected 'storage' in output:\n{output}"
|
|
assert "s3" in output, f"Expected 's3' in output:\n{output}"
|
|
assert "readwrite" in output, f"Expected 'readwrite' in output:\n{output}"
|
|
|
|
|
|
# ===========================================================================
|
|
# Scenario: remove command handles service returning False
|
|
# ===========================================================================
|
|
|
|
|
|
@given("tool cli branch a mock service where remove_tool returns False")
|
|
def step_tool_cli_branch_mock_service_remove_false(context):
|
|
mock_service = MagicMock()
|
|
# get_tool returns a truthy value so we pass the "not found" check
|
|
mock_service.get_tool.return_value = {"name": "local/doomed-tool"}
|
|
# remove_tool returns False → triggers the "Failed to remove" branch
|
|
mock_service.remove_tool.return_value = False
|
|
context.mock_service = mock_service
|
|
|
|
|
|
@when("tool cli branch I invoke the remove command with confirmation")
|
|
def step_tool_cli_branch_invoke_remove(context):
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.tool import app as tool_app
|
|
|
|
runner = CliRunner()
|
|
with patch(
|
|
"cleveragents.cli.commands.tool._get_tool_registry_service",
|
|
return_value=context.mock_service,
|
|
):
|
|
result = runner.invoke(tool_app, ["remove", "--yes", "local/doomed-tool"])
|
|
context.result = result
|
|
|
|
|
|
@then("tool cli branch the output should show failed to remove tool")
|
|
def step_tool_cli_branch_output_failed_remove(context):
|
|
output = context.result.output
|
|
assert "Failed to remove tool" in output, (
|
|
f"Expected 'Failed to remove tool' in output:\n{output}"
|
|
)
|