forked from HAL9000/cleveragents-core
147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
"""Step definitions for uncovered plan command branches."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
from io import StringIO
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.cli.commands.plan import _tell_streaming
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
|
|
|
|
class _StreamingPlanService:
|
|
def __init__(self, events: list[dict[str, object]], error: Exception | None = None):
|
|
self._events = events
|
|
self._error = error
|
|
|
|
async def generate_plan_streaming(self, *args, **kwargs):
|
|
for event in self._events:
|
|
yield event
|
|
if self._error:
|
|
raise self._error
|
|
|
|
|
|
class _PlanTellContainer:
|
|
def __init__(self, plan_service, project_service, actor_service):
|
|
self._plan_service = plan_service
|
|
self._project_service = project_service
|
|
self._actor_service = actor_service
|
|
|
|
def plan_service(self):
|
|
return self._plan_service
|
|
|
|
def project_service(self):
|
|
return self._project_service
|
|
|
|
def actor_service(self):
|
|
return self._actor_service
|
|
|
|
|
|
def _strip_rich_markup(text: str) -> str:
|
|
return re.sub(r"\[[^\]]+\]", "", text)
|
|
|
|
|
|
def _run_streaming_helper(context, plan_service, description: str) -> None:
|
|
output = StringIO()
|
|
test_console = Console(file=output, force_terminal=False, width=120)
|
|
error_message = ""
|
|
success = False
|
|
try:
|
|
with patch("cleveragents.cli.commands.plan.console", test_console):
|
|
asyncio.run(
|
|
_tell_streaming(
|
|
context.streaming_project,
|
|
description,
|
|
None,
|
|
plan_service,
|
|
)
|
|
)
|
|
success = True
|
|
except Exception as exc:
|
|
error_message = str(exc)
|
|
finally:
|
|
rendered = _strip_rich_markup(output.getvalue())
|
|
context.command_output = rendered if success else f"{rendered}{error_message}"
|
|
context.command_success = success
|
|
context.command_error = error_message
|
|
|
|
|
|
@given("I have a stub streaming project")
|
|
def step_stub_streaming_project(context):
|
|
context.streaming_project = SimpleNamespace(name="streaming-project")
|
|
|
|
|
|
@when("I run the streaming plan helper with only an end event")
|
|
def step_run_streaming_end_event(context):
|
|
service = _StreamingPlanService(events=[{"__end__": True}])
|
|
_run_streaming_helper(context, service, "End-only streaming plan")
|
|
|
|
|
|
@when("I run the streaming plan helper with a pre-node exception")
|
|
def step_run_streaming_pre_node_exception(context):
|
|
service = _StreamingPlanService(events=[], error=RuntimeError("boom"))
|
|
_run_streaming_helper(context, service, "Exception streaming plan")
|
|
|
|
|
|
@then("the streaming helper should complete successfully")
|
|
def step_streaming_helper_success(context):
|
|
assert context.command_success, f"Streaming failed: {context.command_output}"
|
|
|
|
|
|
@then("the streaming output should mention plan creation success")
|
|
def step_streaming_output_mentions_success(context):
|
|
output = context.command_output
|
|
assert "Plan created and built" in output or "Plan generated successfully" in output
|
|
|
|
|
|
@then("the streaming helper should fail with an error")
|
|
def step_streaming_helper_failure(context):
|
|
assert not context.command_success
|
|
assert context.command_output or context.command_error
|
|
|
|
|
|
@then("the streaming output should show an error without node failure details")
|
|
def step_streaming_output_no_node_failure(context):
|
|
output = context.command_output.lower()
|
|
assert "error" in output
|
|
assert "failed" not in output
|
|
|
|
|
|
@when("I execute plan tell with testing mode disabled and no actor registry")
|
|
def step_plan_tell_testing_mode_disabled(context):
|
|
runner = CliRunner()
|
|
prompt = "Check branch coverage"
|
|
plan_service = MagicMock(spec=PlanService)
|
|
plan_service.create_plan.return_value = SimpleNamespace(
|
|
name="coverage-plan",
|
|
prompt=prompt,
|
|
)
|
|
project_service = MagicMock()
|
|
project_service.get_current_project.return_value = SimpleNamespace(name="project")
|
|
actor_service = MagicMock()
|
|
container = _PlanTellContainer(plan_service, project_service, actor_service)
|
|
env = os.environ.copy()
|
|
env["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "false"
|
|
|
|
with patch(
|
|
"cleveragents.application.container.get_container", return_value=container
|
|
):
|
|
result = runner.invoke(plan_app, ["tell", prompt], env=env)
|
|
|
|
context.result = result
|
|
context.actor_service = actor_service
|
|
|
|
|
|
@then("the mock actor should not be initialized")
|
|
def step_mock_actor_not_initialized(context):
|
|
context.actor_service.ensure_default_mock_actor.assert_not_called()
|