forked from HAL9000/cleveragents-core
freemo
a0df5a4cd0
Implements the spec-required JSON/YAML output envelope for all CLI commands
that use format_output(). The envelope structure is:
{
"command": "<command that was run>",
"status": "ok" | "warn" | "error",
"exit_code": 0,
"data": { ... command-specific payload ... },
"timing": { "duration_ms": 123 },
"messages": [{ "level": "ok", "text": "..." }]
}
Changes:
- Add _build_envelope() helper to construct the spec-required envelope
- Add optional command, status, exit_code, messages parameters to format_output()
- Wrap json/yaml output in the envelope; plain/table/rich/color unchanged
- Add timing measurement (duration_ms) to all json/yaml outputs
- Add new BDD feature file (cli_json_envelope.feature) with 14 scenarios
testing envelope field presence, values, and data payload
- Update 14 existing step files to unwrap the envelope when checking
specific data keys (backward-compatible via _unwrap_envelope() helper)
Closes #3431
525 lines
20 KiB
Python
525 lines
20 KiB
Python
"""Step definitions for plan use --execution-env-priority feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
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.plan import app as plan_app
|
|
from cleveragents.domain.models.core.action import Action, ActionState
|
|
from cleveragents.domain.models.core.plan import (
|
|
ExecutionEnvPriority,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"
|
|
|
|
_ENVELOPE_KEYS = {"command", "status", "exit_code", "data", "timing", "messages"}
|
|
|
|
|
|
def _unwrap_envelope(parsed: Any) -> Any:
|
|
"""Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is."""
|
|
if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()):
|
|
return parsed["data"]
|
|
return parsed
|
|
|
|
|
|
def _make_plan() -> Plan:
|
|
"""Create a Plan instance for env priority tests."""
|
|
now = datetime.now()
|
|
return Plan(
|
|
identity=PlanIdentity(plan_id=_PLAN_ULID),
|
|
namespaced_name=NamespacedName.parse("local/env-priority-plan"),
|
|
description="Test plan for env priority",
|
|
definition_of_done="All tests pass",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
project_links=[ProjectLink(project_name="proj-a")],
|
|
arguments={},
|
|
arguments_order=[],
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
reusable=True,
|
|
read_only=False,
|
|
created_by=None,
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
)
|
|
|
|
|
|
def _make_action() -> Action:
|
|
"""Create an Action for plan use tests."""
|
|
return Action(
|
|
namespaced_name=NamespacedName.parse("local/test-action"),
|
|
description="Test action",
|
|
long_description=None,
|
|
definition_of_done="All tests pass",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
reusable=True,
|
|
read_only=False,
|
|
state=ActionState.AVAILABLE,
|
|
created_by=None,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a plan env priority CLI runner")
|
|
def step_env_priority_runner(context: Context) -> None:
|
|
"""Set up the CLI runner."""
|
|
context.runner = CliRunner()
|
|
|
|
|
|
@given("a plan env priority mocked lifecycle service")
|
|
def step_env_priority_service(context: Context) -> None:
|
|
"""Set up a mock PlanLifecycleService."""
|
|
context.mock_service = MagicMock()
|
|
context.service_patcher = patch(
|
|
"cleveragents.cli.commands.plan._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)
|
|
|
|
|
|
@given("a plan env priority action exists")
|
|
def step_env_priority_action(context: Context) -> None:
|
|
"""Set up an action and configure mock service for use_action."""
|
|
context.mock_action = _make_action()
|
|
context.mock_service.get_action_by_name.return_value = context.mock_action
|
|
context.mock_plan = _make_plan()
|
|
context.mock_service.use_action.return_value = context.mock_plan
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I run plan use with execution environment "{env}" and priority "{priority}"')
|
|
def step_run_use_with_env_and_priority(
|
|
context: Context, env: str, priority: str
|
|
) -> None:
|
|
"""Run plan use with both --execution-environment and --execution-env-priority."""
|
|
context.result = context.runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"proj-a",
|
|
"--execution-environment",
|
|
env,
|
|
"--execution-env-priority",
|
|
priority,
|
|
],
|
|
)
|
|
|
|
|
|
@when('I run plan use with priority "{priority}" but no execution environment')
|
|
def step_run_use_with_priority_no_env(context: Context, priority: str) -> None:
|
|
"""Run plan use with --execution-env-priority but no --execution-environment."""
|
|
context.result = context.runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"proj-a",
|
|
"--execution-env-priority",
|
|
priority,
|
|
],
|
|
)
|
|
|
|
|
|
@when('I run plan use with execution environment "{env}" but no priority')
|
|
def step_run_use_with_env_no_priority(context: Context, env: str) -> None:
|
|
"""Run plan use with --execution-environment but no --execution-env-priority."""
|
|
context.result = context.runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"proj-a",
|
|
"--execution-environment",
|
|
env,
|
|
],
|
|
)
|
|
|
|
|
|
@when(
|
|
'I run plan use with execution environment "{env}" and priority "{priority}" in json format'
|
|
)
|
|
def step_run_use_with_env_priority_json(
|
|
context: Context, env: str, priority: str
|
|
) -> None:
|
|
"""Run plan use with env, priority, and JSON format."""
|
|
context.result = context.runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"proj-a",
|
|
"--execution-environment",
|
|
env,
|
|
"--execution-env-priority",
|
|
priority,
|
|
"--format",
|
|
"json",
|
|
],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the plan env priority command should succeed")
|
|
def step_env_priority_succeed(context: Context) -> None:
|
|
"""Assert the CLI command succeeded."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected exit code 0, got {context.result.exit_code}. "
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the plan env priority command should fail")
|
|
def step_env_priority_fail(context: Context) -> None:
|
|
"""Assert the CLI command failed."""
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code, got {context.result.exit_code}. "
|
|
f"Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then('the plan should have execution_env_priority set to "{value}"')
|
|
def step_env_priority_value(context: Context, value: str) -> None:
|
|
"""Assert the plan has the expected execution_env_priority."""
|
|
plan = context.mock_service.use_action.return_value
|
|
assert plan.execution_env_priority == value, (
|
|
f"Expected execution_env_priority={value!r}, "
|
|
f"got {plan.execution_env_priority!r}"
|
|
)
|
|
|
|
|
|
@then('the plan should have execution_environment set to "{value}"')
|
|
def step_env_value(context: Context, value: str) -> None:
|
|
"""Assert the plan has the expected execution_environment."""
|
|
plan = context.mock_service.use_action.return_value
|
|
assert plan.execution_environment == value, (
|
|
f"Expected execution_environment={value!r}, got {plan.execution_environment!r}"
|
|
)
|
|
|
|
|
|
@then('the plan env priority output should contain "{text}"')
|
|
def step_env_priority_output_contains(context: Context, text: str) -> None:
|
|
"""Assert the CLI output contains the expected text."""
|
|
assert text in context.result.output, (
|
|
f"Expected output to contain {text!r}. Output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then('the plan env priority json output should include "{key}" as "{value}"')
|
|
def step_env_priority_json_output(context: Context, key: str, value: str) -> None:
|
|
"""Assert the JSON output includes the expected key/value."""
|
|
parsed = json.loads(context.result.output)
|
|
data = _unwrap_envelope(parsed)
|
|
assert key in data, (
|
|
f"Expected JSON to contain key {key!r}. Keys: {list(data.keys())}"
|
|
)
|
|
assert data[key] == value, f"Expected {key}={value!r}, got {data[key]!r}"
|
|
|
|
|
|
@then("the plan env priority service use_action should have been called")
|
|
def step_verify_use_action_called(context: Context) -> None:
|
|
"""Verify that use_action was called on the lifecycle service."""
|
|
context.mock_service.use_action.assert_called_once()
|
|
|
|
|
|
@then("the plan env priority service save_plan should have been called")
|
|
def step_verify_save_plan_called(context: Context) -> None:
|
|
"""Verify that save_plan was called with the correct plan to persist CLI overrides."""
|
|
context.mock_service.save_plan.assert_called_once()
|
|
saved_plan = context.mock_service.save_plan.call_args[0][0]
|
|
assert saved_plan is context.mock_plan, (
|
|
"save_plan should be called with the same plan object returned by use_action"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Domain model validator steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I construct a Plan with execution_env_priority "{priority}" '
|
|
"and no execution_environment"
|
|
)
|
|
def step_construct_plan_priority_no_env(context: Context, priority: str) -> None:
|
|
"""Try to construct a Plan with priority but no environment."""
|
|
context.construction_error = None
|
|
context.constructed_plan = None
|
|
try:
|
|
context.constructed_plan = _make_plan()
|
|
context.constructed_plan.execution_env_priority = ExecutionEnvPriority(priority)
|
|
context.constructed_plan.execution_environment = None
|
|
except (ValueError, Exception) as exc:
|
|
context.construction_error = exc
|
|
# Also try direct construction (the validator fires in __init__)
|
|
if context.construction_error is None:
|
|
try:
|
|
now = datetime.now()
|
|
Plan(
|
|
identity=PlanIdentity(plan_id=_PLAN_ULID),
|
|
namespaced_name=NamespacedName.parse("local/validator-test"),
|
|
description="Validator test",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
project_links=[ProjectLink(project_name="proj-a")],
|
|
arguments={},
|
|
arguments_order=[],
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
reusable=True,
|
|
read_only=False,
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
execution_env_priority=ExecutionEnvPriority(priority),
|
|
execution_environment=None,
|
|
)
|
|
except (ValueError, Exception) as exc:
|
|
context.construction_error = exc
|
|
|
|
|
|
@when(
|
|
'I construct a Plan with execution_environment "{env}" '
|
|
'and execution_env_priority "{priority}"'
|
|
)
|
|
def step_construct_plan_both_fields(context: Context, env: str, priority: str) -> None:
|
|
"""Construct a Plan with both execution_environment and execution_env_priority."""
|
|
context.construction_error = None
|
|
context.constructed_plan = None
|
|
try:
|
|
now = datetime.now()
|
|
context.constructed_plan = Plan(
|
|
identity=PlanIdentity(plan_id=_PLAN_ULID),
|
|
namespaced_name=NamespacedName.parse("local/validator-test"),
|
|
description="Validator test",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
project_links=[ProjectLink(project_name="proj-a")],
|
|
arguments={},
|
|
arguments_order=[],
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
reusable=True,
|
|
read_only=False,
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
execution_environment=env,
|
|
execution_env_priority=ExecutionEnvPriority(priority),
|
|
)
|
|
except (ValueError, Exception) as exc:
|
|
context.construction_error = exc
|
|
|
|
|
|
@then("the plan construction should raise ValueError")
|
|
def step_construction_raises_value_error(context: Context) -> None:
|
|
"""Assert that plan construction raised a ValueError."""
|
|
assert context.construction_error is not None, (
|
|
"Expected ValueError during plan construction, but no error was raised"
|
|
)
|
|
|
|
|
|
@then('the plan construction error should mention "{text}"')
|
|
def step_construction_error_message(context: Context, text: str) -> None:
|
|
"""Assert the construction error message contains expected text."""
|
|
assert context.construction_error is not None
|
|
assert text in str(context.construction_error), (
|
|
f"Expected error to mention {text!r}, got: {context.construction_error}"
|
|
)
|
|
|
|
|
|
@then("the plan construction should succeed")
|
|
def step_construction_succeeds(context: Context) -> None:
|
|
"""Assert that plan construction succeeded."""
|
|
assert context.construction_error is None, (
|
|
f"Expected plan construction to succeed, "
|
|
f"but got error: {context.construction_error}"
|
|
)
|
|
assert context.constructed_plan is not None
|
|
|
|
|
|
@then('the constructed plan should have execution_env_priority "{value}"')
|
|
def step_constructed_plan_priority(context: Context, value: str) -> None:
|
|
"""Assert the constructed plan has the expected execution_env_priority."""
|
|
assert context.constructed_plan is not None
|
|
actual = context.constructed_plan.execution_env_priority
|
|
assert actual is not None and actual.value == value, (
|
|
f"Expected execution_env_priority={value!r}, got {actual!r}"
|
|
)
|
|
|
|
|
|
@then('the constructed plan should have execution_environment "{value}"')
|
|
def step_constructed_plan_env(context: Context, value: str) -> None:
|
|
"""Assert the constructed plan has the expected execution_environment."""
|
|
assert context.constructed_plan is not None
|
|
assert context.constructed_plan.execution_environment == value, (
|
|
f"Expected execution_environment={value!r}, "
|
|
f"got {context.constructed_plan.execution_environment!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ExecutionEnvPriority enum steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the ExecutionEnvPriority enum should have value "{value}"')
|
|
def step_enum_has_value(context: Context, value: str) -> None:
|
|
"""Assert that ExecutionEnvPriority has the given value."""
|
|
values = [e.value for e in ExecutionEnvPriority]
|
|
assert value in values, (
|
|
f"Expected {value!r} in ExecutionEnvPriority values, got {values}"
|
|
)
|
|
|
|
|
|
@then("ExecutionEnvPriority should be a StrEnum subclass")
|
|
def step_enum_is_strenum(context: Context) -> None:
|
|
"""Assert that ExecutionEnvPriority is a StrEnum subclass."""
|
|
assert issubclass(ExecutionEnvPriority, StrEnum), (
|
|
f"ExecutionEnvPriority should be a StrEnum subclass, "
|
|
f"got bases: {ExecutionEnvPriority.__bases__}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan.as_cli_dict execution environment steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a plan with execution_environment "{env}" and execution_env_priority "{priority}"'
|
|
)
|
|
def step_plan_with_env_and_priority(context: Context, env: str, priority: str) -> None:
|
|
"""Create a plan with both execution environment fields set."""
|
|
context.env_plan = _make_plan()
|
|
context.env_plan.execution_environment = env
|
|
context.env_plan.execution_env_priority = ExecutionEnvPriority(priority)
|
|
|
|
|
|
@given("a plan with no execution_environment")
|
|
def step_plan_no_env(context: Context) -> None:
|
|
"""Create a plan with no execution environment."""
|
|
context.env_plan = _make_plan()
|
|
# execution_environment and execution_env_priority default to None
|
|
|
|
|
|
@given('a plan with execution_environment "{env}" but no execution_env_priority')
|
|
def step_plan_env_no_priority(context: Context, env: str) -> None:
|
|
"""Create a plan with execution_environment but no priority (pre-migration).
|
|
|
|
Simulates pre-migration data where execution_environment was set but
|
|
execution_env_priority was not persisted. Uses ``__dict__`` update
|
|
to bypass the model validator that normally pairs the two fields.
|
|
"""
|
|
context.env_plan = _make_plan()
|
|
context.env_plan.__dict__["execution_environment"] = env
|
|
context.env_plan.__dict__["execution_env_priority"] = None
|
|
|
|
|
|
@when("I call as_cli_dict on the env priority plan")
|
|
def step_call_as_cli_dict_env(context: Context) -> None:
|
|
"""Call as_cli_dict() on the env priority plan and store the result."""
|
|
context.env_cli_dict = context.env_plan.as_cli_dict()
|
|
|
|
|
|
@then('the env priority CLI dict should contain key "{key}" with value "{value}"')
|
|
def step_env_cli_dict_has_key_value(context: Context, key: str, value: str) -> None:
|
|
"""Assert the CLI dict contains the specified key with expected value."""
|
|
assert key in context.env_cli_dict, (
|
|
f"Expected key '{key}' in CLI dict, "
|
|
f"got keys: {list(context.env_cli_dict.keys())}"
|
|
)
|
|
assert str(context.env_cli_dict[key]) == value, (
|
|
f"Expected CLI dict['{key}'] == '{value}', got '{context.env_cli_dict[key]}'"
|
|
)
|
|
|
|
|
|
@then('the env priority CLI dict should not contain key "{key}"')
|
|
def step_env_cli_dict_missing_key(context: Context, key: str) -> None:
|
|
"""Assert the CLI dict does not contain the specified key."""
|
|
assert key not in context.env_cli_dict, (
|
|
f"Key '{key}' should not be in CLI dict but it is: {context.env_cli_dict[key]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB round-trip serialization steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I serialize the plan through LifecyclePlanModel from_domain and to_domain")
|
|
def step_round_trip_serialization(context: Context) -> None:
|
|
"""Perform a from_domain -> to_domain round-trip on the env plan."""
|
|
from cleveragents.infrastructure.database.models import LifecyclePlanModel
|
|
|
|
db_model = LifecyclePlanModel.from_domain(context.env_plan)
|
|
context.round_trip_plan = db_model.to_domain()
|
|
|
|
|
|
@then('the round-trip plan should have execution_environment "{value}"')
|
|
def step_round_trip_env(context: Context, value: str) -> None:
|
|
"""Assert the round-trip plan has the expected execution_environment."""
|
|
assert context.round_trip_plan.execution_environment == value, (
|
|
f"Expected execution_environment={value!r}, "
|
|
f"got {context.round_trip_plan.execution_environment!r}"
|
|
)
|
|
|
|
|
|
@then("the round-trip plan should have execution_environment None")
|
|
def step_round_trip_env_none(context: Context) -> None:
|
|
"""Assert the round-trip plan has execution_environment=None."""
|
|
assert context.round_trip_plan.execution_environment is None, (
|
|
f"Expected execution_environment=None, "
|
|
f"got {context.round_trip_plan.execution_environment!r}"
|
|
)
|
|
|
|
|
|
@then('the round-trip plan should have execution_env_priority "{value}"')
|
|
def step_round_trip_priority(context: Context, value: str) -> None:
|
|
"""Assert the round-trip plan has the expected execution_env_priority."""
|
|
actual = context.round_trip_plan.execution_env_priority
|
|
assert actual is not None and actual.value == value, (
|
|
f"Expected execution_env_priority={value!r}, got {actual!r}"
|
|
)
|
|
|
|
|
|
@then("the round-trip plan should have execution_env_priority None")
|
|
def step_round_trip_priority_none(context: Context) -> None:
|
|
"""Assert the round-trip plan has execution_env_priority=None."""
|
|
assert context.round_trip_plan.execution_env_priority is None, (
|
|
f"Expected execution_env_priority=None, "
|
|
f"got {context.round_trip_plan.execution_env_priority!r}"
|
|
)
|