freemo
983eee7a37
CI / lint (pull_request) Successful in 14s
CI / typecheck (pull_request) Successful in 28s
CI / security (pull_request) Successful in 23s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 4m53s
CI / build (pull_request) Successful in 16s
CI / coverage (pull_request) Successful in 7m42s
CI / unit_tests (pull_request) Successful in 13m36s
CI / docker (pull_request) Successful in 40s
CI / lint (push) Successful in 14s
CI / typecheck (push) Successful in 28s
CI / security (push) Successful in 23s
CI / quality (push) Successful in 16s
CI / integration_tests (push) Successful in 4m54s
CI / build (push) Successful in 16s
CI / coverage (push) Successful in 7m46s
CI / unit_tests (push) Successful in 13m46s
CI / docker (push) Successful in 41s
911 lines
32 KiB
Python
911 lines
32 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 typing import TYPE_CHECKING, Any
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
|
|
if TYPE_CHECKING:
|
|
from behave.runner import Context
|
|
else:
|
|
Context = Any
|
|
|
|
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"
|
|
|
|
|
|
@given("I have a YAML file without templates:")
|
|
def step_yaml_file_without_templates(context: Context) -> None:
|
|
context.yaml_file_dir = tempfile.TemporaryDirectory()
|
|
context.yaml_file_path = Path(context.yaml_file_dir.name) / "plain.yaml"
|
|
context.yaml_file_path.write_text(context.text.strip("\n"), encoding="utf-8")
|
|
|
|
|
|
@when("I load the YAML file without context")
|
|
def step_load_yaml_file_without_context(context: Context) -> None:
|
|
try:
|
|
context.no_template_file_result = context.yaml_engine.load_file(
|
|
context.yaml_file_path
|
|
)
|
|
context.no_template_file_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.no_template_file_result = None
|
|
context.no_template_file_error = exc
|
|
|
|
|
|
@then("the YAML file should parse as a mapping")
|
|
def step_assert_yaml_file_mapping(context: Context) -> None:
|
|
assert context.no_template_file_error is None
|
|
assert isinstance(context.no_template_file_result, dict)
|
|
|
|
|
|
@then('the parsed YAML should include config name "{name}"')
|
|
def step_assert_yaml_file_name(context: Context, name: str) -> None:
|
|
assert context.no_template_file_result is not None
|
|
config_section = context.no_template_file_result.get("config", {})
|
|
assert config_section.get("name") == name
|
|
|
|
|
|
@given("I have a YAML string without templates:")
|
|
def step_yaml_string_without_templates(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I load the string directly")
|
|
def step_load_string_directly(context: Context) -> None:
|
|
try:
|
|
context.direct_load_result = context.yaml_engine.load_string(
|
|
context.yaml_content
|
|
)
|
|
context.direct_load_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.direct_load_result = None
|
|
context.direct_load_error = exc
|
|
|
|
|
|
@then("the YAML string should parse as a mapping")
|
|
def step_assert_yaml_string_mapping(context: Context) -> None:
|
|
assert context.direct_load_error is None
|
|
assert isinstance(context.direct_load_result, dict)
|
|
|
|
|
|
@then('the parsed YAML string should include key value "{value}"')
|
|
def step_assert_yaml_string_value(context: Context, value: str) -> None:
|
|
assert context.direct_load_result is not None
|
|
assert context.direct_load_result["simple"]["key"] == value
|
|
|
|
|
|
@given("I have a template using nested context and overrides:")
|
|
def step_template_with_nested_context(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@given("I have a template context with nested overrides")
|
|
def step_template_nested_context_values(context: Context) -> None:
|
|
context.template_context = {
|
|
"context": {"shared": "nested", "explicit": "nested-explicit"},
|
|
"value": "top-level",
|
|
}
|
|
|
|
|
|
@when("I render the YAML with merged context")
|
|
def step_render_with_merged_context(context: Context) -> None:
|
|
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("the render context should include nested and top-level values")
|
|
def step_assert_merged_context_render(context: Context) -> None:
|
|
assert context.render_error is None
|
|
assert isinstance(context.render_result, dict)
|
|
result = context.render_result.get("result", {})
|
|
assert result.get("nested") == "nested"
|
|
assert result.get("explicit") == "top-level"
|
|
assert result.get("context_copy") == "nested-explicit"
|
|
|
|
|
|
@given("I have template using utility functions:")
|
|
def step_template_with_utilities(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I process with complete render context utilities")
|
|
def step_process_with_utilities(context: Context) -> None:
|
|
try:
|
|
context.utility_result = context.yaml_engine.load_string(
|
|
context.yaml_content, {"context": {}}
|
|
)
|
|
context.utility_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.utility_result = None
|
|
context.utility_error = exc
|
|
|
|
|
|
@then("utility functions should be evaluated")
|
|
def step_assert_utility_functions(context: Context) -> None:
|
|
assert context.utility_error is None
|
|
assert isinstance(context.utility_result, dict)
|
|
utilities = context.utility_result.get("utilities", {})
|
|
assert utilities.get("range_values") == [0, 1, 2]
|
|
assert utilities.get("abs_value") == 4
|
|
assert utilities.get("round_value") == 3.14
|
|
|
|
|
|
@given("I have non-summable values for the sum filter")
|
|
def step_non_summable_values(context: Context) -> None:
|
|
context.non_summable_values = [{"value": 1}, {"value": 2}]
|
|
|
|
|
|
@when("I apply the sum filter expecting an error")
|
|
def step_apply_sum_filter_error(context: Context) -> None:
|
|
try:
|
|
context.sum_filter_result = context.yaml_engine._sum_filter(
|
|
context.non_summable_values
|
|
)
|
|
context.sum_filter_error = None
|
|
except Exception as exc: # pragma: no cover - exercised in tests
|
|
context.sum_filter_result = None
|
|
context.sum_filter_error = exc
|
|
|
|
|
|
@then("a ValueError should be raised for the sum filter")
|
|
def step_assert_sum_filter_error(context: Context) -> None:
|
|
assert isinstance(context.sum_filter_error, ValueError)
|
|
|
|
|
|
# -- New step definitions for additional coverage scenarios --
|
|
|
|
|
|
@given("I have rendered YAML with blank lines between sections:")
|
|
def step_yaml_blank_lines_between_sections(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I postprocess the rendered YAML for blank line cleanup")
|
|
def step_postprocess_blank_lines(context: Context) -> None:
|
|
context.postprocessed = context.yaml_engine._postprocess_rendered_yaml(
|
|
context.yaml_content
|
|
)
|
|
|
|
|
|
@then("blank lines before top-level keys should be removed")
|
|
def step_assert_blank_lines_removed(context: Context) -> None:
|
|
# The postprocessor removes blank lines where the next line is not indented
|
|
assert "section_one" in context.postprocessed
|
|
assert "section_two" in context.postprocessed
|
|
lines = context.postprocessed.splitlines()
|
|
# No blank lines should remain between top-level keys
|
|
for i, line in enumerate(lines):
|
|
if line.strip() == "" and i > 0 and i < len(lines) - 1:
|
|
next_line = lines[i + 1]
|
|
# If blank line remains, the next line must be indented
|
|
assert next_line.startswith(" "), (
|
|
f"Blank line at index {i} should have been removed "
|
|
f"(next line is top-level: {next_line!r})"
|
|
)
|
|
|
|
|
|
@given("I have rendered YAML with blank lines before indented content:")
|
|
def step_yaml_blank_lines_before_indented(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("blank lines before indented content should be kept")
|
|
def step_assert_blank_lines_kept(context: Context) -> None:
|
|
lines = context.postprocessed.splitlines()
|
|
# At least one blank line should be preserved (before indented key2)
|
|
has_blank = any(line.strip() == "" for line in lines)
|
|
assert has_blank, "Expected blank lines before indented content to be preserved"
|
|
|
|
|
|
@given("I have a template context without nested context key")
|
|
def step_context_without_nested(context: Context) -> None:
|
|
context.flat_context = {"name": "test", "value": 42}
|
|
|
|
|
|
@when("I create the render context")
|
|
def step_create_render_context(context: Context) -> None:
|
|
context.render_ctx = context.yaml_engine._create_render_context(
|
|
context.flat_context
|
|
)
|
|
|
|
|
|
@then("the render context should have an empty context mapping")
|
|
def step_assert_empty_context_mapping(context: Context) -> None:
|
|
assert isinstance(context.render_ctx["context"], dict)
|
|
# When no "context" key exists in the input, the nested mapping should be empty
|
|
assert context.render_ctx["context"] == {}
|
|
# Top-level keys should still be merged
|
|
assert context.render_ctx["name"] == "test"
|
|
assert context.render_ctx["value"] == 42
|
|
|
|
|
|
@given("I have a template context with non-dict nested context")
|
|
def step_context_with_non_dict_nested(context: Context) -> None:
|
|
context.flat_context = {"context": "not-a-dict", "extra": "data"}
|
|
|
|
|
|
@then("the render context should override context with non-dict value")
|
|
def step_assert_non_dict_context_override(context: Context) -> None:
|
|
# _create_render_context builds base_context = {"context": {}} then merges
|
|
# with input via base_context | context. Since input has context="not-a-dict",
|
|
# the merge overwrites the dict with the string.
|
|
assert context.render_ctx["context"] == "not-a-dict"
|
|
assert context.render_ctx["extra"] == "data"
|
|
|
|
|
|
@given("I have YAML with empty and blank lines for structure analysis:")
|
|
def step_yaml_empty_blank_lines(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("hierarchy should contain root and nested keys")
|
|
def step_assert_hierarchy_keys(context: Context) -> None:
|
|
keys = [entry["key"] for entry in context.analysis["hierarchy"]]
|
|
assert "root" in keys
|
|
assert "nested" in keys
|
|
|
|
|
|
@then("blank lines should be skipped in analysis")
|
|
def step_assert_blank_lines_skipped(context: Context) -> None:
|
|
# Blank lines should not appear in hierarchy
|
|
for entry in context.analysis["hierarchy"]:
|
|
assert entry["key"].strip() != ""
|
|
|
|
|
|
@given("I have YAML with dedented keys for structure analysis:")
|
|
def step_yaml_dedented_keys(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("hierarchy paths should reflect dedentation correctly")
|
|
def step_assert_dedentation_paths(context: Context) -> None:
|
|
hierarchy = context.analysis["hierarchy"]
|
|
# Find the "sibling" entry - it should have popped "child" and "parent" from path
|
|
sibling_entry = next((e for e in hierarchy if e["key"] == "sibling"), None)
|
|
assert sibling_entry is not None
|
|
# "sibling" is at the root level (indent 0), so its path should just be ["sibling"]
|
|
assert sibling_entry["path"] == ["sibling"]
|
|
|
|
|
|
@given("I have YAML with colon-ending tokens:")
|
|
def step_yaml_colon_ending_tokens(context: Context) -> None:
|
|
context.inline_yaml = context.text.strip("\n")
|
|
|
|
|
|
@then("colon-ending tokens should be split onto separate lines")
|
|
def step_assert_colon_tokens_split(context: Context) -> None:
|
|
lines = context.fixed_common.splitlines()
|
|
# The first part should be just "config:"
|
|
assert lines[0] == "config:"
|
|
# Subsequent lines should be indented with colon-ending tokens on their own lines
|
|
assert any("key1:" in line for line in lines[1:])
|
|
assert any("key2:" in line for line in lines[1:])
|
|
|
|
|
|
@given("I have YAML without multiple colons per line:")
|
|
def step_yaml_no_multiple_colons(context: Context) -> None:
|
|
context.inline_yaml = context.text.strip("\n")
|
|
|
|
|
|
@then("lines without multiple colons should remain unchanged")
|
|
def step_assert_no_changes(context: Context) -> None:
|
|
assert context.fixed_common.strip() == context.inline_yaml.strip()
|
|
|
|
|
|
@given("I have YAML with mixed colon tokens for fixing:")
|
|
def step_yaml_mixed_colon_tokens(context: Context) -> None:
|
|
context.inline_yaml = context.text.strip("\n")
|
|
|
|
|
|
@then("the trailing tokens should be handled correctly")
|
|
def step_assert_trailing_tokens(context: Context) -> None:
|
|
lines = context.fixed_common.splitlines()
|
|
# Should have been split due to multiple colons
|
|
assert len(lines) >= 2
|
|
assert lines[0] == "config:"
|
|
|
|
|
|
@given("I have rendered YAML with colon in value words:")
|
|
def step_yaml_colon_in_value_words(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I postprocess the rendered YAML with colon values")
|
|
def step_postprocess_colon_values(context: Context) -> None:
|
|
context.postprocessed = context.yaml_engine._postprocess_rendered_yaml(
|
|
context.yaml_content
|
|
)
|
|
|
|
|
|
@then("the colon value should trigger line splitting")
|
|
def step_assert_colon_value_split(context: Context) -> None:
|
|
# The postprocessor checks for colons in subsequent words of a value
|
|
# and may split lines accordingly
|
|
assert "endpoint" in context.postprocessed
|
|
|
|
|
|
@given("I have YAML with non-loop Jinja blocks:")
|
|
def step_yaml_non_loop_jinja(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("no indentation hints should be added for non-loop blocks")
|
|
def step_assert_no_hints_for_non_loops(context: Context) -> None:
|
|
assert "{# indent:" not in context.preprocessed
|
|
|
|
|
|
@given("I have invalid YAML for simple template extraction:")
|
|
def step_invalid_yaml_for_extraction(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@when("I extract templates from invalid YAML")
|
|
def step_extract_invalid_templates(context: Context) -> None:
|
|
try:
|
|
context.extraction_result = context.yaml_engine._simple_template_extraction(
|
|
context.yaml_content
|
|
)
|
|
context.extraction_error = None
|
|
except Exception as exc:
|
|
context.extraction_result = None
|
|
context.extraction_error = exc
|
|
|
|
|
|
@then("a YAML error should be raised during extraction")
|
|
def step_assert_extraction_error(context: Context) -> None:
|
|
assert context.extraction_error is not None
|
|
assert isinstance(context.extraction_error, yaml.YAMLError)
|
|
|
|
|
|
@given("I have a simple template context with message")
|
|
def step_simple_template_context(context: Context) -> None:
|
|
context.template_context = {"message": "hello-world", "context": {}}
|
|
|
|
|
|
@then("the rendered result should contain the message value")
|
|
def step_assert_message_value(context: Context) -> None:
|
|
assert context.render_result["greeting"] == "hello-world"
|
|
|
|
|
|
@when('I apply the selectattr filter for attribute "{attr}" with default "{default}"')
|
|
def step_apply_selectattr_str(context: Context, attr: str, default: str) -> None:
|
|
context.selected_attrs = context.yaml_engine._selectattr_filter(
|
|
context.objects, attr, default
|
|
)
|
|
|
|
|
|
@given("I have objects that all have the target attribute")
|
|
def step_objects_all_have_attr(context: Context) -> None:
|
|
context.objects = [
|
|
SimpleNamespace(name="alice"),
|
|
SimpleNamespace(name="bob"),
|
|
SimpleNamespace(name="charlie"),
|
|
]
|
|
|
|
|
|
@then("all attribute values should be returned without defaults")
|
|
def step_assert_all_attrs_no_defaults(context: Context) -> None:
|
|
assert context.selected_attrs == ["alice", "bob", "charlie"]
|
|
|
|
|
|
@given("I have objects that all lack the target attribute")
|
|
def step_objects_all_lack_attr(context: Context) -> None:
|
|
context.objects = [
|
|
SimpleNamespace(name="alice"),
|
|
SimpleNamespace(name="bob"),
|
|
]
|
|
|
|
|
|
@then("all values should be the default")
|
|
def step_assert_all_defaults(context: Context) -> None:
|
|
assert context.selected_attrs == [-1, -1]
|
|
|
|
|
|
@given("I have a multiline string for indentation")
|
|
def step_multiline_for_indent(context: Context) -> None:
|
|
context.indent_text = "first line\nsecond line\nthird line"
|
|
|
|
|
|
@when("I apply the indent filter with 4 spaces")
|
|
def step_apply_indent_4_spaces(context: Context) -> None:
|
|
context.indented_result = context.yaml_engine._indent_filter(
|
|
context.indent_text, spaces=4
|
|
)
|
|
|
|
|
|
@then("each line should be indented by 4 spaces")
|
|
def step_assert_4_space_indent(context: Context) -> None:
|
|
for line in context.indented_result.splitlines():
|
|
assert line.startswith(" "), f"Expected 4-space indent, got: {line!r}"
|
|
|
|
|
|
@given("I have a complex nested structure for YAML serialization")
|
|
def step_complex_structure_for_yaml(context: Context) -> None:
|
|
context.complex_structure = {
|
|
"name": "test",
|
|
"nested": {"key": "value", "list": [1, 2, 3]},
|
|
}
|
|
|
|
|
|
@when("I apply the YAML filter to serialize it")
|
|
def step_apply_yaml_filter(context: Context) -> None:
|
|
context.yaml_serialized = context.yaml_engine._yaml_filter(
|
|
context.complex_structure
|
|
)
|
|
|
|
|
|
@then("the serialized output should be valid YAML text")
|
|
def step_assert_valid_yaml_text(context: Context) -> None:
|
|
assert "name: test" in context.yaml_serialized
|
|
assert "nested:" in context.yaml_serialized
|
|
# Verify it can be parsed back
|
|
parsed = yaml.safe_load(context.yaml_serialized)
|
|
assert parsed["name"] == "test"
|
|
assert parsed["nested"]["key"] == "value"
|
|
|
|
|
|
@given("I have YAML with block-level templates for deferred loading:")
|
|
def step_yaml_block_templates_deferred(context: Context) -> None:
|
|
context.yaml_content = context.text.strip("\n")
|
|
|
|
|
|
@then("the deferred result should preserve quoted template strings")
|
|
def step_assert_deferred_preserves_quotes(context: Context) -> None:
|
|
assert context.deferred_error is None
|
|
assert isinstance(context.deferred_result, dict)
|
|
assert "config" in context.deferred_result
|
|
# The quoted template string should be preserved as a literal string
|
|
assert "{{ agent_name }}" in str(context.deferred_result["config"]["name"])
|