forked from HAL9000/cleveragents-core
225 lines
7.9 KiB
Python
225 lines
7.9 KiB
Python
"""
|
|
Direct step definitions for graph templates coverage tests.
|
|
These tests bypass existing mocks to ensure GraphTemplate code execution.
|
|
"""
|
|
|
|
import copy
|
|
from unittest.mock import Mock
|
|
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.templates.base import ComponentReference
|
|
from cleveragents.templates.base import InstantiationContext
|
|
from cleveragents.templates.base import TemplateType
|
|
from cleveragents.templates.graph_templates import GraphTemplate
|
|
|
|
|
|
@given("I have a direct test environment for graph templates")
|
|
def step_direct_environment_graph_templates(context):
|
|
"""Direct test environment setup for graph templates."""
|
|
context.direct_template = None
|
|
context.direct_params = {}
|
|
context.direct_context = None
|
|
context.direct_result = None
|
|
|
|
|
|
@given("I have a direct graph template instance")
|
|
def step_direct_graph_template_instance(context):
|
|
"""Create a direct graph template instance."""
|
|
definition = {
|
|
"name": "direct_test_graph",
|
|
"parameters": {"test_param": {"type": "string", "default": "test_value"}},
|
|
"nodes": {
|
|
"node1": {"type": "task", "config": {"task": "test_task"}},
|
|
"agent_node": {"type": "agent", "agent": "test_agent"},
|
|
"none_node": None,
|
|
},
|
|
"edges": [
|
|
{"from": "node1", "to": "agent_node"},
|
|
None,
|
|
{"from": "agent_node", "to": "node1", "condition": {"type": "test"}},
|
|
],
|
|
}
|
|
|
|
# Create a real GraphTemplate instance
|
|
context.direct_template = GraphTemplate(
|
|
"direct_test", TemplateType.GRAPH, definition
|
|
)
|
|
|
|
# Mock only the base class methods we need
|
|
context.direct_template.validate_params = Mock(
|
|
return_value={"test_param": "test_value"}
|
|
)
|
|
context.direct_template._apply_template_vars = Mock(side_effect=lambda x, y: x)
|
|
|
|
|
|
@given("I have a direct graph template instance with complex nodes")
|
|
def step_direct_graph_template_complex_nodes(context):
|
|
"""Create a direct graph template instance with complex nodes."""
|
|
definition = {
|
|
"name": "complex_nodes_graph",
|
|
"nodes": {
|
|
"agent_with_param": {"type": "agent", "agent": "param_ref"},
|
|
"agent_with_component": {"type": "agent", "agent": "component_ref"},
|
|
"agent_with_template_var": {"type": "agent", "agent": "{{template_var}}"},
|
|
"regular_node": {"type": "task", "config": {"task": "test"}},
|
|
"none_node": None,
|
|
},
|
|
}
|
|
|
|
context.direct_template = GraphTemplate(
|
|
"complex_nodes", TemplateType.GRAPH, definition
|
|
)
|
|
context.direct_template.validate_params = Mock(
|
|
return_value={"param_ref": "resolved_agent"}
|
|
)
|
|
context.direct_template._apply_template_vars = Mock(side_effect=lambda x, y: x)
|
|
|
|
|
|
@given("I have a direct graph template instance with complex edges")
|
|
def step_direct_graph_template_complex_edges(context):
|
|
"""Create a direct graph template instance with complex edges."""
|
|
definition = {
|
|
"name": "complex_edges_graph",
|
|
"nodes": {
|
|
"node1": {"type": "task", "config": {"task": "test1"}},
|
|
"node2": {"type": "task", "config": {"task": "test2"}},
|
|
},
|
|
"edges": [
|
|
{"from": "node1", "to": "node2"},
|
|
None,
|
|
{
|
|
"from": "node2",
|
|
"to": "node1",
|
|
"condition": {"type": "{{condition_type}}"},
|
|
},
|
|
{"from": "node1", "to": "node2", "condition": None},
|
|
],
|
|
}
|
|
|
|
context.direct_template = GraphTemplate(
|
|
"complex_edges", TemplateType.GRAPH, definition
|
|
)
|
|
context.direct_template.validate_params = Mock(
|
|
return_value={"condition_type": "test"}
|
|
)
|
|
context.direct_template._apply_template_vars = Mock(side_effect=lambda x, y: x)
|
|
|
|
|
|
@given("I have direct test parameters")
|
|
def step_direct_test_parameters(context):
|
|
"""Setup direct test parameters."""
|
|
context.direct_params = {
|
|
"test_param": "test_value",
|
|
"param_ref": "resolved_agent",
|
|
"condition_type": "resolved_condition",
|
|
}
|
|
|
|
|
|
@given("I have direct test parameters with agent references")
|
|
def step_direct_test_parameters_with_agent_references(context):
|
|
"""Setup direct test parameters with agent references."""
|
|
context.direct_params = {"param_ref": "resolved_agent_name"}
|
|
|
|
|
|
@given("I have a direct instantiation context")
|
|
def step_direct_instantiation_context(context):
|
|
"""Setup direct instantiation context."""
|
|
context.direct_context = Mock(spec=InstantiationContext)
|
|
context.direct_context.resolve_reference.return_value = None
|
|
|
|
|
|
@given("I have a direct instantiation context with agents")
|
|
def step_direct_instantiation_context_with_agents(context):
|
|
"""Setup direct instantiation context with agents."""
|
|
context.direct_context = Mock(spec=InstantiationContext)
|
|
agent_config = {"type": "llm", "config": {"model": "test-model"}}
|
|
context.direct_context.resolve_reference.return_value = agent_config
|
|
|
|
|
|
@when("I directly instantiate the graph template")
|
|
def step_directly_instantiate_graph_template(context):
|
|
"""Directly instantiate the graph template."""
|
|
registry = Mock()
|
|
context.direct_result = context.direct_template.instantiate(
|
|
context.direct_params, registry, context.direct_context
|
|
)
|
|
|
|
|
|
@when("I directly call _process_nodes")
|
|
def step_directly_call_process_nodes(context):
|
|
"""Directly call the _process_nodes method."""
|
|
nodes = context.direct_template.definition["nodes"]
|
|
context.direct_result = context.direct_template._process_nodes(
|
|
nodes, context.direct_params, context.direct_context
|
|
)
|
|
|
|
|
|
@when("I directly call _process_edges")
|
|
def step_directly_call_process_edges(context):
|
|
"""Directly call the _process_edges method."""
|
|
edges = context.direct_template.definition["edges"]
|
|
context.direct_result = context.direct_template._process_edges(
|
|
edges, context.direct_params
|
|
)
|
|
|
|
|
|
@then("the direct instantiation should succeed")
|
|
def step_direct_instantiation_should_succeed(context):
|
|
"""Verify direct instantiation succeeded."""
|
|
assert context.direct_result is not None
|
|
assert isinstance(context.direct_result, dict)
|
|
assert "name" in context.direct_result
|
|
|
|
|
|
@then("all code paths should be executed")
|
|
def step_all_code_paths_should_be_executed(context):
|
|
"""Verify all code paths were executed."""
|
|
# Check that the main logic was executed
|
|
assert context.direct_result is not None
|
|
|
|
# Check nodes were processed
|
|
if "nodes" in context.direct_result:
|
|
# None nodes should be filtered out
|
|
assert "none_node" not in context.direct_result["nodes"]
|
|
# Regular nodes should be present
|
|
assert "node1" in context.direct_result["nodes"]
|
|
|
|
# Check edges were processed
|
|
if "edges" in context.direct_result:
|
|
# None edges should be filtered out
|
|
edges = context.direct_result["edges"]
|
|
assert all(edge is not None for edge in edges)
|
|
|
|
|
|
@then("node processing should execute all branches")
|
|
def step_node_processing_should_execute_all_branches(context):
|
|
"""Verify node processing executed all branches."""
|
|
assert context.direct_result is not None
|
|
|
|
# None nodes should be filtered out
|
|
assert "none_node" not in context.direct_result
|
|
|
|
# Regular nodes should be present
|
|
assert "regular_node" in context.direct_result
|
|
|
|
# Agent nodes should be processed
|
|
assert "agent_with_param" in context.direct_result
|
|
assert "agent_with_component" in context.direct_result
|
|
assert "agent_with_template_var" in context.direct_result
|
|
|
|
|
|
@then("edge processing should execute all branches")
|
|
def step_edge_processing_should_execute_all_branches(context):
|
|
"""Verify edge processing executed all branches."""
|
|
assert context.direct_result is not None
|
|
assert isinstance(context.direct_result, list)
|
|
|
|
# None edges should be filtered out
|
|
assert all(edge is not None for edge in context.direct_result)
|
|
|
|
# Should have processed edges with and without conditions
|
|
assert len(context.direct_result) > 0
|