218 lines
7.5 KiB
Python
218 lines
7.5 KiB
Python
"""Behave steps to cover reactive config parser paths."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.core.exceptions import ConfigurationError
|
|
from cleveragents.reactive.config_parser import ReactiveConfigParser
|
|
from cleveragents.reactive.route import RouteType
|
|
from cleveragents.reactive.stream_router import StreamType
|
|
|
|
|
|
@given("a temporary directory for reactive config parsing")
|
|
def step_temp_directory(context: Any) -> None:
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
context.parser = ReactiveConfigParser()
|
|
context.config_files: list[Path] = []
|
|
context.parsed_config = None
|
|
context.parse_error: Exception | None = None
|
|
|
|
|
|
@given("reactive config files covering merges, routes, and pipelines")
|
|
def step_reactive_config_files(context: Any) -> None:
|
|
stream_route = {
|
|
"type": "stream",
|
|
"stream_type": "hot",
|
|
"operators": [{"type": "map"}],
|
|
"subscriptions": ["input"],
|
|
"publications": ["output"],
|
|
"agents": ["alpha"],
|
|
"initial_value": 5,
|
|
"buffer_size": 2,
|
|
"template_config": {"tmpl": True},
|
|
"bridge": {
|
|
"upgrade_conditions": {"ready": True},
|
|
"downgrade_conditions": {},
|
|
"state_extractor": "extract",
|
|
"state_flattener": "flatten",
|
|
"preserve_subscriptions": False,
|
|
"preserve_checkpointing": False,
|
|
},
|
|
"metadata": {"kind": "stream"},
|
|
}
|
|
|
|
graph_route = {
|
|
"type": "graph",
|
|
"nodes": {"n1": {"agent": "alpha"}},
|
|
"edges": [{"from": "n1", "to": "n1"}],
|
|
"entry_point": "n1",
|
|
"checkpointing": True,
|
|
"checkpoint_dir": str(context.temp_dir / "ckpts"),
|
|
"enable_time_travel": True,
|
|
"parallel_execution": False,
|
|
"state_class": "State",
|
|
"metadata": {"kind": "graph"},
|
|
}
|
|
|
|
primary_config = {
|
|
"agents": {"alpha": {"type": "llm", "config": {"a": 1}}},
|
|
"routes": [stream_route, graph_route],
|
|
"merges": [{"sources": ["input"], "target": "route_0"}],
|
|
"splits": [{"source": "route_0", "targets": ["route_1"]}],
|
|
}
|
|
|
|
secondary_config = {
|
|
"agents": {"alpha": {"config": {"b": 2}}},
|
|
"merges": [{"sources": ["input2"], "target": "route_0"}],
|
|
"templates": {"t1": {"value": 1}},
|
|
"instances": {"inst": {"config": "val"}},
|
|
"global_context": {"mode": "test"},
|
|
"template_engine": "CUSTOM",
|
|
"prompts": {"p1": "hi"},
|
|
"pipelines": {
|
|
"pipe1": {
|
|
"stages": [
|
|
{"type": "stream", "name": "route_0"},
|
|
{"type": "graph", "config": {"name": "route_1"}},
|
|
],
|
|
"metadata": {"note": "x"},
|
|
}
|
|
},
|
|
}
|
|
|
|
primary_file = context.temp_dir / "primary.yaml"
|
|
secondary_file = context.temp_dir / "secondary.yaml"
|
|
primary_file.write_text(yaml.safe_dump(primary_config))
|
|
secondary_file.write_text(yaml.safe_dump(secondary_config))
|
|
context.config_files = [primary_file, secondary_file]
|
|
|
|
|
|
@when("I parse the reactive config files")
|
|
def step_parse_reactive_config_files(context: Any) -> None:
|
|
try:
|
|
context.parsed_config = context.parser.parse_files(context.config_files)
|
|
context.parse_error = None
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
context.parsed_config = None
|
|
context.parse_error = exc
|
|
|
|
|
|
@then("the merged reactive config should preserve agents and replacements")
|
|
def step_verify_merges_and_agents(context: Any) -> None:
|
|
assert context.parse_error is None, f"Unexpected error: {context.parse_error}"
|
|
rc = context.parsed_config
|
|
assert rc is not None
|
|
assert rc.agents["alpha"].type == "llm"
|
|
assert rc.agents["alpha"].config["a"] == 1
|
|
assert rc.agents["alpha"].config["b"] == 2
|
|
assert len(rc.merges) == 2
|
|
|
|
|
|
@then("stream and graph routes should be built with defaults and bridges")
|
|
def step_verify_routes(context: Any) -> None:
|
|
rc = context.parsed_config
|
|
assert rc is not None
|
|
|
|
stream_route = rc.routes["route_0"]
|
|
assert stream_route.type == RouteType.STREAM
|
|
assert stream_route.stream_type == StreamType.HOT
|
|
assert stream_route.buffer_size == 2
|
|
assert stream_route.template_config == {"tmpl": True}
|
|
assert stream_route.bridge is not None
|
|
assert stream_route.bridge.preserve_subscriptions is False
|
|
assert stream_route.bridge.preserve_checkpointing is False
|
|
|
|
graph_route = rc.routes["route_1"]
|
|
assert graph_route.type == RouteType.GRAPH
|
|
assert graph_route.checkpointing is True
|
|
assert graph_route.enable_time_travel is True
|
|
assert graph_route.parallel_execution is False
|
|
assert graph_route.state_class == "State"
|
|
|
|
|
|
@then("reactive config should include merges splits pipelines and templates")
|
|
def step_verify_other_sections(context: Any) -> None:
|
|
rc = context.parsed_config
|
|
assert rc is not None
|
|
|
|
assert len(rc.splits) == 1
|
|
assert "pipe1" in rc.pipelines
|
|
assert rc.pipelines["pipe1"].metadata["note"] == "x"
|
|
assert "t1" in rc.templates
|
|
assert "inst" in rc.instances
|
|
assert rc.global_context["mode"] == "test"
|
|
assert rc.template_engine == "CUSTOM"
|
|
assert rc.prompts["p1"] == "hi"
|
|
|
|
|
|
@given("a reactive config file with environment placeholders")
|
|
def step_env_config_file(context: Any) -> None:
|
|
config = {
|
|
"agents": {
|
|
"env_agent": {
|
|
"type": "llm",
|
|
"config": {
|
|
"api_key": "${ENV_API_KEY}",
|
|
"with_default": "${MISSING_KEY:defaulted}",
|
|
},
|
|
}
|
|
},
|
|
"routes": {"main": {"type": "stream", "agents": ["env_agent"]}},
|
|
}
|
|
config_file = context.temp_dir / "env.yaml"
|
|
config_file.write_text(yaml.safe_dump(config))
|
|
context.config_files = [config_file]
|
|
|
|
|
|
@given("environment variables are set for interpolation")
|
|
def step_set_env_vars(_: Any) -> None:
|
|
os.environ["ENV_API_KEY"] = "secret-key"
|
|
os.environ["OTHER_VAR"] = "unused"
|
|
|
|
|
|
@then("environment placeholders should resolve using values and defaults")
|
|
def step_verify_env_interpolation(context: Any) -> None:
|
|
assert context.parse_error is None, f"Unexpected error: {context.parse_error}"
|
|
rc = context.parsed_config
|
|
assert rc is not None
|
|
|
|
agent_cfg = rc.agents["env_agent"].config
|
|
assert agent_cfg["api_key"] == "secret-key"
|
|
assert agent_cfg["with_default"] == "defaulted"
|
|
|
|
|
|
@given("a reactive config file with a required environment placeholder")
|
|
def step_env_required_config(context: Any) -> None:
|
|
config = {
|
|
"agents": {"needs_env": {"type": "llm", "config": {"token": "${MUST_HAVE}"}}},
|
|
"routes": {"main": {"type": "stream", "agents": ["needs_env"]}},
|
|
}
|
|
file_path = context.temp_dir / "missing_env.yaml"
|
|
file_path.write_text(yaml.safe_dump(config))
|
|
context.config_files = [file_path]
|
|
if "MUST_HAVE" in os.environ:
|
|
del os.environ["MUST_HAVE"]
|
|
|
|
|
|
@when("I parse the reactive config files expecting failure")
|
|
def step_parse_expect_failure(context: Any) -> None:
|
|
try:
|
|
context.parsed_config = context.parser.parse_files(context.config_files)
|
|
context.parse_error = None
|
|
except Exception as exc:
|
|
context.parsed_config = None
|
|
context.parse_error = exc
|
|
|
|
|
|
@then("a configuration error should be raised for the missing variable")
|
|
def step_verify_missing_env_error(context: Any) -> None:
|
|
assert context.parse_error is not None, "Expected a configuration error"
|
|
assert isinstance(context.parse_error, ConfigurationError)
|