forked from HAL9000/cleveragents-core
602 lines
20 KiB
Python
602 lines
20 KiB
Python
"""
|
|
Unit tests for core/config.py
|
|
|
|
Tests the ConfigurationManager and SchemaValidator classes.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
from cleveragents.core.config import ConfigurationManager, SchemaValidator
|
|
from cleveragents.core.exceptions import ConfigurationError
|
|
|
|
|
|
# Helper function to get fixture file path
|
|
def get_fixture_path(filename: str) -> Path:
|
|
"""Get path to a test fixture file."""
|
|
return Path(__file__).parent.parent.parent / "fixtures" / filename
|
|
|
|
|
|
class TestConfigurationManager:
|
|
"""Test suite for the ConfigurationManager class."""
|
|
|
|
@pytest.fixture
|
|
def config_manager(self):
|
|
"""Create a ConfigurationManager instance."""
|
|
return ConfigurationManager()
|
|
|
|
@pytest.fixture
|
|
def temp_config_file(self):
|
|
"""Get path to researcher config fixture file."""
|
|
return get_fixture_path("researcher_config.yaml")
|
|
|
|
@pytest.fixture
|
|
def temp_config_file_2(self):
|
|
"""Get path to second researcher config fixture file for merging tests."""
|
|
return get_fixture_path("researcher_config_2.yaml")
|
|
|
|
def test_initialization(self, config_manager):
|
|
"""Test ConfigurationManager initialization."""
|
|
assert config_manager.config == {}
|
|
assert config_manager.schema_validator is not None
|
|
|
|
def test_load_single_file(self, config_manager, temp_config_file):
|
|
"""Test loading a single configuration file."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
assert "agents" in config_manager.config
|
|
assert "researcher" in config_manager.config["agents"]
|
|
assert config_manager.config["agents"]["researcher"]["model"] == "gpt-4"
|
|
|
|
def test_load_multiple_files_merge(self, config_manager, temp_config_file, temp_config_file_2):
|
|
"""Test loading and merging multiple configuration files."""
|
|
config_manager.load_files([temp_config_file, temp_config_file_2])
|
|
|
|
# Second file should override first file values
|
|
assert config_manager.config["agents"]["researcher"]["model"] == "gpt-3.5-turbo"
|
|
assert config_manager.config["agents"]["researcher"]["parameters"]["temperature"] == 0.5
|
|
|
|
# But preserve values not in second file
|
|
assert config_manager.config["agents"]["researcher"]["type"] == "llm"
|
|
|
|
# And add new values from second file
|
|
assert "writer" in config_manager.config["agents"]
|
|
assert "application" in config_manager.config
|
|
|
|
def test_load_empty_file(self, config_manager):
|
|
"""Test loading an empty YAML file."""
|
|
empty_file = get_fixture_path("truly_empty.yaml")
|
|
|
|
config_manager.load_files([empty_file])
|
|
# Empty file should result in empty config
|
|
assert config_manager.config == {}
|
|
|
|
def test_load_invalid_yaml(self, config_manager):
|
|
"""Test loading a file with invalid YAML syntax."""
|
|
invalid_file = get_fixture_path("invalid_yaml.yaml")
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.load_files([invalid_file])
|
|
|
|
assert "Failed to parse YAML file" in str(exc_info.value)
|
|
|
|
def test_load_non_dict_yaml(self, config_manager):
|
|
"""Test loading a YAML file that doesn't contain a dictionary."""
|
|
non_dict_file = get_fixture_path("non_dict_yaml.yaml")
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.load_files([non_dict_file])
|
|
|
|
assert "must contain a YAML dictionary" in str(exc_info.value)
|
|
|
|
def test_load_nonexistent_file(self, config_manager):
|
|
"""Test loading a file that doesn't exist."""
|
|
nonexistent = Path("/nonexistent/config.yaml")
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.load_files([nonexistent])
|
|
|
|
assert "Failed to load configuration file" in str(exc_info.value)
|
|
|
|
def test_deep_merge(self, config_manager):
|
|
"""Test deep merging of dictionaries."""
|
|
dict1 = {
|
|
"a": 1,
|
|
"b": {"c": 2, "d": 3},
|
|
"e": [1, 2, 3],
|
|
}
|
|
dict2 = {
|
|
"b": {"d": 4, "f": 5},
|
|
"e": [4, 5],
|
|
"g": 6,
|
|
}
|
|
|
|
result = config_manager._deep_merge(dict1, dict2)
|
|
|
|
assert result["a"] == 1
|
|
assert result["b"]["c"] == 2
|
|
assert result["b"]["d"] == 4
|
|
assert result["b"]["f"] == 5
|
|
assert result["e"] == [4, 5] # Lists are replaced, not merged
|
|
assert result["g"] == 6
|
|
|
|
def test_interpolate_env_vars(self, config_manager):
|
|
"""Test environment variable interpolation."""
|
|
# Save original values if they exist
|
|
old_test_model = os.environ.get("TEST_MODEL")
|
|
old_test_temp = os.environ.get("TEST_TEMP")
|
|
|
|
os.environ["TEST_MODEL"] = "gpt-4-test"
|
|
os.environ["TEST_TEMP"] = "0.9"
|
|
|
|
try:
|
|
config = {
|
|
"model": "${TEST_MODEL}",
|
|
"temperature": "${TEST_TEMP}",
|
|
"nested": {
|
|
"value": "${TEST_MODEL}-nested",
|
|
},
|
|
}
|
|
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
assert result["model"] == "gpt-4-test"
|
|
assert result["temperature"] == 0.9 # Converted to float
|
|
assert result["nested"]["value"] == "gpt-4-test-nested"
|
|
finally:
|
|
# Restore original values or delete if they didn't exist
|
|
if old_test_model is None:
|
|
os.environ.pop("TEST_MODEL", None)
|
|
else:
|
|
os.environ["TEST_MODEL"] = old_test_model
|
|
|
|
if old_test_temp is None:
|
|
os.environ.pop("TEST_TEMP", None)
|
|
else:
|
|
os.environ["TEST_TEMP"] = old_test_temp
|
|
|
|
def test_interpolate_env_vars_missing(self, config_manager):
|
|
"""Test interpolation with missing environment variable raises error."""
|
|
config = {
|
|
"model": "${NONEXISTENT_VAR}",
|
|
}
|
|
|
|
# Missing variables raise ConfigurationError
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.interpolate_env_vars(config)
|
|
|
|
assert "NONEXISTENT_VAR" in str(exc_info.value)
|
|
|
|
def test_get_simple_path(self, config_manager, temp_config_file):
|
|
"""Test getting a configuration value with simple path."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
value = config_manager.get("agents")
|
|
assert isinstance(value, dict)
|
|
assert "researcher" in value
|
|
|
|
def test_get_nested_path(self, config_manager, temp_config_file):
|
|
"""Test getting a configuration value with nested path."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
value = config_manager.get("agents.researcher.model")
|
|
assert value == "gpt-4"
|
|
|
|
value = config_manager.get("agents.researcher.parameters.temperature")
|
|
assert value == 0.7
|
|
|
|
def test_get_with_default(self, config_manager, temp_config_file):
|
|
"""Test getting a configuration value with default."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
value = config_manager.get("nonexistent.path", default="default_value")
|
|
assert value == "default_value"
|
|
|
|
def test_get_nonexistent_no_default(self, config_manager, temp_config_file):
|
|
"""Test getting a nonexistent value without default returns None."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
value = config_manager.get("nonexistent.path")
|
|
assert value is None
|
|
|
|
def test_set_simple_path(self, config_manager):
|
|
"""Test setting a configuration value with simple path."""
|
|
config_manager.set("model", "gpt-4")
|
|
|
|
assert config_manager.config["model"] == "gpt-4"
|
|
|
|
def test_set_nested_path(self, config_manager):
|
|
"""Test setting a configuration value with nested path."""
|
|
config_manager.set("agents.researcher.model", "gpt-4")
|
|
|
|
assert config_manager.config["agents"]["researcher"]["model"] == "gpt-4"
|
|
|
|
def test_set_overwrites_existing(self, config_manager, temp_config_file):
|
|
"""Test that setting a value overwrites existing value."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
original = config_manager.get("agents.researcher.model")
|
|
assert original == "gpt-4"
|
|
|
|
config_manager.set("agents.researcher.model", "claude-3")
|
|
|
|
new_value = config_manager.get("agents.researcher.model")
|
|
assert new_value == "claude-3"
|
|
|
|
def test_to_dict(self, config_manager, temp_config_file):
|
|
"""Test getting the entire configuration as a dictionary."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
config_dict = config_manager.to_dict()
|
|
|
|
assert isinstance(config_dict, dict)
|
|
assert "agents" in config_dict
|
|
|
|
# Verify it's a copy
|
|
config_dict["new_key"] = "value"
|
|
assert "new_key" not in config_manager.config
|
|
|
|
def test_to_json(self, config_manager, temp_config_file):
|
|
"""Test converting configuration to JSON string."""
|
|
config_manager.load_files([temp_config_file])
|
|
|
|
json_str = config_manager.to_json()
|
|
|
|
assert isinstance(json_str, str)
|
|
parsed = json.loads(json_str)
|
|
assert "agents" in parsed
|
|
assert parsed["agents"]["researcher"]["model"] == "gpt-4"
|
|
|
|
def test_set_empty_path_raises_error(self, config_manager):
|
|
"""Test that setting with empty path raises ConfigurationError."""
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.set("", "value")
|
|
|
|
assert "Path cannot be empty" in str(exc_info.value)
|
|
|
|
def test_set_on_non_dict_parent_raises_error(self, config_manager):
|
|
"""Test that setting a nested path when parent is not a dict raises error."""
|
|
config_manager.config = {"agents": "not_a_dict"}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.set("agents.researcher.model", "gpt-4")
|
|
|
|
assert "not a dictionary" in str(exc_info.value)
|
|
|
|
def test_set_nested_on_non_dict_raises_error(self, config_manager):
|
|
"""Test setting when intermediate path exists but is not a dict."""
|
|
config_manager.config = {"agents": {"researcher": "string_value"}}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.set("agents.researcher.model", "gpt-4")
|
|
|
|
assert "not a dictionary" in str(exc_info.value)
|
|
|
|
def test_interpolate_env_vars_with_boolean_default(self, config_manager):
|
|
"""Test environment variable interpolation with boolean default values."""
|
|
config = {
|
|
"enabled": "${NONEXISTENT_BOOL:true}",
|
|
"disabled": "${NONEXISTENT_BOOL2:false}",
|
|
}
|
|
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
# Boolean strings in defaults are converted to bool
|
|
assert result["enabled"] is True
|
|
assert result["disabled"] is False
|
|
|
|
def test_interpolate_env_vars_with_numeric_defaults(self, config_manager):
|
|
"""Test environment variable interpolation with numeric default values."""
|
|
config = {
|
|
"count": "${NONEXISTENT_INT:42}",
|
|
"ratio": "${NONEXISTENT_FLOAT:3.14}",
|
|
}
|
|
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
# Numeric strings in defaults are converted to int/float
|
|
assert result["count"] == 42
|
|
assert result["ratio"] == 3.14
|
|
|
|
def test_interpolate_env_vars_with_string_defaults(self, config_manager):
|
|
"""Test environment variable interpolation with string default values."""
|
|
config = {
|
|
"model": "${NONEXISTENT_MODEL:gpt-4}",
|
|
"name": "${NONEXISTENT_NAME:default_agent}",
|
|
}
|
|
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
# String defaults should be returned as-is
|
|
assert result["model"] == "gpt-4"
|
|
assert result["name"] == "default_agent"
|
|
|
|
def test_interpolate_env_vars_converts_boolean_strings(self, config_manager):
|
|
"""Test that boolean strings are converted to booleans."""
|
|
os.environ["TEST_BOOL_TRUE"] = "true"
|
|
os.environ["TEST_BOOL_FALSE"] = "false"
|
|
|
|
try:
|
|
config = {
|
|
"enabled": "${TEST_BOOL_TRUE}",
|
|
"disabled": "${TEST_BOOL_FALSE}",
|
|
}
|
|
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
assert result["enabled"] is True
|
|
assert result["disabled"] is False
|
|
finally:
|
|
del os.environ["TEST_BOOL_TRUE"]
|
|
del os.environ["TEST_BOOL_FALSE"]
|
|
|
|
def test_interpolate_env_vars_converts_integers(self, config_manager):
|
|
"""Test that integer strings are converted to integers."""
|
|
os.environ["TEST_INT"] = "42"
|
|
|
|
try:
|
|
config = {"count": "${TEST_INT}"}
|
|
result = config_manager.interpolate_env_vars(config)
|
|
|
|
assert result["count"] == 42
|
|
assert isinstance(result["count"], int)
|
|
finally:
|
|
del os.environ["TEST_INT"]
|
|
|
|
def test_validate_calls_schema_validator(self, config_manager):
|
|
"""Test that validate() calls schema_validator."""
|
|
config_manager.config = {
|
|
"agents": {
|
|
"test": {"type": "llm", "model": "gpt-4"}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "main"
|
|
}
|
|
}
|
|
|
|
# Should not raise
|
|
config_manager.validate()
|
|
|
|
def test_validate_raises_configuration_error_on_invalid_config(self, config_manager):
|
|
"""Test that validate() raises ConfigurationError for invalid config."""
|
|
config_manager.config = {} # Invalid: missing agents
|
|
|
|
with pytest.raises(ConfigurationError):
|
|
config_manager.validate()
|
|
|
|
def test_validate_wraps_other_exceptions(self, config_manager, monkeypatch):
|
|
"""Test that validate() wraps non-ConfigurationError exceptions."""
|
|
config_manager.config = {"agents": {}, "routes": {}}
|
|
|
|
def mock_validate(_self, _config):
|
|
raise ValueError("Unexpected error")
|
|
|
|
monkeypatch.setattr(SchemaValidator, "validate", mock_validate)
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
config_manager.validate()
|
|
|
|
assert "validation failed" in str(exc_info.value)
|
|
|
|
|
|
class TestSchemaValidator:
|
|
"""Test suite for the SchemaValidator class."""
|
|
|
|
@pytest.fixture
|
|
def validator(self):
|
|
"""Create a SchemaValidator instance."""
|
|
return SchemaValidator()
|
|
|
|
def test_validate_empty_config_fails(self, validator):
|
|
"""Test that empty configuration fails validation."""
|
|
# Empty config requires agents section
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate({})
|
|
|
|
assert "agents" in str(exc_info.value)
|
|
|
|
def test_validate_basic_config_missing_routes(self, validator):
|
|
"""Test that configuration without routes section fails."""
|
|
config = {
|
|
"agents": {
|
|
"researcher": {
|
|
"type": "llm",
|
|
"model": "gpt-4",
|
|
}
|
|
}
|
|
}
|
|
|
|
# Should raise because routes section is required
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "routes" in str(exc_info.value)
|
|
|
|
def test_validate_complete_config(self, validator):
|
|
"""Test validating a complete configuration."""
|
|
config = {
|
|
"agents": {
|
|
"researcher": {
|
|
"type": "llm",
|
|
"model": "gpt-4",
|
|
}
|
|
},
|
|
"routes": {
|
|
"reactive": {
|
|
"source": "input",
|
|
"target": "researcher",
|
|
}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "reactive",
|
|
}
|
|
}
|
|
|
|
# Should not raise with both agents and routes and default_router
|
|
validator.validate(config)
|
|
|
|
def test_validate_config_with_invalid_agent_structure(self, validator):
|
|
"""Test validating configuration with invalid agent structure."""
|
|
config = {
|
|
"agents": "invalid_value", # Should be dict
|
|
"routes": {}
|
|
}
|
|
|
|
# Should raise due to invalid structure
|
|
with pytest.raises(ConfigurationError):
|
|
validator.validate(config)
|
|
|
|
def test_validate_agent_missing_type_field(self, validator):
|
|
"""Test that agent without 'type' field fails validation."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"model": "gpt-4",
|
|
# Missing 'type' field
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "main"
|
|
}
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "missing required 'type' field" in str(exc_info.value)
|
|
|
|
def test_validate_agent_config_not_dict(self, validator):
|
|
"""Test that agent with invalid config type fails validation."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"type": "llm",
|
|
"config": "not_a_dict" # Should be dict
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "main"
|
|
}
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "'config' must be a dictionary" in str(exc_info.value)
|
|
|
|
def test_validate_auto_creates_config_field(self, validator):
|
|
"""Test that validator auto-creates 'config' field if missing."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"type": "llm",
|
|
"model": "gpt-4"
|
|
# Missing 'config' field - should be auto-created
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "main"
|
|
}
|
|
}
|
|
|
|
# Should not raise
|
|
validator.validate(config)
|
|
|
|
# Config should be auto-created
|
|
assert "config" in config["agents"]["test_agent"]
|
|
assert config["agents"]["test_agent"]["config"] == {}
|
|
|
|
def test_validate_empty_routes_fails(self, validator):
|
|
"""Test that empty routes section fails validation."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"type": "llm",
|
|
"model": "gpt-4"
|
|
}
|
|
},
|
|
"routes": {} # Empty routes
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "'routes' section cannot be empty" in str(exc_info.value)
|
|
|
|
def test_validate_missing_default_router_fails(self, validator):
|
|
"""Test that config with routes but missing default_router fails."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"type": "llm",
|
|
"model": "gpt-4"
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
}
|
|
# Missing cleveragents.default_router
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "default_router" in str(exc_info.value)
|
|
|
|
def test_validate_default_router_not_in_routes_fails(self, validator):
|
|
"""Test that default_router not found in routes fails validation."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": {
|
|
"type": "llm",
|
|
"model": "gpt-4"
|
|
}
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
},
|
|
"cleveragents": {
|
|
"default_router": "nonexistent_router" # Not in routes
|
|
}
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "not found within 'routes'" in str(exc_info.value)
|
|
|
|
def test_validate_agent_config_not_dict_type(self, validator):
|
|
"""Test that agent configuration not being a dict fails validation."""
|
|
config = {
|
|
"agents": {
|
|
"test_agent": "not_a_dict" # Should be dict
|
|
},
|
|
"routes": {
|
|
"main": {}
|
|
}
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError) as exc_info:
|
|
validator.validate(config)
|
|
|
|
assert "must be a dictionary" in str(exc_info.value)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|
|
|