"""Step definitions for context_analysis_coverage_boost.feature. These steps target specific uncovered lines in cleveragents/agents/graphs/context_analysis.py: - Line 183: _with_retry fallback when chain lacks .with_retry - Lines 239-240: Exception in _load_files during TextLoader.load() - Lines 267-276: Exception in _analyze_dependencies when chain.invoke() raises - Lines 351-358: Exception in _score_relevance when chain.invoke() raises - Lines 416-424: Exception in _summarize_context when chain.invoke() raises """ from __future__ import annotations import tempfile from pathlib import Path from typing import Any from unittest.mock import patch from behave import given, then, when from langchain_community.llms import FakeListLLM from langchain_core.documents import Document from cleveragents.agents.graphs.context_analysis import ( ContextAnalysisAgent, ContextAnalysisState, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- DEFAULT_RESPONSES = [ "Dependencies: ['os', 'sys']", "Relevance: 0.7", "Summary: test summary", ] * 20 class _AlwaysRaisingLLM(FakeListLLM): """FakeListLLM subclass that raises on every call.""" def __init__(self, error_message: str = "forced LLM failure"): super().__init__(responses=[]) object.__setattr__(self, "_error_message", error_message) def _call(self, prompt: str, **kwargs: Any) -> str: raise RuntimeError(getattr(self, "_error_message", "forced LLM failure")) def _make_state(**overrides: Any) -> ContextAnalysisState: """Build a minimal valid ContextAnalysisState with optional overrides.""" state: ContextAnalysisState = { "file_paths": [], "documents": [], "dependencies": {}, "summary": "", "relevance_scores": {}, "chunks": [], "error": None, } state.update(overrides) # type: ignore[typeddict-item] return state # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the context analysis coverage module is imported") def step_coverage_module_imported(context: Any) -> None: assert ContextAnalysisAgent is not None assert ContextAnalysisState is not None # --------------------------------------------------------------------------- # Agent creation helpers # --------------------------------------------------------------------------- @given("I have a context analysis agent for coverage boost") def step_create_normal_agent(context: Any) -> None: llm = FakeListLLM(responses=list(DEFAULT_RESPONSES)) context.cb_agent = ContextAnalysisAgent(llm=llm, retry_attempts=1) @given("I have a context analysis agent with a raising LLM for coverage boost") def step_create_raising_agent(context: Any) -> None: """Create an agent whose LLM always raises RuntimeError.""" # Build with a valid LLM first so __init__ completes llm = FakeListLLM(responses=list(DEFAULT_RESPONSES)) agent = ContextAnalysisAgent(llm=llm, retry_attempts=1) # Swap to the raising LLM so node methods hit exception handlers agent.llm = _AlwaysRaisingLLM("forced LLM failure") context.cb_agent = agent # --------------------------------------------------------------------------- # _with_retry fallback (line 183) # --------------------------------------------------------------------------- @given("I have a runnable without a with_retry method") def step_create_runnable_without_retry(context: Any) -> None: """Create a simple callable that does NOT have a with_retry attribute.""" class _BareRunnable: """Minimal object without with_retry.""" def invoke(self, input_data: Any) -> str: return "bare_result" context.cb_bare_runnable = _BareRunnable() @when("I call _with_retry on the runnable") def step_call_with_retry(context: Any) -> None: context.cb_retry_result = context.cb_agent._with_retry(context.cb_bare_runnable) @then("the returned object should be the same runnable") def step_verify_same_runnable(context: Any) -> None: assert context.cb_retry_result is context.cb_bare_runnable # --------------------------------------------------------------------------- # _load_files exception handler (lines 239-240) # --------------------------------------------------------------------------- @given("TextLoader.load is patched to raise an OSError") def step_patch_text_loader(context: Any) -> None: """Store a flag; the patch is applied in the When step.""" context.cb_patch_loader = True @when("I execute load_files with a real file path") def step_execute_load_files_with_exception(context: Any) -> None: # Create a real temp file so Path.exists() and Path.is_file() pass with tempfile.NamedTemporaryFile(suffix=".py", delete=False, mode="w") as tmp: tmp.write("print('hello')") context.cb_tmp_path = tmp.name state = _make_state(file_paths=[tmp.name]) with patch("cleveragents.agents.graphs.context_analysis.TextLoader") as mock_cls: mock_cls.return_value.load.side_effect = OSError("disk read error") context.cb_load_result = context.cb_agent._load_files(state) # Cleanup temp file Path(tmp.name).unlink(missing_ok=True) @then("the load_files result should contain an error mentioning the file") def step_verify_load_error(context: Any) -> None: error = context.cb_load_result.get("error") assert error is not None, "Expected an error in the result" assert "Error loading" in error assert "disk read error" in error @then("the documents list should be empty in the result") def step_verify_empty_documents(context: Any) -> None: docs = context.cb_load_result.get("documents", []) assert len(docs) == 0 # --------------------------------------------------------------------------- # _analyze_dependencies exception handler (lines 267-276) # --------------------------------------------------------------------------- @given("I have a state with one document for dependency analysis") def step_state_with_document_for_deps(context: Any) -> None: doc = Document( page_content="import os\nimport sys\n", metadata={"source": "test_module.py"}, ) context.cb_state = _make_state(documents=[doc]) @given("I have a state with one document and a pre-existing error") def step_state_with_document_and_error(context: Any) -> None: doc = Document( page_content="import os\nimport sys\n", metadata={"source": "test_module.py"}, ) context.cb_state = _make_state( documents=[doc], error="previous error occurred", ) @when("I execute analyze_dependencies on the state") def step_execute_analyze_deps(context: Any) -> None: context.cb_deps_result = context.cb_agent._analyze_dependencies(context.cb_state) @then("the result should have an error about dependency analysis") def step_verify_dep_error(context: Any) -> None: error = context.cb_deps_result.get("error") assert error is not None, "Expected an error in the result" assert "Dependency analysis error" in error @then("the dependencies for the document should be an empty list") def step_verify_deps_empty(context: Any) -> None: deps = context.cb_deps_result.get("dependencies", {}) # The document source is "test_module.py" assert deps.get("test_module.py") == [] @then("the result error should contain both the pre-existing and new error messages") def step_verify_combined_dep_error(context: Any) -> None: error = context.cb_deps_result.get("error") assert error is not None, "Expected an error in the result" assert "previous error occurred" in error assert "Dependency analysis error" in error # --------------------------------------------------------------------------- # _score_relevance exception handler (lines 351-358) # --------------------------------------------------------------------------- @given("I have a state with chunks for relevance scoring") def step_state_with_chunks(context: Any) -> None: chunk = Document( page_content="def main(): pass", metadata={"source": "main.py"}, ) context.cb_state = _make_state(chunks=[chunk]) @given("I have a state with chunks and a pre-existing error for scoring") def step_state_with_chunks_and_error(context: Any) -> None: chunk = Document( page_content="def main(): pass", metadata={"source": "main.py"}, ) context.cb_state = _make_state( chunks=[chunk], error="earlier pipeline error", ) @when("I execute score_relevance on the state") def step_execute_score_relevance(context: Any) -> None: context.cb_score_result = context.cb_agent._score_relevance(context.cb_state) @then("the result should have an error about relevance scoring") def step_verify_score_error(context: Any) -> None: error = context.cb_score_result.get("error") assert error is not None, "Expected an error in the result" assert "Relevance scoring error" in error @then("the relevance score for the file should default to 0.5") def step_verify_default_score(context: Any) -> None: scores = context.cb_score_result.get("relevance_scores", {}) assert scores.get("main.py") == 0.5 @then( "the scoring result error should contain both the pre-existing and new error messages" ) def step_verify_combined_score_error(context: Any) -> None: error = context.cb_score_result.get("error") assert error is not None, "Expected an error in the result" assert "earlier pipeline error" in error assert "Relevance scoring error" in error # --------------------------------------------------------------------------- # _summarize_context exception handler (lines 416-424) # --------------------------------------------------------------------------- @given("I have a complete state ready for summarization") def step_state_for_summarization(context: Any) -> None: doc = Document(page_content="class Foo: pass", metadata={"source": "foo.py"}) context.cb_state = _make_state( documents=[doc], dependencies={"foo.py": ["os"]}, relevance_scores={"foo.py": 0.9}, ) @given("I have a complete state with a pre-existing error for summarization") def step_state_for_summarization_with_error(context: Any) -> None: doc = Document(page_content="class Foo: pass", metadata={"source": "foo.py"}) context.cb_state = _make_state( documents=[doc], dependencies={"foo.py": ["os"]}, relevance_scores={"foo.py": 0.9}, error="upstream error in pipeline", ) @when("I execute summarize_context on the state") def step_execute_summarize(context: Any) -> None: context.cb_summary_result = context.cb_agent._summarize_context(context.cb_state) @then('the summary should be "Context analysis failed"') def step_verify_failed_summary(context: Any) -> None: assert context.cb_summary_result.get("summary") == "Context analysis failed" @then("the result should have an error about summarization") def step_verify_summary_error(context: Any) -> None: error = context.cb_summary_result.get("error") assert error is not None, "Expected an error in the result" assert "Summarization error" in error @then( "the summarization result error should contain both the pre-existing and new error messages" ) def step_verify_combined_summary_error(context: Any) -> None: error = context.cb_summary_result.get("error") assert error is not None, "Expected an error in the result" assert "upstream error in pipeline" in error assert "Summarization error" in error # --------------------------------------------------------------------------- # No LLM raises ValueError (lines 114-118) # --------------------------------------------------------------------------- @when("I try to create a ContextAnalysisAgent with no LLM") def step_create_agent_no_llm(context: Any) -> None: try: ContextAnalysisAgent(llm=None) context.cb_no_llm_error = None except ValueError as exc: context.cb_no_llm_error = exc @then("a ValueError should be raised about missing LLM provider") def step_verify_no_llm_error(context: Any) -> None: assert context.cb_no_llm_error is not None, "Expected a ValueError" assert "No LLM provider configured" in str(context.cb_no_llm_error)