forked from HAL9000/cleveragents-core
hamza.khyari
06428a5db6
Replace flat shutil.copy2 apply with spec-aligned git worktree flow
(specification.md §13225-13276):
Execute phase: creates an isolated git worktree via GitWorktreeSandbox
for the plan's linked git-checkout resource. LLM file output is
written to the worktree and committed on branch
cleveragents/plan-<plan_id> — no merge yet.
Apply phase: merges the worktree branch into the project's current
branch via git merge. Prints spec-aligned panels:
- Apply Summary: Plan ID, artifacts count, insertions/deletions,
project name, applied-at timestamp
- Sandbox Cleanup: worktree removed, branch merged to main
- Footer: ✓ OK Changes applied
Non-git projects (fs-directory resources) fall back to the original
flat directory sandbox with shutil.copy2.
Also fixes:
- context_tier_hydrator metadata types (int/float → string) that
caused Pydantic validation errors during context assembly
- A2A facade duplicate execute dispatch (idempotent handler)
- Logs actual error from context assembly failures
ISSUES CLOSED: #4454
1352 lines
42 KiB
Python
1352 lines
42 KiB
Python
"""Step definitions for V3 lifecycle commands and streaming coverage tests."""
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from io import StringIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
from cleveragents.core.exceptions import (
|
|
CleverAgentsError,
|
|
NotFoundError,
|
|
PlanError,
|
|
ValidationError,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers to build mock lifecycle plan objects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
name="test-plan",
|
|
namespace="local",
|
|
phase="strategize",
|
|
state="queued",
|
|
project_names=None,
|
|
description="Test plan description for coverage",
|
|
error_message=None,
|
|
):
|
|
"""Build a mock that looks like a v3 Plan (LifecyclePlan) from the domain model."""
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
Plan as LifecyclePlan,
|
|
)
|
|
|
|
if project_names is None:
|
|
project_names = ["proj-1"]
|
|
|
|
links = [ProjectLink(project_name=pn) for pn in project_names]
|
|
|
|
plan = LifecyclePlan(
|
|
identity=PlanIdentity(plan_id=plan_id),
|
|
namespaced_name=NamespacedName(namespace=namespace, name=name),
|
|
action_name=f"{namespace}/test-action",
|
|
description=description,
|
|
phase=PlanPhase(phase),
|
|
processing_state=ProcessingState(state),
|
|
project_links=links,
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
timestamps=PlanTimestamps(created_at=datetime.now(), updated_at=datetime.now()),
|
|
error_message=error_message,
|
|
)
|
|
return plan
|
|
|
|
|
|
def _make_mock_action():
|
|
"""Build a minimal mock action with a namespaced_name attribute."""
|
|
action = MagicMock()
|
|
action.namespaced_name = "local/test-action"
|
|
return action
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# use_action scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I invoke use_action with string argument "target_coverage=80"')
|
|
def step_invoke_use_action_string_arg(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--arg",
|
|
"target_coverage=80",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.lifecycle_service_mock = mock_service
|
|
|
|
|
|
@when('I invoke use_action with args "count=42" and "ratio=3.14"')
|
|
def step_invoke_use_action_int_float_args(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--arg",
|
|
"count=42",
|
|
"--arg",
|
|
"ratio=3.14",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.lifecycle_service_mock = mock_service
|
|
|
|
|
|
@when('I invoke use_action with args "verbose=true" and "dry_run=false"')
|
|
def step_invoke_use_action_bool_args(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--arg",
|
|
"verbose=true",
|
|
"--arg",
|
|
"dry_run=false",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.lifecycle_service_mock = mock_service
|
|
|
|
|
|
@when('I invoke use_action with malformed argument "badarg"')
|
|
def step_invoke_use_action_malformed_arg(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--arg",
|
|
"badarg",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke use_action where get_action raises NotFoundError")
|
|
def step_invoke_use_action_name_fallback(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.side_effect = NotFoundError(
|
|
resource_type="action",
|
|
resource_id="local/my-action",
|
|
)
|
|
mock_service.get_action_by_name.return_value = mock_action
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/my-action",
|
|
"--project",
|
|
"proj-1",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.lifecycle_service_mock = mock_service
|
|
|
|
|
|
@when('I invoke use_action with automation profile "full-auto"')
|
|
def step_invoke_use_action_custom_automation(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--automation-profile",
|
|
"full-auto",
|
|
],
|
|
)
|
|
context.result = result
|
|
context.lifecycle_service_mock = mock_service
|
|
|
|
|
|
@when('I invoke use_action with automation profile "super_auto"')
|
|
def step_invoke_use_action_invalid_automation(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
"--automation-profile",
|
|
"super_auto",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke use_action and ActionNotAvailableError is raised")
|
|
def step_invoke_use_action_not_available(context):
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
ActionNotAvailableError,
|
|
)
|
|
from cleveragents.domain.models.core.action import ActionState
|
|
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_service.use_action.side_effect = ActionNotAvailableError(
|
|
"local/test-action",
|
|
ActionState.ARCHIVED,
|
|
)
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke use_action and ValidationError is raised")
|
|
def step_invoke_use_action_validation_error(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_service.use_action.side_effect = ValidationError("Missing required argument")
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke use_action and CleverAgentsError is raised")
|
|
def step_invoke_use_action_general_error(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_action = _make_mock_action()
|
|
mock_service.get_action.return_value = mock_action
|
|
mock_service.use_action.side_effect = CleverAgentsError("Something went wrong")
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"use",
|
|
"local/test-action",
|
|
"--project",
|
|
"proj-1",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# use_action then assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the use_action CLI should succeed")
|
|
def step_use_action_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the lifecycle plan panel should be printed")
|
|
def step_lifecycle_plan_panel_printed(context):
|
|
assert (
|
|
"Plan Created" in context.result.output
|
|
or "plan" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the parsed arguments should contain integer 42 and float 3.14")
|
|
def step_parsed_int_float(context):
|
|
call_kwargs = context.lifecycle_service_mock.use_action.call_args
|
|
args_passed = call_kwargs.kwargs.get("arguments") or call_kwargs[1].get(
|
|
"arguments", {}
|
|
)
|
|
assert args_passed.get("count") == 42
|
|
assert abs(args_passed.get("ratio", 0) - 3.14) < 0.001
|
|
|
|
|
|
@then("the parsed arguments should contain booleans true and false")
|
|
def step_parsed_booleans(context):
|
|
call_kwargs = context.lifecycle_service_mock.use_action.call_args
|
|
args_passed = call_kwargs.kwargs.get("arguments") or call_kwargs[1].get(
|
|
"arguments", {}
|
|
)
|
|
assert args_passed.get("verbose") is True
|
|
assert args_passed.get("dry_run") is False
|
|
|
|
|
|
@then("the use_action CLI should abort")
|
|
def step_use_action_cli_abort(context):
|
|
assert context.result.exit_code != 0, (
|
|
f"Expected non-zero exit code, got {context.result.exit_code}"
|
|
)
|
|
|
|
|
|
@then("the output should mention invalid argument format")
|
|
def step_output_invalid_arg_format(context):
|
|
assert "Invalid argument format" in context.result.output
|
|
|
|
|
|
@then("the use_action CLI should succeed via name fallback")
|
|
def step_use_action_name_fallback_succeed(context):
|
|
assert context.result.exit_code == 0
|
|
context.lifecycle_service_mock.get_action_by_name.assert_called_once()
|
|
|
|
|
|
@then("the lifecycle service should receive full-auto profile")
|
|
def step_service_received_full_automation(context):
|
|
# Verify the service's use_action was called (automation profile is applied
|
|
# after use_action returns, so we just check the call happened).
|
|
context.lifecycle_service_mock.use_action.assert_called_once()
|
|
|
|
|
|
@then("the output should mention invalid automation profile")
|
|
def step_output_invalid_automation(context):
|
|
output = context.result.output.lower()
|
|
assert "automation" in output or context.result.exit_code != 0
|
|
|
|
|
|
@then("the output should mention action not available")
|
|
def step_output_action_not_available(context):
|
|
assert "not available" in context.result.output.lower()
|
|
|
|
|
|
@then("the use_action CLI should abort with lifecycle validation error")
|
|
def step_use_action_abort_validation(context):
|
|
assert context.result.exit_code != 0
|
|
assert "Validation Error" in context.result.output
|
|
|
|
|
|
@then("the use_action CLI should abort with lifecycle general error")
|
|
def step_use_action_abort_general(context):
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# execute_plan scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I invoke execute_plan with a valid plan ID")
|
|
def step_invoke_execute_plan_with_id(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_plan = _make_mock_lifecycle_plan(phase="execute", state="queued")
|
|
mock_service.execute_plan.return_value = mock_plan
|
|
mock_service.get_plan.return_value = mock_plan
|
|
|
|
mock_executor = MagicMock()
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_plan_executor",
|
|
return_value=mock_executor,
|
|
),
|
|
):
|
|
result = runner.invoke(plan_app, ["execute", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke execute_plan without ID and one strategize-complete plan exists")
|
|
def step_invoke_execute_plan_auto_select(context):
|
|
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan = _make_mock_lifecycle_plan(phase="strategize", state="complete")
|
|
mock_service.list_plans.return_value = [plan]
|
|
mock_service.get_plan.return_value = plan
|
|
mock_service.execute_plan.return_value = _make_mock_lifecycle_plan(
|
|
phase="execute",
|
|
state="queued",
|
|
)
|
|
|
|
mock_executor = MagicMock()
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_plan_executor",
|
|
return_value=mock_executor,
|
|
),
|
|
):
|
|
result = runner.invoke(plan_app, ["execute"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke execute_plan without ID and no strategize-complete plans exist")
|
|
def step_invoke_execute_plan_no_ready(context):
|
|
"""Legacy step: kept for backward-compat with older feature files."""
|
|
step_invoke_execute_plan_no_plans(context)
|
|
|
|
|
|
@when("I invoke execute_plan without ID and no strategize plans exist at all")
|
|
def step_invoke_execute_plan_no_plans(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
# Return an empty list — no strategize plans at all
|
|
mock_service.list_plans.return_value = []
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["execute"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke execute_plan without ID and multiple strategize-complete plans exist")
|
|
def step_invoke_execute_plan_multiple_ready(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan1 = _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
phase="strategize",
|
|
state="complete",
|
|
)
|
|
plan2 = _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAB",
|
|
name="test-plan-2",
|
|
phase="strategize",
|
|
state="complete",
|
|
)
|
|
mock_service.list_plans.return_value = [plan1, plan2]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["execute"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke execute_plan and InvalidPhaseTransitionError is raised")
|
|
def step_invoke_execute_plan_invalid_transition(context):
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
InvalidPhaseTransitionError,
|
|
)
|
|
from cleveragents.domain.models.core.plan import PlanPhase
|
|
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
# Provide a real plan so the phase check passes before the error is hit
|
|
mock_plan = _make_mock_lifecycle_plan(phase="strategize", state="complete")
|
|
mock_service.get_plan.return_value = mock_plan
|
|
mock_service.execute_plan.side_effect = InvalidPhaseTransitionError(
|
|
PlanPhase.APPLY,
|
|
PlanPhase.EXECUTE,
|
|
)
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["execute", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke execute_plan and PlanNotReadyError is raised")
|
|
def step_invoke_execute_plan_not_ready(context):
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanNotReadyError,
|
|
)
|
|
from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState
|
|
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
# Provide a real plan so the phase check passes before the error is hit
|
|
mock_plan = _make_mock_lifecycle_plan(phase="strategize", state="complete")
|
|
mock_service.get_plan.return_value = mock_plan
|
|
mock_service.execute_plan.side_effect = PlanNotReadyError(
|
|
"01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
PlanPhase.STRATEGIZE,
|
|
ProcessingState.QUEUED,
|
|
)
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["execute", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
context.result = result
|
|
|
|
|
|
# execute_plan then assertions
|
|
|
|
|
|
@then("the execute_plan CLI should succeed")
|
|
def step_execute_plan_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the execute phase panel should be printed")
|
|
def step_execute_phase_panel(context):
|
|
out = context.result.output.lower()
|
|
assert "plan executing" in out or "execute" in out
|
|
|
|
|
|
@then("the output should mention no plans ready for execution")
|
|
def step_output_no_plans_ready_execute(context):
|
|
assert "No plans ready for execution" in context.result.output
|
|
|
|
|
|
@then("the output should mention multiple plans ready")
|
|
def step_output_multiple_plans_ready(context):
|
|
assert "Multiple plans ready" in context.result.output
|
|
|
|
|
|
@then("the execute_plan CLI should abort")
|
|
def step_execute_plan_cli_abort(context):
|
|
assert context.result.exit_code != 0
|
|
|
|
|
|
@then("the output should mention invalid transition")
|
|
def step_output_invalid_transition(context):
|
|
assert "Invalid transition" in context.result.output
|
|
|
|
|
|
@then("the output should mention plan not ready")
|
|
def step_output_plan_not_ready(context):
|
|
assert "not ready" in context.result.output.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# execute_plan — inline strategize & auto-progress scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I invoke execute_plan with a plan in strategize-queued state")
|
|
def step_invoke_execute_plan_queued(context):
|
|
"""Plan is in STRATEGIZE/QUEUED — CLI should auto-run strategize inline."""
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
|
|
queued_plan = _make_mock_lifecycle_plan(phase="strategize", state="queued")
|
|
executed_plan = _make_mock_lifecycle_plan(phase="execute", state="queued")
|
|
completed_plan = _make_mock_lifecycle_plan(phase="execute", state="complete")
|
|
|
|
# get_plan call sequence (consolidated read-only + phase check):
|
|
# 1. current_plan (null + read-only + strategize state)
|
|
# 2. re-fetch after inline strategize
|
|
# 3. inline execute state inspection
|
|
# 4. re-fetch after inline execute
|
|
mock_service.get_plan.side_effect = [
|
|
queued_plan,
|
|
executed_plan,
|
|
executed_plan,
|
|
completed_plan,
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._create_sandbox_for_plan",
|
|
return_value=(None, None),
|
|
),
|
|
patch(
|
|
"cleveragents.application.services.plan_executor.PlanExecutor",
|
|
) as mock_executor_cls,
|
|
):
|
|
mock_executor = MagicMock()
|
|
mock_executor_cls.return_value = mock_executor
|
|
result = runner.invoke(plan_app, ["execute", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
|
|
context.result = result
|
|
context.mock_executor = mock_executor
|
|
|
|
|
|
@then("the PlanExecutor should have run strategize inline")
|
|
def step_executor_ran_strategize(context):
|
|
context.mock_executor.run_strategize.assert_called_once_with(
|
|
"01JAAAAAAAAAAAAAAAAAAAAAAA"
|
|
)
|
|
|
|
|
|
@when("I invoke execute_plan and auto-progress already moved plan to execute")
|
|
def step_invoke_execute_plan_auto_progressed(context):
|
|
"""After inline strategize, auto_progress moved plan to EXECUTE.
|
|
|
|
The CLI should detect the plan is already in Execute and skip the
|
|
explicit ``execute_plan`` service call.
|
|
"""
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
|
|
queued_plan = _make_mock_lifecycle_plan(phase="strategize", state="queued")
|
|
executed_plan = _make_mock_lifecycle_plan(phase="execute", state="queued")
|
|
completed_plan = _make_mock_lifecycle_plan(phase="execute", state="complete")
|
|
|
|
# get_plan call sequence (consolidated read-only + phase check):
|
|
# 1. current_plan (null + read-only + strategize state)
|
|
# 2. re-fetch after inline strategize → execute/queued
|
|
# 3. inline execute state inspection → execute/queued (triggers run_execute)
|
|
# 4. re-fetch after inline execute → execute/complete
|
|
mock_service.get_plan.side_effect = [
|
|
queued_plan,
|
|
executed_plan,
|
|
executed_plan,
|
|
completed_plan,
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._create_sandbox_for_plan",
|
|
return_value=(None, None),
|
|
),
|
|
patch(
|
|
"cleveragents.application.services.plan_executor.PlanExecutor",
|
|
) as mock_executor_cls,
|
|
):
|
|
mock_executor = MagicMock()
|
|
mock_executor_cls.return_value = mock_executor
|
|
result = runner.invoke(plan_app, ["execute", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
|
|
context.result = result
|
|
# execute_plan should NOT have been called — auto_progress already
|
|
# moved the plan to Execute.
|
|
mock_service.execute_plan.assert_not_called()
|
|
|
|
|
|
@then("the lifecycle service should persist the plan overrides")
|
|
def step_lifecycle_persists_overrides(context):
|
|
context.lifecycle_service_mock._commit_plan.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_plan scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I invoke apply with a valid plan ID")
|
|
def step_invoke_apply_with_id(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_plan = _make_mock_lifecycle_plan(phase="apply", state="queued")
|
|
mock_service.apply_plan.return_value = mock_plan
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._apply_sandbox_changes",
|
|
),
|
|
):
|
|
result = runner.invoke(
|
|
plan_app, ["apply", "--yes", "01JAAAAAAAAAAAAAAAAAAAAAAA"]
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke apply without ID and one execute-complete plan exists")
|
|
def step_invoke_apply_auto_select(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan = _make_mock_lifecycle_plan(phase="execute", state="complete")
|
|
# list_plans is called twice: once for EXECUTE phase, once for APPLY.
|
|
mock_service.list_plans.side_effect = [[plan], []]
|
|
mock_service.apply_plan.return_value = _make_mock_lifecycle_plan(
|
|
phase="apply",
|
|
state="queued",
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.plan._apply_sandbox_changes",
|
|
),
|
|
):
|
|
result = runner.invoke(plan_app, ["apply", "--yes"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke apply without ID and no execute-complete plans exist")
|
|
def step_invoke_apply_no_ready(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan = _make_mock_lifecycle_plan(phase="execute", state="queued")
|
|
# list_plans is called twice: once for EXECUTE phase, once for APPLY.
|
|
# Both return the same execute/queued plan which is ineligible for
|
|
# either filter (not COMPLETE for execute, not in APPLY phase).
|
|
mock_service.list_plans.side_effect = [[plan], []]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["apply"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke apply without ID and multiple execute-complete plans exist")
|
|
def step_invoke_apply_multiple_ready(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan1 = _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
phase="execute",
|
|
state="complete",
|
|
)
|
|
plan2 = _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAB",
|
|
name="test-plan-2",
|
|
phase="execute",
|
|
state="complete",
|
|
)
|
|
# list_plans is called twice: once for EXECUTE phase, once for APPLY.
|
|
mock_service.list_plans.side_effect = [[plan1, plan2], []]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["apply"])
|
|
context.result = result
|
|
|
|
|
|
# apply then assertions
|
|
|
|
|
|
@then("the apply CLI should succeed")
|
|
def step_apply_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the apply phase panel should be printed")
|
|
def step_apply_phase_panel(context):
|
|
out = context.result.output.lower()
|
|
assert "plan applying" in out or "apply" in out
|
|
|
|
|
|
@then("the apply CLI should abort")
|
|
def step_apply_cli_abort(context):
|
|
assert context.result.exit_code != 0
|
|
|
|
|
|
@then("the output should mention no plans ready for apply")
|
|
def step_output_no_plans_ready_apply(context):
|
|
assert "No plans ready for apply" in context.result.output
|
|
|
|
|
|
@then("the output should mention multiple plans ready for apply")
|
|
def step_output_multiple_plans_ready_apply(context):
|
|
assert "Multiple plans ready for apply" in context.result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# plan_status scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I invoke plan status with a specific plan ID")
|
|
def step_invoke_plan_status_with_id(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_plan = _make_mock_lifecycle_plan()
|
|
mock_service.get_plan.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["status", "01JAAAAAAAAAAAAAAAAAAAAAAA"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke plan status without ID and active plans exist")
|
|
def step_invoke_plan_status_list_active(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan1 = _make_mock_lifecycle_plan()
|
|
plan2 = _make_mock_lifecycle_plan(
|
|
plan_id="01JAAAAAAAAAAAAAAAAAAAAAAB",
|
|
name="plan-2",
|
|
phase="execute",
|
|
state="processing",
|
|
)
|
|
mock_service.list_plans.return_value = [plan1, plan2]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["status"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke plan status without ID and no plans exist")
|
|
def step_invoke_plan_status_no_plans(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_service.list_plans.return_value = []
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["status"])
|
|
context.result = result
|
|
|
|
|
|
# plan_status then assertions
|
|
|
|
|
|
@then("the plan status CLI should succeed")
|
|
def step_plan_status_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the plan status panel should be displayed")
|
|
def step_plan_status_panel_displayed(context):
|
|
out = context.result.output
|
|
assert "Plan Status" in out or "plan" in out.lower()
|
|
|
|
|
|
@then("the active plans table should be displayed")
|
|
def step_active_plans_table_displayed(context):
|
|
out = context.result.output
|
|
assert "Active Plans" in out or "total" in out.lower()
|
|
|
|
|
|
@then("the plan status CLI should succeed with no plans message")
|
|
def step_plan_status_no_plans_message(context):
|
|
assert context.result.exit_code == 0
|
|
assert "No v3 lifecycle plans found" in context.result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_plans scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I invoke list without filters")
|
|
def step_invoke_list_no_filters(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan = _make_mock_lifecycle_plan()
|
|
mock_service.list_plans.return_value = [plan]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when('I invoke list with phase filter "strategize"')
|
|
def step_invoke_list_phase_filter(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
plan = _make_mock_lifecycle_plan(phase="strategize", state="complete")
|
|
mock_service.list_plans.return_value = [plan]
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["list", "--phase", "strategize"])
|
|
context.result = result
|
|
|
|
|
|
@when('I invoke list with phase filter "bogus_phase"')
|
|
def step_invoke_list_invalid_phase(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["list", "--phase", "bogus_phase"])
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke list and no plans match")
|
|
def step_invoke_list_empty(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_service.list_plans.return_value = []
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(plan_app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
# list then assertions
|
|
|
|
|
|
@then("the list CLI should succeed")
|
|
def step_list_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the lifecycle plans table should be displayed")
|
|
def step_lifecycle_plans_table_displayed(context):
|
|
out = context.result.output
|
|
assert "V3 Lifecycle Plans" in out or "total" in out.lower()
|
|
|
|
|
|
@then("the list CLI should abort")
|
|
def step_list_cli_abort(context):
|
|
assert context.result.exit_code != 0
|
|
|
|
|
|
@then("the output should mention invalid phase")
|
|
def step_output_invalid_phase(context):
|
|
assert "Invalid phase" in context.result.output
|
|
|
|
|
|
@then("the list CLI should succeed with empty list message")
|
|
def step_list_empty_message(context):
|
|
assert context.result.exit_code == 0
|
|
assert "No plans found" in context.result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# set_automation_level scenarios (command removed - steps kept as no-ops)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the output should mention cannot change level")
|
|
def step_output_cannot_change_level(context):
|
|
assert (
|
|
"Cannot change level" in context.result.output
|
|
or "cannot change" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cancel_plan scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I invoke cancel with a reason "no longer needed"')
|
|
def step_invoke_cancel_with_reason(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_plan = _make_mock_lifecycle_plan(state="cancelled")
|
|
mock_service.cancel_plan.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"cancel",
|
|
"01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
"--reason",
|
|
"no longer needed",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke cancel without a reason")
|
|
def step_invoke_cancel_without_reason(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_plan = _make_mock_lifecycle_plan(state="cancelled")
|
|
mock_service.cancel_plan.return_value = mock_plan
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"cancel",
|
|
"01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
@when("I invoke cancel and PlanError is raised")
|
|
def step_invoke_cancel_plan_error(context):
|
|
runner = CliRunner()
|
|
mock_service = MagicMock()
|
|
mock_service.cancel_plan.side_effect = PlanError("Plan already terminal")
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=mock_service,
|
|
):
|
|
result = runner.invoke(
|
|
plan_app,
|
|
[
|
|
"cancel",
|
|
"01JAAAAAAAAAAAAAAAAAAAAAAA",
|
|
],
|
|
)
|
|
context.result = result
|
|
|
|
|
|
# cancel then assertions
|
|
|
|
|
|
@then("the cancel CLI should succeed")
|
|
def step_cancel_cli_succeed(context):
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then("the output should show cancellation with reason")
|
|
def step_output_cancel_with_reason(context):
|
|
out = context.result.output
|
|
assert "cancelled" in out.lower() or "Plan cancelled" in out
|
|
assert "no longer needed" in out.lower()
|
|
|
|
|
|
@then("the output should show cancellation without reason text")
|
|
def step_output_cancel_no_reason(context):
|
|
out = context.result.output
|
|
assert "cancelled" in out.lower() or "Plan cancelled" in out
|
|
# Should NOT contain "Reason:" line since no reason was given
|
|
assert "Reason:" not in out
|
|
|
|
|
|
@then("the cancel CLI should abort")
|
|
def step_cancel_cli_abort(context):
|
|
assert context.result.exit_code != 0
|
|
|
|
|
|
@then("the output should mention cannot cancel")
|
|
def step_output_cannot_cancel(context):
|
|
assert (
|
|
"Cannot cancel" in context.result.output
|
|
or "cannot cancel" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _tell_streaming scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run _tell_streaming with a successful async event stream")
|
|
def step_run_tell_streaming_success(context):
|
|
from cleveragents.cli.commands.plan import _tell_streaming
|
|
|
|
mock_project = MagicMock()
|
|
mock_plan_service = MagicMock()
|
|
|
|
async def mock_stream(*args, **kwargs):
|
|
yield {"load_context": {"status": "ok"}}
|
|
yield {"generate_plan": {"status": "ok"}}
|
|
yield {"__end__": {"status": "done"}}
|
|
|
|
mock_plan_service.generate_plan_streaming = mock_stream
|
|
|
|
# Capture console output
|
|
string_buffer = StringIO()
|
|
from rich.console import Console
|
|
|
|
test_console = Console(file=string_buffer, force_terminal=True, width=120)
|
|
|
|
context.streaming_error = None
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
try:
|
|
asyncio.run(
|
|
_tell_streaming(
|
|
mock_project, "Test description", None, mock_plan_service
|
|
)
|
|
)
|
|
except Exception as e:
|
|
context.streaming_error = e
|
|
|
|
context.streaming_output = string_buffer.getvalue()
|
|
|
|
|
|
@when("I run _tell_streaming with an error during streaming")
|
|
def step_run_tell_streaming_error(context):
|
|
from cleveragents.cli.commands.plan import _tell_streaming
|
|
|
|
mock_project = MagicMock()
|
|
mock_plan_service = MagicMock()
|
|
|
|
async def mock_stream_with_error(*args, **kwargs):
|
|
yield {"load_context": {"status": "ok"}}
|
|
raise RuntimeError("Provider connection lost")
|
|
|
|
mock_plan_service.generate_plan_streaming = mock_stream_with_error
|
|
|
|
string_buffer = StringIO()
|
|
from rich.console import Console
|
|
|
|
test_console = Console(file=string_buffer, force_terminal=True, width=120)
|
|
|
|
context.streaming_error = None
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
try:
|
|
asyncio.run(
|
|
_tell_streaming(
|
|
mock_project, "Test description", None, mock_plan_service
|
|
)
|
|
)
|
|
except Exception as e:
|
|
context.streaming_error = e
|
|
|
|
context.streaming_output = string_buffer.getvalue()
|
|
|
|
|
|
# streaming then assertions
|
|
|
|
|
|
@then("_tell_streaming should complete without error")
|
|
def step_streaming_no_error(context):
|
|
assert context.streaming_error is None, (
|
|
f"Unexpected error: {context.streaming_error}"
|
|
)
|
|
|
|
|
|
@then("the streaming output should include completion panel")
|
|
def step_streaming_completion_panel(context):
|
|
out = context.streaming_output
|
|
assert (
|
|
"Plan Ready" in out
|
|
or "Plan created and built" in out
|
|
or "generated successfully" in out.lower()
|
|
)
|
|
|
|
|
|
@then("_tell_streaming should raise the streaming error")
|
|
def step_streaming_raised_error(context):
|
|
assert context.streaming_error is not None
|
|
assert isinstance(context.streaming_error, RuntimeError)
|
|
assert "Provider connection lost" in str(context.streaming_error)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _print_lifecycle_plan scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call _print_lifecycle_plan with a LifecyclePlan instance")
|
|
def step_call_print_lifecycle_plan_real(context):
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
plan = _make_mock_lifecycle_plan(
|
|
description="A real lifecycle plan for testing display output",
|
|
)
|
|
|
|
string_buffer = StringIO()
|
|
from rich.console import Console
|
|
|
|
test_console = Console(file=string_buffer, force_terminal=True, width=120)
|
|
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
_print_lifecycle_plan(plan, title="Test Plan")
|
|
|
|
context.print_output = string_buffer.getvalue()
|
|
|
|
|
|
@when("I call _print_lifecycle_plan with a non-LifecyclePlan object")
|
|
def step_call_print_lifecycle_plan_fallback(context):
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
fake_plan = {"id": "fake", "name": "not-a-plan"}
|
|
|
|
string_buffer = StringIO()
|
|
from rich.console import Console
|
|
|
|
test_console = Console(file=string_buffer, force_terminal=True, width=120)
|
|
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
_print_lifecycle_plan(fake_plan, title="Fallback Plan")
|
|
|
|
context.print_output = string_buffer.getvalue()
|
|
|
|
|
|
@when("I call _print_lifecycle_plan with a LifecyclePlan that has an error")
|
|
def step_call_print_lifecycle_plan_with_error(context):
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
plan = _make_mock_lifecycle_plan(
|
|
phase="strategize",
|
|
state="errored",
|
|
error_message="Strategy generation failed: model timeout",
|
|
)
|
|
|
|
string_buffer = StringIO()
|
|
from rich.console import Console
|
|
|
|
test_console = Console(file=string_buffer, force_terminal=True, width=120)
|
|
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
_print_lifecycle_plan(plan, title="Error Plan")
|
|
|
|
context.print_output = string_buffer.getvalue()
|
|
|
|
|
|
# _print_lifecycle_plan then assertions
|
|
|
|
|
|
@then("the lifecycle plan panel should be rendered with plan details")
|
|
def step_lifecycle_plan_panel_rendered(context):
|
|
out = context.print_output
|
|
assert "Test Plan" in out or "test-plan" in out
|
|
assert "ID" in out or "01J" in out
|
|
|
|
|
|
@then("the fallback plan panel should be rendered")
|
|
def step_fallback_plan_panel_rendered(context):
|
|
out = context.print_output
|
|
assert "Fallback Plan" in out or "Plan:" in out
|
|
|
|
|
|
@then("the lifecycle plan panel should include the error message")
|
|
def step_lifecycle_plan_error_shown(context):
|
|
out = context.print_output
|
|
assert "Error" in out
|
|
assert "model timeout" in out or "Strategy generation failed" in out
|