"""Step definitions for plan_generation_coverage_boost.feature. These steps target specific uncovered lines in plan_generation.py: - Lines 90-91: BoundedMemorySaver._prune exception fallback in serde.loads_typed - Lines 428-429: _generate_plan read_text() exception fallback - Line 491: _find_matching_context returns None when no context matches - Lines 631-635: _analyze_contexts defensive fallback when agent.invoke raises """ from __future__ import annotations import tempfile from pathlib import Path from unittest.mock import MagicMock, patch from behave import given, then, when from langchain_community.llms import FakeListLLM from cleveragents.agents.graphs.plan_generation import ( BoundedMemorySaver, PlanGenerationGraph, ) from cleveragents.domain.models.core import Context, Plan, Project def _default_test_llm() -> FakeListLLM: """Create a FakeListLLM for test purposes.""" return FakeListLLM( responses=[ "Requirements: Add error handling", "Generated code with error handling", "Validation passed", ] ) # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the plan generation coverage module is imported") def step_coverage_module_imported(context): """Ensure the plan_generation module is importable.""" assert BoundedMemorySaver is not None assert PlanGenerationGraph is not None # --------------------------------------------------------------------------- # Scenario: BoundedMemorySaver prune with corrupt checkpoint (lines 90-91) # --------------------------------------------------------------------------- @given("a BoundedMemorySaver with max_checkpoints set to 1") def step_create_bounded_saver(context): """Create a BoundedMemorySaver that retains only 1 checkpoint.""" context.saver = BoundedMemorySaver(max_checkpoints=1) @given("the saver storage contains two checkpoints with corrupt data") def step_populate_corrupt_storage(context): """Directly populate internal storage with two checkpoints. The second tuple element is metadata bytes and the third is the parent config; only the first element (checkpoint bytes) matters for the serde.loads_typed call that we want to fail. """ saver = context.saver thread_id = "test-thread" ns = "" # Access internals the same way _prune does storage = saver.storage # defaultdict(lambda: defaultdict(dict)) # Ensure the nested defaultdicts exist storage[thread_id][ns]["ckpt-001"] = (b"corrupt-data", b"", None) storage[thread_id][ns]["ckpt-002"] = (b"also-corrupt", b"", None) # Seed writes and blobs so pop() calls don't error saver.writes[(thread_id, ns, "ckpt-001")] = {} saver.writes[(thread_id, ns, "ckpt-002")] = {} context.thread_id = thread_id context.ns = ns @when("the saver prunes the oldest checkpoint") def step_prune_checkpoints(context): """Monkey-patch serde.loads_typed to raise, then call _prune. This forces the except branch on lines 90-91. """ saver = context.saver original_loads = saver.serde.loads_typed def _failing_loads_typed(*args, **kwargs): raise ValueError("corrupt checkpoint bytes") saver.serde.loads_typed = _failing_loads_typed try: context.prune_error = None saver._prune(context.thread_id, context.ns) except Exception as exc: context.prune_error = exc finally: saver.serde.loads_typed = original_loads @then("the prune should complete without error") def step_verify_prune_no_error(context): assert context.prune_error is None, ( f"Expected no error from _prune, got {context.prune_error!r}" ) @then("the corrupt checkpoint should be removed from storage") def step_verify_checkpoint_removed(context): """After pruning with max_checkpoints=1, only 1 checkpoint should remain.""" remaining = context.saver.storage[context.thread_id][context.ns] assert len(remaining) == 1, f"Expected 1 remaining checkpoint, got {len(remaining)}" # --------------------------------------------------------------------------- # Scenario: _generate_plan read_text exception fallback (lines 428-429) # --------------------------------------------------------------------------- @given("a PlanGenerationGraph instance for coverage testing") def step_create_graph_instance(context): """Create a PlanGenerationGraph with a FakeListLLM.""" context.graph = PlanGenerationGraph(llm=_default_test_llm(), max_retries=1) @given("a generate state with an explicit path pointing to an unreadable file") def step_prepare_unreadable_file_state(context): """Create a temp file and build a state where the prompt uses @. We will mock Path.read_text to raise PermissionError so the except branch on lines 428-429 is hit. """ # Create a real temp file so Path.exists() returns True and is_dir() returns False with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as tmp: tmp.write(b"# existing code") context.temp_file_path = tmp.name context.unreadable_state = { "project": Project(id=1, name="test-project", path=Path("/tmp")), "plan": Plan(id=1, project_id=1, name="plan", prompt="test prompt"), "contexts": [], "context_summary": "summary", "prompt": f"modify @{tmp.name}", "analyzed_requirements": { "description": "add error handling", "operation": "create", }, "generated_changes": [], "validation_result": {}, "retry_count": 0, "error": None, "context_dependencies": {}, "context_relevance": {}, "context_analysis_error": None, "actor_name": None, "actor_options": {}, "actor_graph_descriptor": None, "actor_initial_context": {}, } def cleanup(): import os os.unlink(tmp.name) context.add_cleanup(cleanup) @when("_generate_plan is called with the unreadable file state") def step_call_generate_plan_unreadable(context): """Call _generate_plan with Path.read_text patched to raise.""" graph = context.graph state = context.unreadable_state with patch.object(Path, "read_text", side_effect=PermissionError("no access")): context.gen_result = graph._generate_plan(state) @then("a change should be generated with None as original_content") def step_verify_original_content_none(context): changes = context.gen_result.get("generated_changes", []) assert len(changes) == 1, f"Expected 1 change, got {len(changes)}" assert changes[0].original_content is None, ( f"Expected original_content=None, got {changes[0].original_content!r}" ) @then("the change operation should be MODIFY") def step_verify_operation_modify(context): from cleveragents.domain.models.core.change import OperationType changes = context.gen_result["generated_changes"] assert changes[0].operation == OperationType.MODIFY, ( f"Expected MODIFY, got {changes[0].operation!r}" ) # --------------------------------------------------------------------------- # Scenario: _find_matching_context returns None (line 491) # --------------------------------------------------------------------------- @given("a list of contexts with non-matching paths") def step_create_non_matching_contexts(context): """Create contexts whose paths don't match the search path.""" context.test_contexts = [ Context(plan_id=1, path="src/foo.py", content="foo content"), Context(plan_id=1, path="src/bar.py", content="bar content"), Context(plan_id=1, path="lib/utils.py", content="utils content"), ] @when("_find_matching_context is called with a path that matches no context") def step_call_find_matching_no_match(context): """Call _find_matching_context with a path that no context has.""" context.match_result = context.graph._find_matching_context( context.test_contexts, "completely/different/path.py" ) @then("the matching context result should be None") def step_verify_match_result_none(context): assert context.match_result is None, f"Expected None, got {context.match_result!r}" # --------------------------------------------------------------------------- # Scenario: _analyze_contexts exception fallback (lines 631-635) # --------------------------------------------------------------------------- @given("the ContextAnalysisAgent invoke method is mocked to raise an error") def step_mock_context_analysis_invoke(context): """Prepare a patch that makes ContextAnalysisAgent.invoke raise.""" context.ca_patch = patch( "cleveragents.agents.graphs.plan_generation.ContextAnalysisAgent" ) mock_cls = context.ca_patch.start() mock_instance = MagicMock() mock_instance.invoke.side_effect = RuntimeError("analysis agent exploded") mock_cls.return_value = mock_instance def cleanup(): context.ca_patch.stop() context.add_cleanup(cleanup) @when("_analyze_contexts is called with non-empty contexts") def step_call_analyze_contexts(context): """Call _analyze_contexts with real Context objects.""" test_contexts = [ Context(plan_id=1, path="src/main.py", content="print('hello')"), Context(plan_id=1, path="src/utils.py", content="def helper(): pass"), ] context.analyze_result = context.graph._analyze_contexts(test_contexts) @then("the result should contain the fallback summary") def step_verify_fallback_summary(context): summary = context.analyze_result.get("context_summary", "") # The fallback summary is produced by _format_context_summary which includes "File:" assert "File:" in summary, ( f"Expected fallback summary with 'File:', got: {summary!r}" ) @then("the result should contain an analysis error message") def step_verify_analysis_error(context): error = context.analyze_result.get("context_analysis_error") assert error is not None, "Expected context_analysis_error to be set" assert "Context analysis failed" in error, f"Unexpected error message: {error!r}" assert "analysis agent exploded" in error, ( f"Missing original exception in: {error!r}" ) @then("dependencies and relevance should be empty dicts") def step_verify_empty_dicts(context): deps = context.analyze_result.get("context_dependencies") rel = context.analyze_result.get("context_relevance") assert deps == {}, f"Expected empty dependencies, got {deps!r}" assert rel == {}, f"Expected empty relevance, got {rel!r}"