forked from HAL9000/cleveragents-core
brent.edwards
e801eb1ee8
- Rewrite .forgejo/workflows/ci.yml to route all jobs through nox sessions - Fix coverage_report nox session: serial behave mode replaces broken parallel mode (22% -> 97% accuracy), raise fail-under from 85% to 97% - Pass posargs through format nox session for CI --check support - Add 11 CI workflow validation scenarios (Behave) + Robot smoke test + ASV bench - Add 108 new Behave scenarios covering 6 largest coverage gaps to reach 97%: yaml_template_engine, actor/config, actor/registry, message_router, context_analysis, context_service - Update docs/development/ci-cd.md with nox-based CI docs and 97% threshold - Restore implementation_plan.md verbose style, check off completed CI tasks Verified: 1673 scenarios pass, 97% coverage, lint clean, typecheck clean
388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""Step definitions for actor_config_new_coverage.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.actor.config import ActorConfiguration
|
|
|
|
|
|
@then('an actor config ValueError should mention "{text}"')
|
|
def step_assert_actor_config_valueerror(context: Context, text: str) -> None:
|
|
assert context.error is not None, "Expected ValueError but none was raised"
|
|
assert text.lower() in str(context.error).lower(), (
|
|
f"Expected '{text}' in error: {context.error}"
|
|
)
|
|
|
|
|
|
@given("a temporary JSON config file with provider and model")
|
|
def step_temp_json_config(context: Context) -> None:
|
|
data = {"provider": "openai", "model": "gpt-4", "options": {"temperature": 0.7}}
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".json", delete=False, encoding="utf-8"
|
|
) as tmp:
|
|
json.dump(data, tmp)
|
|
tmp.flush()
|
|
context.temp_config_path = Path(tmp.name)
|
|
|
|
|
|
@when("I load the blob from the JSON file")
|
|
def step_load_blob_json(context: Context) -> None:
|
|
context.blob = ActorConfiguration.load_blob_from_file(context.temp_config_path)
|
|
|
|
|
|
@then("the blob should contain provider and model keys")
|
|
def step_assert_blob_keys(context: Context) -> None:
|
|
assert "provider" in context.blob
|
|
assert "model" in context.blob
|
|
|
|
|
|
@when("I load the blob from a nonexistent path")
|
|
def step_load_blob_missing(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration.load_blob_from_file(
|
|
Path("/tmp/nonexistent_config_12345.json")
|
|
)
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@given("a temporary JSON config file with a list")
|
|
def step_temp_json_list(context: Context) -> None:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".json", delete=False, encoding="utf-8"
|
|
) as tmp:
|
|
json.dump([1, 2, 3], tmp)
|
|
tmp.flush()
|
|
context.temp_list_path = Path(tmp.name)
|
|
|
|
|
|
@when("I load the blob from the list JSON file")
|
|
def step_load_blob_list(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration.load_blob_from_file(context.temp_list_path)
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@given("a temporary JSON config file with null content")
|
|
def step_temp_json_null(context: Context) -> None:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".json", delete=False, encoding="utf-8"
|
|
) as tmp:
|
|
tmp.write("null")
|
|
tmp.flush()
|
|
context.temp_null_path = Path(tmp.name)
|
|
|
|
|
|
@when("I load the blob from the null JSON file")
|
|
def step_load_blob_null(context: Context) -> None:
|
|
context.blob = ActorConfiguration.load_blob_from_file(context.temp_null_path)
|
|
|
|
|
|
@then("the blob should be an empty dict")
|
|
def step_assert_empty_dict(context: Context) -> None:
|
|
assert context.blob == {}
|
|
|
|
|
|
@given("a temporary YAML-only config file with provider and model")
|
|
def step_temp_yaml_config(context: Context) -> None:
|
|
content = "provider: anthropic\nmodel: claude-3\noptions:\n temperature: 0.5\n"
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".yaml", delete=False, encoding="utf-8"
|
|
) as tmp:
|
|
tmp.write(content)
|
|
tmp.flush()
|
|
context.temp_yaml_path = Path(tmp.name)
|
|
|
|
|
|
@when("I load the blob from the YAML file")
|
|
def step_load_blob_yaml(context: Context) -> None:
|
|
context.blob = ActorConfiguration.load_blob_from_file(context.temp_yaml_path)
|
|
|
|
|
|
@when("I load v2 YAML content without templates")
|
|
def step_load_v2_plain(context: Context) -> None:
|
|
text = "provider: openai\nmodel: gpt-4\n"
|
|
context.result = ActorConfiguration._load_v2_yaml_content(text)
|
|
|
|
|
|
@then("the v2 result should be a dict with expected keys")
|
|
def step_assert_dict_expected_keys(context: Context) -> None:
|
|
assert isinstance(context.result, dict)
|
|
assert len(context.result) > 0
|
|
|
|
|
|
@when("I load v2 YAML content that is empty")
|
|
def step_load_v2_empty(context: Context) -> None:
|
|
context.result = ActorConfiguration._load_v2_yaml_content("")
|
|
|
|
|
|
@then("the v2 result should be an empty dict")
|
|
def step_assert_empty_result(context: Context) -> None:
|
|
assert context.result == {}
|
|
|
|
|
|
@when("I load v2 YAML content that is a list")
|
|
def step_load_v2_list(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration._load_v2_yaml_content("- item1\n- item2\n")
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@when("I load v2 YAML content with Jinja2 templates")
|
|
def step_load_v2_templated(context: Context) -> None:
|
|
text = 'provider: openai\nmodel: gpt-4\nsystem_prompt: "Hello {{ context.name }}"\n'
|
|
context.result = ActorConfiguration._load_v2_yaml_content(text)
|
|
|
|
|
|
@then("the actor config result should be a dict")
|
|
def step_assert_is_dict(context: Context) -> None:
|
|
assert isinstance(context.result, dict)
|
|
|
|
|
|
@when("I restore template syntax in a config with protected markers")
|
|
def step_restore_markers(context: Context) -> None:
|
|
config: dict[str, Any] = {
|
|
"system_prompt": "Hello <<<TEMPLATE_START>>>name<<<TEMPLATE_END>>>",
|
|
"other": "<<<BLOCK_START>>> for x <<<BLOCK_END>>>",
|
|
}
|
|
context.result = ActorConfiguration._restore_template_syntax(config)
|
|
|
|
|
|
@then("the system_prompt should have Jinja2 syntax restored")
|
|
def step_assert_restored_prompt(context: Context) -> None:
|
|
assert "{{" in context.result["system_prompt"]
|
|
assert "}}" in context.result["system_prompt"]
|
|
|
|
|
|
@when("I restore template syntax in a list config")
|
|
def step_restore_list(context: Context) -> None:
|
|
config = [
|
|
{"system_prompt": "<<<TEMPLATE_START>>>x<<<TEMPLATE_END>>>"},
|
|
"plain",
|
|
]
|
|
context.result = ActorConfiguration._restore_template_syntax(config)
|
|
|
|
|
|
@then("nested list items should have markers restored")
|
|
def step_assert_list_restored(context: Context) -> None:
|
|
assert isinstance(context.result, list)
|
|
assert "{{" in context.result[0]["system_prompt"]
|
|
|
|
|
|
@given('environment variable "{name}" is set to "{value}"')
|
|
def step_set_env_var(context: Context, name: str, value: str) -> None:
|
|
os.environ[name] = value
|
|
if not hasattr(context, "env_cleanup"):
|
|
context.env_cleanup = []
|
|
context.env_cleanup.append(name)
|
|
|
|
|
|
@when('I interpolate env vars in config with "${{{var_expr}}}"')
|
|
def step_interpolate_env(context: Context, var_expr: str) -> None:
|
|
config = f"${{{var_expr}}}"
|
|
context.result = ActorConfiguration._interpolate_env_vars(config)
|
|
|
|
|
|
@then('the interpolated value should be "{expected}"')
|
|
def step_assert_interpolated(context: Context, expected: str) -> None:
|
|
assert str(context.result) == expected, (
|
|
f"Expected '{expected}', got '{context.result}'"
|
|
)
|
|
|
|
|
|
@when('I interpolate env vars with "${{{var_expr}}}" and no default')
|
|
def step_interpolate_no_default(context: Context, var_expr: str) -> None:
|
|
config = f"${{{var_expr}}}"
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration._interpolate_env_vars(config)
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@when('I interpolate env vars for string "{value}"')
|
|
def step_interpolate_literal(context: Context, value: str) -> None:
|
|
context.result = ActorConfiguration._interpolate_env_vars(value)
|
|
|
|
|
|
@then("the interpolated result should be boolean True")
|
|
def step_assert_bool_true(context: Context) -> None:
|
|
assert context.result is True
|
|
|
|
|
|
@then("the interpolated result should be integer {n:d}")
|
|
def step_assert_int(context: Context, n: int) -> None:
|
|
assert context.result == n
|
|
assert isinstance(context.result, int)
|
|
|
|
|
|
@then("the interpolated result should be float {f}")
|
|
def step_assert_float(context: Context, f: str) -> None:
|
|
assert abs(context.result - float(f)) < 0.001
|
|
assert isinstance(context.result, float)
|
|
|
|
|
|
@then('the interpolated result should be string "{expected}"')
|
|
def step_assert_string(context: Context, expected: str) -> None:
|
|
assert str(context.result) == expected
|
|
|
|
|
|
@when("I interpolate env vars in a nested structure")
|
|
def step_interpolate_nested(context: Context) -> None:
|
|
config: dict[str, Any] = {
|
|
"provider": "openai",
|
|
"items": ["first", "second"],
|
|
"nested": {"key": "value"},
|
|
}
|
|
context.result = ActorConfiguration._interpolate_env_vars(config)
|
|
|
|
|
|
@then("all nested string values should be interpolated")
|
|
def step_assert_nested_interpolated(context: Context) -> None:
|
|
assert isinstance(context.result, dict)
|
|
assert isinstance(context.result["items"], list)
|
|
assert isinstance(context.result["nested"], dict)
|
|
|
|
|
|
@when("I create an ActorConfiguration from a blob with provider and model")
|
|
def step_from_blob(context: Context) -> None:
|
|
blob: dict[str, Any] = {"provider": "openai", "model": "gpt-4"}
|
|
context.actor_cfg = ActorConfiguration.from_blob(blob=blob)
|
|
|
|
|
|
@then("the config provider should be set")
|
|
def step_assert_config_provider(context: Context) -> None:
|
|
assert context.actor_cfg.provider
|
|
|
|
|
|
@then("the config model should be set")
|
|
def step_assert_config_model(context: Context) -> None:
|
|
assert context.actor_cfg.model
|
|
|
|
|
|
@when("I create an ActorConfiguration from a blob without provider")
|
|
def step_from_blob_no_provider(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration.from_blob(blob={"model": "gpt-4"})
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@when("I create an ActorConfiguration from a blob without model")
|
|
def step_from_blob_no_model(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
ActorConfiguration.from_blob(blob={"provider": "openai"})
|
|
except ValueError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@when("I create an ActorConfiguration from a blob with option overrides")
|
|
def step_from_blob_options(context: Context) -> None:
|
|
blob: dict[str, Any] = {"provider": "openai", "model": "gpt-4"}
|
|
context.actor_cfg = ActorConfiguration.from_blob(
|
|
blob=blob, option_overrides={"temperature": 0.9}
|
|
)
|
|
|
|
|
|
@then("the config options should include the overrides")
|
|
def step_assert_options(context: Context) -> None:
|
|
assert context.actor_cfg.options["temperature"] == 0.9
|
|
|
|
|
|
@when("I extract v2 actor from a v2-style config with agents block")
|
|
def step_extract_v2_actor(context: Context) -> None:
|
|
data: dict[str, Any] = {
|
|
"agents": {
|
|
"main_agent": {
|
|
"config": {
|
|
"provider": "anthropic",
|
|
"model": "claude-3",
|
|
"unsafe": True,
|
|
"options": {"max_tokens": 1000},
|
|
}
|
|
}
|
|
},
|
|
"routes": [{"from": "main_agent", "to": "end"}],
|
|
}
|
|
context.extracted = ActorConfiguration._extract_v2_actor(data)
|
|
|
|
|
|
@then("the extracted provider and model should be set")
|
|
def step_assert_extracted(context: Context) -> None:
|
|
provider, model, _graph, unsafe = context.extracted
|
|
assert provider == "anthropic"
|
|
assert model == "claude-3"
|
|
assert unsafe is True
|
|
|
|
|
|
@then("the graph descriptor should contain agent info")
|
|
def step_assert_graph_descriptor(context: Context) -> None:
|
|
_, _, graph, _ = context.extracted
|
|
assert graph is not None
|
|
assert "agent" in graph
|
|
assert "routes" in graph
|
|
|
|
|
|
@when("I extract v2 actor from a config without agents block")
|
|
def step_extract_v2_no_agents(context: Context) -> None:
|
|
context.extracted = ActorConfiguration._extract_v2_actor({"provider": "openai"})
|
|
|
|
|
|
@then("all extracted values should be None")
|
|
def step_assert_all_none(context: Context) -> None:
|
|
provider, model, graph, unsafe = context.extracted
|
|
assert provider is None
|
|
assert model is None
|
|
assert graph is None
|
|
assert unsafe is False
|
|
|
|
|
|
@when("I extract v2 options from a v2-style config")
|
|
def step_extract_v2_options(context: Context) -> None:
|
|
data: dict[str, Any] = {
|
|
"agents": {
|
|
"main_agent": {
|
|
"config": {
|
|
"provider": "openai",
|
|
"model": "gpt-4",
|
|
"options": {"temperature": 0.7, "max_tokens": 500},
|
|
}
|
|
}
|
|
}
|
|
}
|
|
context.options = ActorConfiguration._extract_v2_options(data)
|
|
|
|
|
|
@then("the extracted options should contain expected keys")
|
|
def step_assert_extracted_options(context: Context) -> None:
|
|
assert context.options is not None
|
|
assert "temperature" in context.options
|
|
|
|
|
|
@when("I create an ActorConfiguration from the file")
|
|
def step_from_file(context: Context) -> None:
|
|
context.actor_cfg = ActorConfiguration.from_file(
|
|
path=context.temp_config_path,
|
|
)
|
|
|
|
|
|
@when('I interpolate env vars with "${{{var_expr}}}"')
|
|
def step_interpolate_with_default(context: Context, var_expr: str) -> None:
|
|
config = f"${{{var_expr}}}"
|
|
context.result = ActorConfiguration._interpolate_env_vars(config)
|