forked from cleveragents/cleveragents-core
624 lines
22 KiB
Python
624 lines
22 KiB
Python
"""Step definitions for the Action CLI feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.action import app as action_app
|
|
from cleveragents.core.exceptions import (
|
|
BusinessRuleViolation,
|
|
CleverAgentsError,
|
|
NotFoundError,
|
|
ValidationError,
|
|
)
|
|
from cleveragents.domain.models.core.action import (
|
|
Action,
|
|
ActionArgument,
|
|
ActionState,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import NamespacedName
|
|
|
|
|
|
def _make_cli_action(
|
|
*,
|
|
name: str = "local/test-action",
|
|
state: ActionState = ActionState.AVAILABLE,
|
|
definition_of_done: str = "Test passes",
|
|
strategy_actor: str = "openai/gpt-4",
|
|
execution_actor: str = "openai/gpt-4",
|
|
arguments: list[ActionArgument] | None = None,
|
|
reusable: bool = True,
|
|
read_only: bool = False,
|
|
created_by: str | None = None,
|
|
description: str = "Test action",
|
|
long_description: str | None = "A test action for testing",
|
|
) -> Action:
|
|
"""Create a mock Action for CLI testing."""
|
|
return Action(
|
|
namespaced_name=NamespacedName.parse(name),
|
|
description=description,
|
|
long_description=long_description,
|
|
definition_of_done=definition_of_done,
|
|
strategy_actor=strategy_actor,
|
|
execution_actor=execution_actor,
|
|
arguments=arguments or [],
|
|
reusable=reusable,
|
|
read_only=read_only,
|
|
state=state,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
created_by=created_by,
|
|
)
|
|
|
|
|
|
@given("an action CLI runner with mocks")
|
|
def step_action_cli_runner_with_mocks(context: Context) -> None:
|
|
"""Set up the CLI runner for testing."""
|
|
context.runner = CliRunner()
|
|
|
|
|
|
@given("a mocked plan lifecycle service")
|
|
def step_mocked_lifecycle_service(context: Context) -> None:
|
|
"""Set up a mock PlanLifecycleService."""
|
|
context.mock_service = MagicMock()
|
|
context.service_patcher = patch(
|
|
"cleveragents.cli.commands.action._get_lifecycle_service",
|
|
return_value=context.mock_service,
|
|
)
|
|
context.service_patcher.start()
|
|
|
|
# Store cleanup handler
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(context.service_patcher.stop)
|
|
|
|
|
|
@given("there are mocked existing actions")
|
|
def step_mocked_existing_actions(context: Context) -> None:
|
|
"""Set up existing actions in the mock service."""
|
|
context.actions = [
|
|
_make_cli_action(
|
|
name="local/action-one",
|
|
state=ActionState.AVAILABLE,
|
|
),
|
|
_make_cli_action(
|
|
name="local/action-two",
|
|
state=ActionState.AVAILABLE,
|
|
),
|
|
_make_cli_action(
|
|
name="myorg/action-three",
|
|
state=ActionState.AVAILABLE,
|
|
),
|
|
]
|
|
context.mock_service.list_actions.return_value = context.actions
|
|
|
|
|
|
@given("there are no mocked actions")
|
|
def step_no_mocked_actions(context: Context) -> None:
|
|
"""Set up empty actions list."""
|
|
context.mock_service.list_actions.return_value = []
|
|
|
|
|
|
@given('there is a mocked action with name "{name}"')
|
|
def step_mocked_action_by_name(context: Context, name: str) -> None:
|
|
"""Set up an existing action with specific name."""
|
|
context.existing_action = _make_cli_action(name=name)
|
|
context.mock_service.get_action.return_value = context.existing_action
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
|
|
|
|
@given("there is a mocked action without descriptions")
|
|
def step_mocked_action_without_descriptions(context: Context) -> None:
|
|
"""Set up an action without optional descriptions."""
|
|
arguments = [
|
|
ActionArgument(
|
|
name="target",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="",
|
|
)
|
|
]
|
|
context.existing_action = _make_cli_action(
|
|
name="local/no-description",
|
|
arguments=arguments,
|
|
description="No description action",
|
|
long_description=None,
|
|
)
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
|
|
|
|
@given("there is a mocked draft action")
|
|
def step_mocked_draft_action(context: Context) -> None:
|
|
"""Set up an action to be made available."""
|
|
context.existing_action = _make_cli_action(state=ActionState.AVAILABLE)
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
context.mock_service.make_action_available.return_value = _make_cli_action(
|
|
state=ActionState.AVAILABLE
|
|
)
|
|
|
|
|
|
@given("there is a mocked available action")
|
|
def step_mocked_available_action(context: Context) -> None:
|
|
"""Set up an available action."""
|
|
context.existing_action = _make_cli_action(state=ActionState.AVAILABLE)
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
context.mock_service.archive_action.return_value = _make_cli_action(
|
|
state=ActionState.ARCHIVED
|
|
)
|
|
|
|
|
|
@when('I run action CLI create with name "{name}" and required parameters')
|
|
def step_run_action_cli_create(context: Context, name: str) -> None:
|
|
"""Run action create command with required parameters."""
|
|
created_action = _make_cli_action(name=name)
|
|
context.mock_service.create_action.return_value = created_action
|
|
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
name,
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"openai/gpt-4",
|
|
"--definition-of-done",
|
|
"Test passes",
|
|
"--description",
|
|
"Test action",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.created_action = created_action
|
|
|
|
|
|
@when("I run action CLI create with all parameters")
|
|
def step_run_action_cli_create_all_params(context: Context) -> None:
|
|
"""Run action create command with all parameters."""
|
|
args = [
|
|
ActionArgument(
|
|
name="coverage",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Target coverage",
|
|
)
|
|
]
|
|
created_action = _make_cli_action(
|
|
name="local/full-action",
|
|
arguments=args,
|
|
)
|
|
context.mock_service.create_action.return_value = created_action
|
|
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
"local/full-action",
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"anthropic/claude-3",
|
|
"--definition-of-done",
|
|
"Coverage reaches target",
|
|
"--description",
|
|
"Increase test coverage",
|
|
"--arg",
|
|
"coverage:int:required:Target coverage",
|
|
"--tag",
|
|
"testing",
|
|
"--tag",
|
|
"coverage",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.created_action = created_action
|
|
|
|
|
|
@when("I run action CLI create with the available flag")
|
|
def step_run_action_cli_create_available(context: Context) -> None:
|
|
"""Run action create command (actions default to available, no --available flag needed)."""
|
|
created_action = _make_cli_action(state=ActionState.AVAILABLE)
|
|
|
|
context.mock_service.create_action.return_value = created_action
|
|
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
"local/avail-action",
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"openai/gpt-4",
|
|
"--definition-of-done",
|
|
"Test passes",
|
|
"--description",
|
|
"Available action",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.created_action = created_action
|
|
|
|
|
|
@when("I run action CLI create with validation error")
|
|
def step_run_action_cli_create_validation_error(context: Context) -> None:
|
|
"""Run action create command with service validation error."""
|
|
context.mock_service.create_action.side_effect = ValidationError("Missing fields")
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
"local/invalid-action",
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"openai/gpt-4",
|
|
"--definition-of-done",
|
|
"Test passes",
|
|
"--description",
|
|
"Invalid action",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.expected_error = "Missing fields"
|
|
|
|
|
|
@when("I run action CLI create with service error")
|
|
def step_run_action_cli_create_service_error(context: Context) -> None:
|
|
"""Run action create command with service error."""
|
|
context.mock_service.create_action.side_effect = CleverAgentsError("Create failed")
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
"local/error-action",
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"openai/gpt-4",
|
|
"--definition-of-done",
|
|
"Test passes",
|
|
"--description",
|
|
"Error action",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.expected_error = "Create failed"
|
|
|
|
|
|
@when("I run action CLI create with invalid argument format")
|
|
def step_run_action_cli_create_invalid_arg(context: Context) -> None:
|
|
"""Run action create command with invalid argument format."""
|
|
result = context.runner.invoke(
|
|
action_app,
|
|
[
|
|
"create",
|
|
"local/bad-action",
|
|
"--strategy-actor",
|
|
"openai/gpt-4",
|
|
"--execution-actor",
|
|
"openai/gpt-4",
|
|
"--definition-of-done",
|
|
"Test passes",
|
|
"--description",
|
|
"Bad action",
|
|
"--arg",
|
|
"invalid-format", # Missing type and requirement
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when('I run action CLI list with invalid state "{state}"')
|
|
def step_run_action_cli_list_invalid_state(context: Context, state: str) -> None:
|
|
"""Run action list command with invalid state filter."""
|
|
result = context.runner.invoke(action_app, ["list", "--state", state])
|
|
context.result = result
|
|
context.expected_state = state
|
|
|
|
|
|
@when("I run action CLI list with service error")
|
|
def step_run_action_cli_list_service_error(context: Context) -> None:
|
|
"""Run action list command when service fails."""
|
|
context.mock_service.list_actions.side_effect = CleverAgentsError("List failed")
|
|
result = context.runner.invoke(action_app, ["list"])
|
|
context.result = result
|
|
context.expected_error = "List failed"
|
|
|
|
|
|
@when("I run action CLI list")
|
|
def step_run_action_cli_list(context: Context) -> None:
|
|
"""Run action list command."""
|
|
result = context.runner.invoke(action_app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when('I run action CLI list with namespace filter "{namespace}"')
|
|
def step_run_action_cli_list_namespace(context: Context, namespace: str) -> None:
|
|
"""Run action list command with namespace filter."""
|
|
# Filter the actions to only the specified namespace
|
|
filtered = [a for a in context.actions if a.namespaced_name.namespace == namespace]
|
|
context.mock_service.list_actions.return_value = filtered
|
|
|
|
result = context.runner.invoke(action_app, ["list", "--namespace", namespace])
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI list with available filter")
|
|
def step_run_action_cli_list_available(context: Context) -> None:
|
|
"""Run action list command with available filter."""
|
|
# Filter to only available actions
|
|
filtered = [a for a in context.actions if a.state == ActionState.AVAILABLE]
|
|
context.mock_service.list_actions.return_value = filtered
|
|
|
|
result = context.runner.invoke(action_app, ["list", "--available"])
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI show with that ID")
|
|
def step_run_action_cli_show_id(context: Context) -> None:
|
|
"""Run action show command with the stored action ID."""
|
|
result = context.runner.invoke(
|
|
action_app, ["show", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when('I run action CLI show with name "{name}"')
|
|
def step_run_action_cli_show_name(context: Context, name: str) -> None:
|
|
"""Run action show command with name."""
|
|
result = context.runner.invoke(action_app, ["show", name])
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI show with service error")
|
|
def step_run_action_cli_show_service_error(context: Context) -> None:
|
|
"""Run action show command when service fails."""
|
|
context.mock_service.get_action_by_name.side_effect = CleverAgentsError(
|
|
"Show failed"
|
|
)
|
|
result = context.runner.invoke(action_app, ["show", "local/show-fail"])
|
|
context.result = result
|
|
context.expected_error = "Show failed"
|
|
|
|
|
|
@when("I run action CLI show with unknown ID")
|
|
def step_run_action_cli_show_unknown(context: Context) -> None:
|
|
"""Run action show command with unknown name."""
|
|
context.mock_service.get_action_by_name.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
|
|
result = context.runner.invoke(action_app, ["show", "unknown-id"])
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI available with the action ID")
|
|
def step_run_action_cli_available(context: Context) -> None:
|
|
"""Run action available command."""
|
|
result = context.runner.invoke(
|
|
action_app, ["available", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI available with the action name")
|
|
def step_run_action_cli_available_by_name(context: Context) -> None:
|
|
"""Run action available command using namespaced name."""
|
|
# CLI now always uses get_action_by_name (no fallback from get_action)
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
result = context.runner.invoke(
|
|
action_app, ["available", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI available with unknown ID")
|
|
def step_run_action_cli_available_unknown(context: Context) -> None:
|
|
"""Run action available command with unknown name."""
|
|
context.mock_service.get_action_by_name.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
result = context.runner.invoke(action_app, ["available", "unknown-id"])
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI available with service error")
|
|
def step_run_action_cli_available_service_error(context: Context) -> None:
|
|
"""Run action available command when service fails."""
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
context.mock_service.make_action_available.side_effect = CleverAgentsError(
|
|
"Availability failed"
|
|
)
|
|
result = context.runner.invoke(
|
|
action_app, ["available", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
context.expected_error = "Availability failed"
|
|
|
|
|
|
@when("I run action CLI archive with the action ID")
|
|
def step_run_action_cli_archive(context: Context) -> None:
|
|
"""Run action archive command."""
|
|
result = context.runner.invoke(
|
|
action_app, ["archive", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I run action CLI archive with service error")
|
|
def step_run_action_cli_archive_service_error(context: Context) -> None:
|
|
"""Run action archive command when service fails."""
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
context.mock_service.archive_action.side_effect = CleverAgentsError(
|
|
"Archive failed"
|
|
)
|
|
result = context.runner.invoke(
|
|
action_app, ["archive", str(context.existing_action.namespaced_name)]
|
|
)
|
|
context.result = result
|
|
context.expected_error = "Archive failed"
|
|
|
|
|
|
@when("I run action CLI archive with unknown ID")
|
|
def step_run_action_cli_archive_unknown(context: Context) -> None:
|
|
"""Run action archive command with unknown name."""
|
|
context.mock_service.get_action_by_name.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
|
|
result = context.runner.invoke(action_app, ["archive", "unknown-id"])
|
|
context.result = result
|
|
|
|
|
|
@then("the action CLI create should succeed")
|
|
def step_action_cli_create_success(context: Context) -> None:
|
|
"""Verify action was created."""
|
|
assert context.result.exit_code == 0, f"CLI failed: {context.result.output}"
|
|
context.mock_service.create_action.assert_called_once()
|
|
|
|
|
|
@then('the action CLI created name should be "{name}"')
|
|
def step_action_cli_created_name(context: Context, name: str) -> None:
|
|
"""Verify created action name."""
|
|
call_kwargs = context.mock_service.create_action.call_args[1]
|
|
assert call_kwargs["name"] == name
|
|
|
|
|
|
@then("the action CLI should have specified arguments")
|
|
def step_action_cli_has_arguments(context: Context) -> None:
|
|
"""Verify action has arguments."""
|
|
call_kwargs = context.mock_service.create_action.call_args[1]
|
|
assert len(call_kwargs.get("arguments", [])) > 0
|
|
|
|
|
|
@then("the action CLI should create in available state")
|
|
def step_action_cli_available_state(context: Context) -> None:
|
|
"""Verify action was created (actions default to available, no separate make_available call)."""
|
|
assert context.result.exit_code == 0, f"CLI failed: {context.result.output}"
|
|
context.mock_service.create_action.assert_called_once()
|
|
|
|
|
|
@then("the action CLI command should abort")
|
|
def step_action_cli_abort(context: Context) -> None:
|
|
"""Verify CLI aborted."""
|
|
assert context.result.exit_code != 0
|
|
|
|
|
|
@then("the action CLI should show all actions in a table")
|
|
def step_action_cli_show_all_actions(context: Context) -> None:
|
|
"""Verify all actions are displayed."""
|
|
assert context.result.exit_code == 0
|
|
assert "Actions" in context.result.output
|
|
# The table may truncate names, so check for partial matches
|
|
for action in context.actions:
|
|
# Check for the beginning of each action name (before truncation)
|
|
name_start = str(action.namespaced_name)[:10]
|
|
assert name_start in context.result.output, (
|
|
f"Expected '{name_start}' in output but not found"
|
|
)
|
|
|
|
|
|
@then("the action CLI should show only local namespace actions")
|
|
def step_action_cli_show_local_actions(context: Context) -> None:
|
|
"""Verify only local actions are displayed."""
|
|
assert context.result.exit_code == 0
|
|
# Should contain local actions
|
|
assert "local/" in context.result.output
|
|
|
|
|
|
@then("the action CLI should show only available actions")
|
|
def step_action_cli_show_available_actions(context: Context) -> None:
|
|
"""Verify only available actions are displayed."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("the action CLI should show no actions message")
|
|
def step_action_cli_show_no_actions(context: Context) -> None:
|
|
"""Verify no actions message is displayed."""
|
|
assert context.result.exit_code == 0
|
|
assert "No actions found" in context.result.output
|
|
|
|
|
|
@then("the action CLI should show the action details")
|
|
def step_action_cli_show_action_details(context: Context) -> None:
|
|
"""Verify action details are displayed."""
|
|
assert context.result.exit_code == 0
|
|
assert "Action" in context.result.output
|
|
|
|
|
|
@then("the action CLI command should abort for missing action")
|
|
def step_action_cli_abort_missing(context: Context) -> None:
|
|
"""Verify CLI aborts for missing action."""
|
|
assert context.result.exit_code != 0
|
|
assert "not found" in context.result.output.lower()
|
|
|
|
|
|
@then("the action CLI should make action available")
|
|
def step_action_cli_made_available(context: Context) -> None:
|
|
"""Verify action was made available."""
|
|
assert context.result.exit_code == 0
|
|
context.mock_service.make_action_available.assert_called_once()
|
|
|
|
|
|
@then("the action CLI should resolve action by name")
|
|
def step_action_cli_resolve_by_name(context: Context) -> None:
|
|
"""Verify name-based lookup was used."""
|
|
context.mock_service.get_action_by_name.assert_called_with(
|
|
str(context.existing_action.namespaced_name)
|
|
)
|
|
|
|
|
|
@then("the action CLI command should abort with business rule violation")
|
|
def step_action_cli_abort_business_rule(context: Context) -> None:
|
|
"""Verify CLI aborts with business rule violation."""
|
|
# Set up the mock to raise BusinessRuleViolation
|
|
context.mock_service.get_action_by_name.return_value = context.existing_action
|
|
context.mock_service.make_action_available.side_effect = BusinessRuleViolation(
|
|
"Action is already available"
|
|
)
|
|
|
|
# Re-run the command
|
|
result = context.runner.invoke(
|
|
action_app, ["available", str(context.existing_action.namespaced_name)]
|
|
)
|
|
assert result.exit_code != 0
|
|
|
|
|
|
@then("the action CLI should abort with validation error")
|
|
def step_action_cli_abort_validation_error(context: Context) -> None:
|
|
"""Verify CLI aborts with validation error output."""
|
|
assert context.result.exit_code != 0
|
|
assert "Validation Error" in context.result.output
|
|
assert context.expected_error in context.result.output
|
|
|
|
|
|
@then("the action CLI should abort with service error")
|
|
def step_action_cli_abort_service_error(context: Context) -> None:
|
|
"""Verify CLI aborts with service error output."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
assert context.expected_error in context.result.output
|
|
|
|
|
|
@then("the action CLI should abort with invalid state")
|
|
def step_action_cli_abort_invalid_state(context: Context) -> None:
|
|
"""Verify CLI aborts with invalid state output."""
|
|
assert context.result.exit_code != 0
|
|
assert "Invalid state" in context.result.output
|
|
assert "Valid values" in context.result.output
|
|
|
|
|
|
@then("the action CLI should archive the action")
|
|
def step_action_cli_archived(context: Context) -> None:
|
|
"""Verify action was archived."""
|
|
assert context.result.exit_code == 0
|
|
context.mock_service.archive_action.assert_called_once()
|