forked from cleveragents/cleveragents-core
349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""Step definitions for complex template processing features."""
|
|
|
|
import json
|
|
|
|
import yaml
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.templates.base import TemplateType
|
|
from cleveragents.templates.enhanced_registry import EnhancedTemplateRegistry
|
|
from cleveragents.templates.yaml_preprocessor import YAMLTemplateProcessor
|
|
|
|
|
|
@given("the template processing system is initialized")
|
|
def step_init_template_system(context):
|
|
"""Initialize template processing system."""
|
|
context.processor = YAMLTemplateProcessor()
|
|
context.registry = EnhancedTemplateRegistry()
|
|
|
|
|
|
@given("I have a YAML configuration with templates")
|
|
def step_yaml_templates(context):
|
|
"""Store YAML with templates."""
|
|
context.yaml_content = context.text
|
|
|
|
|
|
@given("I have a YAML configuration with complex templates")
|
|
def step_yaml_complex_templates(context):
|
|
"""Store YAML with complex templates."""
|
|
context.yaml_content = context.text
|
|
|
|
|
|
@given("I have a configuration with template definitions")
|
|
def step_config_template_defs(context):
|
|
"""Store configuration with template definitions."""
|
|
context.yaml_content = context.text
|
|
|
|
|
|
@given("I have registered a complex template")
|
|
def step_register_template(context):
|
|
"""Register a complex template."""
|
|
template_yaml = context.text
|
|
context.registry.register_template_string(
|
|
TemplateType.AGENT, "pipeline_template", template_yaml
|
|
)
|
|
|
|
|
|
@given("I have a YAML with Jinja2 filters")
|
|
def step_yaml_with_filters(context):
|
|
"""Store YAML with Jinja2 filters."""
|
|
context.yaml_content = context.text
|
|
|
|
|
|
@given("I have YAML with edge case scenarios")
|
|
def step_yaml_edge_cases(context):
|
|
"""Store YAML with edge cases."""
|
|
context.yaml_content = context.text
|
|
|
|
|
|
@given("I have a template context")
|
|
def step_create_context(context):
|
|
"""Create template context from table."""
|
|
context.template_context = {}
|
|
for row in context.table:
|
|
key = row["key"]
|
|
value = row["value"]
|
|
|
|
# Parse JSON values
|
|
if value.startswith("[") or value.startswith("{"):
|
|
value = json.loads(value)
|
|
else:
|
|
# Try numeric conversion
|
|
try:
|
|
value = float(value)
|
|
if value.is_integer():
|
|
value = int(value)
|
|
except ValueError:
|
|
# Keep as string
|
|
pass
|
|
|
|
context.template_context[key] = value
|
|
|
|
|
|
@given("I have a context with services and features")
|
|
def step_context_services_features(context):
|
|
"""Create context with services and features."""
|
|
context.template_context = {
|
|
"project_name": "MyApp",
|
|
"version": "2.0.0",
|
|
"debug_mode": True,
|
|
"services": [
|
|
{"name": "api", "port": 8080, "replicas": 3},
|
|
{"name": "web", "port": 3000, "enabled": True},
|
|
{"name": "worker", "port": 9000, "replicas": 2},
|
|
],
|
|
"enable_advanced": True,
|
|
"analytics_level": "detailed",
|
|
}
|
|
|
|
|
|
@given("I have a context with Unicode messages and nested data")
|
|
def step_context_unicode_nested(context):
|
|
"""Create context with Unicode and nested data."""
|
|
context.template_context = {
|
|
"messages": [
|
|
{"text": "Hello", "emoji": "👋"},
|
|
{"text": "Ça va?", "emoji": "🇫🇷"},
|
|
{"text": "你好", "emoji": "🇨🇳"},
|
|
],
|
|
"items": [
|
|
{"name": "service1", "config": {"port": 8080, "enabled": True}},
|
|
{"name": "service2", "config": {"port": 9090, "workers": 4}},
|
|
{"name": "service3", "config": None},
|
|
],
|
|
}
|
|
|
|
|
|
@when("I process the YAML with the template processor")
|
|
def step_process_yaml_processor(context):
|
|
"""Process YAML with template processor."""
|
|
context.result = context.processor.process_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
|
|
|
|
@when("I parse the template configuration")
|
|
def step_parse_config(context):
|
|
"""Parse configuration."""
|
|
# For template definitions, just parse as YAML
|
|
context.result = yaml.safe_load(context.yaml_content)
|
|
|
|
|
|
@when("I instantiate the template with parameters")
|
|
def step_instantiate_template(context):
|
|
"""Instantiate template with parameters."""
|
|
params = {}
|
|
for row in context.table:
|
|
key = row["key"]
|
|
value = row["value"]
|
|
|
|
# Parse JSON values
|
|
if value.startswith("[") or value.startswith("{"):
|
|
value = json.loads(value)
|
|
elif value.lower() == "true":
|
|
value = True
|
|
elif value.lower() == "false":
|
|
value = False
|
|
|
|
params[key] = value
|
|
|
|
context.result = context.registry.instantiate(
|
|
TemplateType.AGENT, "pipeline_template", params
|
|
)
|
|
|
|
|
|
@when("I process the configuration")
|
|
def step_process_config(context):
|
|
"""Process configuration."""
|
|
context.result = context.processor.process_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
|
|
|
|
@when("I process the edge cases")
|
|
def step_process_edge_cases(context):
|
|
"""Process edge case YAML."""
|
|
context.result = context.processor.process_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
|
|
|
|
@then("the result should have {count:d} agents")
|
|
def step_check_agent_count_simple(context, count):
|
|
"""Check agent count."""
|
|
assert "agents" in context.result
|
|
assert len(context.result["agents"]) == count
|
|
|
|
|
|
@then("the agents section should contain {agent_list}")
|
|
def step_check_agents_list(context, agent_list):
|
|
"""Check agents section contains specific agents."""
|
|
# Parse agent list - handle both quoted and unquoted names
|
|
agents = []
|
|
for part in agent_list.split(","):
|
|
agent = part.strip().strip('"').strip("'")
|
|
agents.append(agent)
|
|
|
|
assert "agents" in context.result
|
|
for agent in agents:
|
|
assert agent in context.result["agents"], f"Agent '{agent}' not found"
|
|
|
|
|
|
@then('each agent should have model "{model}" and temperature {temp:f}')
|
|
def step_check_agent_config(context, model, temp):
|
|
"""Check agent configurations."""
|
|
for agent_name, agent_config in context.result["agents"].items():
|
|
if agent_name.endswith("_agent"):
|
|
assert agent_config["config"]["model"] == model
|
|
assert agent_config["config"]["temperature"] == temp
|
|
|
|
|
|
@then("the graph should have {nodes:d} nodes and {edges:d} edges")
|
|
def step_check_graph_structure(context, nodes, edges):
|
|
"""Check graph node and edge count."""
|
|
assert "graph" in context.result
|
|
assert len(context.result["graph"]["nodes"]) == nodes
|
|
assert len(context.result["graph"]["edges"]) == edges
|
|
|
|
|
|
@then('templates should contain "{template_name}"')
|
|
def step_check_template_exists(context, template_name):
|
|
"""Check template exists."""
|
|
assert "templates" in context.result
|
|
assert "agents" in context.result["templates"]
|
|
assert template_name in context.result["templates"]["agents"]
|
|
|
|
|
|
@then('template_strings should contain "{template_name}"')
|
|
def step_check_template_string_exists(context, template_name):
|
|
"""Check template string exists."""
|
|
assert "template_strings" in context.result
|
|
assert "agents" in context.result["template_strings"]
|
|
assert template_name in context.result["template_strings"]["agents"]
|
|
|
|
|
|
@then("agents should reference the dynamic_team template")
|
|
def step_check_agent_reference(context):
|
|
"""Check agents reference templates."""
|
|
assert "agents" in context.result
|
|
assert "my_team" in context.result["agents"]
|
|
assert context.result["agents"]["my_team"]["template"] == "dynamic_team"
|
|
|
|
|
|
@then("the result should contain {count:d} agents: {agent_names}")
|
|
def step_check_result_agents(context, count, agent_names):
|
|
"""Check result contains specific agents."""
|
|
agents = [name.strip(' "') for name in agent_names.split(",")]
|
|
assert len(agents) == count
|
|
|
|
components = context.result["components"]["agents"]
|
|
for agent in agents:
|
|
assert agent in components
|
|
|
|
|
|
@then("the workflow should have parallel execution paths")
|
|
def step_check_parallel_paths(context):
|
|
"""Check workflow has parallel paths."""
|
|
workflow = context.result["components"]["graphs"]["workflow"]
|
|
edges = workflow["edges"]
|
|
|
|
# Count edges from start
|
|
start_edges = [e for e in edges if e["source"] == "start"]
|
|
assert len(start_edges) > 1 # Multiple paths from start = parallel
|
|
|
|
|
|
@then("there should be edges from start to all stages")
|
|
def step_check_all_start_edges(context):
|
|
"""Check edges from start to all stages."""
|
|
workflow = context.result["components"]["graphs"]["workflow"]
|
|
edges = workflow["edges"]
|
|
|
|
# Get all stage names
|
|
stage_nodes = [n for n in workflow["nodes"] if n not in ["start", "end"]]
|
|
|
|
# Check each stage has edge from start
|
|
for stage in stage_nodes:
|
|
has_edge = any(e["source"] == "start" and e["target"] == stage for e in edges)
|
|
assert has_edge, f"No edge from start to {stage}"
|
|
|
|
|
|
@then("config debug should be {value}")
|
|
def step_check_debug_value(context, value):
|
|
"""Check config debug value."""
|
|
expected = value.lower() == "true"
|
|
assert context.result["config"]["debug"] == expected
|
|
|
|
|
|
@then("services should be properly configured with defaults")
|
|
def step_check_services_defaults(context):
|
|
"""Check services have proper defaults."""
|
|
services = context.result["services"]
|
|
|
|
# Check each service
|
|
for service_name, service_config in services.items():
|
|
assert "port" in service_config
|
|
assert "enabled" in service_config
|
|
assert "replicas" in service_config
|
|
|
|
# Check defaults were applied
|
|
if service_name == "web":
|
|
assert service_config["enabled"] is True
|
|
assert service_config["replicas"] == 1 # default
|
|
|
|
|
|
@then("metrics should calculate correct totals")
|
|
def step_check_metrics_totals(context):
|
|
"""Check metrics calculations."""
|
|
metrics = context.result["metrics"]
|
|
|
|
# Verify calculations
|
|
assert metrics["total_services"] == 3
|
|
assert metrics["total_ports"] == 8080 + 3000 + 9000
|
|
assert len(metrics["service_names"]) == 3
|
|
assert set(metrics["service_names"]) == {"api", "web", "worker"}
|
|
|
|
|
|
@then('empty_list should only contain "{content}"')
|
|
def step_check_empty_list(context, content):
|
|
"""Check empty list content."""
|
|
assert "empty_list" in context.result
|
|
assert "static" in context.result["empty_list"]
|
|
assert context.result["empty_list"]["static"] == "value"
|
|
|
|
# Should not have any list items from the empty loop
|
|
list_items = [k for k in context.result["empty_list"] if k != "static"]
|
|
assert len(list_items) == 0
|
|
|
|
|
|
@then("messages should preserve Unicode characters")
|
|
def step_check_unicode_preserved(context):
|
|
"""Check Unicode preservation."""
|
|
messages = context.result["messages"]
|
|
|
|
# Check specific Unicode content
|
|
assert messages[0]["emoji"] == "👋"
|
|
assert messages[1]["text"] == "Ça va?"
|
|
assert messages[2]["text"] == "你好"
|
|
|
|
|
|
@then("complex nested structures should render correctly")
|
|
def step_check_complex_nested(context):
|
|
"""Check complex nested rendering."""
|
|
assert "complex" in context.result
|
|
|
|
# Should have rendered the items
|
|
assert "service1" in context.result["complex"]
|
|
assert "service2" in context.result["complex"]
|
|
|
|
# service1 and service2 should have config
|
|
assert "config" in context.result["complex"]["service1"]
|
|
assert context.result["complex"]["service1"]["config"]["port"] == 8080
|
|
|
|
# service3 should not have config section (it was None)
|
|
if "service3" in context.result.get("complex", {}):
|
|
service3 = context.result["complex"].get("service3")
|
|
if service3 is not None:
|
|
assert "config" not in service3
|