forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
415 lines
15 KiB
Python
415 lines
15 KiB
Python
"""Step definitions for plan_service_coverage_boost_r2.feature.
|
|
|
|
Targets uncovered lines in plan_service.py:
|
|
- Lines 851, 853: auto_debug_build exception path formats code context from
|
|
contexts with content (list comprehension filtering)
|
|
- Lines 922-923, 925-929, 931-932, 935: auto_debug_build successful fix branch
|
|
updates plan prompt and continues to retry
|
|
- Line 1036: apply_changes generic exception handler extracts operation.value
|
|
- Lines 1413-1416: _coerce_change_list defensive guard for invalid dict entry
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import warnings
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context as BehaveContext
|
|
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import PlanError
|
|
from cleveragents.domain.models.core import Change, OperationType, PlanStatus
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_settings() -> MagicMock:
|
|
"""Create a minimal Settings mock."""
|
|
settings = MagicMock(spec=Settings)
|
|
settings.is_langsmith_enabled = False
|
|
settings.langsmith_tags = []
|
|
settings.database_url = "sqlite://:memory:"
|
|
settings.provider_configuration_diagnostics.return_value = {}
|
|
settings.configured_provider_names.return_value = []
|
|
settings.provider_expected_env_vars.return_value = []
|
|
return settings
|
|
|
|
|
|
def _make_plan(
|
|
plan_id: int = 1,
|
|
project_id: int = 1,
|
|
prompt: str | None = "Build a feature",
|
|
) -> MagicMock:
|
|
"""Create a mock Plan with minimal attributes."""
|
|
plan = MagicMock()
|
|
plan.id = plan_id
|
|
plan.project_id = project_id
|
|
plan.prompt = prompt
|
|
plan.name = "test-plan"
|
|
plan.status = PlanStatus.PENDING
|
|
plan.updated_at = datetime.now()
|
|
return plan
|
|
|
|
|
|
def _make_project(project_id: int = 1, tmp_dir: str = "/tmp/test-proj") -> MagicMock:
|
|
"""Create a mock Project."""
|
|
project = MagicMock()
|
|
project.id = project_id
|
|
project.name = "test-project"
|
|
project.path = Path(tmp_dir)
|
|
return project
|
|
|
|
|
|
def _make_context_obj(plan_id: int, path: str, content: str | None) -> SimpleNamespace:
|
|
"""Create a lightweight context object mimicking the Context domain model."""
|
|
return SimpleNamespace(
|
|
id=1,
|
|
plan_id=plan_id,
|
|
path=path,
|
|
content=content,
|
|
)
|
|
|
|
|
|
def _make_uow_with_plan(plan: MagicMock) -> MagicMock:
|
|
"""Create a UoW mock whose transaction ctx returns the given plan."""
|
|
uow = MagicMock()
|
|
ctx = MagicMock()
|
|
ctx.plans.get_current_for_project.return_value = plan
|
|
ctx.plans.update.return_value = None
|
|
ctx.changes.get_for_plan.return_value = []
|
|
|
|
# debug_attempts.add returns an object with an id
|
|
added_attempt = MagicMock()
|
|
added_attempt.id = 100
|
|
ctx.debug_attempts.add.return_value = added_attempt
|
|
ctx.debug_attempts.get.return_value = None
|
|
ctx.debug_attempts.update.return_value = None
|
|
|
|
# contexts
|
|
ctx.contexts.get_for_plan.return_value = []
|
|
|
|
# Make the transaction context manager work
|
|
tx_cm = MagicMock()
|
|
tx_cm.__enter__ = MagicMock(return_value=ctx)
|
|
tx_cm.__exit__ = MagicMock(return_value=False)
|
|
uow.transaction.return_value = tx_cm
|
|
|
|
return uow, ctx
|
|
|
|
|
|
def _build_service(
|
|
uow: MagicMock,
|
|
llm: MagicMock | None = None,
|
|
) -> PlanService:
|
|
"""Build a PlanService suppressing the deprecation warning."""
|
|
settings = _make_settings()
|
|
actor_service = MagicMock()
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
svc = PlanService(
|
|
settings=settings,
|
|
unit_of_work=uow,
|
|
actor_service=actor_service,
|
|
llm=llm or MagicMock(),
|
|
)
|
|
return svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I have a plan service configured for r2 coverage testing")
|
|
def step_plan_service_r2(context: BehaveContext) -> None:
|
|
"""Set up default plan and service for each scenario."""
|
|
context.r2_plan = _make_plan()
|
|
context.r2_uow, context.r2_ctx = _make_uow_with_plan(context.r2_plan)
|
|
context.r2_service = _build_service(context.r2_uow)
|
|
context.r2_project = _make_project()
|
|
context.r2_error = None
|
|
context.r2_result = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: auto_debug_build contexts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the current plan has contexts with file content")
|
|
def step_plan_with_contexts(context: BehaveContext) -> None:
|
|
"""Populate the plan with contexts that have content (hits lines 851, 853)."""
|
|
contexts = [
|
|
_make_context_obj(1, "src/app.py", "print('hello')"),
|
|
_make_context_obj(1, "src/util.py", "def helper(): pass"),
|
|
# One context without content — should be filtered out by `if c.content`
|
|
_make_context_obj(1, "src/empty.py", None),
|
|
]
|
|
context.r2_ctx.contexts.get_for_plan.return_value = contexts
|
|
|
|
|
|
@given("the current plan has contexts where some lack content")
|
|
def step_plan_with_mixed_contexts(context: BehaveContext) -> None:
|
|
"""Populate with contexts where only some have content."""
|
|
contexts = [
|
|
_make_context_obj(1, "src/main.py", "import os"),
|
|
_make_context_obj(1, "src/no_content.py", None),
|
|
_make_context_obj(1, "src/also_empty.py", ""),
|
|
]
|
|
context.r2_ctx.contexts.get_for_plan.return_value = contexts
|
|
|
|
|
|
@given("the current plan has no prompt set")
|
|
def step_plan_no_prompt(context: BehaveContext) -> None:
|
|
"""Create a plan with no prompt so the prompt-append branch is skipped."""
|
|
context.r2_plan.prompt = None
|
|
# Re-wire contexts (same as with-content scenario)
|
|
contexts = [
|
|
_make_context_obj(1, "src/app.py", "print('hello')"),
|
|
]
|
|
context.r2_ctx.contexts.get_for_plan.return_value = contexts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: build_plan behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("build_plan raises an exception on every attempt")
|
|
def step_build_always_fails(context: BehaveContext) -> None:
|
|
"""Patch build_plan to always raise."""
|
|
context.r2_service.build_plan = MagicMock(
|
|
side_effect=RuntimeError("Simulated build failure")
|
|
)
|
|
|
|
|
|
@given("build_plan fails on the first attempt then succeeds")
|
|
def step_build_fail_then_succeed(context: BehaveContext) -> None:
|
|
"""Patch build_plan to fail once then succeed."""
|
|
call_count = {"n": 0}
|
|
|
|
def _build_plan(project: Any, callback: Any = None) -> list:
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise RuntimeError("First attempt build failure")
|
|
return [] # empty changes = success
|
|
|
|
context.r2_service.build_plan = MagicMock(side_effect=_build_plan)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: AutoDebugAgent behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the auto debug agent returns an unsuccessful fix")
|
|
def step_agent_no_fix(context: BehaveContext) -> None:
|
|
"""Configure the mock agent to return an unsuccessful result."""
|
|
context.r2_agent_result = {
|
|
"error_message": "Simulated build failure",
|
|
"code_context": "",
|
|
"attempted_fixes": [],
|
|
"current_fix": {},
|
|
"fix_validated": False,
|
|
"messages": [],
|
|
"context": {},
|
|
"result": {"success": False, "fix": {}},
|
|
"error": None,
|
|
"metadata": {},
|
|
}
|
|
|
|
|
|
@given("the auto debug agent returns a successful fix")
|
|
def step_agent_successful_fix(context: BehaveContext) -> None:
|
|
"""Configure the mock agent to return a successful result (hits 922-935)."""
|
|
context.r2_agent_result = {
|
|
"error_message": "First attempt build failure",
|
|
"code_context": "",
|
|
"attempted_fixes": [],
|
|
"current_fix": {},
|
|
"fix_validated": True,
|
|
"messages": [],
|
|
"context": {},
|
|
"result": {
|
|
"success": True,
|
|
"fix": {
|
|
"description": "Fixed the import statement",
|
|
"code": "import os\nimport sys",
|
|
},
|
|
},
|
|
"error": None,
|
|
"metadata": {},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: apply_changes setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the current plan has a pending CREATE change with a poison path")
|
|
def step_pending_change_poison(context: BehaveContext) -> None:
|
|
"""Set up a pending change that will trigger a generic exception on apply."""
|
|
change = MagicMock(spec=Change)
|
|
change.id = 10
|
|
change.plan_id = 1
|
|
change.file_path = "src/poison.py"
|
|
change.operation = OperationType.CREATE
|
|
change.new_content = "content"
|
|
change.new_path = None
|
|
change.applied = False
|
|
context.r2_ctx.changes.get_for_plan.return_value = [change]
|
|
|
|
# Update the plan mock for apply_changes
|
|
context.r2_plan.status = PlanStatus.BUILT
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call auto_debug_build")
|
|
def step_call_auto_debug_build(context: BehaveContext) -> None:
|
|
"""Execute auto_debug_build with mocked AutoDebugAgent."""
|
|
mock_agent_cls = MagicMock()
|
|
mock_agent_instance = MagicMock()
|
|
mock_agent_instance.invoke.return_value = context.r2_agent_result
|
|
mock_agent_cls.return_value = mock_agent_instance
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.agents.AutoDebugAgent",
|
|
mock_agent_cls,
|
|
),
|
|
patch(
|
|
"cleveragents.agents.AutoDebugState",
|
|
dict,
|
|
),
|
|
):
|
|
# Also patch _prepare_langsmith_config to return empty dict
|
|
context.r2_service._prepare_langsmith_config = MagicMock(return_value={})
|
|
|
|
try:
|
|
result = context.r2_service.auto_debug_build(
|
|
context.r2_project, max_attempts=3
|
|
)
|
|
context.r2_result = result
|
|
except Exception as exc:
|
|
context.r2_error = exc
|
|
|
|
|
|
@when("I call apply_changes and expect an error")
|
|
def step_call_apply_changes_error(context: BehaveContext) -> None:
|
|
"""Call apply_changes with a change that triggers a generic exception.
|
|
|
|
Uses a real temporary directory for path resolution, and patches only
|
|
Path.write_text to raise OSError (not PlanError) so the generic
|
|
exception handler on line 1036 is reached.
|
|
"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
context.r2_project.path = Path(tmpdir)
|
|
with patch.object(Path, "write_text", side_effect=OSError("disk full")):
|
|
try:
|
|
context.r2_service.apply_changes(context.r2_project)
|
|
context.r2_error = None
|
|
except PlanError as exc:
|
|
context.r2_error = exc
|
|
except Exception as exc:
|
|
context.r2_error = exc
|
|
|
|
|
|
@when("I call _coerce_change_list with a dict that fails Change validation")
|
|
def step_call_coerce_change_list_invalid(context: BehaveContext) -> None:
|
|
"""Call _coerce_change_list with a dict that will fail Change(**entry_dict).
|
|
|
|
An empty dict lacks required fields (plan_id, file_path, operation),
|
|
triggering a validation error inside the try/except guard (lines 1413-1416).
|
|
"""
|
|
try:
|
|
context.r2_service._coerce_change_list(
|
|
raw_changes=[{"not_a_valid_field": True}],
|
|
plan_name="test-plan",
|
|
provider_name="test-provider",
|
|
)
|
|
context.r2_error = None
|
|
except PlanError as exc:
|
|
context.r2_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the auto_debug_build should return failure with the build error")
|
|
def step_assert_auto_debug_failure(context: BehaveContext) -> None:
|
|
"""Verify auto_debug_build returned (False, changes, error_msg)."""
|
|
assert context.r2_error is None, f"Unexpected exception: {context.r2_error!r}"
|
|
assert context.r2_result is not None, "Expected a result tuple"
|
|
success, _changes, error_msg = context.r2_result
|
|
assert success is False, f"Expected failure, got success={success}"
|
|
assert error_msg is not None, "Expected an error message"
|
|
assert "Simulated build failure" in error_msg
|
|
|
|
|
|
@then("the auto_debug_build should return success")
|
|
def step_assert_auto_debug_success(context: BehaveContext) -> None:
|
|
"""Verify auto_debug_build returned (True, changes, None)."""
|
|
assert context.r2_error is None, f"Unexpected exception: {context.r2_error!r}"
|
|
assert context.r2_result is not None, "Expected a result tuple"
|
|
success, _changes, error_msg = context.r2_result
|
|
assert success is True, f"Expected success, got success={success}"
|
|
assert error_msg is None, f"Expected no error, got: {error_msg}"
|
|
|
|
|
|
@then("the plan prompt should contain the auto-debug fix description")
|
|
def step_assert_prompt_updated(context: BehaveContext) -> None:
|
|
"""Verify the plan prompt was updated with the fix description (lines 926-929)."""
|
|
prompt = context.r2_plan.prompt
|
|
assert prompt is not None, "Expected plan prompt to be set"
|
|
assert "Auto-debug fix attempt" in prompt, (
|
|
f"Expected 'Auto-debug fix attempt' in prompt, got: {prompt!r}"
|
|
)
|
|
assert "Fixed the import statement" in prompt, (
|
|
f"Expected fix description in prompt, got: {prompt!r}"
|
|
)
|
|
|
|
|
|
@then("the apply error should be a PlanError with operation value in details")
|
|
def step_assert_apply_error(context: BehaveContext) -> None:
|
|
"""Verify apply_changes raised a PlanError (line 1036)."""
|
|
err = context.r2_error
|
|
assert err is not None, "Expected an error from apply_changes"
|
|
assert isinstance(err, PlanError), (
|
|
f"Expected PlanError, got {type(err).__name__}: {err}"
|
|
)
|
|
assert "Failed to apply change" in err.message, (
|
|
f"Expected 'Failed to apply change' in message: {err.message}"
|
|
)
|
|
assert "operation" in (err.details or {}), (
|
|
f"Expected 'operation' in details: {err.details}"
|
|
)
|
|
|
|
|
|
@then("a PlanError should be raised about unable to parse streamed change entry")
|
|
def step_assert_coerce_error(context: BehaveContext) -> None:
|
|
"""Verify _coerce_change_list raised PlanError (lines 1413-1416)."""
|
|
err = context.r2_error
|
|
assert err is not None, "Expected a PlanError"
|
|
assert isinstance(err, PlanError), (
|
|
f"Expected PlanError, got {type(err).__name__}: {err}"
|
|
)
|
|
assert "Unable to parse streamed change entry" in err.message, (
|
|
f"Expected parse error message, got: {err.message}"
|
|
)
|