forked from cleveragents/cleveragents-core
455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""Step definitions for the Action CLI feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
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,
|
|
NotFoundError,
|
|
)
|
|
from cleveragents.domain.models.core.action import Action, ActionArgument
|
|
from cleveragents.domain.models.core.plan import ActionState, NamespacedName
|
|
|
|
|
|
def _make_cli_action(
|
|
*,
|
|
action_id: str = "01ARZ3NDEKTSV4RRFFQ69G5FAV",
|
|
name: str = "local/test-action",
|
|
state: ActionState = ActionState.DRAFT,
|
|
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,
|
|
) -> Action:
|
|
"""Create a mock Action for CLI testing."""
|
|
return Action(
|
|
action_id=action_id,
|
|
namespaced_name=NamespacedName.parse(name),
|
|
short_description="Test action",
|
|
long_description="A test action for testing",
|
|
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(
|
|
action_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
|
|
name="local/action-one",
|
|
state=ActionState.AVAILABLE,
|
|
),
|
|
_make_cli_action(
|
|
action_id="01ARZ3NDEKTSV4RRFFQ69G5FAW",
|
|
name="local/action-two",
|
|
state=ActionState.DRAFT,
|
|
),
|
|
_make_cli_action(
|
|
action_id="01ARZ3NDEKTSV4RRFFQ69G5FAX",
|
|
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 ID "{action_id}"')
|
|
def step_mocked_action_by_id(context: Context, action_id: str) -> None:
|
|
"""Set up an existing action with specific ID."""
|
|
context.existing_action = _make_cli_action(action_id=action_id)
|
|
context.mock_service.get_action.return_value = context.existing_action
|
|
|
|
|
|
@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.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
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 a draft action."""
|
|
context.existing_action = _make_cli_action(state=ActionState.DRAFT)
|
|
context.mock_service.get_action.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.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",
|
|
],
|
|
)
|
|
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="int",
|
|
requirement="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 with --available flag."""
|
|
created_action = _make_cli_action(state=ActionState.DRAFT)
|
|
available_action = _make_cli_action(state=ActionState.AVAILABLE)
|
|
|
|
context.mock_service.create_action.return_value = created_action
|
|
context.mock_service.make_action_available.return_value = available_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",
|
|
"--available",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.created_action = available_action
|
|
|
|
|
|
@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",
|
|
"--arg",
|
|
"invalid-format", # Missing type and requirement
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@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", context.existing_action.action_id]
|
|
)
|
|
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 unknown ID")
|
|
def step_run_action_cli_show_unknown(context: Context) -> None:
|
|
"""Run action show command with unknown ID."""
|
|
context.mock_service.get_action.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
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", context.existing_action.action_id]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@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", context.existing_action.action_id]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@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 ID."""
|
|
context.mock_service.get_action.side_effect = NotFoundError(
|
|
resource_type="action", resource_id="unknown"
|
|
)
|
|
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 made available."""
|
|
context.mock_service.make_action_available.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 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.make_action_available.side_effect = BusinessRuleViolation(
|
|
"Action is already available"
|
|
)
|
|
|
|
# Re-run the command
|
|
result = context.runner.invoke(
|
|
action_app, ["available", context.existing_action.action_id]
|
|
)
|
|
assert result.exit_code != 0
|
|
|
|
|
|
@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()
|