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
454 lines
17 KiB
Python
454 lines
17 KiB
Python
"""Behave steps for PlanGenerationGraph LangGraph coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from langchain_community.llms import FakeListLLM
|
|
|
|
PLAN_GEN_MODULE_PATH = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "src"
|
|
/ "cleveragents"
|
|
/ "agents"
|
|
/ "plan_generation.py"
|
|
)
|
|
|
|
|
|
def _default_test_llm() -> FakeListLLM:
|
|
"""Create a FakeListLLM for test purposes."""
|
|
return FakeListLLM(
|
|
responses=[
|
|
"Requirements: Add error handling with try-except blocks",
|
|
"Generated code with proper error handling implementation",
|
|
"Validation passed: Code follows best practices",
|
|
]
|
|
)
|
|
|
|
|
|
def _load_plan_generation_module(context: Any) -> None:
|
|
"""Load the plan_generation module dynamically."""
|
|
if hasattr(context, "plan_generation_module"):
|
|
return
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"cleveragents.agents.plan_generation", PLAN_GEN_MODULE_PATH
|
|
)
|
|
if spec and spec.loader:
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules["cleveragents.agents.plan_generation"] = module
|
|
spec.loader.exec_module(module)
|
|
context.plan_generation_module = module
|
|
|
|
|
|
@given("the langgraph plan generation module is importable")
|
|
def step_langgraph_module_importable(context: Any) -> None:
|
|
"""Ensure the plan generation module can be imported."""
|
|
_load_plan_generation_module(context)
|
|
assert hasattr(context, "plan_generation_module")
|
|
assert hasattr(context.plan_generation_module, "PlanGenerationGraph")
|
|
|
|
|
|
@when("I create a langgraph PlanGenerationGraph with no LLM")
|
|
def step_create_langgraph_graph_no_llm(context: Any) -> None:
|
|
"""Create graph with explicit test LLM."""
|
|
_load_plan_generation_module(context)
|
|
PlanGenerationGraph = context.plan_generation_module.PlanGenerationGraph
|
|
context.graph = PlanGenerationGraph(llm=_default_test_llm())
|
|
|
|
|
|
@when("I create a langgraph PlanGenerationGraph with max_retries of {retries:d}")
|
|
def step_create_langgraph_graph_with_retries(context: Any, retries: int) -> None:
|
|
"""Create graph with custom max_retries."""
|
|
_load_plan_generation_module(context)
|
|
PlanGenerationGraph = context.plan_generation_module.PlanGenerationGraph
|
|
context.graph = PlanGenerationGraph(llm=_default_test_llm(), max_retries=retries)
|
|
|
|
|
|
@then("the langgraph graph should be initialized successfully")
|
|
def step_langgraph_graph_initialized(context: Any) -> None:
|
|
"""Verify graph is initialized."""
|
|
assert context.graph is not None
|
|
assert hasattr(context.graph, "llm")
|
|
assert hasattr(context.graph, "graph")
|
|
assert hasattr(context.graph, "app")
|
|
|
|
|
|
@then("the langgraph graph should have a default FakeListLLM configured")
|
|
def step_langgraph_graph_has_fake_llm(context: Any) -> None:
|
|
"""Verify FakeListLLM is used when explicitly provided."""
|
|
assert isinstance(context.graph.llm, FakeListLLM)
|
|
|
|
|
|
@then("the langgraph graph should have max_retries set to {retries:d}")
|
|
def step_langgraph_graph_max_retries(context: Any, retries: int) -> None:
|
|
"""Verify max_retries value."""
|
|
assert context.graph.max_retries == retries
|
|
|
|
|
|
@then("the langgraph graph max_retries should be {retries:d}")
|
|
def step_verify_langgraph_max_retries(context: Any, retries: int) -> None:
|
|
"""Verify max_retries value."""
|
|
assert context.graph.max_retries == retries
|
|
|
|
|
|
@then("the langgraph graph should have an analyze_prompt template")
|
|
def step_has_langgraph_analyze_prompt(context: Any) -> None:
|
|
"""Verify analyze_prompt exists."""
|
|
assert hasattr(context.graph, "analyze_prompt")
|
|
assert context.graph.analyze_prompt is not None
|
|
|
|
|
|
@then("the langgraph graph should have a generate_prompt template")
|
|
def step_has_langgraph_generate_prompt(context: Any) -> None:
|
|
"""Verify generate_prompt exists."""
|
|
assert hasattr(context.graph, "generate_prompt")
|
|
assert context.graph.generate_prompt is not None
|
|
|
|
|
|
@then("the langgraph graph should have a validate_prompt template")
|
|
def step_has_langgraph_validate_prompt(context: Any) -> None:
|
|
"""Verify validate_prompt exists."""
|
|
assert hasattr(context.graph, "validate_prompt")
|
|
assert context.graph.validate_prompt is not None
|
|
|
|
|
|
@then('the langgraph workflow graph should contain node "{node_name}"')
|
|
def step_langgraph_graph_has_node(context: Any, node_name: str) -> None:
|
|
"""Verify graph has specific node."""
|
|
nodes = context.graph.graph.nodes
|
|
assert node_name in nodes
|
|
|
|
|
|
@given("I have a langgraph PlanGenerationGraph instance")
|
|
def step_have_langgraph_graph_instance(context: Any) -> None:
|
|
"""Create a PlanGenerationGraph instance."""
|
|
_load_plan_generation_module(context)
|
|
PlanGenerationGraph = context.plan_generation_module.PlanGenerationGraph
|
|
context.graph = PlanGenerationGraph(llm=_default_test_llm())
|
|
|
|
|
|
@when("I format the langgraph context summary with no contexts")
|
|
def step_format_langgraph_summary_no_contexts(context: Any) -> None:
|
|
"""Format context summary with empty list."""
|
|
summary = context.graph._format_context_summary([])
|
|
context.summary = summary
|
|
|
|
|
|
@when("I format the langgraph context summary with {count:d} contexts")
|
|
def step_format_langgraph_summary_n_contexts(context: Any, count: int) -> None:
|
|
"""Format context summary with N contexts."""
|
|
from cleveragents.domain.models.core import Context
|
|
|
|
contexts = [
|
|
Context(
|
|
plan_id=1,
|
|
path=f"file{i}.py",
|
|
content=f"# File {i} content\n" * 30,
|
|
)
|
|
for i in range(count)
|
|
]
|
|
summary = context.graph._format_context_summary(contexts)
|
|
context.summary = summary
|
|
context.context_count = count
|
|
|
|
|
|
@then('the langgraph summary should be "{expected}"')
|
|
def step_langgraph_summary_is(context: Any, expected: str) -> None:
|
|
"""Verify exact summary text."""
|
|
assert context.summary == expected
|
|
|
|
|
|
@then("the langgraph summary should include all {count:d} file paths")
|
|
def step_langgraph_summary_includes_files(context: Any, count: int) -> None:
|
|
"""Verify summary includes all files."""
|
|
expected = min(count, 5) # Max 5 files shown
|
|
for i in range(expected):
|
|
assert f"file{i}.py" in context.summary
|
|
|
|
|
|
@then('the langgraph summary should indicate "and {count:d} more files"')
|
|
def step_langgraph_summary_more_files(context: Any, count: int) -> None:
|
|
"""Verify 'more files' indicator."""
|
|
assert f"{count} more files" in context.summary
|
|
|
|
|
|
@when("I execute the langgraph load_context node")
|
|
def step_execute_langgraph_load_context(context: Any) -> None:
|
|
"""Execute load_context node."""
|
|
state: dict[str, Any] = {}
|
|
result = context.graph._load_context(state)
|
|
context.node_result = result
|
|
|
|
|
|
@when("I execute the langgraph load_context node with sample contexts")
|
|
def step_execute_langgraph_load_context_with_samples(context: Any) -> None:
|
|
"""Execute load_context node with example contexts."""
|
|
from cleveragents.domain.models.core import Context
|
|
|
|
contexts = [
|
|
Context(
|
|
plan_id=1,
|
|
path="src/app.py",
|
|
content="def app():\n return 1",
|
|
),
|
|
Context(plan_id=1, path="src/utils.py", content="VALUE = 42"),
|
|
]
|
|
state: dict[str, Any] = {"contexts": contexts}
|
|
context.node_result = context.graph._load_context(state)
|
|
|
|
|
|
@then("the langgraph node result should have retry_count set to {count:d}")
|
|
def step_langgraph_node_retry_count(context: Any, count: int) -> None:
|
|
"""Verify retry_count in result."""
|
|
assert context.node_result.get("retry_count") == count
|
|
|
|
|
|
@then("the langgraph node result should have error set to None")
|
|
def step_langgraph_node_error_none(context: Any) -> None:
|
|
"""Verify error is None."""
|
|
assert context.node_result.get("error") is None
|
|
|
|
|
|
@then("the langgraph node result should include context metadata defaults")
|
|
def step_langgraph_node_defaults(context: Any) -> None:
|
|
"""Verify context metadata defaults are present."""
|
|
assert context.node_result.get("context_summary") == "No context files provided"
|
|
assert context.node_result.get("context_dependencies") == {}
|
|
assert context.node_result.get("context_relevance") == {}
|
|
assert context.node_result.get("context_analysis_error") is None
|
|
|
|
|
|
@then("the langgraph node result should include an analyzed context summary")
|
|
def step_langgraph_node_has_summary(context: Any) -> None:
|
|
"""Ensure context summary is populated from analysis."""
|
|
summary = context.node_result.get("context_summary", "")
|
|
assert summary
|
|
assert summary != "No context files provided"
|
|
|
|
|
|
@then("the langgraph node result should include context dependencies")
|
|
def step_langgraph_node_has_dependencies(context: Any) -> None:
|
|
"""Ensure dependency metadata exists."""
|
|
deps = context.node_result.get("context_dependencies")
|
|
assert isinstance(deps, dict)
|
|
assert deps
|
|
|
|
|
|
@given("I have a langgraph PlanGenerationGraph instance with max_retries {retries:d}")
|
|
def step_langgraph_graph_with_max_retries(context: Any, retries: int) -> None:
|
|
"""Create graph with specific max_retries."""
|
|
_load_plan_generation_module(context)
|
|
PlanGenerationGraph = context.plan_generation_module.PlanGenerationGraph
|
|
context.graph = PlanGenerationGraph(llm=_default_test_llm(), max_retries=retries)
|
|
|
|
|
|
@when(
|
|
"I check langgraph should_retry with {status} validation and retry_count {count:d}"
|
|
)
|
|
def step_check_langgraph_should_retry(context: Any, status: str, count: int) -> None:
|
|
"""Check should_retry decision."""
|
|
state: dict[str, Any] = {
|
|
"validation_result": {"status": status},
|
|
"retry_count": count,
|
|
}
|
|
decision = context.graph._should_retry(state)
|
|
context.retry_decision = decision
|
|
context.final_retry_count = state.get("retry_count", count)
|
|
|
|
|
|
@then('the langgraph retry decision should be "{decision}"')
|
|
def step_langgraph_decision_is(context: Any, decision: str) -> None:
|
|
"""Verify retry decision."""
|
|
assert context.retry_decision == decision
|
|
|
|
|
|
@when("I execute the langgraph validate node with no changes")
|
|
def step_execute_langgraph_validate_no_changes(context: Any) -> None:
|
|
"""Execute validate with no changes."""
|
|
state: dict[str, Any] = {"generated_changes": []}
|
|
result = context.graph._validate(state)
|
|
context.node_result = result
|
|
|
|
|
|
@then('the langgraph validation status should be "{status}"')
|
|
def step_langgraph_validation_status(context: Any, status: str) -> None:
|
|
"""Verify validation status."""
|
|
validation = context.node_result.get("validation_result", {})
|
|
assert validation.get("status") == status
|
|
|
|
|
|
@then('the langgraph validation message should contain "{text}"')
|
|
def step_langgraph_validation_message_contains(context: Any, text: str) -> None:
|
|
"""Verify validation message contains text."""
|
|
validation = context.node_result.get("validation_result", {})
|
|
message = validation.get("message", "")
|
|
assert text in message
|
|
|
|
|
|
@when("I execute the langgraph generate_plan node with no requirements")
|
|
def step_execute_langgraph_generate_no_requirements(context: Any) -> None:
|
|
"""Execute generate_plan with no requirements."""
|
|
state: dict[str, Any] = {"analyzed_requirements": {}}
|
|
result = context.graph._generate_plan(state)
|
|
context.node_result = result
|
|
|
|
|
|
@then("the langgraph generated_changes should be empty")
|
|
def step_langgraph_changes_empty(context: Any) -> None:
|
|
"""Verify changes list is empty."""
|
|
changes = context.node_result.get("generated_changes", [])
|
|
assert len(changes) == 0
|
|
|
|
|
|
@when(
|
|
"I execute the langgraph analyze_requirements node with a flaky LLM that fails once"
|
|
)
|
|
def step_langgraph_analyze_with_flaky_llm(context: Any) -> None:
|
|
"""Execute analyze_requirements with a flaky LLM."""
|
|
from cleveragents.domain.models.core import Context as PlanContext
|
|
|
|
class FlakyLLM(FakeListLLM):
|
|
def __init__(self) -> None:
|
|
super().__init__(responses=["Requirements succeeded after retry"])
|
|
object.__setattr__(self, "_call_count", 0)
|
|
|
|
@property
|
|
def call_count(self) -> int:
|
|
return getattr(self, "_call_count", 0)
|
|
|
|
def _call(self, prompt: str, stop: list[str] | None = None) -> str:
|
|
object.__setattr__(self, "_call_count", self.call_count + 1)
|
|
if self.call_count == 1:
|
|
raise RuntimeError("transient failure")
|
|
return super()._call(prompt, stop=stop)
|
|
|
|
original_llm = context.graph.llm
|
|
flaky_llm = FlakyLLM()
|
|
context.graph.llm = flaky_llm
|
|
contexts = [
|
|
PlanContext(plan_id=1, path="retry.py", content="print('retry')"),
|
|
]
|
|
state: dict[str, Any] = {
|
|
"prompt": "Add retry support",
|
|
"contexts": contexts,
|
|
"context_summary": "",
|
|
}
|
|
try:
|
|
context.node_result = context.graph._analyze_requirements(state)
|
|
finally:
|
|
context.graph.llm = original_llm
|
|
context.flaky_llm_calls = flaky_llm.call_count
|
|
|
|
|
|
@then("the langgraph analyze node should succeed after retry")
|
|
def step_langgraph_analyze_retry_success(context: Any) -> None:
|
|
"""Verify analyze_requirements succeeded after retry."""
|
|
result = context.node_result
|
|
assert result.get("analyzed_requirements")
|
|
assert not result.get("error")
|
|
assert context.flaky_llm_calls >= 2
|
|
|
|
|
|
@then('the langgraph error should contain "{text}"')
|
|
def step_langgraph_error_contains(context: Any, text: str) -> None:
|
|
"""Verify error message contains text."""
|
|
error = context.node_result.get("error")
|
|
assert error is not None
|
|
assert text in error
|
|
|
|
|
|
@given("I have langgraph workflow inputs with project plan and contexts")
|
|
def step_have_langgraph_workflow_inputs(context: Any) -> None:
|
|
"""Create workflow inputs."""
|
|
from cleveragents.domain.models.core import Context, Plan, Project
|
|
|
|
context.project = Project(id=1, name="test_project", path=Path("/tmp/test"))
|
|
context.plan = Plan(id=1, project_id=1, name="test_plan", prompt="Test")
|
|
context.contexts = [Context(plan_id=1, path="test.py", content="# test")]
|
|
|
|
|
|
@when("I invoke the langgraph workflow synchronously")
|
|
def step_invoke_langgraph_workflow_sync(context: Any) -> None:
|
|
"""Invoke workflow synchronously."""
|
|
result = context.graph.invoke(context.project, context.plan, context.contexts)
|
|
context.result = result
|
|
|
|
|
|
@then("the langgraph workflow result should contain all expected fields")
|
|
def step_langgraph_result_has_all_fields(context: Any) -> None:
|
|
"""Verify all expected fields."""
|
|
expected_fields = [
|
|
"project",
|
|
"plan",
|
|
"contexts",
|
|
"context_summary",
|
|
"context_dependencies",
|
|
"context_relevance",
|
|
"context_analysis_error",
|
|
"prompt",
|
|
"analyzed_requirements",
|
|
"generated_changes",
|
|
"validation_result",
|
|
"retry_count",
|
|
"error",
|
|
]
|
|
for field in expected_fields:
|
|
assert field in context.result
|
|
|
|
|
|
@when("I stream the langgraph workflow execution")
|
|
def step_stream_langgraph_workflow(context: Any) -> None:
|
|
"""Stream workflow execution."""
|
|
events = list(context.graph.stream(context.project, context.plan, context.contexts))
|
|
context.stream_events = events
|
|
|
|
|
|
@then("the langgraph stream should yield multiple events")
|
|
def step_langgraph_stream_yields_events(context: Any) -> None:
|
|
"""Verify stream yields events."""
|
|
assert len(context.stream_events) > 0
|
|
|
|
|
|
@given("the langgraph graphs package is importable")
|
|
def step_langgraph_graphs_package_importable(context: Any) -> None:
|
|
"""Import the graphs package for LangGraph workflows."""
|
|
context.langgraph_graphs_package = importlib.import_module(
|
|
"cleveragents.agents.graphs"
|
|
)
|
|
|
|
|
|
@then('the langgraph graphs exports should include "{symbol}"')
|
|
def step_langgraph_graphs_exports_include(context: Any, symbol: str) -> None:
|
|
"""Verify the graphs package exports include the symbol."""
|
|
package = getattr(context, "langgraph_graphs_package", None)
|
|
assert package is not None, "LangGraph graphs package not imported"
|
|
exports = getattr(package, "__all__", [])
|
|
assert symbol in exports, f"{symbol} not listed in __all__"
|
|
assert getattr(package, symbol, None) is not None, (
|
|
f"Package missing attribute {symbol}"
|
|
)
|
|
|
|
|
|
@given("the agents package is importable")
|
|
def step_agents_package_importable(context: Any) -> None:
|
|
"""Import the top-level agents package."""
|
|
context.agents_package = importlib.import_module("cleveragents.agents")
|
|
|
|
|
|
@then('the agents package exports should include "{symbol}"')
|
|
def step_agents_package_exports_include(context: Any, symbol: str) -> None:
|
|
"""Verify the agents package exports include the symbol."""
|
|
package = getattr(context, "agents_package", None)
|
|
assert package is not None, "Agents package was not imported"
|
|
exports = getattr(package, "__all__", [])
|
|
assert symbol in exports, f"{symbol} not in agents __all__"
|
|
assert getattr(package, symbol, None) is not None, (
|
|
f"Agents package missing attribute {symbol}"
|
|
)
|