forked from HAL9000/cleveragents-core
1044 lines
32 KiB
Plaintext
1044 lines
32 KiB
Plaintext
"""
|
|
Comprehensive step definitions for inline YAML Jinja coverage tests.
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.templates.inline_yaml_jinja import InlineYAMLJinja
|
|
|
|
|
|
@given("I have an InlineYAMLJinja processor")
|
|
def step_have_processor(context):
|
|
"""Initialize test context."""
|
|
context.processor = None
|
|
context.result = None
|
|
context.error = None
|
|
context.temp_files = []
|
|
|
|
|
|
@when("I create a new InlineYAMLJinja instance")
|
|
def step_create_instance(context):
|
|
"""Create a new InlineYAMLJinja instance."""
|
|
try:
|
|
context.processor = InlineYAMLJinja()
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the Jinja2 environment should be configured correctly")
|
|
def step_jinja_env_configured(context):
|
|
"""Verify Jinja2 environment is configured correctly."""
|
|
assert context.processor is not None
|
|
assert context.processor.env.block_start_string == "{%"
|
|
assert context.processor.env.block_end_string == "%}"
|
|
assert context.processor.env.variable_start_string == "{{"
|
|
assert context.processor.env.variable_end_string == "}}"
|
|
assert context.processor.env.comment_start_string == "{#"
|
|
assert context.processor.env.comment_end_string == "#}"
|
|
|
|
|
|
@then("custom filters should be available")
|
|
def step_custom_filters_available(context):
|
|
"""Verify custom filters are available."""
|
|
assert "yaml" in context.processor.env.filters
|
|
assert "json" in context.processor.env.filters
|
|
assert "indent" in context.processor.env.filters
|
|
|
|
# Test filters work
|
|
yaml_filter = context.processor.env.filters["yaml"]
|
|
json_filter = context.processor.env.filters["json"]
|
|
indent_filter = context.processor.env.filters["indent"]
|
|
|
|
test_data = {"key": "value"}
|
|
yaml_result = yaml_filter(test_data)
|
|
json_result = json_filter(test_data)
|
|
indent_result = indent_filter("line1\nline2", 2)
|
|
|
|
assert "key: value" in yaml_result
|
|
assert json_result == '{"key": "value"}'
|
|
assert indent_result == " line1\n line2"
|
|
|
|
|
|
@then("custom tests should be available")
|
|
def step_custom_tests_available(context):
|
|
"""Verify custom tests are available."""
|
|
assert "list" in context.processor.env.tests
|
|
assert "dict" in context.processor.env.tests
|
|
assert "none" in context.processor.env.tests
|
|
|
|
# Test the tests work
|
|
list_test = context.processor.env.tests["list"]
|
|
dict_test = context.processor.env.tests["dict"]
|
|
none_test = context.processor.env.tests["none"]
|
|
|
|
assert list_test([1, 2, 3]) == True
|
|
assert list_test("not a list") == False
|
|
assert dict_test({"key": "value"}) == True
|
|
assert dict_test("not a dict") == False
|
|
assert none_test(None) == True
|
|
assert none_test("not none") == False
|
|
|
|
|
|
@given("I have a YAML file without Jinja templates")
|
|
def step_yaml_file_no_templates(context):
|
|
"""Create a YAML file without Jinja templates."""
|
|
context.processor = InlineYAMLJinja()
|
|
yaml_content = """
|
|
name: test_agent
|
|
type: llm
|
|
config:
|
|
model: gpt-3.5-turbo
|
|
temperature: 0.7
|
|
"""
|
|
temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
|
|
temp_file.write(yaml_content)
|
|
temp_file.close()
|
|
context.temp_files.append(temp_file.name)
|
|
context.yaml_file_path = Path(temp_file.name)
|
|
|
|
|
|
@when("I process the file")
|
|
def step_process_file(context):
|
|
"""Process the YAML file."""
|
|
try:
|
|
context.result = context.processor.process_file(context.yaml_file_path)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should return parsed YAML without rendering")
|
|
def step_return_parsed_yaml(context):
|
|
"""Verify it returns parsed YAML without rendering."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert "name" in context.result
|
|
assert context.result["name"] == "test_agent"
|
|
|
|
|
|
@given("I have a YAML string without Jinja templates")
|
|
def step_yaml_string_no_templates(context):
|
|
"""Create a YAML string without Jinja templates."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
name: test_agent
|
|
type: llm
|
|
config:
|
|
model: gpt-3.5-turbo
|
|
temperature: 0.7
|
|
"""
|
|
|
|
|
|
@when("I process the string")
|
|
def step_process_string(context):
|
|
"""Process the YAML string."""
|
|
try:
|
|
context.result = context.processor.process_string(context.yaml_string)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@given("I have a YAML file with Jinja templates")
|
|
def step_yaml_file_with_templates(context):
|
|
"""Create a YAML file with Jinja templates."""
|
|
context.processor = InlineYAMLJinja()
|
|
yaml_content = """
|
|
name: {{ agent_name }}
|
|
type: {{ agent_type }}
|
|
config:
|
|
model: {{ model_name | default('gpt-3.5-turbo') }}
|
|
temperature: {{ temperature | default(0.7) }}
|
|
"""
|
|
temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
|
|
temp_file.write(yaml_content)
|
|
temp_file.close()
|
|
context.temp_files.append(temp_file.name)
|
|
context.yaml_file_path = Path(temp_file.name)
|
|
|
|
|
|
@given("I have a rendering context")
|
|
def step_have_rendering_context(context):
|
|
"""Create a rendering context."""
|
|
context.render_context = {
|
|
"agent_name": "test_agent",
|
|
"agent_type": "llm",
|
|
"model_name": "gpt-4",
|
|
"temperature": 0.8,
|
|
"items": [1, 2, 3],
|
|
"config": {"key": "value"},
|
|
"enabled": True,
|
|
}
|
|
|
|
|
|
@when("I process the file with immediate rendering")
|
|
def step_process_file_immediate(context):
|
|
"""Process the file with immediate rendering."""
|
|
try:
|
|
context.result = context.processor.process_file(
|
|
context.yaml_file_path, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should render templates and return parsed YAML")
|
|
def step_render_and_parse(context):
|
|
"""Verify it renders templates and returns parsed YAML."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert context.result["name"] == "test_agent"
|
|
assert context.result["type"] == "llm"
|
|
assert context.result["config"]["model"] == "gpt-4"
|
|
|
|
|
|
@then("it should render and return parsed YAML")
|
|
def step_render_and_return_parsed(context):
|
|
"""Verify it renders and returns parsed YAML."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert context.result["name"] == "test_agent"
|
|
assert context.result["type"] == "llm"
|
|
|
|
|
|
@given("I have a YAML string with Jinja templates")
|
|
def step_yaml_string_with_templates(context):
|
|
"""Create a YAML string with Jinja templates."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
name: {{ agent_name }}
|
|
type: {{ agent_type }}
|
|
config:
|
|
model: {{ model_name | default('gpt-3.5-turbo') }}
|
|
temperature: {{ temperature | default(0.7) }}
|
|
"""
|
|
|
|
|
|
@when("I process the string with immediate rendering")
|
|
def step_process_string_immediate(context):
|
|
"""Process the string with immediate rendering."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I process the string without context")
|
|
def step_process_string_no_context(context):
|
|
"""Process the string without context for deferred rendering."""
|
|
try:
|
|
context.result = context.processor.process_string(context.yaml_string)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should store the template for deferred rendering")
|
|
def step_store_for_deferred(context):
|
|
"""Verify it stores the template for deferred rendering."""
|
|
assert context.error is None
|
|
assert "__yaml_template__" in context.result
|
|
template_info = context.result["__yaml_template__"]
|
|
assert "content" in template_info
|
|
assert "type" in template_info
|
|
assert "has_jinja2" in template_info
|
|
assert template_info["has_jinja2"] == True
|
|
|
|
|
|
@given("I have a deferred template configuration")
|
|
def step_have_deferred_config(context):
|
|
"""Create a deferred template configuration."""
|
|
context.processor = InlineYAMLJinja()
|
|
yaml_content = """
|
|
name: {{ agent_name }}
|
|
type: {{ agent_type }}
|
|
"""
|
|
context.deferred_config = {
|
|
"__yaml_template__": {
|
|
"content": yaml_content,
|
|
"type": "full",
|
|
"has_jinja2": True,
|
|
}
|
|
}
|
|
|
|
|
|
@when("I render the deferred template")
|
|
def step_render_deferred(context):
|
|
"""Render the deferred template."""
|
|
try:
|
|
context.result = context.processor.render_deferred(
|
|
context.deferred_config, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@given("I have a regular configuration without template markers")
|
|
def step_regular_config_no_markers(context):
|
|
"""Create a regular configuration without template markers."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.regular_config = {
|
|
"name": "test_agent",
|
|
"type": "llm",
|
|
"config": {"model": "gpt-3.5-turbo"},
|
|
}
|
|
|
|
|
|
@when("I render it as if it were deferred")
|
|
def step_render_as_deferred(context):
|
|
"""Render it as if it were deferred."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {
|
|
"agent_name": "test_agent",
|
|
"agent_type": "llm",
|
|
"model_name": "gpt-4",
|
|
"temperature": 0.8,
|
|
"items": [1, 2, 3],
|
|
"config": {"key": "value"},
|
|
"enabled": True,
|
|
}
|
|
context.result = context.processor.render_deferred(
|
|
context.regular_config, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should return the configuration unchanged")
|
|
def step_return_unchanged(context):
|
|
"""Verify it returns the configuration unchanged."""
|
|
if context.error is not None:
|
|
print(f"Error: {context.error}")
|
|
assert context.error is None
|
|
assert context.result == context.regular_config
|
|
|
|
|
|
@given("I have YAML with simple variable substitutions")
|
|
def step_yaml_simple_variables(context):
|
|
"""Create YAML with simple variable substitutions."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
name: {{ agent_name }}
|
|
model: {{ model_name }}
|
|
temperature: {{ temperature }}
|
|
"""
|
|
|
|
|
|
@when("I process the YAML with inline templates")
|
|
def step_process_inline_templates(context):
|
|
"""Process the YAML with inline templates."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should render variables correctly")
|
|
def step_render_variables_correctly(context):
|
|
"""Verify variables are rendered correctly."""
|
|
assert context.error is None
|
|
assert context.result["name"] == "test_agent"
|
|
assert context.result["model"] == "gpt-4"
|
|
assert context.result["temperature"] == 0.8
|
|
|
|
|
|
@given("I have YAML with for loops and conditionals")
|
|
def step_yaml_block_templates(context):
|
|
"""Create YAML with for loops and conditionals."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
agents:
|
|
{% for item in items %}
|
|
- name: agent_{{ item }}
|
|
id: {{ item }}
|
|
{% endfor %}
|
|
features:
|
|
{% if enabled %}
|
|
- name: feature1
|
|
enabled: true
|
|
{% else %}
|
|
- name: feature1
|
|
enabled: false
|
|
{% endif %}
|
|
"""
|
|
|
|
|
|
@when("I process the YAML with block templates")
|
|
def step_process_block_templates(context):
|
|
"""Process the YAML with block templates."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should render control structures correctly")
|
|
def step_render_control_structures(context):
|
|
"""Verify control structures are rendered correctly."""
|
|
assert context.error is None
|
|
assert "agents" in context.result
|
|
assert len(context.result["agents"]) == 3
|
|
assert context.result["agents"][0]["name"] == "agent_1"
|
|
assert "features" in context.result
|
|
assert context.result["features"][0]["enabled"] == True
|
|
|
|
|
|
@given("I have YAML that would cause parsing errors")
|
|
def step_yaml_parsing_errors(context):
|
|
"""Create YAML that would cause parsing errors."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
config:
|
|
{% for item in items %}
|
|
key{{ item }}: value{{ item }} key{{ item + 10 }}: value{{ item + 10 }}
|
|
{% endfor %}
|
|
"""
|
|
|
|
|
|
@when("I process the problematic YAML")
|
|
def step_process_problematic_yaml(context):
|
|
"""Process the problematic YAML."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should fix common YAML issues and parse successfully")
|
|
def step_fix_yaml_issues(context):
|
|
"""Verify YAML issues are fixed and parsing succeeds."""
|
|
# This should succeed even with the problematic structure
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert "config" in context.result
|
|
|
|
|
|
@given("I have YAML with nested templates and multiple mappings")
|
|
def step_yaml_complex_structures(context):
|
|
"""Create YAML with complex structures."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
{% set base_config = {'timeout': 2, 'retries': 3} %}
|
|
agents:
|
|
{% for item in items %}
|
|
- name: agent_{{ item }}
|
|
config: {{ base_config | yaml }}
|
|
{% if item == 2 %}
|
|
special: true enabled: {{ enabled }}
|
|
{% endif %}
|
|
{% endfor %}
|
|
"""
|
|
|
|
|
|
@when("I process the complex templates")
|
|
def step_process_complex_templates(context):
|
|
"""Process the complex templates."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should handle structure analysis and rendering")
|
|
def step_handle_structure_analysis(context):
|
|
"""Verify structure analysis and rendering work."""
|
|
assert context.error is None
|
|
assert "agents" in context.result
|
|
assert len(context.result["agents"]) == 3
|
|
|
|
|
|
@given("I have YAML using custom yaml, json, and indent filters")
|
|
def step_yaml_custom_filters(context):
|
|
"""Create YAML using custom filters."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
config_yaml: {{ config | yaml }}
|
|
config_json: {{ config | json }}
|
|
indented_text: |
|
|
{{ "line1\\nline2" | indent(4) }}
|
|
"""
|
|
|
|
|
|
@given("I have appropriate data for filtering")
|
|
def step_data_for_filtering(context):
|
|
"""Set up data for filtering."""
|
|
# render_context already has config data
|
|
pass
|
|
|
|
|
|
@when("I process the YAML with custom filters")
|
|
def step_process_custom_filters(context):
|
|
"""Process YAML with custom filters."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {
|
|
"agent_name": "test_agent",
|
|
"config": {"key": "value"},
|
|
}
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the filters should be applied correctly")
|
|
def step_filters_applied_correctly(context):
|
|
"""Verify filters are applied correctly."""
|
|
assert context.error is None
|
|
assert "config_yaml" in context.result
|
|
assert "config_json" in context.result
|
|
assert "indented_text" in context.result
|
|
# The JSON filter was applied but YAML parsing converted it back to dict
|
|
# So we need to check if it's either a dict or contains the JSON string
|
|
config_json = context.result["config_json"]
|
|
if isinstance(config_json, dict):
|
|
assert "key" in config_json
|
|
assert config_json["key"] == "value"
|
|
else:
|
|
assert "key" in str(config_json)
|
|
assert "value" in str(config_json)
|
|
|
|
|
|
@given("I have YAML using custom list, dict, and none tests")
|
|
def step_yaml_custom_tests(context):
|
|
"""Create YAML using custom tests."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
{% if items is list %}
|
|
items_type: list
|
|
{% endif %}
|
|
{% if config is dict %}
|
|
config_type: dict
|
|
{% endif %}
|
|
{% if missing_value is none %}
|
|
missing_handled: true
|
|
{% endif %}
|
|
"""
|
|
|
|
|
|
@given("I have appropriate data for testing")
|
|
def step_data_for_testing(context):
|
|
"""Set up data for testing."""
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {
|
|
"agent_name": "test_agent",
|
|
"agent_type": "llm",
|
|
"model_name": "gpt-4",
|
|
"temperature": 0.8,
|
|
"items": [1, 2, 3],
|
|
"config": {"key": "value"},
|
|
"enabled": True,
|
|
}
|
|
context.render_context["missing_value"] = None
|
|
|
|
|
|
@when("I process the YAML with custom tests")
|
|
def step_process_custom_tests(context):
|
|
"""Process YAML with custom tests."""
|
|
try:
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the tests should work correctly")
|
|
def step_tests_work_correctly(context):
|
|
"""Verify tests work correctly."""
|
|
assert context.error is None
|
|
assert "items_type" in context.result
|
|
assert context.result["items_type"] == "list"
|
|
assert "config_type" in context.result
|
|
assert context.result["config_type"] == "dict"
|
|
assert "missing_handled" in context.result
|
|
assert context.result["missing_handled"] == True
|
|
|
|
|
|
@given("I have an invalid file path")
|
|
def step_invalid_file_path(context):
|
|
"""Set up invalid file path."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.invalid_path = Path("/nonexistent/path/file.yaml")
|
|
|
|
|
|
@when("I try to process the file")
|
|
def step_try_process_invalid_file(context):
|
|
"""Try to process the invalid file."""
|
|
try:
|
|
context.result = context.processor.process_file(context.invalid_path)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should raise an appropriate error")
|
|
def step_raise_appropriate_error(context):
|
|
"""Verify an appropriate error is raised."""
|
|
assert context.error is not None
|
|
assert isinstance(context.error, FileNotFoundError)
|
|
|
|
|
|
@given("I have empty YAML content")
|
|
def step_empty_yaml_content(context):
|
|
"""Create empty YAML content."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = ""
|
|
|
|
|
|
@when("I process the empty content")
|
|
def step_process_empty_content(context):
|
|
"""Process the empty content."""
|
|
try:
|
|
context.result = context.processor.process_string(context.yaml_string)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should handle it gracefully")
|
|
def step_handle_gracefully(context):
|
|
"""Verify empty content is handled gracefully."""
|
|
assert context.error is None
|
|
assert context.result is None
|
|
|
|
|
|
@given("I have templates that produce invalid YAML")
|
|
def step_templates_invalid_yaml(context):
|
|
"""Create templates that produce invalid YAML."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
config:
|
|
{% for item in items %}
|
|
{{ item }}: value: invalid: syntax
|
|
{% endfor %}
|
|
"""
|
|
|
|
|
|
@when("I process the malformed templates")
|
|
def step_process_malformed_templates(context):
|
|
"""Process the malformed templates."""
|
|
try:
|
|
# Mock yaml.safe_load to raise YAMLError on first attempt
|
|
original_safe_load = yaml.safe_load
|
|
|
|
def mock_safe_load(content):
|
|
if hasattr(mock_safe_load, "_called"):
|
|
return original_safe_load(content)
|
|
mock_safe_load._called = True
|
|
raise yaml.YAMLError("Test YAML error")
|
|
|
|
with patch("yaml.safe_load", side_effect=mock_safe_load):
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should attempt to fix YAML issues")
|
|
def step_attempt_fix_yaml(context):
|
|
"""Verify it attempts to fix YAML issues."""
|
|
# The processor should handle the error and try to fix issues
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
|
|
|
|
@given("I have YAML with various template block types")
|
|
def step_yaml_various_blocks(context):
|
|
"""Create YAML with various template block types."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
{% if enabled %}
|
|
section1:
|
|
name: conditional_section
|
|
{% endif %}
|
|
{% for item in items %}
|
|
item_{{ item }}:
|
|
value: {{ item }}
|
|
{% endfor %}
|
|
inline_var: {{ agent_name }}
|
|
"""
|
|
|
|
|
|
@when("I analyze the structure")
|
|
def step_analyze_structure(context):
|
|
"""Analyze the structure."""
|
|
try:
|
|
context.structure = context.processor._analyze_structure(context.yaml_string)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should correctly identify block and inline templates")
|
|
def step_identify_templates(context):
|
|
"""Verify template identification."""
|
|
assert context.error is None
|
|
assert "has_block_templates" in context.structure
|
|
assert "has_inline_templates" in context.structure
|
|
assert context.structure["has_block_templates"] == True
|
|
assert context.structure["has_inline_templates"] == True
|
|
|
|
|
|
@given("I have rendered YAML with multiple mappings per line")
|
|
def step_rendered_multiple_mappings(context):
|
|
"""Create rendered YAML with multiple mappings per line."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.rendered_content = """
|
|
config:
|
|
key1: value1 key2: value2
|
|
key3: value3 key4: value4
|
|
"""
|
|
|
|
|
|
@when("I fix YAML structural issues")
|
|
def step_fix_structural_issues(context):
|
|
"""Fix YAML structural issues."""
|
|
try:
|
|
context.result = context.processor._fix_yaml_issues(context.rendered_content)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should split mappings to separate lines")
|
|
def step_split_mappings(context):
|
|
"""Verify mappings are split to separate lines."""
|
|
assert context.error is None
|
|
assert "key1: value1" in context.result
|
|
assert "key2: value2" in context.result
|
|
# Should have separate lines
|
|
lines = context.result.split("\n")
|
|
key1_line = next((line for line in lines if "key1:" in line), None)
|
|
key2_line = next((line for line in lines if "key2:" in line), None)
|
|
assert key1_line is not None
|
|
assert key2_line is not None
|
|
assert key1_line != key2_line
|
|
|
|
|
|
@given("I have a basic context dictionary")
|
|
def step_basic_context(context):
|
|
"""Create a basic context dictionary."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.basic_context = {"name": "test", "value": 42}
|
|
|
|
|
|
@when("I prepare the full rendering context")
|
|
def step_prepare_full_context(context):
|
|
"""Prepare the full rendering context."""
|
|
try:
|
|
context.full_context = context.processor._prepare_context(context.basic_context)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should include built-in functions and utilities")
|
|
def step_include_builtins(context):
|
|
"""Verify built-in functions and utilities are included."""
|
|
assert context.error is None
|
|
assert "range" in context.full_context
|
|
assert "len" in context.full_context
|
|
assert "enumerate" in context.full_context
|
|
assert "join" in context.full_context
|
|
assert "keys" in context.full_context
|
|
assert "name" in context.full_context # Original context
|
|
assert "value" in context.full_context # Original context
|
|
|
|
|
|
@given("I have YAML with mixed template blocks and regular content")
|
|
def step_yaml_mixed_content(context):
|
|
"""Create YAML with mixed content."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
regular_key: regular_value
|
|
{% for item in items %}
|
|
dynamic_{{ item }}: value_{{ item }}
|
|
{% endfor %}
|
|
another_regular: another_value
|
|
inline_var: {{ agent_name }}
|
|
"""
|
|
|
|
|
|
@when("I split into sections and process")
|
|
def step_split_and_process(context):
|
|
"""Split into sections and process."""
|
|
try:
|
|
context.sections = context.processor._split_into_sections(context.yaml_string)
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {
|
|
"agent_name": "test_agent",
|
|
"agent_type": "llm",
|
|
"items": [1, 2, 3],
|
|
}
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("it should handle each section appropriately")
|
|
def step_handle_sections(context):
|
|
"""Verify sections are handled appropriately."""
|
|
if context.error is not None:
|
|
print(f"Error: {context.error}")
|
|
assert context.error is None
|
|
assert context.sections is not None
|
|
assert len(context.sections) > 0
|
|
assert context.result is not None
|
|
|
|
|
|
@given("I have template blocks at different indentation levels")
|
|
def step_template_blocks_indentation(context):
|
|
"""Create template blocks at different indentation levels."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
level1:
|
|
{% for item in items %}
|
|
item_{{ item }}:
|
|
{% if item == 2 %}
|
|
special: true
|
|
{% endif %}
|
|
value: {{ item }}
|
|
{% endfor %}
|
|
"""
|
|
|
|
|
|
@when("I render the blocks")
|
|
def step_render_blocks(context):
|
|
"""Render the blocks."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {"agent_name": "test_agent", "items": [1, 2, 3]}
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("indentation should be preserved correctly")
|
|
def step_indentation_preserved(context):
|
|
"""Verify indentation is preserved correctly."""
|
|
if context.error is not None:
|
|
print(f"Error: {context.error}")
|
|
assert context.error is None
|
|
assert "level1" in context.result
|
|
assert "item_1" in context.result["level1"]
|
|
assert "item_2" in context.result["level1"]
|
|
assert "special" in context.result["level1"]["item_2"]
|
|
|
|
|
|
@given("I have YAML with template blocks and lines without colons")
|
|
def step_yaml_blocks_no_colons(context):
|
|
"""Create YAML with template blocks and lines without colons."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
config:
|
|
{% for item in items %}
|
|
- agent_{{ item }}
|
|
- name_agent_{{ item }}
|
|
{% endfor %}
|
|
# Just a comment line
|
|
data: []
|
|
"""
|
|
|
|
|
|
@when("I process the YAML with structured rendering")
|
|
def step_process_structured_rendering(context):
|
|
"""Process YAML with structured rendering."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {"items": [1, 2]}
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("non-colon lines should be preserved")
|
|
def step_non_colon_preserved(context):
|
|
"""Verify non-colon lines are preserved."""
|
|
if context.error is not None:
|
|
print(f"Error: {context.error}")
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
|
|
|
|
@given("I have YAML with both inline and block templates mixed")
|
|
def step_yaml_mixed_templates(context):
|
|
"""Create YAML with mixed inline and block templates."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
name: {{ agent_name }}
|
|
config:
|
|
{% for item in items %}
|
|
item_{{ item }}: value_{{ item }}
|
|
{% endfor %}
|
|
inline_value: {{ temperature }}
|
|
"""
|
|
|
|
|
|
@when("I split and process the mixed content")
|
|
def step_split_process_mixed(context):
|
|
"""Split and process mixed content."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {
|
|
"agent_name": "test",
|
|
"items": [1, 2],
|
|
"temperature": 0.8,
|
|
}
|
|
context.sections = context.processor._split_into_sections(context.yaml_string)
|
|
context.result = context.processor.process_string(
|
|
context.yaml_string, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("inline templates should be handled in regular sections")
|
|
def step_inline_handled_regular(context):
|
|
"""Verify inline templates are handled in regular sections."""
|
|
assert context.error is None
|
|
assert context.sections is not None
|
|
assert len(context.sections) > 0
|
|
assert context.result is not None
|
|
|
|
|
|
@given("I have a template block that needs rendering")
|
|
def step_template_block_rendering(context):
|
|
"""Create a template block that needs rendering."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.block_content = """
|
|
{% for item in items %}
|
|
item_{{ item }}: value_{{ item }} extra_{{ item }}: data_{{ item }}
|
|
{% endfor %}
|
|
"""
|
|
|
|
|
|
@when("I render the block directly")
|
|
def step_render_block_directly(context):
|
|
"""Render the block directly."""
|
|
try:
|
|
if not hasattr(context, "render_context"):
|
|
context.render_context = {"items": [1, 2]}
|
|
context.result = context.processor._render_block(
|
|
context.block_content, 0, context.render_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the block should be rendered with proper structure")
|
|
def step_block_rendered_structure(context):
|
|
"""Verify block is rendered with proper structure."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
assert "item_1:" in context.result
|
|
assert "item_2:" in context.result
|
|
|
|
|
|
@given("I have YAML with various formatting issues")
|
|
def step_yaml_formatting_issues(context):
|
|
"""Create YAML with various formatting issues."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.problematic_yaml = """
|
|
config:
|
|
key1: value1 key2: value2
|
|
key3: key4:
|
|
key5: value5 key6:
|
|
"""
|
|
|
|
|
|
@when("I apply YAML issue fixing")
|
|
def step_apply_yaml_fixing(context):
|
|
"""Apply YAML issue fixing."""
|
|
try:
|
|
context.result = context.processor._fix_yaml_issues(context.problematic_yaml)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("all edge cases should be handled correctly")
|
|
def step_edge_cases_handled(context):
|
|
"""Verify all edge cases are handled correctly."""
|
|
assert context.error is None
|
|
assert context.result is not None
|
|
# Check that multiple mappings per line are fixed
|
|
lines = context.result.split("\n")
|
|
for line in lines:
|
|
if ":" in line:
|
|
# Should not have multiple colons indicating multiple mappings
|
|
colon_count = line.count(":")
|
|
if colon_count > 1:
|
|
# This is acceptable if it's just one key:value pair
|
|
# The fix should ensure proper YAML structure
|
|
pass
|
|
|
|
|
|
@given("I have YAML with macro template blocks")
|
|
def step_yaml_macro_blocks(context):
|
|
"""Create YAML with macro template blocks."""
|
|
context.processor = InlineYAMLJinja()
|
|
context.yaml_string = """
|
|
{% macro render_agent(name, type) %}
|
|
agent_{{ name }}:
|
|
type: {{ type }}
|
|
enabled: true
|
|
{% endmacro %}
|
|
|
|
agents:
|
|
{{ render_agent('test', 'llm') }}
|
|
"""
|
|
|
|
|
|
@when("I split the content into sections")
|
|
def step_split_content_sections(context):
|
|
"""Split content into sections."""
|
|
try:
|
|
context.sections = context.processor._split_into_sections(context.yaml_string)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@then("macro blocks should be identified correctly")
|
|
def step_macro_blocks_identified(context):
|
|
"""Verify macro blocks are identified correctly."""
|
|
assert context.error is None
|
|
assert context.sections is not None
|
|
assert len(context.sections) > 0
|
|
# Check that we have both template blocks and regular sections
|
|
has_template_block = any(
|
|
section.get("type") == "template_block" for section in context.sections
|
|
)
|
|
has_regular_section = any(
|
|
section.get("type") == "regular" for section in context.sections
|
|
)
|
|
assert has_template_block or has_regular_section
|
|
|
|
|
|
def after_scenario(context, scenario):
|
|
"""Clean up after each scenario."""
|
|
if hasattr(context, "temp_files"):
|
|
for temp_file in context.temp_files:
|
|
try:
|
|
Path(temp_file).unlink()
|
|
except:
|
|
pass
|