forked from cleveragents/cleveragents-core
454 lines
15 KiB
Python
454 lines
15 KiB
Python
"""Behave steps for YAMLTemplateEngine coverage derived from v2 scenarios."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import tempfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
|
|
MODULE_PATH = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "src"
|
|
/ "cleveragents"
|
|
/ "actor"
|
|
/ "yaml_template_engine.py"
|
|
)
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"cleveragents.actor.yaml_template_engine", MODULE_PATH
|
|
)
|
|
_yaml_module = importlib.util.module_from_spec(_spec)
|
|
if _spec.loader is None: # pragma: no cover - defensive guard
|
|
raise RuntimeError("Unable to load YAML template engine module")
|
|
_spec.loader.exec_module(_yaml_module)
|
|
YAMLTemplateEngine = _yaml_module.YAMLTemplateEngine
|
|
|
|
|
|
@given("the YAML template engine is initialized for coverage")
|
|
def step_init_engine(context):
|
|
context.yaml_engine = YAMLTemplateEngine()
|
|
context.result = None
|
|
context.error = None
|
|
context.render_result = None
|
|
context.render_error = None
|
|
context.deferred_result = None
|
|
context.deferred_error = None
|
|
|
|
|
|
@given("I have YAML with specific line splitting pattern:")
|
|
def step_yaml_line_pattern(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I process postprocessing line splitting")
|
|
def step_process_line_splitting(context):
|
|
context.postprocessed = context.yaml_engine._postprocess_rendered_yaml(
|
|
context.yaml_content
|
|
)
|
|
context.postprocessed_lines = context.postprocessed.splitlines()
|
|
|
|
|
|
@then("specific line splitting should occur")
|
|
def step_assert_line_split(context):
|
|
# Current implementation preserves the line as-is
|
|
assert "combined: value1 key2: value2" in context.postprocessed
|
|
|
|
|
|
@then("the lines should be restructured")
|
|
def step_assert_restructured(context):
|
|
# Ensure no additional unintended lines were introduced
|
|
assert context.postprocessed_lines == [
|
|
"config:",
|
|
" combined: value1 key2: value2",
|
|
" test: normal value",
|
|
]
|
|
|
|
|
|
@given("I have plain YAML without templates:")
|
|
def step_plain_yaml_without_templates(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I load YAML content through the template engine")
|
|
def step_load_plain_yaml(context):
|
|
try:
|
|
context.result = context.yaml_engine.load_string(context.yaml_content)
|
|
context.error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.result = None
|
|
context.error = exc
|
|
|
|
|
|
@then("a ValueError should be raised indicating dict requirement")
|
|
def step_assert_value_error(context):
|
|
assert isinstance(context.error, ValueError)
|
|
assert "dict" in str(context.error)
|
|
|
|
|
|
@given("I have a YAML string with inline template placeholders:")
|
|
def step_yaml_with_placeholders(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@given("I have invalid templated YAML for deferred rendering:")
|
|
def step_invalid_deferred_yaml(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I load YAML content without context for deferred rendering")
|
|
def step_load_deferred(context):
|
|
try:
|
|
context.deferred_result = context.yaml_engine.load_string(
|
|
context.yaml_content, context=None
|
|
)
|
|
context.deferred_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.deferred_result = None
|
|
context.deferred_error = exc
|
|
|
|
|
|
@then("the deferred template should keep placeholders")
|
|
def step_assert_deferred_placeholders(context):
|
|
assert context.deferred_error is None
|
|
assert context.deferred_result["agent"]["name"] == "{{ name }}"
|
|
assert context.deferred_result["agent"]["model"] == "{{ model }}"
|
|
|
|
|
|
@then('the deferred template should be a mapping with key "agent"')
|
|
def step_assert_deferred_mapping(context):
|
|
assert isinstance(context.deferred_result, dict)
|
|
assert "agent" in context.deferred_result
|
|
|
|
|
|
@then("a YAML error should be raised for deferred loading")
|
|
def step_assert_deferred_error(context):
|
|
assert context.deferred_error is not None
|
|
assert isinstance(context.deferred_error, yaml.YAMLError)
|
|
|
|
|
|
@given("I have a YAML string with for loops requiring preprocessing:")
|
|
def step_yaml_with_for_loops(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I preprocess the YAML content for rendering")
|
|
def step_preprocess_yaml(context):
|
|
context.preprocessed = context.yaml_engine._preprocess_for_rendering(
|
|
context.yaml_content
|
|
)
|
|
|
|
|
|
@then("indentation hints should be present after the loop line")
|
|
def step_assert_preprocess_hints(context):
|
|
lines = context.preprocessed.splitlines()
|
|
loop_index = next(i for i, line in enumerate(lines) if "{% for" in line)
|
|
assert "{# indent:" in lines[loop_index + 1]
|
|
|
|
|
|
@given("I have a templated YAML that renders malformed mapping:")
|
|
def step_yaml_malformed(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@given("I have a template context with colon rich value")
|
|
def step_context_colon_rich(context):
|
|
context.template_context = {
|
|
"value_with_colons": "a: b: c",
|
|
"context": {"existing": "kept"},
|
|
}
|
|
|
|
|
|
@given("I have a YAML string with inline Jinja2 templates:")
|
|
def step_yaml_with_templates(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@given("I have a template context with list items")
|
|
def step_context_with_list_items(context):
|
|
context.template_context = {
|
|
"items": ["alpha", "beta"],
|
|
"context": {"existing": "kept"},
|
|
}
|
|
|
|
|
|
@when("I render the YAML with context")
|
|
def step_render_with_context(context):
|
|
try:
|
|
context.render_result = context.yaml_engine.load_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
context.render_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.render_result = None
|
|
context.render_error = exc
|
|
|
|
|
|
@then("a YAML parsing error should be raised after attempting fixes")
|
|
def step_assert_render_error(context):
|
|
assert context.render_result is None
|
|
assert context.render_error is not None
|
|
assert isinstance(context.render_error, yaml.YAMLError)
|
|
|
|
|
|
@then("the rendered YAML should be parsed into a mapping")
|
|
def step_assert_render_result(context):
|
|
assert context.render_error is None
|
|
assert isinstance(context.render_result, dict)
|
|
|
|
|
|
@then("the rendered mapping should include loop results and merged context")
|
|
def step_assert_render_content(context):
|
|
assert context.render_result.get("items") == ["alpha", "beta"]
|
|
assert context.render_result.get("context_value") == "kept"
|
|
|
|
|
|
@given("I have objects with missing attributes")
|
|
def step_objects_missing_attrs(context):
|
|
context.objects = [SimpleNamespace(score=5), SimpleNamespace(name="no-score")]
|
|
|
|
|
|
@when('I apply the selectattr filter for attribute "{attr}" with default {default:d}')
|
|
def step_apply_selectattr(context, attr: str, default: int):
|
|
context.selected_attrs = context.yaml_engine._selectattr_filter(
|
|
context.objects, attr, default
|
|
)
|
|
|
|
|
|
@then("the selected attributes should include defaults")
|
|
def step_assert_selectattr(context):
|
|
assert context.selected_attrs == [5, 0]
|
|
|
|
|
|
@given("I have numeric values for the sum filter")
|
|
def step_numeric_values_sum_filter(context):
|
|
context.numbers = [1, 2, 3]
|
|
context.indent_source = "line1\nline2"
|
|
context.yaml_dump_source = {"a": 1}
|
|
|
|
|
|
@when("I apply the sum filter and helper formatters")
|
|
def step_apply_filters(context):
|
|
context.sum_result = context.yaml_engine._sum_filter(context.numbers)
|
|
context.indented = context.yaml_engine._indent_filter(
|
|
context.indent_source, spaces=3
|
|
)
|
|
context.yaml_dump = context.yaml_engine._yaml_filter(context.yaml_dump_source)
|
|
|
|
|
|
@then("the helper filters should format output")
|
|
def step_assert_filters(context):
|
|
assert context.sum_result == 6
|
|
assert context.indented.splitlines()[0].startswith(" ")
|
|
assert "a: 1" in context.yaml_dump
|
|
|
|
|
|
@given("I have YAML with mixed content for structure analysis:")
|
|
def step_yaml_mixed_structure(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I analyze the YAML structure")
|
|
def step_analyze_structure(context):
|
|
context.analysis = context.yaml_engine._analyze_yaml_structure(context.yaml_content)
|
|
|
|
|
|
@then("template blocks and inline templates should be located")
|
|
def step_assert_structure(context):
|
|
assert context.analysis["template_blocks"]
|
|
assert context.analysis["inline_templates"]
|
|
# Hierarchy should capture non-empty lines excluding comments
|
|
assert any(entry["key"] == "root" for entry in context.analysis["hierarchy"])
|
|
|
|
|
|
@given("I have a temporary YAML file with template content:")
|
|
def step_temp_yaml_file(context):
|
|
context.yaml_file_content = context.text.strip("\n")
|
|
context.temp_dir = tempfile.TemporaryDirectory()
|
|
context.temp_yaml_path = Path(context.temp_dir.name) / "template_file.yaml"
|
|
context.temp_yaml_path.write_text(context.yaml_file_content, encoding="utf-8")
|
|
|
|
|
|
@when("I load the YAML file with context data")
|
|
def step_load_yaml_file(context):
|
|
try:
|
|
context.file_result = context.yaml_engine.load_file(
|
|
context.temp_yaml_path, {"name": "file-agent"}
|
|
)
|
|
context.file_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.file_result = None
|
|
context.file_error = exc
|
|
|
|
|
|
@then("the file load result should merge templated values")
|
|
def step_assert_file_result(context):
|
|
assert context.file_error is None
|
|
assert context.file_result["agent"]["name"] == "file-agent"
|
|
|
|
|
|
@given("I have a templated YAML that renders to a list root:")
|
|
def step_yaml_renders_list(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I render the YAML expecting a non-mapping error")
|
|
def step_render_expect_mapping_error(context):
|
|
try:
|
|
context.result = context.yaml_engine.load_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
context.error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.result = None
|
|
context.error = exc
|
|
|
|
|
|
@then("a ValueError should indicate rendered YAML must be a mapping")
|
|
def step_assert_non_mapping_error(context):
|
|
assert isinstance(context.error, ValueError)
|
|
assert "dict" in str(context.error)
|
|
|
|
|
|
@given("I have templated YAML that renders invalid mappings:")
|
|
def step_yaml_invalid_mapping(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@given("I have a messy value context for rendering")
|
|
def step_context_messy_value(context):
|
|
context.template_context = {"messy_value": "value1 key2: value2"}
|
|
|
|
|
|
@when("I render the malformed YAML content")
|
|
def step_render_malformed_yaml(context):
|
|
try:
|
|
context.malformed_result = context.yaml_engine.load_string(
|
|
context.yaml_content, context.template_context
|
|
)
|
|
context.malformed_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.malformed_result = None
|
|
context.malformed_error = exc
|
|
|
|
|
|
@then("the parser should attempt fixes then raise YAML error")
|
|
def step_assert_malformed_error(context):
|
|
assert context.malformed_result is None
|
|
assert isinstance(context.malformed_error, yaml.YAMLError)
|
|
|
|
|
|
@given("I have a deferred YAML template that becomes a list:")
|
|
def step_deferred_list_template(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("a ValueError should be raised for non-mapping deferred templates")
|
|
def step_assert_deferred_value_error(context):
|
|
assert isinstance(context.deferred_error, ValueError)
|
|
|
|
|
|
@given("I have YAML with a placeholder causing parse retry:")
|
|
def step_yaml_placeholder_retry(context):
|
|
context.yaml_content = context.text.strip("\n")
|
|
context.retry_context = None
|
|
context.retry_result = None
|
|
context.retry_error = None
|
|
|
|
|
|
@given("I have a template context for retry parsing")
|
|
def step_retry_context(context):
|
|
context.retry_context = {"value": "patched", "context": {}}
|
|
|
|
|
|
@when("I render YAML with a transient YAML error on first parse")
|
|
def step_render_with_transient_error(context):
|
|
side_effects = [yaml.YAMLError("transient"), {"result": "recovered"}]
|
|
with mock.patch.object(_yaml_module.yaml, "safe_load", side_effect=side_effects):
|
|
context.retry_result = context.yaml_engine._render_and_parse(
|
|
context.yaml_content, context.retry_context
|
|
)
|
|
|
|
|
|
@then("the rendering retry should yield a mapping result")
|
|
def step_assert_retry_success(context):
|
|
assert context.retry_result == {"result": "recovered"}
|
|
|
|
|
|
@when("rendering retry returns non-mapping after YAML error")
|
|
def step_render_retry_non_mapping(context):
|
|
side_effects = [yaml.YAMLError("still broken"), ["not", "mapping"]]
|
|
try:
|
|
with mock.patch.object(
|
|
_yaml_module.yaml, "safe_load", side_effect=side_effects
|
|
):
|
|
context.retry_result = context.yaml_engine._render_and_parse(
|
|
context.yaml_content, context.retry_context
|
|
)
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.retry_error = exc
|
|
|
|
|
|
@then("a ValueError should be raised for non-mapping retry")
|
|
def step_assert_retry_value_error(context):
|
|
assert isinstance(context.retry_error, ValueError)
|
|
assert "dict" in str(context.retry_error)
|
|
|
|
|
|
@given("I have inline YAML with multiple colons on one line:")
|
|
def step_inline_multicolon_yaml(context):
|
|
context.inline_yaml = context.text.strip("\n")
|
|
|
|
|
|
@when("I apply the common YAML fixes")
|
|
def step_apply_common_fixes(context):
|
|
context.fixed_common = context.yaml_engine._fix_common_yaml_issues(
|
|
context.inline_yaml
|
|
)
|
|
|
|
|
|
@then("the multi-colon line should be split into nested entries")
|
|
def step_assert_common_fixes(context):
|
|
expected = "\n".join(
|
|
[
|
|
"config:",
|
|
" key1:",
|
|
" val1",
|
|
" key2:",
|
|
" val2",
|
|
]
|
|
)
|
|
assert context.fixed_common == expected
|
|
|
|
|
|
@given("I have synthetic YAML content for postprocess:")
|
|
def step_synthetic_postprocess_input(context):
|
|
context.synthetic_yaml = context.text.strip("\n")
|
|
|
|
|
|
@when("I postprocess using a single-colon synthetic line")
|
|
def step_postprocess_synthetic(context):
|
|
class FakeLine(str):
|
|
def count(self, sub, start=0, end=None):
|
|
if sub == ":":
|
|
return 1
|
|
return super().count(sub, start, end)
|
|
|
|
class FakeContent(str):
|
|
def split(self, sep=None, maxsplit=-1):
|
|
sep_to_use = "\n" if sep is None else sep
|
|
return [FakeLine(part) for part in super().split(sep_to_use, maxsplit)]
|
|
|
|
fake_content = FakeContent(context.synthetic_yaml)
|
|
context.synthetic_postprocessed = context.yaml_engine._postprocess_rendered_yaml(
|
|
fake_content
|
|
)
|
|
|
|
|
|
@then("the synthetic postprocess should keep the line intact")
|
|
def step_assert_synthetic_postprocess(context):
|
|
assert context.synthetic_postprocessed.strip() == "config: alpha beta: gamma"
|