brent.edwards
c5004ae918
CI / lint (pull_request) Failing after 16s
CI / typecheck (pull_request) Successful in 27s
CI / coverage (pull_request) Has been skipped
CI / security (pull_request) Successful in 19s
CI / quality (pull_request) Failing after 15s
CI / behave (3.11) (pull_request) Failing after 11s
CI / behave (3.12) (pull_request) Failing after 10s
CI / behave (3.13) (pull_request) Failing after 14s
CI / docker (pull_request) Has been skipped
CI / helm (pull_request) Has been skipped
CI / build (pull_request) Failing after 13s
1539 lines
56 KiB
Python
1539 lines
56 KiB
Python
"""Step definitions for plan commands coverage tests."""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import typer
|
|
from behave import given, then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
from cleveragents.core.exceptions import (
|
|
CleverAgentsError,
|
|
PlanError,
|
|
ValidationError,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MockPlan:
|
|
"""Mock plan object for testing."""
|
|
|
|
name: str
|
|
prompt: str | None
|
|
status: str
|
|
created_at: datetime
|
|
current: bool = False
|
|
|
|
|
|
@dataclass
|
|
class MockChange:
|
|
"""Mock change object for testing."""
|
|
|
|
file_path: str
|
|
operation: str
|
|
content: str | None = None
|
|
|
|
|
|
def _mock_plan_container(project):
|
|
"""Create a mock container with plan and project services."""
|
|
|
|
mock_container = MagicMock()
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_project_service = MagicMock()
|
|
mock_project_service.get_current_project.return_value = project
|
|
mock_container.plan_service.return_value = mock_plan_service
|
|
mock_container.project_service.return_value = mock_project_service
|
|
return mock_container, mock_plan_service, mock_project_service
|
|
|
|
|
|
@given("I have a plan test project initialized")
|
|
def step_plan_initialized_project(context):
|
|
"""Create an initialized project for plan tests."""
|
|
project_dir = Path(context.temp_dir) / ".cleveragents"
|
|
project_dir.mkdir(exist_ok=True)
|
|
project_file = project_dir / "project.json"
|
|
project_file.write_text('{"name": "test_project", "version": "1.0.0"}')
|
|
|
|
|
|
@given("I have an initialized project with current plan")
|
|
def step_initialized_project_with_plan(context):
|
|
"""Create an initialized project with a current plan."""
|
|
step_plan_initialized_project(context)
|
|
context.current_plan = MockPlan(
|
|
name="current-plan",
|
|
prompt="Build a feature",
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
current=True,
|
|
)
|
|
|
|
|
|
@given("I have an initialized project with built plan")
|
|
def step_initialized_project_with_built_plan(context):
|
|
"""Create an initialized project with a built plan."""
|
|
step_initialized_project_with_plan(context)
|
|
context.current_plan.status = "built"
|
|
context.pending_changes = [
|
|
MockChange("file1.py", "create"),
|
|
MockChange("file2.py", "modify"),
|
|
MockChange("file3.py", "delete"),
|
|
]
|
|
|
|
|
|
@given("I have an initialized project with empty plan")
|
|
def step_initialized_project_with_empty_plan(context):
|
|
"""Create an initialized project with an empty plan."""
|
|
step_plan_initialized_project(context)
|
|
context.current_plan = MockPlan(
|
|
name="empty-plan",
|
|
prompt=None,
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
current=True,
|
|
)
|
|
|
|
|
|
@given("I have an initialized project with no pending changes")
|
|
def step_initialized_project_no_pending_changes(context):
|
|
"""Create an initialized project with no pending changes."""
|
|
step_initialized_project_with_plan(context)
|
|
context.pending_changes = []
|
|
|
|
|
|
@given("I have an initialized project with no plans")
|
|
def step_initialized_project_no_plans(context):
|
|
"""Create an initialized project with no plans."""
|
|
step_plan_initialized_project(context)
|
|
context.current_plan = None
|
|
|
|
|
|
@given("I have an initialized project with multiple plans")
|
|
def step_initialized_project_multiple_plans(context):
|
|
"""Create an initialized project with multiple plans."""
|
|
step_plan_initialized_project(context)
|
|
context.plans = [
|
|
MockPlan("plan-1", "First plan", "draft", datetime.now(), current=True),
|
|
MockPlan("plan-2", "Second plan", "built", datetime.now()),
|
|
MockPlan("other-plan", "Third plan", "applied", datetime.now()),
|
|
]
|
|
context.current_plan = context.plans[0]
|
|
|
|
|
|
@given("I have an initialized project with many pending changes")
|
|
def step_initialized_project_many_changes(context):
|
|
"""Create an initialized project with many pending changes."""
|
|
step_initialized_project_with_plan(context)
|
|
context.pending_changes = [
|
|
MockChange(f"file{i}.py", ["create", "modify", "delete"][i % 3])
|
|
for i in range(15)
|
|
]
|
|
|
|
|
|
@when('I execute plan tell with prompt "{prompt}" and name "{name}"')
|
|
def step_run_plan_tell_with_name(context, prompt, name):
|
|
"""Run plan tell command with prompt and name."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan = MockPlan(
|
|
name=name, prompt=prompt, status="draft", created_at=datetime.now()
|
|
)
|
|
mock_plan_service.create_plan.return_value = mock_plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["tell", prompt, "--name", name])
|
|
context.result = result
|
|
context.created_plan = mock_plan
|
|
|
|
|
|
@when('I execute plan tell with prompt "{prompt}"')
|
|
def step_run_plan_tell(context, prompt):
|
|
"""Run plan tell command with a prompt."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan = MockPlan(
|
|
name=f"plan_{prompt[:10].replace(' ', '_')}",
|
|
prompt=prompt,
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
)
|
|
mock_plan_service.create_plan.return_value = mock_plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["tell", prompt])
|
|
context.result = result
|
|
context.created_plan = mock_plan
|
|
|
|
|
|
@when("I execute plan tell with invalid prompt causing validation error")
|
|
def step_run_plan_tell_validation_error(context):
|
|
"""Run plan tell with validation error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.create_plan.side_effect = ValidationError("Invalid prompt")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["tell", ""])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan tell causing plan error")
|
|
def step_run_plan_tell_plan_error(context):
|
|
"""Run plan tell with plan error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.create_plan.side_effect = PlanError("Plan creation failed")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["tell", "test"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan tell causing general error")
|
|
def step_run_plan_tell_general_error(context):
|
|
"""Run plan tell with general error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.create_plan.side_effect = CleverAgentsError("General error")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["tell", "test"])
|
|
context.result = result
|
|
|
|
|
|
@when('I execute plan tell streaming with prompt "{prompt}"')
|
|
def step_run_plan_tell_streaming(context, prompt):
|
|
"""Run plan tell command in streaming mode."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
mock_project_service = MagicMock()
|
|
mock_project = MagicMock()
|
|
mock_project_service.get_current_project.return_value = mock_project
|
|
mock_container.return_value.project_service.return_value = mock_project_service
|
|
|
|
streaming_mock = AsyncMock()
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._tell_streaming",
|
|
new=streaming_mock,
|
|
):
|
|
result = runner.invoke(plan_app, ["tell", prompt, "--stream"])
|
|
|
|
context.result = result
|
|
context.stream_call_count = streaming_mock.await_count
|
|
context.stream_call_args = streaming_mock.await_args_list
|
|
context.stream_prompt = prompt
|
|
context.stream_plan_service = mock_plan_service
|
|
|
|
|
|
@when("I execute plan build")
|
|
def step_run_plan_build(context):
|
|
"""Run plan build command."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
changes = [
|
|
MockChange("api.py", "create"),
|
|
MockChange("models.py", "create"),
|
|
]
|
|
mock_plan_service.build_plan.return_value = changes
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["build"])
|
|
context.result = result
|
|
context.generated_changes = changes
|
|
|
|
|
|
@when("I execute plan build with verbose flag")
|
|
def step_run_plan_build_verbose(context):
|
|
"""Run plan build command with verbose flag."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
changes = [MockChange("api.py", "create")]
|
|
mock_plan_service.build_plan.return_value = changes
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["build", "--verbose"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan build generating no changes")
|
|
def step_run_plan_build_no_changes(context):
|
|
"""Run plan build generating no changes."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.build_plan.return_value = []
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["build"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan build causing plan error")
|
|
def step_run_plan_build_plan_error(context):
|
|
"""Run plan build with plan error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.build_plan.side_effect = PlanError("Build failed")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["build"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan build causing general error")
|
|
def step_run_plan_build_general_error(context):
|
|
"""Run plan build with general error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.build_plan.side_effect = CleverAgentsError("Build error")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["build"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan apply with confirmation")
|
|
def step_run_plan_apply_confirm(context):
|
|
"""Run plan apply with user confirmation."""
|
|
import os
|
|
from io import StringIO
|
|
|
|
from rich.console import Console
|
|
|
|
runner = CliRunner()
|
|
|
|
# Create a string buffer to capture console output
|
|
string_buffer = StringIO()
|
|
test_console = Console(
|
|
file=string_buffer, force_terminal=True, width=120, force_jupyter=False
|
|
)
|
|
|
|
# Force confirmation prompt even in testing mode
|
|
# This ensures the truncated changes list is shown regardless of CLEVERAGENTS_TESTING_USE_MOCK_AI
|
|
env = os.environ.copy()
|
|
env["CLEVERAGENTS_TEST_FORCE_CONFIRMATION"] = "true"
|
|
|
|
with (
|
|
patch("cleveragents.application.container.get_container") as mock_container,
|
|
patch("cleveragents.cli.commands.plan.console", test_console),
|
|
):
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.return_value = context.pending_changes
|
|
mock_plan_service.apply_changes.return_value = len(context.pending_changes)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["apply"], input="y\n", env=env)
|
|
|
|
# Store both outputs for testing
|
|
context.result = result
|
|
context.console_output = string_buffer.getvalue()
|
|
|
|
|
|
@when("I execute plan apply with auto-confirm")
|
|
def step_run_plan_apply_auto_confirm(context):
|
|
"""Run plan apply with auto-confirm flag."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.return_value = context.pending_changes
|
|
mock_plan_service.apply_changes.return_value = len(context.pending_changes)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["apply", "--yes"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan apply")
|
|
def step_run_plan_apply_no_changes(context):
|
|
"""Run plan apply with no changes."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.return_value = []
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["apply"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan apply and cancel confirmation")
|
|
def step_run_plan_apply_cancel(context):
|
|
"""Run plan apply and cancel confirmation."""
|
|
import os
|
|
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.return_value = context.pending_changes
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
# Force the confirmation prompt to appear even in test mode
|
|
old_value = os.environ.get("CLEVERAGENTS_TEST_FORCE_CONFIRMATION")
|
|
os.environ["CLEVERAGENTS_TEST_FORCE_CONFIRMATION"] = "true"
|
|
|
|
try:
|
|
result = runner.invoke(plan_app, ["apply"], input="n\n")
|
|
context.result = result
|
|
finally:
|
|
# Restore the old value or remove the variable
|
|
if old_value is None:
|
|
os.environ.pop("CLEVERAGENTS_TEST_FORCE_CONFIRMATION", None)
|
|
else:
|
|
os.environ["CLEVERAGENTS_TEST_FORCE_CONFIRMATION"] = old_value
|
|
|
|
|
|
@when("I execute plan apply causing plan error")
|
|
def step_run_plan_apply_plan_error(context):
|
|
"""Run plan apply with plan error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.side_effect = PlanError("Apply failed")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["apply"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan apply causing general error")
|
|
def step_run_plan_apply_general_error(context):
|
|
"""Run plan apply with general error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_pending_changes.side_effect = CleverAgentsError(
|
|
"Apply error"
|
|
)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["apply"])
|
|
context.result = result
|
|
|
|
|
|
@when('I run plan new with name "{name}"')
|
|
def step_run_plan_new(context, name):
|
|
"""Run plan new command."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan = MockPlan(
|
|
name=name, prompt=None, status="draft", created_at=datetime.now()
|
|
)
|
|
mock_plan_service.create_empty_plan.return_value = mock_plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["new", name])
|
|
context.result = result
|
|
context.created_plan = mock_plan
|
|
|
|
|
|
@when('I execute plan new with name "{name}"')
|
|
def step_run_plan_new_with_name(context, name):
|
|
"""Run plan new command with a name."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan = MockPlan(
|
|
name=name, prompt=None, status="draft", created_at=datetime.now()
|
|
)
|
|
mock_plan_service.new_plan.return_value = mock_plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
# Also mock project service for _get_current_project
|
|
mock_project_service = MagicMock()
|
|
mock_project = MagicMock()
|
|
mock_project_service.get_current_project.return_value = mock_project
|
|
mock_container.return_value.project_service.return_value = mock_project_service
|
|
|
|
result = runner.invoke(plan_app, ["new", name])
|
|
context.result = result
|
|
context.created_plan = mock_plan
|
|
|
|
|
|
@when("I execute plan new without an initialized project")
|
|
def step_run_plan_new_without_project(context):
|
|
"""Run plan new when no project exists."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
mock_project_service = MagicMock()
|
|
mock_project_service.get_current_project.return_value = None
|
|
mock_container.return_value.project_service.return_value = mock_project_service
|
|
|
|
result = runner.invoke(plan_app, ["new", "missing-project"])
|
|
context.result = result
|
|
context.new_plan_service_mock = mock_plan_service
|
|
|
|
|
|
@when("I execute plan new with invalid name")
|
|
def step_run_plan_new_invalid(context):
|
|
"""Run plan new with invalid name."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.new_plan.side_effect = ValidationError("Invalid name")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
# Also mock project service
|
|
mock_project_service = MagicMock()
|
|
mock_project = MagicMock()
|
|
mock_project_service.get_current_project.return_value = mock_project
|
|
mock_container.return_value.project_service.return_value = mock_project_service
|
|
|
|
result = runner.invoke(plan_app, ["new", ""])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan new causing general error")
|
|
def step_run_plan_new_error(context):
|
|
"""Run plan new with general error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.new_plan.side_effect = CleverAgentsError("New plan error")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
# Also mock project service
|
|
mock_project_service = MagicMock()
|
|
mock_project = MagicMock()
|
|
mock_project_service.get_current_project.return_value = mock_project
|
|
mock_container.return_value.project_service.return_value = mock_project_service
|
|
|
|
result = runner.invoke(plan_app, ["new", "test"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan current")
|
|
def step_run_plan_current(context):
|
|
"""Run plan current command."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_current_plan.return_value = context.current_plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["current"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan current causing error")
|
|
def step_run_plan_current_error(context):
|
|
"""Run plan current with error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.get_current_plan.side_effect = CleverAgentsError(
|
|
"Current error"
|
|
)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["current"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan list")
|
|
def step_run_plan_list(context):
|
|
"""Run plan list command."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
plans = getattr(context, "plans", [])
|
|
mock_plan_service.list_plans.return_value = plans
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan list causing error")
|
|
def step_run_plan_list_error(context):
|
|
"""Run plan list with error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.list_plans.side_effect = CleverAgentsError("List error")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["list"])
|
|
context.result = result
|
|
|
|
|
|
@when('I run plan cd to "{name}"')
|
|
def step_run_plan_cd(context, name):
|
|
"""Run plan cd command."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
# Find the plan by name
|
|
plan = next((p for p in context.plans if p.name == name), None)
|
|
if plan:
|
|
mock_plan_service.switch_to_plan.return_value = plan
|
|
else:
|
|
mock_plan_service.switch_to_plan.side_effect = ValidationError(
|
|
"Plan not found"
|
|
)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["cd", name])
|
|
context.result = result
|
|
|
|
|
|
@when('I execute plan cd to "{name}"')
|
|
def step_execute_plan_cd_to_name(context, name):
|
|
"""Execute plan cd to a specific plan."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
# Find the plan by name
|
|
plan = next((p for p in getattr(context, "plans", []) if p.name == name), None)
|
|
if plan:
|
|
mock_plan_service.switch_to_plan.return_value = plan
|
|
else:
|
|
# Create a mock plan for the scenario
|
|
plan = MockPlan(
|
|
name=name, prompt="test", status="draft", created_at=datetime.now()
|
|
)
|
|
mock_plan_service.switch_to_plan.return_value = plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["cd", name])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan cd to non-existent plan")
|
|
def step_run_plan_cd_not_found(context):
|
|
"""Run plan cd to non-existent plan."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.switch_to_plan.side_effect = ValidationError("Plan not found")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["cd", "non-existent"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan cd causing error")
|
|
def step_run_plan_cd_error(context):
|
|
"""Run plan cd with error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.switch_to_plan.side_effect = CleverAgentsError("CD error")
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["cd", "test"])
|
|
context.result = result
|
|
|
|
|
|
@when('I execute plan continue with prompt "{prompt}"')
|
|
def step_execute_plan_continue_with_prompt(context, prompt):
|
|
"""Execute plan continue with prompt."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.continue_plan.return_value = None
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["continue", prompt])
|
|
context.result = result
|
|
|
|
|
|
@when('I run plan continue with prompt "{prompt}"')
|
|
def step_run_plan_continue_with_prompt(context, prompt):
|
|
"""Run plan continue with prompt."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.continue_plan.return_value = None
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["continue", prompt])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan continue without prompt")
|
|
def step_run_plan_continue_no_prompt(context):
|
|
"""Run plan continue without prompt."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
plan = getattr(context, "current_plan", None)
|
|
mock_plan_service.get_current_plan.return_value = plan
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["continue"])
|
|
context.result = result
|
|
|
|
|
|
@when("I execute plan continue causing error")
|
|
def step_run_plan_continue_error(context):
|
|
"""Run plan continue with error."""
|
|
runner = CliRunner()
|
|
with patch("cleveragents.application.container.get_container") as mock_container:
|
|
mock_plan_service = MagicMock(spec=PlanService)
|
|
mock_plan_service.continue_plan.side_effect = CleverAgentsError(
|
|
"Continue error"
|
|
)
|
|
mock_container.return_value.plan_service.return_value = mock_plan_service
|
|
|
|
result = runner.invoke(plan_app, ["continue", "test"])
|
|
context.result = result
|
|
|
|
|
|
@when("I call plan tell_command without a project")
|
|
def step_call_plan_tell_command_without_project(context):
|
|
"""Call tell_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import tell_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
tell_command("Automate improvements")
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan build_command without a project")
|
|
def step_call_plan_build_command_without_project(context):
|
|
"""Call build_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import build_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
build_command()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan apply_command successfully")
|
|
def step_call_plan_apply_command_success(context):
|
|
"""Call apply_command when pending changes exist."""
|
|
from cleveragents.cli.commands.plan import apply_command
|
|
|
|
project = object()
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.apply_changes.return_value = 3
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.call_exception = None
|
|
context.call_result = apply_command()
|
|
|
|
|
|
@when("I call plan apply_command without a project")
|
|
def step_call_plan_apply_command_without_project(context):
|
|
"""Call apply_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import apply_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
apply_command()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan new_command without a project")
|
|
def step_call_plan_new_command_without_project(context):
|
|
"""Call new_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import new_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
new_command("feature-x")
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan current_command successfully")
|
|
def step_call_plan_current_command_success(context):
|
|
"""Call current_command when a project has an active plan."""
|
|
from cleveragents.cli.commands.plan import current_command
|
|
|
|
project = object()
|
|
current_plan = MockPlan(
|
|
name="demo-plan",
|
|
prompt="Ship features",
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
current=True,
|
|
)
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.get_current_plan.return_value = current_plan
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.mock_current_plan = current_plan
|
|
context.call_exception = None
|
|
context.call_result = current_command()
|
|
|
|
|
|
@when("I call plan current_command without a project")
|
|
def step_call_plan_current_command_without_project(context):
|
|
"""Call current_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import current_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
current_command()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan list_command with no stored plans")
|
|
def step_call_plan_list_command_with_no_plans(context):
|
|
"""Call list_command when the plan service returns no plans."""
|
|
from cleveragents.cli.commands.plan import list_command
|
|
|
|
project = object()
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.list_plans.return_value = None
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.call_exception = None
|
|
context.call_result = list_command()
|
|
|
|
|
|
@when("I call plan list_command without a project")
|
|
def step_call_plan_list_command_without_project(context):
|
|
"""Call list_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import list_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
list_command()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan cd_command without a project")
|
|
def step_call_plan_cd_command_without_project(context):
|
|
"""Call cd_command when no project exists."""
|
|
from cleveragents.cli.commands.plan import cd_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
cd_command("feature-stream")
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when('I call plan cd_command successfully to "{name}"')
|
|
def step_call_plan_cd_command_success(context, name):
|
|
"""Call cd_command when a project exists."""
|
|
from cleveragents.cli.commands.plan import cd_command
|
|
|
|
project = object()
|
|
switched_plan = MockPlan(
|
|
name=name,
|
|
prompt="Switch plan",
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
current=True,
|
|
)
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.switch_to_plan.return_value = switched_plan
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.call_exception = None
|
|
context.call_result = cd_command(name)
|
|
|
|
|
|
@when('I call plan continue_command with prompt "{prompt}"')
|
|
def step_call_plan_continue_command_with_prompt(context, prompt):
|
|
"""Call continue_command with explicit prompt."""
|
|
from cleveragents.cli.commands.plan import continue_command
|
|
|
|
project = object()
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.continue_plan.return_value = None
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.call_exception = None
|
|
context.call_result = continue_command(prompt)
|
|
|
|
|
|
@when("I call plan continue_command without prompt and an active plan")
|
|
def step_call_plan_continue_command_with_active_plan(context):
|
|
"""Call continue_command without prompt when a current plan exists."""
|
|
from cleveragents.cli.commands.plan import continue_command
|
|
|
|
project = object()
|
|
active_plan = MockPlan(
|
|
name="active-plan",
|
|
prompt="Keep going",
|
|
status="draft",
|
|
created_at=datetime.now(),
|
|
current=True,
|
|
)
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.get_current_plan.return_value = active_plan
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.active_plan = active_plan
|
|
context.call_exception = None
|
|
context.call_result = continue_command()
|
|
|
|
|
|
@when("I call plan continue_command without prompt and no current plan")
|
|
def step_call_plan_continue_command_without_plan(context):
|
|
"""Call continue_command without prompt when no plan exists."""
|
|
from cleveragents.cli.commands.plan import continue_command
|
|
|
|
project = object()
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, _project_service = _mock_plan_container(
|
|
project=project
|
|
)
|
|
plan_service.get_current_plan.return_value = None
|
|
mock_get_container.return_value = container
|
|
|
|
context.project_mock = project
|
|
context.plan_service_mock = plan_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
continue_command()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call plan continue_command without an initialized project")
|
|
def step_call_plan_continue_command_without_project(context):
|
|
"""Call continue_command when no project is initialized."""
|
|
from cleveragents.cli.commands.plan import continue_command
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
continue_command("Stream coverage")
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
@when("I call the plan helper to fetch current project without initialization")
|
|
def step_call_plan_helper_without_project(context):
|
|
"""Call _get_current_project when no project exists."""
|
|
from cleveragents.cli.commands import plan as plan_module
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container, plan_service, project_service = _mock_plan_container(project=None)
|
|
mock_get_container.return_value = container
|
|
|
|
context.plan_service_mock = plan_service
|
|
context.project_service_mock = project_service
|
|
context.call_result = None
|
|
context.call_exception = None
|
|
try:
|
|
plan_module._get_current_project()
|
|
except Exception as exc:
|
|
context.call_exception = exc
|
|
|
|
|
|
# Then steps for assertions
|
|
|
|
|
|
@then("the programmatic plan apply_command should report {count:d} applied changes")
|
|
def step_assert_programmatic_apply_changes(context, count):
|
|
"""Assert apply_command returned the expected change count."""
|
|
assert context.call_exception is None
|
|
assert context.call_result == count
|
|
context.plan_service_mock.apply_changes.assert_called_once()
|
|
kwargs = context.plan_service_mock.apply_changes.call_args.kwargs
|
|
assert kwargs["project"] is context.project_mock
|
|
|
|
|
|
@then("the programmatic current_command should return the mock plan")
|
|
def step_assert_programmatic_current_command(context):
|
|
"""Assert current_command returned the mock plan."""
|
|
assert context.call_exception is None
|
|
assert context.call_result is context.mock_current_plan
|
|
|
|
|
|
@then("the programmatic list command should return an empty list")
|
|
def step_assert_programmatic_list_command_empty(context):
|
|
"""Assert list_command returned an empty list."""
|
|
assert context.call_exception is None
|
|
assert context.call_result == []
|
|
context.plan_service_mock.list_plans.assert_called_once()
|
|
|
|
|
|
@then('the programmatic cd command should switch to plan "{name}"')
|
|
def step_assert_programmatic_cd_command(context, name):
|
|
"""Assert cd_command switched to the requested plan."""
|
|
assert context.call_exception is None
|
|
context.plan_service_mock.switch_to_plan.assert_called_once()
|
|
kwargs = context.plan_service_mock.switch_to_plan.call_args.kwargs
|
|
assert kwargs["project"] is context.project_mock
|
|
assert kwargs["name"] == name
|
|
|
|
|
|
@then('the programmatic continue command should forward prompt "{prompt}"')
|
|
def step_assert_programmatic_continue_forward(context, prompt):
|
|
"""Assert continue_command forwarded the prompt to the plan service."""
|
|
assert context.call_exception is None
|
|
context.plan_service_mock.continue_plan.assert_called_once_with(
|
|
project=context.project_mock,
|
|
prompt=prompt,
|
|
)
|
|
|
|
|
|
@then("the programmatic continue command should keep the current plan active")
|
|
def step_assert_programmatic_continue_keeps_plan(context):
|
|
"""Assert continue_command validated the current plan without new instructions."""
|
|
assert context.call_exception is None
|
|
context.plan_service_mock.get_current_plan.assert_called_once_with(
|
|
project=context.project_mock
|
|
)
|
|
context.plan_service_mock.continue_plan.assert_not_called()
|
|
|
|
|
|
@then("a typer Abort should be raised for missing project")
|
|
def step_assert_typer_abort(context):
|
|
"""Assert _get_current_project raised a typer.Abort exception."""
|
|
assert isinstance(context.call_exception, typer.Abort)
|
|
|
|
|
|
@then("the plan tell should execute successfully")
|
|
def step_plan_tell_success(context):
|
|
"""Assert plan tell executed successfully."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code was {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
assert "Plan created" in context.result.output or "✓" in context.result.output
|
|
|
|
|
|
@then("the plan should be created with name and prompt")
|
|
def step_plan_created_with_details(context):
|
|
"""Assert plan was created with expected details."""
|
|
assert context.created_plan is not None
|
|
assert context.created_plan.prompt is not None
|
|
|
|
|
|
@then('the plan should be created with custom name "{name}"')
|
|
def step_plan_created_with_name(context, name):
|
|
"""Assert plan was created with custom name."""
|
|
assert context.created_plan.name == name
|
|
|
|
|
|
@then("the plan tell should abort with validation error")
|
|
def step_plan_tell_validation_abort(context):
|
|
"""Assert plan tell aborted with validation error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"Validation Error" in context.result.output
|
|
or "validation" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan tell should abort with plan error")
|
|
def step_plan_tell_plan_abort(context):
|
|
"""Assert plan tell aborted with plan error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"Plan Error" in context.result.output or "plan" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan tell should abort with general error")
|
|
def step_plan_tell_general_abort(context):
|
|
"""Assert plan tell aborted with general error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan tell streaming path should execute successfully")
|
|
def step_plan_tell_streaming_success(context):
|
|
"""Assert plan tell streaming path executed via asyncio."""
|
|
assert context.result.exit_code == 0, (
|
|
f"Exit code was {context.result.exit_code}, output: {context.result.output}"
|
|
)
|
|
assert getattr(context, "stream_call_count", 0) == 1, (
|
|
"Streaming coroutine was not awaited"
|
|
)
|
|
assert context.stream_call_args, "No streaming call arguments recorded"
|
|
first_call = context.stream_call_args[0]
|
|
assert first_call.args[1] == context.stream_prompt
|
|
assert first_call.args[3] is context.stream_plan_service
|
|
|
|
|
|
@then("the plan build should execute successfully")
|
|
def step_plan_build_success(context):
|
|
"""Assert plan build executed successfully."""
|
|
assert context.result.exit_code == 0, f"Exit code was {context.result.exit_code}"
|
|
|
|
|
|
@then("plan changes should be generated")
|
|
def step_changes_generated(context):
|
|
"""Assert changes were generated."""
|
|
assert (
|
|
"change" in context.result.output.lower()
|
|
or "generated" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan build should execute successfully with verbose output")
|
|
def step_plan_build_verbose_success(context):
|
|
"""Assert plan build executed with verbose output."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("the build should complete with no changes message")
|
|
def step_build_no_changes_message(context):
|
|
"""Assert build completed with no changes."""
|
|
assert context.result.exit_code == 0
|
|
assert (
|
|
"No changes" in context.result.output
|
|
or "no changes" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan build should abort with plan error")
|
|
def step_plan_build_plan_abort(context):
|
|
"""Assert plan build aborted with plan error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"Build Error" in context.result.output
|
|
or "error" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan build should abort with general error")
|
|
def step_plan_build_general_abort(context):
|
|
"""Assert plan build aborted with general error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan apply should execute successfully")
|
|
def step_plan_apply_success(context):
|
|
"""Assert plan apply executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("plan changes should be applied")
|
|
def step_changes_applied(context):
|
|
"""Assert changes were applied."""
|
|
output = context.result.output
|
|
console_output = getattr(context, "console_output", "")
|
|
combined_output = output + console_output
|
|
assert "applied" in combined_output.lower() or "✓" in combined_output
|
|
|
|
|
|
@then("the plan apply should execute successfully without prompting")
|
|
def step_plan_apply_auto_success(context):
|
|
"""Assert plan apply executed without prompting."""
|
|
assert context.result.exit_code == 0
|
|
assert "Apply these changes?" not in context.result.output
|
|
|
|
|
|
@then("the apply should exit with no changes message")
|
|
def step_apply_no_changes_exit(context):
|
|
"""Assert apply exited with no changes."""
|
|
assert context.result.exit_code == 0
|
|
assert (
|
|
"No changes" in context.result.output
|
|
or "no changes" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the apply should exit with cancelled message")
|
|
def step_apply_cancelled_exit(context):
|
|
"""Assert apply was cancelled."""
|
|
assert context.result.exit_code == 0
|
|
assert (
|
|
"Cancelled" in context.result.output
|
|
or "cancelled" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan apply should abort with plan error")
|
|
def step_plan_apply_plan_abort(context):
|
|
"""Assert plan apply aborted with plan error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"Apply Error" in context.result.output
|
|
or "error" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan apply should abort with general error")
|
|
def step_plan_apply_general_abort(context):
|
|
"""Assert plan apply aborted with general error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan new should execute successfully")
|
|
def step_plan_new_success(context):
|
|
"""Assert plan new executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then('empty plan "{name}" should be created')
|
|
def step_empty_plan_created(context, name):
|
|
"""Assert empty plan was created."""
|
|
assert context.created_plan.name == name
|
|
assert context.created_plan.prompt is None
|
|
|
|
|
|
@then("the plan new should abort due to missing project")
|
|
def step_plan_new_missing_project_abort(context):
|
|
"""Assert plan new aborted when no project exists."""
|
|
assert context.result.exit_code != 0
|
|
assert "No project found" in context.result.output
|
|
if hasattr(context, "new_plan_service_mock"):
|
|
context.new_plan_service_mock.new_plan.assert_not_called()
|
|
|
|
|
|
@then("the plan new should abort with validation error")
|
|
def step_plan_new_validation_abort(context):
|
|
"""Assert plan new aborted with validation error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"Validation Error" in context.result.output
|
|
or "validation" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan new should abort with general error")
|
|
def step_plan_new_general_abort(context):
|
|
"""Assert plan new aborted with general error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan current should execute successfully")
|
|
def step_plan_current_success(context):
|
|
"""Assert plan current executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("current plan details should be displayed")
|
|
def step_current_plan_details_displayed(context):
|
|
"""Assert current plan details are displayed."""
|
|
assert "Current Plan" in context.result.output or (
|
|
hasattr(context, "current_plan")
|
|
and context.current_plan
|
|
and context.current_plan.name in context.result.output
|
|
)
|
|
|
|
|
|
@then("plan command current plan details should be displayed")
|
|
def step_current_plan_displayed(context):
|
|
"""Assert current plan details are displayed."""
|
|
assert (
|
|
"Current Plan" in context.result.output
|
|
or context.current_plan.name in context.result.output
|
|
)
|
|
|
|
|
|
@then("the current should exit with no plan message")
|
|
def step_current_no_plan_exit(context):
|
|
"""Assert current exited with no plan message."""
|
|
assert context.result.exit_code == 0
|
|
assert (
|
|
"No current plan" in context.result.output
|
|
or "no current plan" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan current should abort with error")
|
|
def step_plan_current_abort(context):
|
|
"""Assert plan current aborted with error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan list should execute successfully")
|
|
def step_plan_list_success(context):
|
|
"""Assert plan list executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("all plans should be displayed in table")
|
|
def step_all_plans_displayed_in_table(context):
|
|
"""Assert plans are displayed in table."""
|
|
if hasattr(context, "plans") and context.plans:
|
|
for plan in context.plans:
|
|
assert plan.name in context.result.output
|
|
|
|
|
|
@then("plan command all plans should be displayed in table")
|
|
def step_plans_displayed_table(context):
|
|
"""Assert plans are displayed in table."""
|
|
if hasattr(context, "plans") and context.plans:
|
|
for plan in context.plans:
|
|
assert plan.name in context.result.output
|
|
|
|
|
|
@then("the list should show no plans message")
|
|
def step_list_no_plans_message(context):
|
|
"""Assert list shows no plans message."""
|
|
assert (
|
|
"No plans" in context.result.output
|
|
or "no plans" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan list should abort with error")
|
|
def step_plan_list_abort(context):
|
|
"""Assert plan list aborted with error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan cd should execute successfully")
|
|
def step_plan_cd_success(context):
|
|
"""Assert plan cd executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then('current plan should be "{name}"')
|
|
def step_current_plan_is(context, name):
|
|
"""Assert current plan is the expected one."""
|
|
assert "Switched to plan" in context.result.output or name in context.result.output
|
|
|
|
|
|
@then("the plan cd should abort with validation error")
|
|
def step_plan_cd_validation_abort(context):
|
|
"""Assert plan cd aborted with validation error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"not found" in context.result.output.lower()
|
|
or "validation" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan cd should abort with error")
|
|
def step_plan_cd_abort(context):
|
|
"""Assert plan cd aborted with error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the plan continue should execute successfully")
|
|
def step_plan_continue_success(context):
|
|
"""Assert plan continue executed successfully."""
|
|
assert context.result.exit_code == 0
|
|
|
|
|
|
@then("instructions should be added to plan")
|
|
def step_instructions_added_to_plan(context):
|
|
"""Assert instructions were added to plan."""
|
|
assert (
|
|
"Added instructions" in context.result.output
|
|
or "added" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("plan command instructions should be added to plan")
|
|
def step_instructions_added(context):
|
|
"""Assert instructions were added to plan."""
|
|
assert (
|
|
"Added instructions" in context.result.output
|
|
or "added" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("plan command continue message should be displayed")
|
|
def step_continue_message_displayed(context):
|
|
"""Assert continue message is displayed."""
|
|
assert (
|
|
"Continuing" in context.result.output
|
|
or "continue" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("continue message should be displayed")
|
|
def step_continue_message_should_be_displayed(context):
|
|
"""Assert continue message is displayed."""
|
|
assert (
|
|
"Continuing" in context.result.output
|
|
or "continue" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the continue should abort with no plan error")
|
|
def step_continue_no_plan_abort(context):
|
|
"""Assert continue aborted with no plan error."""
|
|
assert context.result.exit_code != 0
|
|
assert (
|
|
"No current plan" in context.result.output
|
|
or "no current plan" in context.result.output.lower()
|
|
)
|
|
|
|
|
|
@then("the plan continue should abort with error")
|
|
def step_plan_continue_abort(context):
|
|
"""Assert plan continue aborted with error."""
|
|
assert context.result.exit_code != 0
|
|
assert "Error" in context.result.output
|
|
|
|
|
|
@then("the apply should show truncated changes list")
|
|
def step_apply_truncated_changes(context):
|
|
"""Assert apply shows truncated changes list."""
|
|
# Check both possible outputs (runner output and console output)
|
|
output = context.result.output
|
|
console_output = getattr(context, "console_output", "")
|
|
combined_output = output + console_output
|
|
|
|
# More flexible check for truncation - looking for the pattern of truncated output
|
|
has_truncation = (
|
|
("... and" in combined_output and "more" in combined_output)
|
|
or ("more changes" in combined_output and "..." in combined_output)
|
|
or (combined_output.count("file") >= 10 and "more" in combined_output)
|
|
)
|
|
|
|
if not has_truncation:
|
|
# If the assertion will fail, provide helpful debug info
|
|
print("DEBUG: Output does not show truncation.")
|
|
print(f"DEBUG: Runner output:\n{output}")
|
|
print(f"DEBUG: Console output:\n{console_output}")
|
|
print(f"DEBUG: File count in combined output: {combined_output.count('file')}")
|
|
print(f"DEBUG: Contains '...': {'...' in combined_output}")
|
|
print(f"DEBUG: Contains 'more': {'more' in combined_output}")
|
|
|
|
assert has_truncation, "Expected truncated list indicator in output"
|
|
|
|
|
|
@then("plan changes should be applied after confirmation")
|
|
def step_changes_applied_after_confirm(context):
|
|
"""Assert changes were applied after confirmation."""
|
|
output = context.result.output
|
|
console_output = getattr(context, "console_output", "")
|
|
combined_output = (output + console_output).lower()
|
|
assert "applied" in combined_output
|