forked from HAL9000/cleveragents-core
freemo
4f2aa4189c
Click/Typer CliRunner.Result.stderr is a property that raises ValueError when stderr was not separately captured (mix_stderr=True is the default). Wrap all result.stderr accesses in try/except to handle this gracefully.
371 lines
12 KiB
Python
371 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from contextlib import ExitStack, contextmanager
|
|
from typing import Any
|
|
from unittest.mock import 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.auto_debug import (
|
|
_get_current_project,
|
|
auto_debug_command,
|
|
)
|
|
from cleveragents.cli.commands.auto_debug import (
|
|
app as auto_debug_app,
|
|
)
|
|
from cleveragents.core.exceptions import CleverAgentsError, PlanError
|
|
from cleveragents.domain.models.core.project_legacy import Project
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
class _FakeLive:
|
|
"""Minimal Live replacement that writes updates to the console."""
|
|
|
|
def __init__(self, *_args, console=None, **_kwargs):
|
|
self._console = console
|
|
|
|
def __enter__(self) -> _FakeLive:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
|
return False
|
|
|
|
def update(self, renderable) -> None:
|
|
if self._console is not None:
|
|
self._console.print(renderable)
|
|
|
|
|
|
@contextmanager
|
|
def _patched_container(context, project: Project | None):
|
|
with patch(
|
|
"cleveragents.application.container.get_container"
|
|
) as mock_get_container:
|
|
container = MagicMock()
|
|
mock_get_container.return_value = container
|
|
container.plan_service.return_value = context.plan_service_mock
|
|
container.project_service.return_value.get_current_project.return_value = (
|
|
project
|
|
)
|
|
yield container
|
|
|
|
|
|
def _capture_output(context: Any) -> str:
|
|
result = getattr(context, "result", None)
|
|
if isinstance(result, dict):
|
|
return result.get("output", "")
|
|
if result is not None:
|
|
stdout = getattr(result, "stdout", "") or ""
|
|
try:
|
|
stderr = result.stderr or ""
|
|
except (ValueError, AttributeError):
|
|
stderr = ""
|
|
output_attr = getattr(result, "output", "") or ""
|
|
combined = stdout + stderr
|
|
if combined:
|
|
return combined
|
|
if output_attr:
|
|
return output_attr
|
|
if hasattr(context, "output"):
|
|
return context.output
|
|
if hasattr(context, "command_output"):
|
|
return context.command_output
|
|
raise AssertionError("No CLI output captured in context")
|
|
|
|
|
|
def _capture_exit_code(context: Any) -> int:
|
|
result = getattr(context, "result", None)
|
|
if isinstance(result, dict):
|
|
return result.get("exit_code", 0)
|
|
if hasattr(result, "exit_code"):
|
|
return result.exit_code
|
|
if hasattr(context, "exit_code"):
|
|
return context.exit_code
|
|
raise AssertionError("No exit code recorded in context")
|
|
|
|
|
|
def _get_recorded_exception(context: Any) -> Exception | None:
|
|
return getattr(context, "exception", None) or getattr(
|
|
context, "call_exception", None
|
|
)
|
|
|
|
|
|
def _run_auto_debug_cli(context, extra_args: Iterable[str] | None = None) -> None:
|
|
project = context.project
|
|
assert project is not None, "Project must be initialized before running the CLI"
|
|
args: list[str] = []
|
|
if extra_args:
|
|
args.extend(extra_args)
|
|
|
|
with _patched_container(context, project), ExitStack() as stack:
|
|
stack.enter_context(
|
|
patch("cleveragents.cli.commands.auto_debug.Live", _FakeLive)
|
|
)
|
|
if getattr(context, "fail_fix_generation", False):
|
|
stack.enter_context(_patch_time_sleep_failure())
|
|
context.result = runner.invoke(auto_debug_app, args)
|
|
context.command_output = _capture_output(context)
|
|
|
|
|
|
@contextmanager
|
|
def _patch_time_sleep_failure():
|
|
calls = {"count": 0}
|
|
|
|
def _sleep_override(seconds):
|
|
if calls["count"] == 0:
|
|
calls["count"] += 1
|
|
raise RuntimeError("Could not generate fix")
|
|
|
|
with patch("time.sleep", side_effect=_sleep_override):
|
|
yield
|
|
|
|
|
|
@given("I have a clean auto_debug test environment")
|
|
def step_clean_environment(context):
|
|
context.project = None
|
|
context.result = None
|
|
context.exception = None
|
|
context.command_output = ""
|
|
context.fail_fix_generation = False
|
|
context.plan_service_mock = MagicMock(spec=PlanService)
|
|
context.plan_service_mock.auto_debug_build.return_value = (True, [], None)
|
|
context.plan_service_mock.build_plan.return_value = []
|
|
context.plan_service_mock.build_plan.side_effect = None
|
|
|
|
|
|
@given("I have an initialized project for auto_debug")
|
|
def step_initialized_project(context):
|
|
context.project = Project(
|
|
id=1,
|
|
name="Test Project",
|
|
description="A test project",
|
|
path="/tmp",
|
|
)
|
|
|
|
|
|
@given("I have no project initialized")
|
|
def step_no_project(context):
|
|
context.project = None
|
|
|
|
|
|
@given("the plan service auto_debug_build will succeed")
|
|
def step_autodebug_success(context):
|
|
context.plan_service_mock.auto_debug_build.return_value = (True, [], None)
|
|
|
|
|
|
@given("the plan service auto_debug_build will fail")
|
|
def step_autodebug_failure(context):
|
|
context.plan_service_mock.auto_debug_build.return_value = (
|
|
False,
|
|
[],
|
|
"Simulated auto-debug failure",
|
|
)
|
|
|
|
|
|
@when("I call the auto_debug_command programmatic interface")
|
|
def step_call_autodebug_command(context):
|
|
with _patched_container(context, context.project):
|
|
context.result = auto_debug_command()
|
|
|
|
|
|
@when("I call the auto_debug_command with max_attempts {max_attempts:d}")
|
|
def step_call_autodebug_command_with_limit(context, max_attempts):
|
|
with _patched_container(context, context.project):
|
|
context.result = auto_debug_command(max_attempts=max_attempts)
|
|
|
|
|
|
@when("I call the auto_debug_command programmatic interface expecting error")
|
|
def step_call_autodebug_command_error(context):
|
|
try:
|
|
with _patched_container(context, context.project):
|
|
auto_debug_command()
|
|
except Exception as exc:
|
|
context.exception = exc
|
|
|
|
|
|
@then("the auto_debug_command should return success True")
|
|
def step_assert_autodebug_success(context):
|
|
assert context.result[0] is True
|
|
|
|
|
|
@then("the auto_debug_command should return success False")
|
|
def step_assert_autodebug_failure(context):
|
|
assert context.result[0] is False
|
|
|
|
|
|
@then("the attempts made should be {count:d}")
|
|
def step_assert_attempts(context, count):
|
|
assert context.result[1] == count
|
|
|
|
|
|
@then('a CleverAgentsError should be raised with message "{message}"')
|
|
def step_assert_cleveragents_error(context, message):
|
|
exception = _get_recorded_exception(context)
|
|
assert exception is not None, (
|
|
"Expected CleverAgentsError but no exception was recorded"
|
|
)
|
|
assert isinstance(exception, CleverAgentsError)
|
|
assert message in str(exception)
|
|
|
|
|
|
@when("I call _get_current_project")
|
|
def step_call_get_current_project(context):
|
|
with _patched_container(context, context.project):
|
|
context.result = _get_current_project()
|
|
|
|
|
|
@when("I call _get_current_project expecting abort")
|
|
def step_call_get_current_project_abort(context):
|
|
try:
|
|
with _patched_container(context, None):
|
|
_get_current_project()
|
|
except typer.Abort as exc:
|
|
context.exception = exc
|
|
|
|
|
|
@then("the project should be returned successfully")
|
|
def step_assert_project_returned(context):
|
|
assert context.result == context.project
|
|
|
|
|
|
@then("typer.Abort should be raised")
|
|
def step_assert_abort(context):
|
|
assert isinstance(_get_recorded_exception(context), typer.Abort)
|
|
|
|
|
|
@given("the build will succeed with {changes:d} changes")
|
|
def step_build_success(context, changes):
|
|
context.plan_service_mock.build_plan.side_effect = None
|
|
context.plan_service_mock.build_plan.return_value = [
|
|
object() for _ in range(changes)
|
|
]
|
|
|
|
|
|
@given("the build will fail then succeed after fix")
|
|
def step_build_fail_then_succeed(context):
|
|
context.plan_service_mock.build_plan.side_effect = [
|
|
RuntimeError("Build failed"),
|
|
[object()],
|
|
]
|
|
|
|
|
|
@given('the build will always fail with error "{error_message}"')
|
|
def step_build_always_fail(context, error_message):
|
|
context.plan_service_mock.build_plan.side_effect = RuntimeError(error_message)
|
|
|
|
|
|
@given("the build will always fail with empty error message")
|
|
def step_build_fail_empty_error(context):
|
|
context.plan_service_mock.build_plan.side_effect = RuntimeError("")
|
|
|
|
|
|
@given("the plan service will raise a PlanError with details")
|
|
def step_planerror_details(context):
|
|
context.plan_service_mock.build_plan.side_effect = PlanError(
|
|
message="Plan Error",
|
|
details={"phase": "build"},
|
|
)
|
|
|
|
|
|
@given("the plan service will raise a PlanError without details")
|
|
def step_planerror_no_details(context):
|
|
context.plan_service_mock.build_plan.side_effect = PlanError(
|
|
message="Plan Error",
|
|
details=None,
|
|
)
|
|
|
|
|
|
@given("the plan service will raise a CleverAgentsError")
|
|
def step_plan_cleveragents_error(context):
|
|
context.plan_service_mock.build_plan.side_effect = CleverAgentsError(
|
|
"Generic Error"
|
|
)
|
|
|
|
|
|
@given("the build will fail with fix generation error")
|
|
def step_fix_generation_error(context):
|
|
context.fail_fix_generation = True
|
|
context.plan_service_mock.build_plan.side_effect = [
|
|
RuntimeError("Build failed"),
|
|
[object()],
|
|
]
|
|
|
|
|
|
@when("I run the auto_debug run command")
|
|
def step_run_cli(context):
|
|
_run_auto_debug_cli(context)
|
|
|
|
|
|
@when("I run the auto_debug run command with max_attempts {max_attempts:d}")
|
|
def step_run_cli_with_limit(context, max_attempts):
|
|
_run_auto_debug_cli(context, ["--max-attempts", str(max_attempts)])
|
|
|
|
|
|
@then("the command should exit with code {code:d}")
|
|
def step_assert_exit_code(context, code):
|
|
actual = _capture_exit_code(context)
|
|
if actual != code:
|
|
output = _capture_output(context)
|
|
assert actual == code, (
|
|
f"Expected exit code {code}, got {actual}. CLI output:\n{output}"
|
|
)
|
|
|
|
|
|
@then("the command should be aborted")
|
|
def step_assert_command_aborted(context):
|
|
step_assert_exit_code(context, 1)
|
|
|
|
|
|
@then('the output should contain "{text}"')
|
|
def step_assert_output_contains(context, text):
|
|
import re
|
|
|
|
output = _capture_output(context)
|
|
# Strip ANSI escape codes and normalize whitespace to handle
|
|
# Rich formatting and text wrapping in narrow CI terminals
|
|
raw = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", output)
|
|
normalized = " ".join(raw.split())
|
|
assert text in normalized, f"Expected to find '{text}' in CLI output: {output}"
|
|
|
|
|
|
@given(
|
|
"the build will fail {num_failures:d} times then succeed with {changes:d} changes"
|
|
)
|
|
def step_build_fail_then_succeed_with_changes(context, num_failures, changes):
|
|
"""Configure build to fail num_failures times then succeed."""
|
|
failures = [
|
|
RuntimeError(f"Build failed attempt {i + 1}") for i in range(num_failures)
|
|
]
|
|
success_result = [object() for _ in range(changes)]
|
|
context.plan_service_mock.build_plan.side_effect = [*failures, success_result]
|
|
|
|
|
|
@given("the build will raise PlanError after one attempt")
|
|
def step_build_raises_planerror_after_attempt(context):
|
|
"""Configure build to raise PlanError after one failed attempt."""
|
|
context.plan_service_mock.build_plan.side_effect = [
|
|
RuntimeError("Initial failure"),
|
|
PlanError(message="Plan Error during retry", details={"phase": "build"}),
|
|
]
|
|
|
|
|
|
@given("the build will raise CleverAgentsError after one attempt")
|
|
def step_build_raises_cleveragentserror_after_attempt(context):
|
|
"""Configure build to raise CleverAgentsError after one failed attempt."""
|
|
context.plan_service_mock.build_plan.side_effect = [
|
|
RuntimeError("Initial failure"),
|
|
CleverAgentsError("Critical error during retry"),
|
|
]
|
|
|
|
|
|
@given("the plan service will raise a PlanError with multiple details")
|
|
def step_planerror_multiple_details(context):
|
|
context.plan_service_mock.build_plan.side_effect = PlanError(
|
|
message="Plan Error",
|
|
details={"phase": "build", "step": "validation", "file": "test.py"},
|
|
)
|