forked from HAL9000/cleveragents-core
1235 lines
44 KiB
Python
1235 lines
44 KiB
Python
"""
|
|
Step definitions for AgentTemplate and CompositeAgentTemplate testing.
|
|
"""
|
|
|
|
import copy
|
|
from typing import Any
|
|
from typing import Dict
|
|
from unittest.mock import MagicMock
|
|
from unittest.mock import Mock
|
|
from unittest.mock import patch
|
|
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.templates.agent_templates import AgentTemplate
|
|
from cleveragents.templates.agent_templates import CompositeAgentTemplate
|
|
from cleveragents.templates.base import BaseTemplate
|
|
from cleveragents.templates.base import ComponentReference
|
|
from cleveragents.templates.base import InstantiationContext
|
|
from cleveragents.templates.base import TemplateType
|
|
from cleveragents.templates.registry import TemplateRegistry
|
|
|
|
|
|
# Mock template classes for testing
|
|
class MockAgentTemplate(BaseTemplate):
|
|
"""Mock agent template for testing."""
|
|
|
|
def __init__(
|
|
self, name: str, template_type: TemplateType, definition: Dict[str, Any]
|
|
):
|
|
super().__init__(name, template_type, definition)
|
|
self.instantiate_calls = []
|
|
|
|
def instantiate(
|
|
self, params: Dict[str, Any], registry, context: InstantiationContext
|
|
) -> Any:
|
|
self.instantiate_calls.append((params, registry, context))
|
|
return {
|
|
"name": self.name,
|
|
"type": self.definition.get("type", "llm"),
|
|
"config": self.definition.get("config", {}),
|
|
"instantiated": f"agent:{self.name}",
|
|
"params": params,
|
|
}
|
|
|
|
|
|
class MockGraphTemplate(BaseTemplate):
|
|
"""Mock graph template for testing."""
|
|
|
|
def __init__(
|
|
self, name: str, template_type: TemplateType, definition: Dict[str, Any]
|
|
):
|
|
super().__init__(name, template_type, definition)
|
|
self.instantiate_calls = []
|
|
|
|
def instantiate(
|
|
self, params: Dict[str, Any], registry, context: InstantiationContext
|
|
) -> Any:
|
|
self.instantiate_calls.append((params, registry, context))
|
|
return {
|
|
"name": self.name,
|
|
"type": "graph",
|
|
"config": self.definition.get("config", {}),
|
|
"instantiated": f"graph:{self.name}",
|
|
"params": params,
|
|
}
|
|
|
|
|
|
class MockStreamTemplate(BaseTemplate):
|
|
"""Mock stream template for testing."""
|
|
|
|
def __init__(
|
|
self, name: str, template_type: TemplateType, definition: Dict[str, Any]
|
|
):
|
|
super().__init__(name, template_type, definition)
|
|
self.instantiate_calls = []
|
|
|
|
def instantiate(
|
|
self, params: Dict[str, Any], registry, context: InstantiationContext
|
|
) -> Any:
|
|
self.instantiate_calls.append((params, registry, context))
|
|
return {
|
|
"name": self.name,
|
|
"type": "stream",
|
|
"config": self.definition.get("config", {}),
|
|
"instantiated": f"stream:{self.name}",
|
|
"params": params,
|
|
}
|
|
|
|
|
|
@given("I have a clean test environment for agent templates")
|
|
def step_clean_environment_agent_templates(context):
|
|
"""Set up clean test environment for agent templates."""
|
|
context.registry = None
|
|
context.agent_template = None
|
|
context.composite_template = None
|
|
context.template_definition = {}
|
|
context.template_params = {}
|
|
context.instantiation_context = None
|
|
context.result = None
|
|
context.error = None
|
|
context.original_definition = None
|
|
context.graph_definition = {}
|
|
|
|
|
|
@given("I have an agent template registry")
|
|
def step_agent_template_registry(context):
|
|
"""Create agent template registry."""
|
|
context.registry = TemplateRegistry()
|
|
|
|
|
|
@given("I have a composite agent template registry")
|
|
def step_composite_agent_template_registry(context):
|
|
"""Create composite agent template registry."""
|
|
context.registry = TemplateRegistry()
|
|
|
|
|
|
@given("I have a basic agent template")
|
|
def step_basic_agent_template(context):
|
|
"""Create basic agent template."""
|
|
context.template_definition = {
|
|
"type": "llm",
|
|
"config": {"model": "gpt-3.5-turbo", "temperature": 0.7},
|
|
"parameters": {"required": ["model"], "optional": ["temperature"]},
|
|
}
|
|
context.original_definition = copy.deepcopy(context.template_definition)
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have an agent template without type")
|
|
def step_agent_template_without_type(context):
|
|
"""Create agent template without type."""
|
|
context.template_definition = {
|
|
"config": {"model": "gpt-3.5-turbo", "temperature": 0.7}
|
|
}
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have an agent template with custom type")
|
|
def step_agent_template_with_custom_type(context):
|
|
"""Create agent template with custom type."""
|
|
context.template_definition = {
|
|
"type": "custom_tool",
|
|
"config": {"tool_name": "calculator"},
|
|
}
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have an agent template with required parameters")
|
|
def step_agent_template_with_required_parameters(context):
|
|
"""Create agent template with required parameters."""
|
|
context.template_definition = {
|
|
"type": "llm",
|
|
"config": {"model": "{{model}}", "temperature": "{{temperature}}"},
|
|
"parameters": {
|
|
"model": {
|
|
"description": "The model to use",
|
|
"required": True,
|
|
"type": "string",
|
|
},
|
|
"temperature": {
|
|
"description": "Temperature setting",
|
|
"required": False,
|
|
"default": 0.5,
|
|
"type": "float",
|
|
},
|
|
},
|
|
}
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have an agent template with template variables")
|
|
def step_agent_template_with_template_variables(context):
|
|
"""Create agent template with template variables."""
|
|
context.template_definition = {
|
|
"type": "llm",
|
|
"config": {
|
|
"model": "{{model}}",
|
|
"temperature": "{{temperature}}",
|
|
"system_prompt": "You are a {{role}} assistant",
|
|
},
|
|
}
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have an agent template")
|
|
def step_agent_template(context):
|
|
"""Create generic agent template."""
|
|
context.template_definition = {"type": "llm", "config": {"model": "gpt-3.5-turbo"}}
|
|
context.original_definition = copy.deepcopy(context.template_definition)
|
|
context.agent_template = AgentTemplate(
|
|
"test_agent", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a basic composite agent template")
|
|
def step_basic_composite_agent_template(context):
|
|
"""Create basic composite agent template."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {"agent1": {"type": "llm", "config": {"model": "gpt-3.5-turbo"}}},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
"routing": {"default": "agent1"},
|
|
"parameters": {"required": [], "optional": []},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template")
|
|
def step_composite_agent_template(context):
|
|
"""Create generic composite agent template."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {"agents": {}, "graphs": {}, "streams": {}},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with required parameters")
|
|
def step_composite_agent_template_with_required_parameters(context):
|
|
"""Create composite agent template with required parameters."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {"agents": {}, "graphs": {}, "streams": {}},
|
|
"parameters": {
|
|
"coordination_type": {
|
|
"description": "Type of coordination",
|
|
"required": True,
|
|
"type": "string",
|
|
},
|
|
"timeout": {
|
|
"description": "Timeout setting",
|
|
"required": False,
|
|
"default": 30,
|
|
"type": "int",
|
|
},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with template variables in components")
|
|
def step_composite_agent_template_with_template_variables_in_components(context):
|
|
"""Create composite agent template with template variables in components."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {
|
|
"agent1": {"type": "{{agent_type}}", "config": {"model": "{{model}}"}}
|
|
},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with agent templates")
|
|
def step_composite_agent_template_with_agent_templates(context):
|
|
"""Create composite agent template with agent templates."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {
|
|
"agent1": {
|
|
"template": "mock_agent_template",
|
|
"params": {"model": "gpt-4"},
|
|
}
|
|
},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with direct agent definitions")
|
|
def step_composite_agent_template_with_direct_agent_definitions(context):
|
|
"""Create composite agent template with direct agent definitions."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {"agent1": {"type": "llm", "config": {"model": "gpt-3.5-turbo"}}},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with None agent values")
|
|
def step_composite_agent_template_with_none_agent_values(context):
|
|
"""Create composite agent template with None agent values."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {
|
|
"agent1": {"type": "llm", "config": {"model": "gpt-3.5-turbo"}},
|
|
"agent2": None,
|
|
},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with agent templates and params")
|
|
def step_composite_agent_template_with_agent_templates_and_params(context):
|
|
"""Create composite agent template with agent templates and params."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {
|
|
"agent1": {
|
|
"template": "mock_agent_template",
|
|
"params": {"model": "gpt-4", "temperature": 0.8},
|
|
}
|
|
},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with graph templates")
|
|
def step_composite_agent_template_with_graph_templates(context):
|
|
"""Create composite agent template with graph templates."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {
|
|
"graph1": {"template": "mock_graph_template", "params": {"timeout": 60}}
|
|
},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with direct graph definitions")
|
|
def step_composite_agent_template_with_direct_graph_definitions(context):
|
|
"""Create composite agent template with direct graph definitions."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {
|
|
"graph1": {
|
|
"nodes": {"node1": {"type": "agent", "agent": "test_agent"}},
|
|
"edges": [],
|
|
}
|
|
},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with None graph values")
|
|
def step_composite_agent_template_with_none_graph_values(context):
|
|
"""Create composite agent template with None graph values."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {"graph1": {"nodes": {}, "edges": []}, "graph2": None},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with graph templates and params")
|
|
def step_composite_agent_template_with_graph_templates_and_params(context):
|
|
"""Create composite agent template with graph templates and params."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {
|
|
"graph1": {
|
|
"template": "mock_graph_template",
|
|
"params": {"timeout": 60, "max_retries": 3},
|
|
}
|
|
},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with stream templates")
|
|
def step_composite_agent_template_with_stream_templates(context):
|
|
"""Create composite agent template with stream templates."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {},
|
|
"streams": {
|
|
"stream1": {
|
|
"template": "mock_stream_template",
|
|
"params": {"buffer_size": 100},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with direct stream definitions")
|
|
def step_composite_agent_template_with_direct_stream_definitions(context):
|
|
"""Create composite agent template with direct stream definitions."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {},
|
|
"streams": {
|
|
"stream1": {
|
|
"operators": [{"type": "map", "agent": "test_agent"}],
|
|
"publications": ["output"],
|
|
}
|
|
},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with None stream values")
|
|
def step_composite_agent_template_with_none_stream_values(context):
|
|
"""Create composite agent template with None stream values."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {},
|
|
"streams": {
|
|
"stream1": {"operators": [], "publications": []},
|
|
"stream2": None,
|
|
},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with stream templates and params")
|
|
def step_composite_agent_template_with_stream_templates_and_params(context):
|
|
"""Create composite agent template with stream templates and params."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {},
|
|
"graphs": {},
|
|
"streams": {
|
|
"stream1": {
|
|
"template": "mock_stream_template",
|
|
"params": {"buffer_size": 200, "timeout": 30},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with routing configuration")
|
|
def step_composite_agent_template_with_routing_configuration(context):
|
|
"""Create composite agent template with routing configuration."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {"agents": {}, "graphs": {}, "streams": {}},
|
|
"routing": {
|
|
"default": "{{default_agent}}",
|
|
"rules": [{"condition": "type == 'query'", "target": "query_agent"}],
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template with pending references")
|
|
def step_composite_agent_template_with_pending_references(context):
|
|
"""Create composite agent template with pending references."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"components": {
|
|
"agents": {"agent1": {"type": "llm", "config": {"model": "gpt-3.5-turbo"}}},
|
|
"graphs": {},
|
|
"streams": {},
|
|
},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a composite agent template without components section")
|
|
def step_composite_agent_template_without_components_section(context):
|
|
"""Create composite agent template without components section."""
|
|
context.template_definition = {
|
|
"type": "composite",
|
|
"routing": {"default": "agent1"},
|
|
}
|
|
context.composite_template = CompositeAgentTemplate(
|
|
"test_composite", TemplateType.AGENT, context.template_definition
|
|
)
|
|
|
|
|
|
@given("I have a graph definition")
|
|
def step_graph_definition(context):
|
|
"""Create basic graph definition."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "start"}, "node2": {"type": "end"}},
|
|
"edges": [{"from": "node1", "to": "node2"}],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition with agent nodes")
|
|
def step_graph_definition_with_agent_nodes(context):
|
|
"""Create graph definition with agent nodes."""
|
|
context.graph_definition = {
|
|
"nodes": {
|
|
"node1": {"type": "agent", "agent": "test_agent"},
|
|
"node2": {"type": "agent", "agent": "another_agent"},
|
|
},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition with nodes without agent type")
|
|
def step_graph_definition_with_nodes_without_agent_type(context):
|
|
"""Create graph definition with nodes without agent type."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "start"}, "node2": {"type": "end"}},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition with template variable agent references")
|
|
def step_graph_definition_with_template_variable_agent_references(context):
|
|
"""Create graph definition with template variable agent references."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "agent", "agent": "{{dynamic_agent}}"}},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition with unresolved agent references")
|
|
def step_graph_definition_with_unresolved_agent_references(context):
|
|
"""Create graph definition with unresolved agent references."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "agent", "agent": "nonexistent_agent"}},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition without nodes section")
|
|
def step_graph_definition_without_nodes_section(context):
|
|
"""Create graph definition without nodes section."""
|
|
context.graph_definition = {"edges": []}
|
|
|
|
|
|
@given("I have a graph definition with None node config")
|
|
def step_graph_definition_with_none_node_config(context):
|
|
"""Create graph definition with None node config."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "agent", "agent": "test_agent"}, "node2": None},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have a graph definition with empty agent reference")
|
|
def step_graph_definition_with_empty_agent_reference(context):
|
|
"""Create graph definition with empty agent reference."""
|
|
context.graph_definition = {
|
|
"nodes": {"node1": {"type": "agent", "agent": ""}},
|
|
"edges": [],
|
|
}
|
|
|
|
|
|
@given("I have template parameters for agent templates")
|
|
def step_template_parameters_agent_templates(context):
|
|
"""Create template parameters for agent templates."""
|
|
context.template_params = {"model": "gpt-4", "temperature": 0.8}
|
|
|
|
|
|
@given("I have valid template parameters for agent templates")
|
|
def step_valid_template_parameters_agent_templates(context):
|
|
"""Create valid template parameters for agent templates."""
|
|
context.template_params = {
|
|
"model": "gpt-4",
|
|
"temperature": 0.5,
|
|
"coordination_type": "sequential",
|
|
}
|
|
|
|
|
|
@given("I have template parameters with variable values")
|
|
def step_template_parameters_with_variable_values(context):
|
|
"""Create template parameters with variable values."""
|
|
context.template_params = {
|
|
"model": "gpt-4",
|
|
"temperature": 0.9,
|
|
"role": "helpful",
|
|
"agent_type": "llm",
|
|
"default_agent": "agent1",
|
|
}
|
|
|
|
|
|
@given("I have an instantiation context for agent templates")
|
|
def step_instantiation_context_agent_templates(context):
|
|
"""Create instantiation context for agent templates."""
|
|
context.instantiation_context = InstantiationContext()
|
|
|
|
|
|
@given("I have an instantiation context with registered agents")
|
|
def step_instantiation_context_with_registered_agents(context):
|
|
"""Create instantiation context with registered agents."""
|
|
context.instantiation_context = InstantiationContext()
|
|
# Add some mock agents to the context
|
|
context.instantiation_context.add_component(
|
|
"agent",
|
|
"test_agent",
|
|
{"name": "test_agent", "type": "llm", "config": {"model": "gpt-3.5-turbo"}},
|
|
)
|
|
|
|
|
|
@given("I have an instantiation context without registered agents")
|
|
def step_instantiation_context_without_registered_agents(context):
|
|
"""Create instantiation context without registered agents."""
|
|
context.instantiation_context = InstantiationContext()
|
|
|
|
|
|
@given("I have registered agent templates")
|
|
def step_registered_agent_templates(context):
|
|
"""Register agent templates in registry."""
|
|
with patch(
|
|
"cleveragents.templates.agent_templates.AgentTemplate", MockAgentTemplate
|
|
):
|
|
context.registry.register_template(
|
|
TemplateType.AGENT, "mock_agent_template", {"type": "llm"}
|
|
)
|
|
|
|
|
|
@given("I have registered graph templates")
|
|
def step_registered_graph_templates(context):
|
|
"""Register graph templates in registry."""
|
|
with patch(
|
|
"cleveragents.templates.graph_templates.GraphTemplate", MockGraphTemplate
|
|
):
|
|
context.registry.register_template(
|
|
TemplateType.GRAPH, "mock_graph_template", {"nodes": []}
|
|
)
|
|
|
|
|
|
@given("I have registered stream templates")
|
|
def step_registered_stream_templates(context):
|
|
"""Register stream templates in registry."""
|
|
with patch(
|
|
"cleveragents.templates.stream_templates.StreamTemplate", MockStreamTemplate
|
|
):
|
|
context.registry.register_template(
|
|
TemplateType.STREAM, "mock_stream_template", {"operators": []}
|
|
)
|
|
|
|
|
|
@when("I instantiate the agent template")
|
|
def step_instantiate_agent_template(context):
|
|
"""Instantiate the agent template."""
|
|
try:
|
|
context.result = context.agent_template.instantiate(
|
|
context.template_params, context.registry, context.instantiation_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I instantiate the composite agent template")
|
|
def step_instantiate_composite_agent_template(context):
|
|
"""Instantiate the composite agent template."""
|
|
try:
|
|
with patch(
|
|
"cleveragents.templates.agent_templates.AgentTemplate", MockAgentTemplate
|
|
), patch(
|
|
"cleveragents.templates.graph_templates.GraphTemplate", MockGraphTemplate
|
|
), patch(
|
|
"cleveragents.templates.stream_templates.StreamTemplate", MockStreamTemplate
|
|
):
|
|
context.result = context.composite_template.instantiate(
|
|
context.template_params, context.registry, context.instantiation_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I process the graph definition")
|
|
def step_process_graph_definition(context):
|
|
"""Process the graph definition."""
|
|
try:
|
|
context.result = context.composite_template._process_graph_definition(
|
|
context.graph_definition,
|
|
context.template_params,
|
|
context.instantiation_context,
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the agent configuration should be returned")
|
|
def step_agent_configuration_should_be_returned(context):
|
|
"""Verify agent configuration is returned."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, dict)
|
|
|
|
|
|
@then("the parameters section should be removed from config")
|
|
def step_parameters_section_should_be_removed(context):
|
|
"""Verify parameters section is removed from config."""
|
|
assert "parameters" not in context.result.get("config", {})
|
|
|
|
|
|
@then("template variables should be applied to the config")
|
|
def step_template_variables_should_be_applied_to_config(context):
|
|
"""Verify template variables are applied to config."""
|
|
# This is verified by checking that the original definition still has template vars
|
|
# but the result has them replaced (via mocking of _apply_template_vars)
|
|
assert context.result is not None
|
|
|
|
|
|
@then("the result should have correct agent structure")
|
|
def step_result_should_have_correct_agent_structure(context):
|
|
"""Verify result has correct agent structure."""
|
|
assert "name" in context.result
|
|
assert "type" in context.result
|
|
assert "config" in context.result
|
|
assert context.result["name"] == "test_agent"
|
|
|
|
|
|
@then("the agent type should default to llm")
|
|
def step_agent_type_should_default_to_llm(context):
|
|
"""Verify agent type defaults to llm."""
|
|
assert context.result["type"] == "llm"
|
|
|
|
|
|
@then("the agent type should match the custom type")
|
|
def step_agent_type_should_match_custom_type(context):
|
|
"""Verify agent type matches custom type."""
|
|
assert context.result["type"] == "custom_tool"
|
|
|
|
|
|
@then("parameter validation should be called")
|
|
def step_parameter_validation_should_be_called(context):
|
|
"""Verify parameter validation is called."""
|
|
# This is implicitly tested by the instantiation succeeding
|
|
assert context.error is None
|
|
|
|
|
|
@then("the filled parameters should be used")
|
|
def step_filled_parameters_should_be_used(context):
|
|
"""Verify filled parameters are used."""
|
|
assert context.result is not None
|
|
|
|
|
|
@then("template variables should be replaced with parameter values")
|
|
def step_template_variables_should_be_replaced(context):
|
|
"""Verify template variables are replaced with parameter values."""
|
|
# This would be verified through mocking _apply_template_vars
|
|
assert context.result is not None
|
|
|
|
|
|
@then("the original definition should not be modified for agent templates")
|
|
def step_original_definition_should_not_be_modified_agent_templates(context):
|
|
"""Verify original definition is not modified for agent templates."""
|
|
# Check that the original definition hasn't changed
|
|
if hasattr(context, "original_definition") and context.original_definition:
|
|
assert (
|
|
context.template_definition == context.original_definition
|
|
or "parameters" not in context.template_definition
|
|
)
|
|
|
|
|
|
@then("the returned config should be a separate copy")
|
|
def step_returned_config_should_be_separate_copy(context):
|
|
"""Verify returned config is a separate copy."""
|
|
# This is verified by the deep copy behavior in the implementation
|
|
assert context.result is not None
|
|
|
|
|
|
@then("a composite agent configuration should be returned")
|
|
def step_composite_agent_configuration_should_be_returned(context):
|
|
"""Verify composite agent configuration is returned."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert context.result["type"] == "composite"
|
|
|
|
|
|
@then("the configuration should include components")
|
|
def step_configuration_should_include_components(context):
|
|
"""Verify configuration includes components."""
|
|
assert "config" in context.result
|
|
assert "components" in context.result["config"]
|
|
config_components = context.result["config"]["components"]
|
|
assert "agents" in config_components
|
|
assert "graphs" in config_components
|
|
assert "streams" in config_components
|
|
|
|
|
|
@then("the configuration should include routing")
|
|
def step_configuration_should_include_routing(context):
|
|
"""Verify configuration includes routing."""
|
|
assert "config" in context.result
|
|
assert "routing" in context.result["config"]
|
|
|
|
|
|
@then("the configuration should include expose_params")
|
|
def step_configuration_should_include_expose_params(context):
|
|
"""Verify configuration includes expose_params."""
|
|
assert "config" in context.result
|
|
assert "expose_params" in context.result["config"]
|
|
|
|
|
|
@then("a local instantiation context should be created")
|
|
def step_local_instantiation_context_should_be_created(context):
|
|
"""Verify local instantiation context is created."""
|
|
# This is verified by the successful instantiation
|
|
assert context.error is None
|
|
|
|
|
|
@then("the local context should have parent context")
|
|
def step_local_context_should_have_parent_context(context):
|
|
"""Verify local context has parent context."""
|
|
# This is verified by the implementation creating InstantiationContext(parent=context)
|
|
assert context.error is None
|
|
|
|
|
|
@then("parameter validation should be called for composite")
|
|
def step_parameter_validation_should_be_called_for_composite(context):
|
|
"""Verify parameter validation is called for composite."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("template variables should be applied to components section")
|
|
def step_template_variables_should_be_applied_to_components_section(context):
|
|
"""Verify template variables are applied to components section."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("agent templates should be instantiated")
|
|
def step_agent_templates_should_be_instantiated(context):
|
|
"""Verify agent templates are instantiated."""
|
|
assert context.error is None
|
|
assert "agents" in context.result["config"]["components"]
|
|
|
|
|
|
@then("agents should be added to local context")
|
|
def step_agents_should_be_added_to_local_context(context):
|
|
"""Verify agents are added to local context."""
|
|
# This is verified by the successful instantiation
|
|
assert context.error is None
|
|
|
|
|
|
@then("instantiated agents should be in components")
|
|
def step_instantiated_agents_should_be_in_components(context):
|
|
"""Verify instantiated agents are in components."""
|
|
agents = context.result["config"]["components"]["agents"]
|
|
assert len(agents) > 0
|
|
|
|
|
|
@then("direct agent definitions should be processed")
|
|
def step_direct_agent_definitions_should_be_processed(context):
|
|
"""Verify direct agent definitions are processed."""
|
|
assert context.error is None
|
|
agents = context.result["config"]["components"]["agents"]
|
|
assert "agent1" in agents
|
|
|
|
|
|
@then("agents should be added to local context without template lookup")
|
|
def step_agents_should_be_added_to_local_context_without_template_lookup(context):
|
|
"""Verify agents are added to local context without template lookup."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("None agent values should be skipped")
|
|
def step_none_agent_values_should_be_skipped(context):
|
|
"""Verify None agent values are skipped."""
|
|
assert context.error is None
|
|
agents = context.result["config"]["components"]["agents"]
|
|
assert "agent2" not in agents or agents.get("agent2") is None
|
|
|
|
|
|
@then("no errors should occur for None agents")
|
|
def step_no_errors_should_occur_for_none_agents(context):
|
|
"""Verify no errors occur for None agents."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("template parameters should be merged with agent params")
|
|
def step_template_parameters_should_be_merged_with_agent_params(context):
|
|
"""Verify template parameters are merged with agent params."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("merged parameters should be passed to agent instantiation")
|
|
def step_merged_parameters_should_be_passed_to_agent_instantiation(context):
|
|
"""Verify merged parameters are passed to agent instantiation."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("graph templates should be instantiated")
|
|
def step_graph_templates_should_be_instantiated(context):
|
|
"""Verify graph templates are instantiated."""
|
|
assert context.error is None
|
|
assert "graphs" in context.result["config"]["components"]
|
|
|
|
|
|
@then("graphs should be added to local context")
|
|
def step_graphs_should_be_added_to_local_context(context):
|
|
"""Verify graphs are added to local context."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("instantiated graphs should be in components")
|
|
def step_instantiated_graphs_should_be_in_components(context):
|
|
"""Verify instantiated graphs are in components."""
|
|
graphs = context.result["config"]["components"]["graphs"]
|
|
assert len(graphs) > 0
|
|
|
|
|
|
@then("direct graph definitions should be processed through process_graph_definition")
|
|
def step_direct_graph_definitions_should_be_processed_through_process_graph_definition(
|
|
context,
|
|
):
|
|
"""Verify direct graph definitions are processed through process_graph_definition."""
|
|
assert context.error is None
|
|
graphs = context.result["config"]["components"]["graphs"]
|
|
assert "graph1" in graphs
|
|
|
|
|
|
@then("None graph values should be skipped")
|
|
def step_none_graph_values_should_be_skipped(context):
|
|
"""Verify None graph values are skipped."""
|
|
assert context.error is None
|
|
graphs = context.result["config"]["components"]["graphs"]
|
|
assert "graph2" not in graphs or graphs.get("graph2") is None
|
|
|
|
|
|
@then("no errors should occur for None graphs")
|
|
def step_no_errors_should_occur_for_none_graphs(context):
|
|
"""Verify no errors occur for None graphs."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("template parameters should be merged with graph params")
|
|
def step_template_parameters_should_be_merged_with_graph_params(context):
|
|
"""Verify template parameters are merged with graph params."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("merged parameters should be passed to graph instantiation")
|
|
def step_merged_parameters_should_be_passed_to_graph_instantiation(context):
|
|
"""Verify merged parameters are passed to graph instantiation."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("stream templates should be instantiated")
|
|
def step_stream_templates_should_be_instantiated(context):
|
|
"""Verify stream templates are instantiated."""
|
|
assert context.error is None
|
|
assert "streams" in context.result["config"]["components"]
|
|
|
|
|
|
@then("streams should be added to local context")
|
|
def step_streams_should_be_added_to_local_context(context):
|
|
"""Verify streams are added to local context."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("instantiated streams should be in components")
|
|
def step_instantiated_streams_should_be_in_components(context):
|
|
"""Verify instantiated streams are in components."""
|
|
streams = context.result["config"]["components"]["streams"]
|
|
assert len(streams) > 0
|
|
|
|
|
|
@then("direct stream definitions should be processed")
|
|
def step_direct_stream_definitions_should_be_processed(context):
|
|
"""Verify direct stream definitions are processed."""
|
|
assert context.error is None
|
|
streams = context.result["config"]["components"]["streams"]
|
|
assert "stream1" in streams
|
|
|
|
|
|
@then("streams should be added to local context without template lookup")
|
|
def step_streams_should_be_added_to_local_context_without_template_lookup(context):
|
|
"""Verify streams are added to local context without template lookup."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("None stream values should be skipped")
|
|
def step_none_stream_values_should_be_skipped(context):
|
|
"""Verify None stream values are skipped."""
|
|
assert context.error is None
|
|
streams = context.result["config"]["components"]["streams"]
|
|
assert "stream2" not in streams or streams.get("stream2") is None
|
|
|
|
|
|
@then("no errors should occur for None streams")
|
|
def step_no_errors_should_occur_for_none_streams(context):
|
|
"""Verify no errors occur for None streams."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("template parameters should be merged with stream params")
|
|
def step_template_parameters_should_be_merged_with_stream_params(context):
|
|
"""Verify template parameters are merged with stream params."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("merged parameters should be passed to stream instantiation")
|
|
def step_merged_parameters_should_be_passed_to_stream_instantiation(context):
|
|
"""Verify merged parameters are passed to stream instantiation."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("routing configuration should be processed")
|
|
def step_routing_configuration_should_be_processed(context):
|
|
"""Verify routing configuration is processed."""
|
|
assert context.error is None
|
|
assert "routing" in context.result["config"]
|
|
|
|
|
|
@then("template variables should be applied to routing")
|
|
def step_template_variables_should_be_applied_to_routing(context):
|
|
"""Verify template variables are applied to routing."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("routing should be included in final config")
|
|
def step_routing_should_be_included_in_final_config(context):
|
|
"""Verify routing is included in final config."""
|
|
assert "routing" in context.result["config"]
|
|
|
|
|
|
@then("pending references should be resolved in local context")
|
|
def step_pending_references_should_be_resolved_in_local_context(context):
|
|
"""Verify pending references are resolved in local context."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("empty components should be created")
|
|
def step_empty_components_should_be_created(context):
|
|
"""Verify empty components are created."""
|
|
assert context.error is None
|
|
components = context.result["config"]["components"]
|
|
assert "agents" in components
|
|
assert "graphs" in components
|
|
assert "streams" in components
|
|
|
|
|
|
@then("no errors should occur for missing components")
|
|
def step_no_errors_should_occur_for_missing_components(context):
|
|
"""Verify no errors occur for missing components."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("template variables should be applied to graph definition")
|
|
def step_template_variables_should_be_applied_to_graph_definition(context):
|
|
"""Verify template variables are applied to graph definition."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
|
|
|
|
@then("the processed graph should be returned")
|
|
def step_processed_graph_should_be_returned(context):
|
|
"""Verify processed graph is returned."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert isinstance(context.result, dict)
|
|
|
|
|
|
@then("agent references should be resolved")
|
|
def step_agent_references_should_be_resolved(context):
|
|
"""Verify agent references are resolved."""
|
|
assert context.error is None
|
|
# Check if nodes with agent references were processed
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if node_config and node_config.get("type") == "agent":
|
|
# The reference resolution would add agent_config
|
|
assert True # Resolution was attempted
|
|
|
|
|
|
@then("agent_config should be added to resolved nodes")
|
|
def step_agent_config_should_be_added_to_resolved_nodes(context):
|
|
"""Verify agent_config is added to resolved nodes."""
|
|
assert context.error is None
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if (
|
|
node_config
|
|
and node_config.get("type") == "agent"
|
|
and node_config.get("agent") == "test_agent"
|
|
):
|
|
assert "agent_config" in node_config
|
|
|
|
|
|
@then("non-agent nodes should not be processed for agent resolution")
|
|
def step_non_agent_nodes_should_not_be_processed_for_agent_resolution(context):
|
|
"""Verify non-agent nodes are not processed for agent resolution."""
|
|
assert context.error is None
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if node_config and node_config.get("type") != "agent":
|
|
assert "agent_config" not in node_config
|
|
|
|
|
|
@then(
|
|
"template variable agent references should not be resolved as component references"
|
|
)
|
|
def step_template_variable_agent_references_should_not_be_resolved_as_component_references(
|
|
context,
|
|
):
|
|
"""Verify template variable agent references are not resolved as component references."""
|
|
assert context.error is None
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if node_config and node_config.get("agent", "").startswith("{{"):
|
|
assert "agent_config" not in node_config
|
|
|
|
|
|
@then("unresolved agent references should be handled gracefully")
|
|
def step_unresolved_agent_references_should_be_handled_gracefully(context):
|
|
"""Verify unresolved agent references are handled gracefully."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("no agent_config should be added for unresolved references")
|
|
def step_no_agent_config_should_be_added_for_unresolved_references(context):
|
|
"""Verify no agent_config is added for unresolved references."""
|
|
assert context.error is None
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if node_config and node_config.get("agent") == "nonexistent_agent":
|
|
assert "agent_config" not in node_config
|
|
|
|
|
|
@then("processing should complete without errors")
|
|
def step_processing_should_complete_without_errors(context):
|
|
"""Verify processing completes without errors."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("no node processing should occur")
|
|
def step_no_node_processing_should_occur(context):
|
|
"""Verify no node processing occurs."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("None node configs should be handled gracefully")
|
|
def step_none_node_configs_should_be_handled_gracefully(context):
|
|
"""Verify None node configs are handled gracefully."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("no agent resolution should occur for None nodes")
|
|
def step_no_agent_resolution_should_occur_for_none_nodes(context):
|
|
"""Verify no agent resolution occurs for None nodes."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("empty agent references should be handled gracefully")
|
|
def step_empty_agent_references_should_be_handled_gracefully(context):
|
|
"""Verify empty agent references are handled gracefully."""
|
|
assert context.error is None
|
|
|
|
|
|
@then("no resolution should occur for empty references")
|
|
def step_no_resolution_should_occur_for_empty_references(context):
|
|
"""Verify no resolution occurs for empty references."""
|
|
assert context.error is None
|
|
if "nodes" in context.result:
|
|
for node_name, node_config in context.result["nodes"].items():
|
|
if node_config and node_config.get("agent") == "":
|
|
assert "agent_config" not in node_config
|