Files
temp/tests/features/steps/graph_templates_direct_steps.py

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