forked from HAL9000/cleveragents-core
547 lines
16 KiB
Python
547 lines
16 KiB
Python
"""Simplified step definitions for InlineYAMLJinja coverage testing."""
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
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 instance for coverage")
|
|
def step_have_instance(context):
|
|
"""Create InlineYAMLJinja instance."""
|
|
context.jinja = InlineYAMLJinja()
|
|
context.results = {}
|
|
context.test_files = []
|
|
|
|
|
|
@when("I test the initialization for coverage")
|
|
def step_test_initialization(context):
|
|
"""Test initialization."""
|
|
# Test __init__ method
|
|
jinja = InlineYAMLJinja()
|
|
context.results["init"] = jinja is not None
|
|
context.results["env"] = jinja.env is not None
|
|
|
|
|
|
@then("the instance should be properly configured for coverage")
|
|
def step_instance_configured(context):
|
|
"""Verify instance configuration."""
|
|
assert context.results["init"]
|
|
assert context.results["env"]
|
|
assert context.jinja.env is not None
|
|
|
|
|
|
@then("custom filters should be available for coverage")
|
|
def step_custom_filters_available(context):
|
|
"""Test custom filters."""
|
|
filters = context.jinja.env.filters
|
|
|
|
# Test yaml filter
|
|
test_data = {"key": "value", "list": [1, 2, 3]}
|
|
yaml_result = filters["yaml"](test_data)
|
|
assert "key: value" in yaml_result
|
|
|
|
# Test json filter
|
|
json_result = filters["json"](test_data)
|
|
assert '"key": "value"' in json_result
|
|
|
|
# Test indent filter
|
|
text = "line1\\nline2"
|
|
indent_result = filters["indent"](text, 4)
|
|
assert " line1" in indent_result
|
|
|
|
context.results["filters"] = True
|
|
|
|
|
|
@then("custom tests should be available for coverage")
|
|
def step_custom_tests_available(context):
|
|
"""Test custom tests."""
|
|
tests = context.jinja.env.tests
|
|
|
|
# Test list test
|
|
assert tests["list"]([1, 2, 3])
|
|
assert not tests["list"]({"key": "value"})
|
|
|
|
# Test dict test
|
|
assert tests["dict"]({"key": "value"})
|
|
assert not tests["dict"]([1, 2, 3])
|
|
|
|
# Test none test
|
|
assert tests["none"](None)
|
|
assert not tests["none"]("value")
|
|
|
|
context.results["tests"] = True
|
|
|
|
|
|
@given("I have a test YAML file with templates for coverage")
|
|
def step_have_test_file(context):
|
|
"""Create test YAML file."""
|
|
content = """
|
|
config:
|
|
name: "{{ app_name }}"
|
|
version: "{{ version }}"
|
|
port: {{ port }}
|
|
features:
|
|
{% for feature in features %}
|
|
- name: "{{ feature.name }}"
|
|
enabled: {{ feature.enabled }}
|
|
{% endfor %}
|
|
"""
|
|
|
|
# Create temporary file
|
|
temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
|
|
temp_file.write(content.strip())
|
|
temp_file.close()
|
|
|
|
context.test_file = Path(temp_file.name)
|
|
context.test_files.append(context.test_file)
|
|
|
|
context.test_context = {
|
|
"app_name": "TestApp",
|
|
"version": "1.0",
|
|
"port": 8080,
|
|
"features": [
|
|
{"name": "auth", "enabled": True},
|
|
{"name": "logging", "enabled": False},
|
|
],
|
|
}
|
|
|
|
|
|
@when("I call process_file for coverage")
|
|
def step_call_process_file(context):
|
|
"""Test process_file method."""
|
|
# Test without context (deferred)
|
|
result1 = context.jinja.process_file(context.test_file)
|
|
context.results["process_file_deferred"] = result1
|
|
|
|
# Test with context (immediate)
|
|
result2 = context.jinja.process_file(context.test_file, context.test_context)
|
|
context.results["process_file_immediate"] = result2
|
|
|
|
|
|
@then("the file should be processed correctly for coverage")
|
|
def step_file_processed(context):
|
|
"""Verify file processing."""
|
|
# Check deferred result
|
|
deferred = context.results["process_file_deferred"]
|
|
assert "__yaml_template__" in deferred
|
|
|
|
# Check immediate result
|
|
immediate = context.results["process_file_immediate"]
|
|
assert "config" in immediate
|
|
assert immediate["config"]["name"] == "TestApp"
|
|
assert immediate["config"]["port"] == 8080
|
|
assert len(immediate["features"]) == 2
|
|
|
|
|
|
@when("I test process_string with various inputs for coverage")
|
|
def step_test_process_string(context):
|
|
"""Test process_string method with various inputs."""
|
|
# Test simple YAML without templates
|
|
simple_yaml = "config:\n host: localhost\n port: 5432"
|
|
result1 = context.jinja.process_string(simple_yaml)
|
|
context.results["simple"] = result1
|
|
|
|
# Test YAML with templates and context
|
|
template_yaml = "config:\n host: '{{ host }}'\n port: {{ port }}"
|
|
test_context = {"host": "example.com", "port": 443}
|
|
result2 = context.jinja.process_string(template_yaml, test_context)
|
|
context.results["with_context"] = result2
|
|
|
|
# Test YAML with templates but no context (deferred)
|
|
result3 = context.jinja.process_string(template_yaml)
|
|
context.results["deferred"] = result3
|
|
|
|
# Test complex template
|
|
complex_yaml = """
|
|
items:
|
|
{% for item in items %}
|
|
- name: "{{ item }}"
|
|
index: {{ loop.index }}
|
|
{% endfor %}
|
|
config:
|
|
total: {{ items | length }}
|
|
"""
|
|
complex_context = {"items": ["a", "b", "c"]}
|
|
result4 = context.jinja.process_string(complex_yaml, complex_context)
|
|
context.results["complex"] = result4
|
|
|
|
|
|
@then("all variations should work correctly for coverage")
|
|
def step_all_variations_work(context):
|
|
"""Verify all process_string variations."""
|
|
# Simple YAML
|
|
assert "config" in context.results["simple"]
|
|
assert context.results["simple"]["config"]["host"] == "localhost"
|
|
|
|
# With context
|
|
assert context.results["with_context"]["config"]["host"] == "example.com"
|
|
assert context.results["with_context"]["config"]["port"] == 443
|
|
|
|
# Deferred
|
|
assert "__yaml_template__" in context.results["deferred"]
|
|
|
|
# Complex
|
|
assert len(context.results["complex"]["items"]) == 3
|
|
assert context.results["complex"]["config"]["total"] == 3
|
|
|
|
|
|
@when("I test _has_templates with various content for coverage")
|
|
def step_test_has_templates(context):
|
|
"""Test _has_templates method."""
|
|
test_cases = [
|
|
("plain yaml without templates", False),
|
|
("value: {{ variable }}", True),
|
|
("{% for item in items %}", True),
|
|
("value: plain text", False),
|
|
("mixed: {{ var }} and plain", True),
|
|
("block: {% if condition %}", True),
|
|
("{# comment #} value: test", False),
|
|
]
|
|
|
|
context.results["has_templates"] = {}
|
|
for content, expected in test_cases:
|
|
result = context.jinja._has_templates(content)
|
|
context.results["has_templates"][content] = (result, expected)
|
|
|
|
|
|
@then("template detection should work correctly for coverage")
|
|
def step_template_detection_works(context):
|
|
"""Verify template detection."""
|
|
for content, (result, expected) in context.results["has_templates"].items():
|
|
assert result == expected, f"Failed for: {content}"
|
|
|
|
|
|
@when("I test _render_and_parse with templates for coverage")
|
|
def step_test_render_and_parse(context):
|
|
"""Test _render_and_parse method."""
|
|
template_content = """
|
|
server:
|
|
host: "{{ server_host }}"
|
|
port: {{ server_port }}
|
|
ssl: {{ ssl_enabled }}
|
|
database:
|
|
url: "{{ db_url }}"
|
|
"""
|
|
|
|
test_context = {
|
|
"server_host": "localhost",
|
|
"server_port": 8080,
|
|
"ssl_enabled": False,
|
|
"db_url": "postgresql://localhost/test",
|
|
}
|
|
|
|
result = context.jinja._render_and_parse(template_content, test_context)
|
|
context.results["render_parse"] = result
|
|
|
|
|
|
@then("rendering and parsing should work correctly for coverage")
|
|
def step_render_parse_works(context):
|
|
"""Verify render and parse."""
|
|
result = context.results["render_parse"]
|
|
assert result["server"]["host"] == "localhost"
|
|
assert result["server"]["port"] == 8080
|
|
assert result["server"]["ssl"] is False
|
|
assert "postgresql" in result["database"]["url"]
|
|
|
|
|
|
@when("I test _analyze_structure for coverage")
|
|
def step_test_analyze_structure(context):
|
|
"""Test _analyze_structure method."""
|
|
content = """
|
|
global_config:
|
|
app_name: "MyApp"
|
|
|
|
dynamic_routes:
|
|
{% for route in routes %}
|
|
- path: "{{ route.path }}"
|
|
method: "{{ route.method }}"
|
|
{% endfor %}
|
|
|
|
static_config:
|
|
database:
|
|
host: localhost
|
|
|
|
conditional_features:
|
|
{% if enable_auth %}
|
|
auth:
|
|
provider: oauth2
|
|
{% endif %}
|
|
"""
|
|
|
|
result = context.jinja._analyze_structure(content)
|
|
context.results["structure"] = result
|
|
|
|
|
|
@then("structure analysis should work correctly for coverage")
|
|
def step_structure_analysis_works(context):
|
|
"""Verify structure analysis."""
|
|
result = context.results["structure"]
|
|
assert "has_block_templates" in result
|
|
assert "has_inline_templates" in result
|
|
assert "max_indent" in result
|
|
assert "template_blocks" in result
|
|
assert result["has_block_templates"] is True # Should detect {% for %} and {% if %}
|
|
assert isinstance(result["max_indent"], int)
|
|
|
|
|
|
@when("I test _render_structured for coverage")
|
|
def step_test_render_structured(context):
|
|
"""Test _render_structured method."""
|
|
content = """
|
|
products:
|
|
{% for product in products %}
|
|
- id: {{ product.id }}
|
|
name: "{{ product.name }}"
|
|
price: {{ product.price }}
|
|
{% endfor %}
|
|
summary:
|
|
total: {{ products | length }}
|
|
"""
|
|
|
|
test_context = {
|
|
"products": [
|
|
{"id": 1, "name": "Widget A", "price": 100},
|
|
{"id": 2, "name": "Widget B", "price": 50},
|
|
]
|
|
}
|
|
|
|
result = context.jinja._render_structured(content, test_context)
|
|
context.results["structured"] = result
|
|
|
|
|
|
@then("structured rendering should work correctly for coverage")
|
|
def step_structured_rendering_works(context):
|
|
"""Verify structured rendering."""
|
|
result = context.results["structured"]
|
|
assert "products:" in result
|
|
assert "Widget A" in result
|
|
assert "Widget B" in result
|
|
assert "total: 2" in result
|
|
|
|
|
|
@when("I test _split_into_sections for coverage")
|
|
def step_test_split_sections(context):
|
|
"""Test _split_into_sections method."""
|
|
content = """
|
|
global_config:
|
|
app_name: "MyApp"
|
|
|
|
dynamic_routes:
|
|
{% for route in routes %}
|
|
- path: "{{ route.path }}"
|
|
{% endfor %}
|
|
|
|
static_config:
|
|
database:
|
|
host: localhost
|
|
"""
|
|
|
|
result = context.jinja._split_into_sections(content)
|
|
context.results["sections"] = result
|
|
|
|
|
|
@then("section splitting should work correctly for coverage")
|
|
def step_section_splitting_works(context):
|
|
"""Verify section splitting."""
|
|
sections = context.results["sections"]
|
|
assert isinstance(sections, list)
|
|
assert len(sections) > 0
|
|
|
|
# Should have both regular and template sections
|
|
section_types = [s["type"] for s in sections]
|
|
assert "regular" in section_types
|
|
assert "template_block" in section_types
|
|
|
|
|
|
@when("I test _render_block for coverage")
|
|
def step_test_render_block(context):
|
|
"""Test _render_block method."""
|
|
block_content = """
|
|
{% for user in users %}
|
|
- name: "{{ user.name }}"
|
|
email: "{{ user.email }}"
|
|
active: {{ user.active }}
|
|
{% endfor %}
|
|
"""
|
|
|
|
test_context = {
|
|
"users": [
|
|
{"name": "Alice", "email": "alice@example.com", "active": True},
|
|
{"name": "Bob", "email": "bob@example.com", "active": False},
|
|
]
|
|
}
|
|
|
|
result = context.jinja._render_block(block_content, 2, test_context)
|
|
context.results["block"] = result
|
|
|
|
|
|
@then("block rendering should work correctly for coverage")
|
|
def step_block_rendering_works(context):
|
|
"""Verify block rendering."""
|
|
result = context.results["block"]
|
|
assert "Alice" in result
|
|
assert "Bob" in result
|
|
assert "alice@example.com" in result
|
|
assert "true" in result.lower() or "True" in result
|
|
|
|
|
|
@when("I test _fix_yaml_issues for coverage")
|
|
def step_test_fix_yaml_issues(context):
|
|
"""Test _fix_yaml_issues method."""
|
|
problematic_content = """
|
|
config:
|
|
key1: value1 key2: value2
|
|
key3: value3 key4: value4
|
|
normal_key: normal_value
|
|
"""
|
|
|
|
result = context.jinja._fix_yaml_issues(problematic_content)
|
|
context.results["fixed"] = result
|
|
|
|
|
|
@then("YAML issues should be fixed for coverage")
|
|
def step_yaml_issues_fixed(context):
|
|
"""Verify YAML issues are fixed."""
|
|
result = context.results["fixed"]
|
|
# Should separate multiple mappings on same line
|
|
lines = result.split("\\n")
|
|
key_lines = [line for line in lines if "key1:" in line or "key2:" in line]
|
|
assert len(key_lines) >= 1 # Should have separated the keys
|
|
|
|
|
|
@when("I test _prepare_context for coverage")
|
|
def step_test_prepare_context(context):
|
|
"""Test _prepare_context method."""
|
|
basic_context = {"name": "test", "items": [1, 2, 3, 4, 5], "data": {"a": 1, "b": 2}}
|
|
|
|
result = context.jinja._prepare_context(basic_context)
|
|
context.results["prepared"] = result
|
|
|
|
|
|
@then("context should be prepared correctly for coverage")
|
|
def step_context_prepared(context):
|
|
"""Verify context preparation."""
|
|
result = context.results["prepared"]
|
|
|
|
# Should include original context
|
|
assert result["name"] == "test"
|
|
assert len(result["items"]) == 5
|
|
|
|
# Should include built-in functions
|
|
assert "range" in result
|
|
assert "len" in result
|
|
assert "int" in result
|
|
assert "str" in result
|
|
assert "join" in result
|
|
|
|
|
|
@when("I test deferred processing for coverage")
|
|
def step_test_deferred_processing(context):
|
|
"""Test _store_for_deferred and render_deferred methods."""
|
|
template_content = """
|
|
server:
|
|
host: "{{ host }}"
|
|
port: {{ port }}
|
|
database:
|
|
url: "{{ db_url }}"
|
|
"""
|
|
|
|
# Test _store_for_deferred
|
|
deferred = context.jinja._store_for_deferred(template_content)
|
|
context.results["stored"] = deferred
|
|
|
|
# Test render_deferred with deferred template
|
|
render_context = {
|
|
"host": "production.example.com",
|
|
"port": 443,
|
|
"db_url": "postgresql://prod/db",
|
|
}
|
|
rendered = context.jinja.render_deferred(deferred, render_context)
|
|
context.results["rendered"] = rendered
|
|
|
|
# Test render_deferred with non-deferred config
|
|
regular_config = {"server": {"host": "localhost", "port": 8080}}
|
|
unchanged = context.jinja.render_deferred(regular_config, render_context)
|
|
context.results["unchanged"] = unchanged
|
|
|
|
|
|
@then("deferred processing should work correctly for coverage")
|
|
def step_deferred_processing_works(context):
|
|
"""Verify deferred processing."""
|
|
# Check stored deferred template
|
|
stored = context.results["stored"]
|
|
assert "__yaml_template__" in stored
|
|
assert stored["__yaml_template__"]["type"] == "full"
|
|
assert stored["__yaml_template__"]["has_jinja2"] is True
|
|
|
|
# Check rendered result
|
|
rendered = context.results["rendered"]
|
|
assert rendered["server"]["host"] == "production.example.com"
|
|
assert rendered["server"]["port"] == 443
|
|
|
|
# Check unchanged result
|
|
unchanged = context.results["unchanged"]
|
|
assert unchanged["server"]["host"] == "localhost"
|
|
assert unchanged["server"]["port"] == 8080
|
|
|
|
|
|
@when("I test error conditions for coverage")
|
|
def step_test_error_conditions(context):
|
|
"""Test error handling paths."""
|
|
context.results["errors"] = {}
|
|
|
|
# Test invalid YAML
|
|
try:
|
|
invalid_yaml = "invalid: yaml: structure: [broken"
|
|
context.jinja._render_and_parse(invalid_yaml, {})
|
|
context.results["errors"]["yaml_error"] = False
|
|
except Exception as e:
|
|
context.results["errors"]["yaml_error"] = True
|
|
context.results["errors"]["yaml_exception"] = str(e)
|
|
|
|
# Test template with missing variables (should not crash)
|
|
try:
|
|
template_with_missing = "config: {{ missing_variable }}"
|
|
result = context.jinja._render_and_parse(template_with_missing, {})
|
|
context.results["errors"]["missing_var"] = "handled"
|
|
except Exception as e:
|
|
context.results["errors"]["missing_var"] = str(e)
|
|
|
|
# Test empty content
|
|
try:
|
|
result = context.jinja.process_string("")
|
|
context.results["errors"]["empty"] = result
|
|
except Exception as e:
|
|
context.results["errors"]["empty"] = str(e)
|
|
|
|
|
|
@then("errors should be handled correctly for coverage")
|
|
def step_errors_handled(context):
|
|
"""Verify error handling."""
|
|
errors = context.results["errors"]
|
|
|
|
# Should handle invalid YAML gracefully
|
|
assert "yaml_error" in errors
|
|
|
|
# Should handle missing variables
|
|
assert "missing_var" in errors
|
|
|
|
# Should handle empty content
|
|
assert "empty" in errors
|
|
|
|
|
|
def after_scenario(context, scenario):
|
|
"""Clean up after each scenario."""
|
|
# Clean up test files
|
|
if hasattr(context, "test_files"):
|
|
for test_file in context.test_files:
|
|
if test_file.exists():
|
|
test_file.unlink()
|