"""Step definitions for auto_debug_coverage_boost.feature. These steps target specific uncovered lines in src/cleveragents/agents/graphs/auto_debug.py: - Line 124: _analyze_error when content == "Mock LLM response" (true branch) - Lines 128-130: _analyze_error exception handler - Lines 200-203: _generate_fix when content == "Mock LLM response" (true branch) - Lines 209-213: _generate_fix JSONDecodeError fallback - Lines 215-220: _generate_fix exception handler - Line 266: _validate_fix when content == "Mock LLM response" - Lines 269-270: _validate_fix JSON parse with is_valid - Lines 271-276: _validate_fix JSONDecodeError keyword scan - Lines 277-279: _validate_fix exception handler - Lines 283-286: _validate_fix invalid → append to attempted_fixes - Lines 290-295: _should_retry_fix routing logic - Lines 297-305: _finalize result construction """ import json from typing import Any from behave import given, then, when from cleveragents.agents.graphs.auto_debug import AutoDebugAgent, AutoDebugState # --------------------------------------------------------------------------- # Helpers: mock LLM classes # --------------------------------------------------------------------------- class _MockResponse: """Minimal response object with a content attribute.""" def __init__(self, content: str) -> None: self.content = content class _MockLLM: """Stub LLM that returns a fixed content string.""" def __init__(self, content: str = "Mock LLM response") -> None: self._content = content def invoke(self, messages: Any) -> _MockResponse: return _MockResponse(self._content) class _FailingLLM: """Stub LLM whose invoke always raises an exception.""" def invoke(self, messages: Any) -> None: raise RuntimeError("LLM service unavailable for testing") def _base_state(**overrides: Any) -> AutoDebugState: """Create a minimal valid AutoDebugState with optional overrides.""" state: AutoDebugState = { "messages": [], "context": {}, "result": None, "error": None, "metadata": {}, "error_message": "NameError: name 'x' is not defined", "code_context": "print(x)", "attempted_fixes": [], "current_fix": {}, "fix_validated": False, } state.update(overrides) # type: ignore[typeddict-item] return state # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the auto debug graph module is imported") def step_auto_debug_module_imported(context): """Verify the auto_debug module is importable.""" assert AutoDebugAgent is not None assert AutoDebugState is not None # --------------------------------------------------------------------------- # Agent construction helpers # --------------------------------------------------------------------------- @given('an auto debug agent with a mock LLM returning "{content}"') def step_create_agent_mock_llm(context, content): """Create an AutoDebugAgent with a mock LLM returning the given content.""" context.ad_agent = AutoDebugAgent(llm=_MockLLM(content), max_fix_attempts=3) @given( 'an auto debug agent with a mock LLM returning "{content}"' " and max {n:d} fix attempts" ) def step_create_agent_mock_llm_max_attempts(context, content, n): """Create an AutoDebugAgent with a mock LLM and custom max_fix_attempts.""" context.ad_agent = AutoDebugAgent(llm=_MockLLM(content), max_fix_attempts=n) @given("an auto debug agent with a failing LLM") def step_create_agent_failing_llm(context): """Create an AutoDebugAgent with an LLM that always raises.""" context.ad_agent = AutoDebugAgent(llm=_FailingLLM(), max_fix_attempts=3) @given("an auto debug agent with a mock LLM returning valid fix JSON") def step_create_agent_valid_fix_json(context): """Create an agent whose LLM returns well-formed fix JSON.""" fix_json = json.dumps( { "description": "Parsed JSON fix", "code": "print('fixed')", "files_to_modify": ["main.py"], } ) context.ad_agent = AutoDebugAgent(llm=_MockLLM(fix_json), max_fix_attempts=3) @given("an auto debug agent with a mock LLM returning valid validation JSON true") def step_create_agent_valid_validation_json_true(context): """Create an agent whose LLM returns validation JSON with is_valid=true.""" validation_json = json.dumps( { "is_valid": True, "reasoning": "Fix resolves the error", "issues": [], } ) context.ad_agent = AutoDebugAgent(llm=_MockLLM(validation_json), max_fix_attempts=3) @given("an auto debug agent with a mock LLM returning valid validation JSON false") def step_create_agent_valid_validation_json_false(context): """Create an agent whose LLM returns validation JSON with is_valid=false.""" validation_json = json.dumps( { "is_valid": False, "reasoning": "Fix does not resolve the error", "issues": ["error persists"], } ) context.ad_agent = AutoDebugAgent(llm=_MockLLM(validation_json), max_fix_attempts=3) # --------------------------------------------------------------------------- # State preparation helpers # --------------------------------------------------------------------------- @given("a state with an existing error analysis message") def step_state_with_analysis(context): """Prepare a state that already has an error_analysis message.""" context.ad_state = _base_state( messages=[ { "role": "assistant", "content": "Detected a NameError due to undefined variable", "type": "error_analysis", } ] ) @given("a state with two previous attempted fixes") def step_state_with_previous_attempts(context): """Prepare a state with two prior attempted fixes.""" context.ad_state = _base_state( messages=[ { "role": "assistant", "content": "Analysis result", "type": "error_analysis", } ], attempted_fixes=[ {"description": "First attempt", "code": "# v1", "files_to_modify": []}, {"description": "Second attempt", "code": "# v2", "files_to_modify": []}, ], ) @given("a state with a current fix to validate") def step_state_with_current_fix(context): """Prepare a state that has a current_fix ready for validation.""" context.ad_state = _base_state( current_fix={ "description": "Proposed fix", "code": "x = 42\nprint(x)", "files_to_modify": ["main.py"], } ) @given("a finalize state with fix_validated {validated} and {n:d} attempted fixes") def step_finalize_state(context, validated, n): """Prepare a state for _finalize with the given validation status.""" is_valid = validated.lower() == "true" fixes = [ {"description": f"Attempt {i + 1}", "code": f"# v{i + 1}"} for i in range(n) ] context.ad_state = _base_state( fix_validated=is_valid, attempted_fixes=fixes, current_fix={"description": "Latest fix", "code": "# latest"}, ) # --------------------------------------------------------------------------- # When: call individual node methods # --------------------------------------------------------------------------- @when("I call _analyze_error with an error state") def step_call_analyze_error(context): """Invoke _analyze_error directly on the agent.""" state = _base_state() context.ad_result_state = context.ad_agent._analyze_error(state) @when("I call _generate_fix with the prepared state") def step_call_generate_fix(context): """Invoke _generate_fix directly on the agent.""" context.ad_result_state = context.ad_agent._generate_fix(context.ad_state) @when("I call _validate_fix with the prepared state") def step_call_validate_fix(context): """Invoke _validate_fix directly on the agent.""" context.ad_result_state = context.ad_agent._validate_fix(context.ad_state) @when("I call _finalize with the prepared state") def step_call_finalize(context): """Invoke _finalize directly on the agent.""" context.ad_result_state = context.ad_agent._finalize(context.ad_state) @when("I check should_retry_fix with validated {validated} and {n:d} attempt") def step_check_should_retry_singular(context, validated, n): """Invoke _should_retry_fix with a crafted state (singular form).""" is_valid = validated.lower() == "true" fixes = [{"description": f"fix {i + 1}"} for i in range(n)] state = _base_state(fix_validated=is_valid, attempted_fixes=fixes) context.routing_decision = context.ad_agent._should_retry_fix(state) @when("I check should_retry_fix with validated {validated} and {n:d} attempts") def step_check_should_retry_plural(context, validated, n): """Invoke _should_retry_fix with a crafted state (plural form).""" is_valid = validated.lower() == "true" fixes = [{"description": f"fix {i + 1}"} for i in range(n)] state = _base_state(fix_validated=is_valid, attempted_fixes=fixes) context.routing_decision = context.ad_agent._should_retry_fix(state) @when("I try to create an AutoDebugAgent with no LLM") def step_create_agent_no_llm(context): """Attempt to create an AutoDebugAgent with llm=None.""" try: AutoDebugAgent(llm=None) context.constructor_error = None except ValueError as exc: context.constructor_error = exc # --------------------------------------------------------------------------- # Then: assertions # --------------------------------------------------------------------------- @then('the analysis message should be "{expected}"') def step_verify_analysis_message(context, expected): """Verify the last error_analysis message content.""" messages = context.ad_result_state.get("messages", []) analysis_msgs = [m for m in messages if m.get("type") == "error_analysis"] assert analysis_msgs, "No error_analysis message found in state" actual = analysis_msgs[-1]["content"] assert actual == expected, f"Expected '{expected}', got '{actual}'" @then('the current fix description should be "{expected}"') def step_verify_fix_description(context, expected): """Verify the current_fix description.""" fix = context.ad_result_state.get("current_fix", {}) actual = fix.get("description", "") assert actual == expected, f"Expected '{expected}', got '{actual}'" @then('the current fix code should be "{expected}"') def step_verify_fix_code(context, expected): """Verify the current_fix code.""" fix = context.ad_result_state.get("current_fix", {}) actual = fix.get("code", "") assert actual == expected, f"Expected '{expected}', got '{actual}'" @then("the fix should be marked as validated") def step_fix_validated_true(context): """Verify fix_validated is True.""" assert context.ad_result_state["fix_validated"] is True @then("the fix should not be marked as validated") def step_fix_validated_false(context): """Verify fix_validated is False.""" assert context.ad_result_state["fix_validated"] is False @then("the current fix should be appended to attempted fixes") def step_fix_appended_to_attempted(context): """Verify the current fix was moved to attempted_fixes.""" attempted = context.ad_result_state.get("attempted_fixes", []) assert len(attempted) > 0, "Expected at least one attempted fix" last = attempted[-1] assert last.get("description") == "Proposed fix", ( f"Expected 'Proposed fix', got '{last.get('description')}'" ) @then('the routing decision should be "{expected}"') def step_verify_routing(context, expected): """Verify the _should_retry_fix routing result.""" assert context.routing_decision == expected, ( f"Expected '{expected}', got '{context.routing_decision}'" ) @then("the result should indicate success {expected}") def step_verify_result_success(context, expected): """Verify the result dict success field.""" is_success = expected.lower() == "true" result = context.ad_result_state.get("result") assert result is not None, "No result dict found in state" assert result["success"] is is_success, ( f"Expected success={is_success}, got {result['success']}" ) @then("the result attempts count should be {n:d}") def step_verify_result_attempts(context, n): """Verify the result dict attempts count.""" result = context.ad_result_state.get("result") assert result is not None, "No result dict found in state" assert result["attempts"] == n, f"Expected {n} attempts, got {result['attempts']}" @then("a ValueError should be raised with the missing provider message") def step_verify_constructor_valueerror(context): """Verify the constructor raised a ValueError about missing LLM.""" assert context.constructor_error is not None, "Expected ValueError but none raised" assert isinstance(context.constructor_error, ValueError) assert "No LLM provider configured" in str(context.constructor_error)