"""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}" )