forked from cleveragents/cleveragents-core
91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.agents.graphs.auto_debug import AutoDebugAgent, AutoDebugState
|
|
|
|
|
|
class _SequentialLLM:
|
|
"""Stub LLM that returns sequential responses with a content attribute."""
|
|
|
|
def __init__(self, responses: list[str]):
|
|
self._responses = list(responses)
|
|
self.invocations: list[list[Any]] = []
|
|
|
|
def invoke(self, messages):
|
|
self.invocations.append(messages)
|
|
if not self._responses:
|
|
raise AssertionError("No responses left in stub LLM")
|
|
content = self._responses.pop(0)
|
|
return type("Response", (), {"content": content})()
|
|
|
|
|
|
@given("a custom auto debug LLM yields JSON responses")
|
|
def step_custom_llm(context):
|
|
context.custom_llm = _SequentialLLM(
|
|
[
|
|
"Custom analysis output",
|
|
json.dumps(
|
|
{
|
|
"description": "JSON described fix",
|
|
"code": "# patched code",
|
|
"files_to_modify": ["file.py"],
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"is_valid": True,
|
|
"reasoning": "Structured JSON shows success",
|
|
"issues": [],
|
|
}
|
|
),
|
|
]
|
|
)
|
|
|
|
|
|
@when("I run the auto debug agent graph with sample state")
|
|
def step_run_auto_debug_graph(context):
|
|
context.agent = AutoDebugAgent(llm=context.custom_llm, max_fix_attempts=2)
|
|
|
|
input_state: AutoDebugState = {
|
|
"messages": [],
|
|
"context": {},
|
|
"result": None,
|
|
"error": None,
|
|
"metadata": {},
|
|
"error_message": "NameError: missing value",
|
|
"code_context": "value = missing",
|
|
"attempted_fixes": [],
|
|
"current_fix": {},
|
|
"fix_validated": False,
|
|
}
|
|
|
|
context.final_state = context.agent.invoke(input_state)
|
|
|
|
|
|
@then("the provided LLM instance should be used")
|
|
def step_assert_llm_used(context):
|
|
assert context.agent.llm is context.custom_llm
|
|
assert len(context.custom_llm.invocations) == 3
|
|
|
|
|
|
@then("the fix JSON should populate the current fix")
|
|
def step_assert_fix_parsed(context):
|
|
fix = context.final_state["current_fix"]
|
|
assert fix["description"] == "JSON described fix"
|
|
assert fix["code"] == "# patched code"
|
|
assert fix["files_to_modify"] == ["file.py"]
|
|
|
|
|
|
@then("the validation JSON should mark the fix as valid")
|
|
def step_assert_validation_parsed(context):
|
|
assert context.final_state["fix_validated"] is True
|
|
result = context.final_state.get("result")
|
|
assert result is not None
|
|
assert result["success"] is True
|
|
assert result["fix"] == context.final_state["current_fix"]
|
|
assert result["attempts"] == 0
|