forked from HAL9000/cleveragents-core
2d423bdfcd
The `action create` CLI command was the only action subcommand missing the `--format`/`-f` parameter. All other action subcommands (`list`, `show`, `archive`) already accepted `--format` and routed through `_print_action()`. Running `action create --config action.yaml --format plain` failed with a Typer unrecognized-option error. Added the `fmt` parameter (with `--format`/`-f` aliases, defaulting to `rich`) to the `create()` function signature and passed it through to the existing `_print_action()` helper which already handles all output formats. Added Behave BDD scenarios for `--format plain` and `--format json` to `action_cli_spec_alignment.feature`. ISSUES CLOSED: #959
396 lines
14 KiB
Python
396 lines
14 KiB
Python
"""Step definitions for action CLI spec alignment feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
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.domain.models.core.action import (
|
|
Action,
|
|
ActionArgument,
|
|
ActionState,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import NamespacedName
|
|
|
|
_VALID_YAML = """\
|
|
name: local/spec-action
|
|
description: A spec-aligned action
|
|
strategy_actor: openai/gpt-4
|
|
execution_actor: openai/gpt-4
|
|
definition_of_done: All tests pass
|
|
arguments:
|
|
- name: target
|
|
type: integer
|
|
required: true
|
|
description: Target value
|
|
"""
|
|
|
|
_INVALID_YAML = """\
|
|
not: valid: yaml: [broken
|
|
"""
|
|
|
|
_MISSING_FIELDS_YAML = """\
|
|
description: Missing name and actors
|
|
"""
|
|
|
|
|
|
def _make_action(
|
|
*,
|
|
name: str = "local/spec-action",
|
|
state: ActionState = ActionState.AVAILABLE,
|
|
definition_of_done: str = "All tests pass",
|
|
strategy_actor: str = "openai/gpt-4",
|
|
execution_actor: str = "openai/gpt-4",
|
|
arguments: list[ActionArgument] | None = None,
|
|
) -> Action:
|
|
"""Create an Action instance for spec alignment tests."""
|
|
return Action(
|
|
namespaced_name=NamespacedName.parse(name),
|
|
description="A spec-aligned action",
|
|
long_description=None,
|
|
definition_of_done=definition_of_done,
|
|
strategy_actor=strategy_actor,
|
|
execution_actor=execution_actor,
|
|
arguments=arguments or [],
|
|
reusable=True,
|
|
read_only=False,
|
|
state=state,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
|
|
|
|
def _write_temp_yaml(context: Context, content: str) -> str:
|
|
"""Write YAML content to a temporary file and register cleanup."""
|
|
fd, path = tempfile.mkstemp(suffix=".yaml")
|
|
with os.fdopen(fd, "w") as fh:
|
|
fh.write(content)
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(
|
|
lambda p=path: os.unlink(p) if os.path.exists(p) else None
|
|
)
|
|
return path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a spec alignment CLI runner")
|
|
def step_spec_runner(context: Context) -> None:
|
|
"""Set up the CLI runner."""
|
|
context.runner = CliRunner()
|
|
|
|
|
|
@given("a spec alignment mocked lifecycle service")
|
|
def step_spec_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()
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(context.service_patcher.stop)
|
|
|
|
# Patch the module-level Rich Console so it never injects ANSI escape
|
|
# codes. Without this, Rich detects a colour terminal and embeds
|
|
# escape sequences in the CLI output, causing plain-text assertions
|
|
# (e.g. ``"Action Created" in output``) to fail on real terminals
|
|
# while passing inside headless CI containers.
|
|
from rich.console import Console as _Console
|
|
|
|
_plain_console = _Console(no_color=True, highlight=False, width=300)
|
|
context.console_patcher = patch(
|
|
"cleveragents.cli.commands.action.console", _plain_console
|
|
)
|
|
context.console_patcher.start()
|
|
context._cleanup_handlers.append(context.console_patcher.stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a spec alignment valid config file")
|
|
def step_spec_valid_config(context: Context) -> None:
|
|
"""Write a valid action config YAML."""
|
|
context.config_path = _write_temp_yaml(context, _VALID_YAML)
|
|
created = _make_action(
|
|
arguments=[
|
|
ActionArgument(
|
|
name="target",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Target value",
|
|
),
|
|
],
|
|
)
|
|
context.mock_service.create_action.return_value = created
|
|
|
|
|
|
@given("spec alignment actions exist")
|
|
def step_spec_actions_exist(context: Context) -> None:
|
|
"""Set up multiple actions in mock service."""
|
|
context.spec_actions = [
|
|
_make_action(name="local/action-a"),
|
|
_make_action(name="local/action-b"),
|
|
_make_action(name="myorg/action-c"),
|
|
]
|
|
context.mock_service.list_actions.return_value = context.spec_actions
|
|
|
|
|
|
@given('a spec alignment action exists with name "{name}"')
|
|
def step_spec_action_by_name(context: Context, name: str) -> None:
|
|
"""Set up an action findable by name."""
|
|
context.spec_action = _make_action(name=name)
|
|
context.mock_service.get_action_by_name.return_value = context.spec_action
|
|
|
|
|
|
@given('a spec alignment action exists for archive "{name}"')
|
|
def step_spec_action_for_archive(context: Context, name: str) -> None:
|
|
"""Set up an action that can be archived."""
|
|
context.spec_action = _make_action(name=name)
|
|
context.mock_service.get_action_by_name.return_value = context.spec_action
|
|
context.mock_service.archive_action.return_value = _make_action(
|
|
name=name, state=ActionState.ARCHIVED
|
|
)
|
|
|
|
|
|
@given("a spec alignment invalid YAML config file")
|
|
def step_spec_invalid_yaml(context: Context) -> None:
|
|
"""Write an invalid YAML config."""
|
|
context.config_path = _write_temp_yaml(context, _INVALID_YAML)
|
|
|
|
|
|
@given("a spec alignment config with missing required fields")
|
|
def step_spec_missing_fields(context: Context) -> None:
|
|
"""Write a config missing required fields."""
|
|
context.config_path = _write_temp_yaml(context, _MISSING_FIELDS_YAML)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run spec alignment create with --config")
|
|
def step_spec_create(context: Context) -> None:
|
|
"""Run create --config with the prepared config file."""
|
|
context.result = context.runner.invoke(
|
|
action_app, ["create", "--config", context.config_path]
|
|
)
|
|
|
|
|
|
@when('I run spec alignment create with legacy flag "{flag}" "{value}"')
|
|
def step_spec_create_legacy_flag(context: Context, flag: str, value: str) -> None:
|
|
"""Run create with a legacy inline flag (should be rejected)."""
|
|
context.result = context.runner.invoke(action_app, ["create", flag, value])
|
|
|
|
|
|
@when('I run spec alignment list with namespace "{namespace}"')
|
|
def step_spec_list_namespace(context: Context, namespace: str) -> None:
|
|
"""Run list with --namespace filter."""
|
|
filtered = [
|
|
a for a in context.spec_actions if a.namespaced_name.namespace == namespace
|
|
]
|
|
context.mock_service.list_actions.return_value = filtered
|
|
context.result = context.runner.invoke(
|
|
action_app, ["list", "--namespace", namespace]
|
|
)
|
|
|
|
|
|
@when('I run spec alignment list with state "{state}"')
|
|
def step_spec_list_state(context: Context, state: str) -> None:
|
|
"""Run list with --state filter."""
|
|
context.result = context.runner.invoke(action_app, ["list", "--state", state])
|
|
|
|
|
|
@when('I run spec alignment list with regex "{pattern}"')
|
|
def step_spec_list_regex(context: Context, pattern: str) -> None:
|
|
"""Run list with a regex positional argument."""
|
|
context.result = context.runner.invoke(action_app, ["list", pattern])
|
|
|
|
|
|
@when('I run spec alignment show with name "{name}"')
|
|
def step_spec_show(context: Context, name: str) -> None:
|
|
"""Run show with a namespaced name."""
|
|
context.result = context.runner.invoke(action_app, ["show", name])
|
|
|
|
|
|
@when('I run spec alignment archive with name "{name}"')
|
|
def step_spec_archive(context: Context, name: str) -> None:
|
|
"""Run archive with a namespaced name."""
|
|
context.result = context.runner.invoke(action_app, ["archive", name])
|
|
|
|
|
|
@when('I run spec alignment create with --config and --format "{fmt}"')
|
|
def step_spec_create_with_format(context: Context, fmt: str) -> None:
|
|
"""Run create --config with --format flag."""
|
|
context.result = context.runner.invoke(
|
|
action_app, ["create", "--config", context.config_path, "--format", fmt]
|
|
)
|
|
|
|
|
|
@when("I run spec alignment create with missing config")
|
|
def step_spec_create_missing(context: Context) -> None:
|
|
"""Run create with a nonexistent config file."""
|
|
context.result = context.runner.invoke(
|
|
action_app, ["create", "--config", "/tmp/nonexistent_spec_action.yaml"]
|
|
)
|
|
|
|
|
|
@given("a spec alignment action with long definition of done")
|
|
def step_spec_long_dod_show(context: Context) -> None:
|
|
"""Create an action with a definition_of_done longer than 120 chars."""
|
|
long_dod = "A" * 150
|
|
action = _make_action(name="local/long-dod", definition_of_done=long_dod)
|
|
context.mock_service.get_action_by_name.return_value = action
|
|
|
|
|
|
@given("spec alignment actions with long definition of done")
|
|
def step_spec_long_dod_list(context: Context) -> None:
|
|
"""Create actions with long definition_of_done for list display."""
|
|
long_dod = "B" * 50
|
|
context.mock_service.list_actions.return_value = [
|
|
_make_action(name="local/long-dod-1", definition_of_done=long_dod),
|
|
]
|
|
|
|
|
|
@when("I run spec alignment list all")
|
|
def step_spec_list_all(context: Context) -> None:
|
|
"""Run list with no filters."""
|
|
context.result = context.runner.invoke(action_app, ["list"])
|
|
|
|
|
|
@given("a spec alignment config that triggers value error")
|
|
def step_spec_config_value_error(context: Context) -> None:
|
|
"""Write a config that will cause a ValueError from Action.from_config.
|
|
|
|
We patch Action.from_config to raise ValueError so the CLI handler
|
|
catches it and shows 'Config validation error'.
|
|
"""
|
|
context.config_path = _write_temp_yaml(context, _VALID_YAML)
|
|
context.from_config_patcher = patch(
|
|
"cleveragents.cli.commands.action.Action.from_config",
|
|
side_effect=ValueError("Bad config data"),
|
|
)
|
|
context.from_config_patcher.start()
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(context.from_config_patcher.stop)
|
|
|
|
|
|
@when("I run spec alignment available command")
|
|
def step_spec_available_cmd(context: Context) -> None:
|
|
"""Try running the removed 'available' subcommand."""
|
|
context.result = context.runner.invoke(action_app, ["available", "local/test"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the spec alignment create should succeed")
|
|
def step_spec_create_ok(context: Context) -> None:
|
|
"""Verify create succeeded."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Create failed ({context.result.exit_code}): {context.result.output}"
|
|
)
|
|
|
|
|
|
@then('the spec alignment output should contain "{text}"')
|
|
def step_spec_output_contains(context: Context, text: str) -> None:
|
|
"""Verify the output contains expected text."""
|
|
output_lower = context.result.output.lower()
|
|
text_lower = text.lower()
|
|
assert text_lower in output_lower, (
|
|
f"Expected '{text}' in output but got:\n{context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the spec alignment command should fail with unrecognized option")
|
|
def step_spec_fail_unrecognized(context: Context) -> None:
|
|
"""Verify the command failed due to an unrecognized option."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected failure but got exit code 0: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the spec alignment list should show filtered results")
|
|
def step_spec_list_filtered(context: Context) -> None:
|
|
"""Verify list returned results."""
|
|
assert context.result.exit_code == 0, f"List failed: {context.result.output}"
|
|
assert "Actions" in context.result.output or "No actions" in context.result.output
|
|
|
|
|
|
@then('the spec alignment service list should use namespace "{namespace}"')
|
|
def step_spec_list_ns(context: Context, namespace: str) -> None:
|
|
"""Verify list_actions was called with the namespace."""
|
|
call_kwargs = context.mock_service.list_actions.call_args[1]
|
|
assert call_kwargs["namespace"] == namespace
|
|
|
|
|
|
@then("the spec alignment list should succeed")
|
|
def step_spec_list_ok(context: Context) -> None:
|
|
"""Verify list succeeded."""
|
|
assert context.result.exit_code == 0, f"List failed: {context.result.output}"
|
|
|
|
|
|
@then("the spec alignment show should display details")
|
|
def step_spec_show_ok(context: Context) -> None:
|
|
"""Verify show displayed details."""
|
|
assert context.result.exit_code == 0, f"Show failed: {context.result.output}"
|
|
assert "Action Details" in context.result.output
|
|
|
|
|
|
@then("the spec alignment archive should succeed")
|
|
def step_spec_archive_ok(context: Context) -> None:
|
|
"""Verify archive succeeded."""
|
|
assert context.result.exit_code == 0, f"Archive failed: {context.result.output}"
|
|
assert "archived" in context.result.output.lower()
|
|
|
|
|
|
@then("the spec alignment command should abort")
|
|
def step_spec_abort(context: Context) -> None:
|
|
"""Verify the command aborted."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected abort but got exit code 0: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the spec alignment list output should contain truncated dod")
|
|
def step_spec_list_truncated_dod(context: Context) -> None:
|
|
"""Verify the list output contains a truncated definition of done.
|
|
|
|
Rich table truncates long text with the unicode ellipsis (…) or our
|
|
code adds '...' — either way, the full 50-char string should NOT appear.
|
|
"""
|
|
full_dod = "B" * 50
|
|
assert full_dod not in context.result.output, (
|
|
"Expected truncated DoD in list output but full string appeared"
|
|
)
|
|
|
|
|
|
@then("the spec alignment command should fail with unrecognized command")
|
|
def step_spec_fail_unrecognized_cmd(context: Context) -> None:
|
|
"""Verify 'available' subcommand is no longer recognized."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected failure but got exit code 0: {context.result.output}"
|
|
)
|