CoreRasurae
38fe40553d
Adds 104 Robot Framework integration tests exercising the full 6-agent email categorization actor graph across five configuration strategies: local, remote, mixed, component-decomposed (merge_configs), and namespaced (local: refs via ReferenceResolver + LocalPackageStore). - 7 new email graph YAML fixtures under tests/fixtures/email_graph/ - 7 namespaced component YAMLs with local namespace identifiers - 9 robot test suite files (email_categorization_graph, per-agent, negative tests, merge_configs validation, components, namespaced) - EmailGraphLib.py keyword library (executor, graph, merge, components) - CleverActorsLib.py refactored into per-module libs: robot/lib/config_lib.py, app_lib.py, registry_lib.py - email_testdata.resource with 12 realistic email scenarios - All 104 integration tests pass, all 23 existing tests preserved Closes #30
240 lines
9.1 KiB
Python
240 lines
9.1 KiB
Python
"""Robot Framework keyword library for configuration and template rendering.
|
|
|
|
Covers: package metadata, configuration manager, merge_configs,
|
|
schema validation, and template rendering/registry.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from cleveractors import __version__, merge_configs
|
|
from cleveractors.core.config import ConfigurationManager, SchemaValidator
|
|
from cleveractors.templates.enhanced_registry import EnhancedTemplateRegistry
|
|
from cleveractors.templates.registry import TemplateRegistry
|
|
from cleveractors.templates.renderer import TemplateEngine, TemplateRenderer
|
|
|
|
|
|
class ConfigLib: # pragma: no cover - integration test library
|
|
"""Keyword library for configuration and templates."""
|
|
|
|
ROBOT_LIBRARY_SCOPE = "TEST SUITE"
|
|
|
|
# ── package metadata ──────────────────────────────────────────────
|
|
|
|
def package_version(self) -> str:
|
|
return __version__
|
|
|
|
def version_matches(self, expected: str) -> None:
|
|
actual = self.package_version()
|
|
if actual != expected:
|
|
raise AssertionError(
|
|
f"Version mismatch: expected {expected!r}, got {actual!r}"
|
|
)
|
|
|
|
# ── configuration ─────────────────────────────────────────────────
|
|
|
|
def create_config_manager(self) -> None:
|
|
self._cm = ConfigurationManager()
|
|
|
|
def load_config_files(self, *paths: str) -> None:
|
|
self._cm.load_files([Path(p) for p in paths])
|
|
|
|
def config_value_equals(self, path: str, expected: str) -> None:
|
|
value = self._cm.get(path)
|
|
if str(value) != str(expected):
|
|
raise AssertionError(f"Config {path}: expected {expected!r}, got {value!r}")
|
|
|
|
def config_path_exists(self, path: str) -> None:
|
|
value = self._cm.get(path)
|
|
if value is None:
|
|
raise AssertionError(f"Config path {path!r} not found or is None")
|
|
|
|
def load_config_should_raise(self, *paths: str) -> None:
|
|
try:
|
|
self._cm = ConfigurationManager()
|
|
self._cm.load_files([Path(p) for p in paths])
|
|
except Exception:
|
|
return
|
|
raise AssertionError("Expected an exception but none was raised")
|
|
|
|
def interpolate_env_var(self, var_name: str, var_value: str) -> None:
|
|
os.environ[var_name] = var_value
|
|
data = {"key": f"${{{var_name}}}"}
|
|
result = self._cm.interpolate_env_vars(data)
|
|
if result.get("key") != var_value:
|
|
raise AssertionError(
|
|
f"Interpolation failed: expected {var_value!r}, got {result.get('key')!r}"
|
|
)
|
|
|
|
# ── merge_configs public API ──────────────────────────────────────
|
|
|
|
def merge_configs_returns_empty_for_no_args(self) -> None:
|
|
result = merge_configs()
|
|
if result != {}:
|
|
raise AssertionError(f"Expected {{}}, got {result!r}")
|
|
|
|
def merge_configs_two_dicts(
|
|
self,
|
|
base_key: str,
|
|
base_val: str,
|
|
overlay_key: str,
|
|
overlay_val: str,
|
|
) -> None:
|
|
base = {base_key: base_val}
|
|
overlay = {overlay_key: overlay_val}
|
|
result = merge_configs(base, overlay)
|
|
if len(result) != 2:
|
|
raise AssertionError(
|
|
f"Expected result length 2, got {len(result)}: {result!r}"
|
|
)
|
|
if result.get(base_key) != base_val:
|
|
raise AssertionError(f"Expected {base_key}={base_val!r}, got {result!r}")
|
|
if result.get(overlay_key) != overlay_val:
|
|
raise AssertionError(
|
|
f"Expected {overlay_key}={overlay_val!r}, got {result!r}"
|
|
)
|
|
if base != {base_key: base_val}:
|
|
raise AssertionError(f"Base dict was mutated: {base!r}")
|
|
if overlay != {overlay_key: overlay_val}:
|
|
raise AssertionError(f"Overlay dict was mutated: {overlay!r}")
|
|
|
|
def merge_configs_deep_merge(self) -> None:
|
|
base = {"config": {"a": 1, "b": 2}}
|
|
overlay = {"config": {"b": 99, "c": 3}}
|
|
result = merge_configs(base, overlay)
|
|
expected = {"config": {"a": 1, "b": 99, "c": 3}}
|
|
if result != expected:
|
|
raise AssertionError(f"Expected {expected!r}, got {result!r}")
|
|
|
|
def merge_configs_sequence_append(self) -> None:
|
|
base = {"items": [1, 2]}
|
|
overlay = {"items": [3, 4]}
|
|
result = merge_configs(base, overlay)
|
|
if result != {"items": [1, 2, 3, 4]}:
|
|
raise AssertionError(f"Expected items=[1,2,3,4], got {result!r}")
|
|
|
|
def merge_configs_through_config_pipeline(
|
|
self, base_key: str, overlay_val: str
|
|
) -> None:
|
|
base = Path(__file__).resolve().parent.parent.parent / "tests" / "fixtures"
|
|
|
|
cm_base = ConfigurationManager()
|
|
cm_base.load_files([base / "test_config.yaml"])
|
|
base_dict = cm_base.to_dict()
|
|
|
|
overlay: dict[str, Any] = {base_key: overlay_val, "meta": {"merged": True}}
|
|
result = merge_configs(base_dict, overlay)
|
|
|
|
if result.get(base_key) != overlay_val:
|
|
raise AssertionError(
|
|
f"Overlay key {base_key!r} expected {overlay_val!r},"
|
|
f" got {result.get(base_key)!r}"
|
|
)
|
|
if result.get("meta", {}).get("merged") is not True:
|
|
raise AssertionError(
|
|
f"Deep-merged overlay meta missing: {result.get('meta')!r}"
|
|
)
|
|
if "agents" not in result:
|
|
raise AssertionError(
|
|
f"Base fixture key 'agents' lost during merge: {sorted(result.keys())}"
|
|
)
|
|
if "cleveragents" not in result:
|
|
raise AssertionError("Base fixture key 'cleveragents' lost during merge")
|
|
|
|
def schema_validator_accepts_minimum_config(self) -> None:
|
|
config = {
|
|
"cleveragents": {"default_router": "main"},
|
|
"agents": {
|
|
"echo": {
|
|
"type": "tool",
|
|
"config": {"tools": ["echo"]},
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {
|
|
"type": "stream",
|
|
"operators": [{"type": "map", "params": {"agent": "echo"}}],
|
|
"publications": ["__output__"],
|
|
}
|
|
},
|
|
}
|
|
sv = SchemaValidator()
|
|
sv.validate(config)
|
|
|
|
def schema_validator_rejects(self, yaml_text: str) -> None:
|
|
sv = SchemaValidator()
|
|
config = yaml.safe_load(yaml_text)
|
|
try:
|
|
sv.validate(config)
|
|
except Exception:
|
|
return
|
|
raise AssertionError(
|
|
"Expected schema validation to reject config, but it passed"
|
|
)
|
|
|
|
# ── templates ─────────────────────────────────────────────────────
|
|
|
|
def create_template_renderer(self, engine: str = "SIMPLE") -> None:
|
|
eng = TemplateEngine[engine.upper()]
|
|
self._renderer = TemplateRenderer(eng)
|
|
|
|
def template_engine_is_string(self, engine: str) -> None:
|
|
assert engine.lower() in ("simple", "jinja2", "mustache"), f"Unknown: {engine}"
|
|
|
|
def register_template(self, name: str, content: str) -> None:
|
|
self._renderer.register_template(name, content)
|
|
|
|
def render_template_equals(
|
|
self, name: str, context_json: str, expected: str
|
|
) -> None:
|
|
ctx = json.loads(context_json)
|
|
result = self._renderer.render(name, ctx)
|
|
if str(result) != expected:
|
|
raise AssertionError(
|
|
f"Render mismatch: expected {expected!r}, got {result!r}"
|
|
)
|
|
|
|
def render_string_equals(
|
|
self, template: str, context_json: str, expected: str
|
|
) -> None:
|
|
ctx = json.loads(context_json)
|
|
result = self._renderer.render_string(template, ctx)
|
|
if str(result) != expected:
|
|
raise AssertionError(
|
|
f"Render mismatch: expected {expected!r}, got {result!r}"
|
|
)
|
|
|
|
def render_string_contains(
|
|
self, template: str, context_json: str, expected: str
|
|
) -> None:
|
|
ctx = json.loads(context_json)
|
|
result = self._renderer.render_string(template, ctx)
|
|
if expected not in str(result):
|
|
raise AssertionError(f"Expected {expected!r} in result, got {result!r}")
|
|
|
|
def render_unknown_template_should_raise(self, name: str) -> None:
|
|
try:
|
|
self._renderer.render(name, {})
|
|
except Exception:
|
|
return
|
|
raise AssertionError("Expected exception for unknown template, but none raised")
|
|
|
|
def list_templates_count(self, expected_count: str) -> None:
|
|
templates = self._renderer.list_templates()
|
|
if len(templates) != int(expected_count):
|
|
raise AssertionError(
|
|
f"Expected {expected_count} templates, got {len(templates)}: {templates}"
|
|
)
|
|
|
|
def create_template_registry(self) -> None:
|
|
self._registry = TemplateRegistry()
|
|
|
|
def create_enhanced_registry(self) -> None:
|
|
self._reg_enhanced = EnhancedTemplateRegistry()
|