forked from HAL9000/cleveragents-core
freemo
a074b4846f
Remove FakeListLLM as a silent fallback in agent graph constructors (plan_generation.py, context_analysis.py, auto_debug.py). All three now raise ValueError when llm=None, making missing-provider errors explicit. Add Settings.mock_providers flag and validate_provider_availability() method. Update container.get_ai_provider() to check Settings.mock_providers first, with env-var fallback for backward compatibility. Add resolve_provider_by_name() helper to the provider registry and export it from cleveragents.providers. Add structlog trace logging to ProviderRegistry.get_default_provider_type() to record selection reasoning. Update all existing behave step files, robot tests, and benchmarks that relied on the implicit FakeListLLM default to pass an explicit LLM instance instead. Add new BDD tests (features/provider_fixes.feature with 17 scenarios), Robot Framework integration tests (robot/provider_detection_smoke.robot), and ASV benchmarks (benchmarks/provider_selection_bench.py). ISSUES CLOSED: #323
395 lines
12 KiB
Python
395 lines
12 KiB
Python
"""Step definitions for context_service_new_coverage.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.application.services.context_service import ContextService
|
|
from cleveragents.core.exceptions import ConfigurationError
|
|
from cleveragents.domain.models.core import (
|
|
Context as CtxModel,
|
|
)
|
|
from cleveragents.domain.models.core import (
|
|
ContextType,
|
|
Plan,
|
|
Project,
|
|
)
|
|
|
|
|
|
def _make_project(
|
|
project_id: int | None = 1,
|
|
) -> Project:
|
|
return Project(
|
|
id=project_id,
|
|
name="test-project",
|
|
path=Path("/tmp/test-project"),
|
|
)
|
|
|
|
|
|
def _make_plan(plan_id: int = 10) -> Plan:
|
|
plan = MagicMock(spec=Plan)
|
|
plan.id = plan_id
|
|
plan.name = "test-plan"
|
|
return plan
|
|
|
|
|
|
def _make_uow(
|
|
*, plan: Plan | None = None, contexts: list[CtxModel] | None = None
|
|
) -> MagicMock:
|
|
uow = MagicMock()
|
|
ctx_mgr = MagicMock()
|
|
ctx_mgr.plans.get_current_for_project.return_value = plan
|
|
ctx_mgr.contexts.get_for_plan.return_value = contexts or []
|
|
uow.transaction.return_value.__enter__ = MagicMock(return_value=ctx_mgr)
|
|
uow.transaction.return_value.__exit__ = MagicMock(return_value=False)
|
|
return uow
|
|
|
|
|
|
def _make_settings(**overrides: Any) -> MagicMock:
|
|
settings = MagicMock()
|
|
settings.is_langsmith_enabled = overrides.get("langsmith", False)
|
|
settings.build_langsmith_config.return_value = overrides.get("langsmith_config", {})
|
|
settings.cleveragents_max_context_size = 50 * 1024 * 1024
|
|
return settings
|
|
|
|
|
|
@given("a context service with mocked dependencies")
|
|
def step_context_service(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
context.uow = _make_uow()
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@given("a project without an id")
|
|
def step_project_no_id(context: Context) -> None:
|
|
context.project = _make_project(project_id=None)
|
|
|
|
|
|
@when("I get the current plan")
|
|
def step_get_plan(context: Context) -> None:
|
|
context.result = context.svc._get_current_plan(context.project)
|
|
|
|
|
|
@then("the context result should be None")
|
|
def step_assert_none(context: Context) -> None:
|
|
assert context.result is None
|
|
|
|
|
|
@given("a project with id {pid:d}")
|
|
def step_project_with_id(context: Context, pid: int) -> None:
|
|
context.project = _make_project(project_id=pid)
|
|
|
|
|
|
@when("I build langsmith config with langsmith disabled")
|
|
def step_build_langsmith_disabled(context: Context) -> None:
|
|
context.result = context.svc._build_langsmith_config(
|
|
context.project,
|
|
run_name="test",
|
|
file_paths=["a.py"],
|
|
mode="sync",
|
|
)
|
|
|
|
|
|
@then("the config should be an empty dict")
|
|
def step_assert_empty_config(context: Context) -> None:
|
|
assert context.result == {}
|
|
|
|
|
|
@given("a context service with langsmith enabled")
|
|
def step_svc_langsmith(context: Context) -> None:
|
|
context.settings = _make_settings(
|
|
langsmith=True,
|
|
langsmith_config={
|
|
"metadata": {"project_id": 1},
|
|
"tags": ["context-analysis"],
|
|
},
|
|
)
|
|
context.uow = _make_uow(plan=_make_plan())
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@when("I build langsmith config with langsmith enabled")
|
|
def step_build_langsmith_enabled(context: Context) -> None:
|
|
context.result = context.svc._build_langsmith_config(
|
|
context.project,
|
|
run_name="test",
|
|
file_paths=["a.py"],
|
|
mode="sync",
|
|
)
|
|
|
|
|
|
@then("the config should contain metadata and tags")
|
|
def step_assert_langsmith_config(context: Context) -> None:
|
|
assert isinstance(context.result, dict)
|
|
assert len(context.result) > 0
|
|
|
|
|
|
@when("I prepare analysis config")
|
|
def step_prepare_config(context: Context) -> None:
|
|
context.result = context.svc._prepare_analysis_config(
|
|
context.project,
|
|
run_name="test",
|
|
file_paths=["a.py"],
|
|
mode="sync",
|
|
)
|
|
|
|
|
|
@then("the config should have a configurable thread_id")
|
|
def step_assert_thread_id(context: Context) -> None:
|
|
assert "configurable" in context.result
|
|
assert "thread_id" in context.result["configurable"]
|
|
|
|
|
|
@given("a context service with no vector store")
|
|
def step_svc_no_vector(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
context.uow = _make_uow()
|
|
context.svc = ContextService(
|
|
context.settings, context.uow, vector_store_service=None
|
|
)
|
|
|
|
|
|
@when("I check vector store availability")
|
|
def step_check_vector(context: Context) -> None:
|
|
context.result = context.svc._vector_store_available()
|
|
|
|
|
|
@then("the vector store check should return False")
|
|
def step_assert_false(context: Context) -> None:
|
|
assert context.result is False
|
|
|
|
|
|
@when("I refresh vector index for plan {pid:d}")
|
|
def step_refresh_vector(context: Context, pid: int) -> None:
|
|
context.error = None
|
|
try:
|
|
context.svc._refresh_vector_index(pid)
|
|
except Exception as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then("no exception should be raised")
|
|
def step_assert_no_error(context: Context) -> None:
|
|
assert context.error is None
|
|
|
|
|
|
@given("a context service with a failing vector store")
|
|
def step_svc_failing_vector(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
context.uow = _make_uow()
|
|
vs = MagicMock()
|
|
vs.is_enabled.return_value = True
|
|
vs.refresh_for_plan.side_effect = ConfigurationError(message="vector store error")
|
|
context.svc = ContextService(context.settings, context.uow, vector_store_service=vs)
|
|
|
|
|
|
@when("I get the context agent")
|
|
def step_get_agent(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
context.result = context.svc._get_context_agent(
|
|
llm=FakeListLLM(
|
|
responses=["Dependencies: ['os']", "Relevance: High", "Summary: test"] * 3
|
|
),
|
|
)
|
|
|
|
|
|
@then("the context agent should be a ContextAnalysisAgent instance")
|
|
def step_assert_agent(context: Context) -> None:
|
|
from cleveragents.agents.graphs.context_analysis import ContextAnalysisAgent
|
|
|
|
assert isinstance(context.result, ContextAnalysisAgent)
|
|
|
|
|
|
@given("a project that has no context files")
|
|
def step_project_no_files(context: Context) -> None:
|
|
context.project = _make_project()
|
|
# list_files returns empty when no plan or empty contexts
|
|
context.uow = _make_uow(plan=None)
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@when("I analyze context")
|
|
def step_analyze_context(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
_fake_llm = FakeListLLM(
|
|
responses=["Dependencies: ['os']", "Relevance: High", "Summary: test"] * 3
|
|
)
|
|
context.result = context.svc.analyze_context(context.project, llm=_fake_llm)
|
|
|
|
|
|
@then("the result summary should indicate no files")
|
|
def step_assert_no_files_summary(context: Context) -> None:
|
|
assert (
|
|
"no context" in context.result["summary"].lower() or context.result["summary"]
|
|
)
|
|
|
|
|
|
@given("a project with context files")
|
|
def step_project_with_files(context: Context) -> None:
|
|
context.project = _make_project()
|
|
# Create a real temp file for the agent to load
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".py", delete=False, encoding="utf-8"
|
|
) as tmp:
|
|
tmp.write("import os\n\ndef main():\n pass\n")
|
|
tmp.flush()
|
|
context.temp_file = tmp.name
|
|
|
|
plan = _make_plan()
|
|
ctx_entry = CtxModel(
|
|
id=1,
|
|
plan_id=10,
|
|
path=tmp.name,
|
|
type=ContextType.FILE,
|
|
content="import os",
|
|
file_hash="abc123",
|
|
size=10,
|
|
)
|
|
context.uow = _make_uow(plan=plan, contexts=[ctx_entry])
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@then("the result should have documents and summary")
|
|
def step_assert_docs_summary(context: Context) -> None:
|
|
assert "summary" in context.result
|
|
|
|
|
|
@when("I retrieve the context summary")
|
|
def step_get_summary(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
_fake_llm = FakeListLLM(
|
|
responses=["Dependencies: ['os']", "Relevance: High", "Summary: test"] * 3
|
|
)
|
|
context.result = context.svc.get_context_summary(context.project, llm=_fake_llm)
|
|
|
|
|
|
@then("the context summary should be a nonempty string")
|
|
def step_assert_nonempty_summary(context: Context) -> None:
|
|
assert isinstance(context.result, str)
|
|
assert len(context.result) > 0
|
|
|
|
|
|
@when("I get context dependencies")
|
|
def step_get_deps(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
_fake_llm = FakeListLLM(
|
|
responses=["Dependencies: ['os']", "Relevance: High", "Summary: test"] * 3
|
|
)
|
|
context.result = context.svc.get_context_dependencies(
|
|
context.project, llm=_fake_llm
|
|
)
|
|
|
|
|
|
@then("the context deps result should be a dict")
|
|
def step_assert_dict(context: Context) -> None:
|
|
assert isinstance(context.result, dict)
|
|
|
|
|
|
@when("I fetch relevant files above threshold {t}")
|
|
def step_get_relevant(context: Context, t: str) -> None:
|
|
try:
|
|
context.result = context.svc.get_relevant_files(
|
|
context.project, threshold=float(t)
|
|
)
|
|
except Exception:
|
|
context.result = []
|
|
|
|
|
|
@then("the relevant files should be a list of tuples")
|
|
def step_assert_tuples(context: Context) -> None:
|
|
assert isinstance(context.result, list)
|
|
|
|
|
|
@when("I search context with empty query")
|
|
def step_search_empty(context: Context) -> None:
|
|
context.result = context.svc.search_context(context.project, " ")
|
|
|
|
|
|
@then("the context search result should be an empty list")
|
|
def step_assert_empty_list(context: Context) -> None:
|
|
assert context.result == []
|
|
|
|
|
|
@given("a project with id {pid:d} but no current plan")
|
|
def step_project_no_plan(context: Context, pid: int) -> None:
|
|
context.project = _make_project(project_id=pid)
|
|
context.uow = _make_uow(plan=None)
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@when('I search context with query "{query}"')
|
|
def step_search_query(context: Context, query: str) -> None:
|
|
context.result = context.svc.search_context(context.project, query)
|
|
|
|
|
|
@given("a project with id {pid:d} and a current plan")
|
|
def step_project_with_plan(context: Context, pid: int) -> None:
|
|
context.project = _make_project(project_id=pid)
|
|
# Only create new svc if one doesn't already exist (preserve vector store setup)
|
|
if not hasattr(context, "svc") or context.svc is None:
|
|
plan = _make_plan()
|
|
context.uow = _make_uow(plan=plan)
|
|
context.svc = ContextService(context.settings, context.uow)
|
|
|
|
|
|
@given("a context service with a working vector store")
|
|
def step_svc_vector(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
plan = _make_plan()
|
|
context.uow = _make_uow(plan=plan)
|
|
vs = MagicMock()
|
|
vs.is_enabled.return_value = True
|
|
vs.search.return_value = [{"path": "a.py", "score": 0.9}]
|
|
context.vs = vs
|
|
context.svc = ContextService(context.settings, context.uow, vector_store_service=vs)
|
|
|
|
|
|
@then("the vector store search should be called")
|
|
def step_assert_vs_called(context: Context) -> None:
|
|
context.vs.search.assert_called()
|
|
|
|
|
|
@given("a context service with a vector store that raises ConfigurationError")
|
|
def step_svc_vector_config_error(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
plan = _make_plan()
|
|
context.uow = _make_uow(plan=plan)
|
|
vs = MagicMock()
|
|
vs.is_enabled.return_value = True
|
|
vs.search.side_effect = ConfigurationError(message="config error")
|
|
context.svc = ContextService(context.settings, context.uow, vector_store_service=vs)
|
|
|
|
|
|
@when("I stream analyze context")
|
|
def step_stream_analyze(context: Context) -> None:
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
_fake_llm = FakeListLLM(
|
|
responses=["Dependencies: ['os']", "Relevance: High", "Summary: test"] * 3
|
|
)
|
|
context.events = list(
|
|
context.svc.analyze_context_streaming(context.project, llm=_fake_llm)
|
|
)
|
|
|
|
|
|
@then("the first event should indicate no files")
|
|
def step_assert_first_event_no_files(context: Context) -> None:
|
|
assert len(context.events) > 0
|
|
first = context.events[0]
|
|
assert "complete" in str(first.get("type", "")) or "summary" in first
|
|
|
|
|
|
@then("at least one streaming event should be yielded")
|
|
def step_assert_events_yielded(context: Context) -> None:
|
|
assert len(context.events) > 0
|