forked from HAL9000/cleveragents-core
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Step definitions for plan diff/artifacts CLI command coverage.
|
|
|
|
Covers:
|
|
- Lines 1815-1822: _get_apply_service()
|
|
- Lines 1845-1855: plan_diff command (success, PlanError, CleverAgentsError)
|
|
- Lines 1878-1888: plan_artifacts command (success, PlanError, CleverAgentsError)
|
|
- Branch L646->648: build command when actor_registry is None
|
|
- Branch L648->652: build command when testing_mode is False
|
|
- Branch L1130->1129: _print_lifecycle_plan when arguments_order has extra key
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from io import StringIO
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, use_step_matcher, when
|
|
from behave.runner import Context
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
from ulid import ULID
|
|
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
from cleveragents.core.exceptions import CleverAgentsError, PlanError
|
|
|
|
runner = CliRunner()
|
|
|
|
_PATCH_GET_APPLY = "cleveragents.cli.commands.plan._get_apply_service"
|
|
_PATCH_CONSOLE = "cleveragents.cli.commands.plan.console"
|
|
|
|
|
|
def _make_mock_apply_service(
|
|
diff_return=None,
|
|
diff_side_effect=None,
|
|
artifacts_return=None,
|
|
artifacts_side_effect=None,
|
|
):
|
|
svc = MagicMock()
|
|
if diff_side_effect:
|
|
svc.diff.side_effect = diff_side_effect
|
|
else:
|
|
svc.diff.return_value = diff_return or ""
|
|
if artifacts_side_effect:
|
|
svc.artifacts.side_effect = artifacts_side_effect
|
|
else:
|
|
svc.artifacts.return_value = artifacts_return or ""
|
|
return svc
|
|
|
|
|
|
# ---- GIVEN (parse matcher) ----
|
|
|
|
use_step_matcher("parse")
|
|
|
|
|
|
@given('plan_diff_artifacts the apply service returns diff output "{output}"')
|
|
def step_pda_g_diff_out_dq(context: Context, output: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(diff_return=output)
|
|
|
|
|
|
@given("plan_diff_artifacts the apply service returns diff output '{output}'")
|
|
def step_pda_g_diff_out_sq(context: Context, output: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(diff_return=output)
|
|
|
|
|
|
@given('plan_diff_artifacts the apply service diff raises a PlanError "{msg}"')
|
|
def step_pda_g_diff_planerr(context: Context, msg: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(
|
|
diff_side_effect=PlanError(msg),
|
|
)
|
|
|
|
|
|
@given('plan_diff_artifacts the apply service diff raises a CleverAgentsError "{msg}"')
|
|
def step_pda_g_diff_caerr(context: Context, msg: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(
|
|
diff_side_effect=CleverAgentsError(msg),
|
|
)
|
|
|
|
|
|
@given('plan_diff_artifacts the apply service returns artifacts output "{output}"')
|
|
def step_pda_g_art_out_dq(context: Context, output: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(artifacts_return=output)
|
|
|
|
|
|
@given("plan_diff_artifacts the apply service returns artifacts output '{output}'")
|
|
def step_pda_g_art_out_sq(context: Context, output: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(artifacts_return=output)
|
|
|
|
|
|
@given('plan_diff_artifacts the apply service artifacts raises a PlanError "{msg}"')
|
|
def step_pda_g_art_planerr(context: Context, msg: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(
|
|
artifacts_side_effect=PlanError(msg),
|
|
)
|
|
|
|
|
|
@given(
|
|
'plan_diff_artifacts the apply service artifacts raises a CleverAgentsError "{msg}"'
|
|
)
|
|
def step_pda_g_art_caerr(context: Context, msg: str) -> None:
|
|
context.pda_mock_service = _make_mock_apply_service(
|
|
artifacts_side_effect=CleverAgentsError(msg),
|
|
)
|
|
|
|
|
|
@given("plan_diff_artifacts a mock lifecycle service")
|
|
def step_pda_g_mock_lifecycle(context: Context) -> None:
|
|
context.pda_mock_lifecycle = MagicMock()
|
|
|
|
|
|
@given("plan_diff_artifacts a container without actor_registry")
|
|
def step_pda_g_container_no_ar(context: Context) -> None:
|
|
container = MagicMock()
|
|
del container.actor_registry
|
|
|
|
plan_service = MagicMock()
|
|
plan_service.build_plan.return_value = [
|
|
SimpleNamespace(file_path="a.py", change_type="create")
|
|
]
|
|
container.plan_service.return_value = plan_service
|
|
container.actor_service.return_value = MagicMock()
|
|
|
|
context.pda_container = container
|
|
context.pda_project = SimpleNamespace(name="test-project")
|
|
context.pda_plan_service = plan_service
|
|
|
|
|
|
@given("plan_diff_artifacts a container with actor_registry but testing mode disabled")
|
|
def step_pda_g_container_no_testing(context: Context) -> None:
|
|
container = MagicMock()
|
|
actor_registry = MagicMock()
|
|
container.actor_registry.return_value = actor_registry
|
|
|
|
plan_service = MagicMock()
|
|
plan_service.build_plan.return_value = [
|
|
SimpleNamespace(file_path="b.py", change_type="modify")
|
|
]
|
|
container.plan_service.return_value = plan_service
|
|
actor_service = MagicMock()
|
|
container.actor_service.return_value = actor_service
|
|
|
|
context.pda_container = container
|
|
context.pda_project = SimpleNamespace(name="test-project-2")
|
|
context.pda_plan_service = plan_service
|
|
context.pda_actor_service = actor_service
|
|
context.pda_actor_registry = actor_registry
|
|
|
|
|
|
@given("plan_diff_artifacts a lifecycle plan with arguments_order having an extra key")
|
|
def step_pda_g_plan_extra_arg(context: Context) -> None:
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
)
|
|
|
|
valid_ulid = str(ULID())
|
|
plan = Plan(
|
|
identity=PlanIdentity(plan_id=valid_ulid),
|
|
namespaced_name=NamespacedName.parse("test/args-plan"),
|
|
description="Test plan for arguments_order branch",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
arguments={"keep_me": "value1"},
|
|
arguments_order=["keep_me", "ghost_key"],
|
|
)
|
|
context.pda_plan = plan
|
|
|
|
|
|
# ---- WHEN (regex matcher for parameterized steps) ----
|
|
|
|
use_step_matcher("re")
|
|
|
|
|
|
@when(
|
|
r'plan_diff_artifacts I run the diff subcommand for plan "(?P<plan_id>[^"]+)" using format "(?P<fmt>[^"]+)"'
|
|
)
|
|
def step_pda_w_diff_fmt(context: Context, plan_id: str, fmt: str) -> None:
|
|
with patch(_PATCH_GET_APPLY, return_value=context.pda_mock_service):
|
|
context.pda_result = runner.invoke(plan_app, ["diff", plan_id, "--format", fmt])
|
|
|
|
|
|
@when(r'plan_diff_artifacts I run the diff subcommand for plan "(?P<plan_id>[^"]+)"')
|
|
def step_pda_w_diff(context: Context, plan_id: str) -> None:
|
|
with patch(_PATCH_GET_APPLY, return_value=context.pda_mock_service):
|
|
context.pda_result = runner.invoke(plan_app, ["diff", plan_id])
|
|
|
|
|
|
@when(
|
|
r'plan_diff_artifacts I run the artifacts subcommand for plan "(?P<plan_id>[^"]+)" using format "(?P<fmt>[^"]+)"'
|
|
)
|
|
def step_pda_w_artifacts_fmt(context: Context, plan_id: str, fmt: str) -> None:
|
|
with patch(_PATCH_GET_APPLY, return_value=context.pda_mock_service):
|
|
context.pda_result = runner.invoke(
|
|
plan_app, ["artifacts", plan_id, "--format", fmt]
|
|
)
|
|
|
|
|
|
@when(
|
|
r'plan_diff_artifacts I run the artifacts subcommand for plan "(?P<plan_id>[^"]+)"'
|
|
)
|
|
def step_pda_w_artifacts(context: Context, plan_id: str) -> None:
|
|
with patch(_PATCH_GET_APPLY, return_value=context.pda_mock_service):
|
|
context.pda_result = runner.invoke(plan_app, ["artifacts", plan_id])
|
|
|
|
|
|
# Switch back to parse matcher for non-parameterized steps
|
|
use_step_matcher("parse")
|
|
|
|
|
|
@when("plan_diff_artifacts I call _get_apply_service")
|
|
def step_pda_w_get_apply(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import _get_apply_service
|
|
|
|
mock_pas_instance = MagicMock()
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=context.pda_mock_lifecycle,
|
|
),
|
|
patch(
|
|
"cleveragents.application.services.plan_apply_service.PlanApplyService",
|
|
return_value=mock_pas_instance,
|
|
),
|
|
):
|
|
context.pda_apply_service_result = _get_apply_service()
|
|
|
|
|
|
@when("plan_diff_artifacts I invoke the build command")
|
|
def step_pda_w_build(context: Context) -> None:
|
|
with (
|
|
patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=context.pda_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_current_project",
|
|
return_value=context.pda_project,
|
|
),
|
|
patch.dict(os.environ, {"CLEVERAGENTS_TESTING_USE_MOCK_AI": ""}, clear=False),
|
|
):
|
|
context.pda_result = runner.invoke(plan_app, ["build"])
|
|
|
|
|
|
@when("plan_diff_artifacts I print the lifecycle plan")
|
|
def step_pda_w_print_plan(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
output = StringIO()
|
|
test_console = Console(file=output, force_terminal=False, width=120)
|
|
with patch(_PATCH_CONSOLE, test_console):
|
|
_print_lifecycle_plan(context.pda_plan, title="Test Plan")
|
|
context.pda_console_output = output.getvalue()
|
|
|
|
|
|
# ---- THEN ----
|
|
|
|
|
|
@then("plan_diff_artifacts the exit code should be {code:d}")
|
|
def step_pda_t_exit_code(context: Context, code: int) -> None:
|
|
assert context.pda_result.exit_code == code, (
|
|
f"Expected exit code {code}, got {context.pda_result.exit_code}. "
|
|
f"Output: {context.pda_result.output}"
|
|
)
|
|
|
|
|
|
@then('plan_diff_artifacts the output should contain "{text}"')
|
|
def step_pda_t_output_contains(context: Context, text: str) -> None:
|
|
full = context.pda_result.output
|
|
assert text in full, f"Expected '{text}' in output. Got:\n{full}"
|
|
|
|
|
|
@then("plan_diff_artifacts the command should be aborted")
|
|
def step_pda_t_aborted(context: Context) -> None:
|
|
assert context.pda_result.exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.pda_result.exit_code}. "
|
|
f"Output: {context.pda_result.output}"
|
|
)
|
|
|
|
|
|
@then("plan_diff_artifacts a PlanApplyService should be returned")
|
|
def step_pda_t_apply_svc(context: Context) -> None:
|
|
assert context.pda_apply_service_result is not None
|
|
|
|
|
|
@then("plan_diff_artifacts the build should complete without actor registry calls")
|
|
def step_pda_t_no_ar(context: Context) -> None:
|
|
assert not hasattr(context.pda_container, "actor_registry")
|
|
|
|
|
|
@then("plan_diff_artifacts ensure_default_mock_actor should not be called")
|
|
def step_pda_t_no_mock_actor(context: Context) -> None:
|
|
context.pda_actor_service.ensure_default_mock_actor.assert_not_called()
|
|
|
|
|
|
@then("plan_diff_artifacts the output should contain the valid argument")
|
|
def step_pda_t_valid_arg(context: Context) -> None:
|
|
assert "keep_me" in context.pda_console_output, (
|
|
f"Expected 'keep_me' in: {context.pda_console_output}"
|
|
)
|
|
assert "value1" in context.pda_console_output, (
|
|
f"Expected 'value1' in: {context.pda_console_output}"
|
|
)
|
|
|
|
|
|
@then("plan_diff_artifacts the output should not contain the missing key value")
|
|
def step_pda_t_no_ghost(context: Context) -> None:
|
|
assert "ghost_key =" not in context.pda_console_output, (
|
|
f"Did not expect 'ghost_key =' in: {context.pda_console_output}"
|
|
)
|