"""Step definitions targeting remaining uncovered lines in context_service.py. Specifically covers: - Line 566: `or {}` fallback in _build_langsmith_config when Settings.build_langsmith_config returns None despite is_langsmith_enabled being True. - Lines 603-607: `except Exception` in _refresh_vector_index (generic non-ConfigurationError raised by vector store refresh). - Line 923: Defensive `service is None` guard inside search_context when _vector_store_available() returns True but the actual service attribute is None. - Lines 938-939: `except Exception` in search_context (generic non-ConfigurationError raised by vector store search). """ from __future__ import annotations import shutil import tempfile from pathlib import Path from typing import Any from unittest.mock import patch from behave import given, then, when from cleveragents.application.services.context_service import ContextService from cleveragents.config.settings import Settings from cleveragents.domain.models.core import Plan, Project from features.steps.service_steps import add_cleanup # --------------------------------------------------------------------------- # Helpers - lightweight in-memory fakes # --------------------------------------------------------------------------- class _BoostPlansRepo: def __init__(self, plan: Plan | None) -> None: self._plan = plan def get_current_for_project(self, project_id: int | None) -> Plan | None: return self._plan if project_id else None class _BoostContextRepo: def __init__(self) -> None: self._store: dict[int, list] = {} def get_for_plan(self, plan_id: int) -> list: return list(self._store.get(plan_id, [])) class _BoostTransaction: def __init__(self, plans: _BoostPlansRepo, contexts: _BoostContextRepo): self.plans = plans self.contexts = contexts def __enter__(self): return self def __exit__(self, *args): return False class _BoostUnitOfWork: def __init__(self, plan: Plan | None) -> None: self._plans = _BoostPlansRepo(plan) self._contexts = _BoostContextRepo() def transaction(self): return _BoostTransaction(self._plans, self._contexts) class _BoostVectorStore: """Configurable vector store double for boost scenarios.""" def __init__(self, *, enabled: bool = True) -> None: self.enabled = enabled self.refresh_error: Exception | None = None self.search_error: Exception | None = None self.results: list[dict[str, Any]] = [] def is_enabled(self) -> bool: return self.enabled def refresh_for_plan(self, plan_id: int) -> int: if self.refresh_error: raise self.refresh_error return 0 def search( self, plan_id: int, query: str, *, top_k: int = 5, refresh_if_missing: bool = True, ) -> list[dict[str, Any]]: if self.search_error: raise self.search_error return list(self.results) def _make_boost_workspace(context, *, vector_stub=None): """Build a minimal workspace for boost coverage scenarios.""" temp_dir = Path(tempfile.mkdtemp(prefix="boost-cov-")) add_cleanup(context, lambda: shutil.rmtree(temp_dir, ignore_errors=True)) plan = Plan( id=501, project_id=1, name="boost-plan", prompt="boost coverage", current=True, ) project = Project(id=1, name="boost-project", path=temp_dir) settings = Settings() uow = _BoostUnitOfWork(plan) service = ContextService( settings=settings, unit_of_work=uow, vector_store_service=vector_stub, ) context.boost_temp_dir = temp_dir context.boost_plan = plan context.boost_project = project context.boost_settings = settings context.boost_uow = uow context.boost_service = service # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a boost workspace with LangSmith enabled but build returning None") def step_boost_langsmith_returns_none(context): """Set up a service where is_langsmith_enabled=True but build_langsmith_config returns None, exercising the `or {}` fallback on line 566.""" _make_boost_workspace(context) # Patch at the Settings *class* level because Pydantic models do not # allow arbitrary attribute assignment on instances. # is_langsmith_enabled must be True so the early-return on line 540 # is skipped and execution reaches build_langsmith_config. # build_langsmith_config must return None so line 566 fires. patcher_enabled = patch.object( Settings, "is_langsmith_enabled", new_callable=lambda: property(lambda self: True), ) patcher_build = patch.object( Settings, "build_langsmith_config", return_value=None, ) patcher_enabled.start() patcher_build.start() add_cleanup(context, patcher_enabled.stop) add_cleanup(context, patcher_build.stop) @given("a boost workspace with a vector store that raises RuntimeError on refresh") def step_boost_vector_refresh_runtime_error(context): """Vector store whose refresh_for_plan raises RuntimeError (not ConfigurationError), exercising lines 603-607.""" stub = _BoostVectorStore(enabled=True) stub.refresh_error = RuntimeError("unexpected refresh failure") _make_boost_workspace(context, vector_stub=stub) @given("a boost workspace where vector_store_available is true but service is None") def step_boost_vector_available_but_none(context): """_vector_store_available returns True yet _vector_store_service is None, exercising the defensive guard on line 923.""" _make_boost_workspace(context, vector_stub=None) # Override _vector_store_available to lie and return True even though # the actual service attribute is None. patcher = patch.object( context.boost_service, "_vector_store_available", return_value=True, ) patcher.start() add_cleanup(context, patcher.stop) @given("a boost workspace with a vector store that raises RuntimeError on search") def step_boost_vector_search_runtime_error(context): """Vector store whose search raises RuntimeError (not ConfigurationError), exercising lines 938-939.""" stub = _BoostVectorStore(enabled=True) stub.search_error = RuntimeError("unexpected search failure") _make_boost_workspace(context, vector_stub=stub) # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when("I build a LangSmith config via the boost service") def step_boost_build_langsmith(context): context.boost_langsmith_result = context.boost_service._build_langsmith_config( context.boost_project, run_name="BoostRun", file_paths=["a.py"], mode="sync", ) @when("I trigger a boost vector index refresh") def step_boost_trigger_refresh(context): context.boost_refresh_exception = None try: context.boost_service._refresh_vector_index(context.boost_plan.id) except Exception as exc: context.boost_refresh_exception = exc @when('I search the boost context for "{query}"') def step_boost_search_context(context, query: str): context.boost_search_results = context.boost_service.search_context( context.boost_project, query ) # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then("the boost LangSmith config should be an empty dict") def step_boost_langsmith_empty(context): assert context.boost_langsmith_result == {}, ( f"Expected empty dict, got {context.boost_langsmith_result}" ) @then("the boost refresh should complete without raising") def step_boost_refresh_no_exception(context): assert context.boost_refresh_exception is None, ( f"Unexpected exception: {context.boost_refresh_exception}" ) @then("the boost search results should be empty") def step_boost_search_empty(context): assert context.boost_search_results == [], ( f"Expected empty results, got {context.boost_search_results}" )