430 lines
16 KiB
Python
430 lines
16 KiB
Python
"""Unit tests for config_manager.py - Configuration loading, validation, defaults."""
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.core.providers.config_manager import (
|
|
ProviderConfigManager,
|
|
ProviderSettings,
|
|
get_config_manager,
|
|
)
|
|
|
|
|
|
class TestProviderSettings:
|
|
"""Test ProviderSettings dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""ProviderSettings should have sensible defaults."""
|
|
settings = ProviderSettings(name="test_provider")
|
|
assert settings.name == "test_provider"
|
|
assert settings.enabled is True
|
|
assert settings.priority == 0
|
|
assert settings.timeout_seconds == 30
|
|
assert settings.max_retries == 3
|
|
assert settings.retry_delay_seconds == 1.0
|
|
assert settings.max_concurrent_downloads == 3
|
|
assert settings.bandwidth_limit_mbps is None
|
|
assert settings.custom_headers is None
|
|
assert settings.custom_params is None
|
|
|
|
def test_to_dict(self):
|
|
"""to_dict should convert settings to dict, excluding None values."""
|
|
settings = ProviderSettings(
|
|
name="test",
|
|
enabled=True,
|
|
priority=1,
|
|
)
|
|
result = settings.to_dict()
|
|
assert result["name"] == "test"
|
|
assert result["enabled"] is True
|
|
assert result["priority"] == 1
|
|
# None values should be excluded
|
|
assert "bandwidth_limit_mbps" not in result
|
|
assert "custom_headers" not in result
|
|
|
|
def test_to_dict_with_optional_fields(self):
|
|
"""to_dict with optional fields set should include them."""
|
|
settings = ProviderSettings(
|
|
name="test",
|
|
bandwidth_limit_mbps=10.0,
|
|
custom_headers={"X-Custom": "value"},
|
|
)
|
|
result = settings.to_dict()
|
|
assert result["bandwidth_limit_mbps"] == 10.0
|
|
assert result["custom_headers"] == {"X-Custom": "value"}
|
|
|
|
def test_from_dict(self):
|
|
"""from_dict should create settings from fields with defaults."""
|
|
# Note: from_dict uses hasattr(cls, k) which only matches fields
|
|
# with defaults on the class. The 'name' field has no default,
|
|
# so it must be passed explicitly.
|
|
data = {
|
|
"name": "test_provider",
|
|
"enabled": False,
|
|
"priority": 5,
|
|
"timeout_seconds": 60,
|
|
}
|
|
# The from_dict filters with hasattr which excludes 'name'
|
|
# (no default), so this should raise TypeError
|
|
with pytest.raises(TypeError):
|
|
ProviderSettings.from_dict(data)
|
|
|
|
def test_from_dict_with_only_defaults_fields(self):
|
|
"""from_dict works when all fields have defaults (except name)."""
|
|
# Directly construct to test fields with defaults
|
|
data = {
|
|
"enabled": False,
|
|
"priority": 5,
|
|
}
|
|
# This will fail because 'name' is required but filtered out
|
|
with pytest.raises(TypeError):
|
|
ProviderSettings.from_dict(data)
|
|
|
|
def test_from_dict_ignores_unknown_fields(self):
|
|
"""from_dict should ignore fields not in the dataclass."""
|
|
data = {
|
|
"name": "test",
|
|
"unknown_field": "value",
|
|
"another_unknown": 42,
|
|
}
|
|
# name gets filtered by hasattr → TypeError for missing name
|
|
with pytest.raises(TypeError):
|
|
ProviderSettings.from_dict(data)
|
|
|
|
def test_from_dict_with_defaults(self):
|
|
"""from_dict with only-defaults data loses required 'name'."""
|
|
data = {"name": "minimal"}
|
|
with pytest.raises(TypeError):
|
|
ProviderSettings.from_dict(data)
|
|
|
|
|
|
class TestProviderConfigManager:
|
|
"""Test ProviderConfigManager class."""
|
|
|
|
def test_init_without_config_file(self):
|
|
"""Should initialize with empty provider settings."""
|
|
manager = ProviderConfigManager()
|
|
assert manager._provider_settings == {}
|
|
|
|
def test_init_with_nonexistent_config_file(self):
|
|
"""Should initialize cleanly when config file doesn't exist."""
|
|
manager = ProviderConfigManager(
|
|
config_file=Path("/nonexistent/config.json")
|
|
)
|
|
assert manager._provider_settings == {}
|
|
|
|
def test_global_settings_defaults(self):
|
|
"""Should have sensible global defaults."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.get_global_setting("default_timeout") == 30
|
|
assert manager.get_global_setting("default_max_retries") == 3
|
|
assert manager.get_global_setting("enable_health_monitoring") is True
|
|
assert manager.get_global_setting("enable_failover") is True
|
|
|
|
def test_get_provider_settings_none_for_unknown(self):
|
|
"""Should return None for unknown provider."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.get_provider_settings("unknown") is None
|
|
|
|
def test_set_and_get_provider_settings(self):
|
|
"""Should store and retrieve provider settings."""
|
|
manager = ProviderConfigManager()
|
|
settings = ProviderSettings(name="test", priority=1)
|
|
manager.set_provider_settings("test", settings)
|
|
result = manager.get_provider_settings("test")
|
|
assert result is not None
|
|
assert result.name == "test"
|
|
assert result.priority == 1
|
|
|
|
def test_update_provider_settings_existing(self):
|
|
"""Should update existing provider settings."""
|
|
manager = ProviderConfigManager()
|
|
settings = ProviderSettings(name="test", priority=1)
|
|
manager.set_provider_settings("test", settings)
|
|
|
|
result = manager.update_provider_settings("test", priority=5)
|
|
assert result is True
|
|
updated = manager.get_provider_settings("test")
|
|
assert updated.priority == 5
|
|
|
|
def test_update_provider_settings_new(self):
|
|
"""Should create new settings when provider doesn't exist."""
|
|
manager = ProviderConfigManager()
|
|
result = manager.update_provider_settings(
|
|
"new_provider", priority=3, timeout_seconds=60
|
|
)
|
|
assert result is True
|
|
settings = manager.get_provider_settings("new_provider")
|
|
assert settings is not None
|
|
assert settings.priority == 3
|
|
assert settings.timeout_seconds == 60
|
|
|
|
def test_get_all_provider_settings(self):
|
|
"""Should return copy of all provider settings."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"p1", ProviderSettings(name="p1")
|
|
)
|
|
manager.set_provider_settings(
|
|
"p2", ProviderSettings(name="p2")
|
|
)
|
|
|
|
all_settings = manager.get_all_provider_settings()
|
|
assert len(all_settings) == 2
|
|
assert "p1" in all_settings
|
|
assert "p2" in all_settings
|
|
|
|
def test_get_all_returns_copy(self):
|
|
"""get_all_provider_settings should return a copy."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"p1", ProviderSettings(name="p1")
|
|
)
|
|
all_settings = manager.get_all_provider_settings()
|
|
all_settings["p2"] = ProviderSettings(name="p2")
|
|
assert "p2" not in manager.get_all_provider_settings()
|
|
|
|
|
|
class TestProviderEnableDisable:
|
|
"""Test enable/disable provider functionality."""
|
|
|
|
def test_enable_provider(self):
|
|
"""Should enable a disabled provider."""
|
|
manager = ProviderConfigManager()
|
|
settings = ProviderSettings(name="test", enabled=False)
|
|
manager.set_provider_settings("test", settings)
|
|
|
|
result = manager.enable_provider("test")
|
|
assert result is True
|
|
assert manager.get_provider_settings("test").enabled is True
|
|
|
|
def test_disable_provider(self):
|
|
"""Should disable an enabled provider."""
|
|
manager = ProviderConfigManager()
|
|
settings = ProviderSettings(name="test", enabled=True)
|
|
manager.set_provider_settings("test", settings)
|
|
|
|
result = manager.disable_provider("test")
|
|
assert result is True
|
|
assert manager.get_provider_settings("test").enabled is False
|
|
|
|
def test_enable_unknown_provider_returns_false(self):
|
|
"""Should return False when enabling unknown provider."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.enable_provider("unknown") is False
|
|
|
|
def test_disable_unknown_provider_returns_false(self):
|
|
"""Should return False when disabling unknown provider."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.disable_provider("unknown") is False
|
|
|
|
def test_get_enabled_providers(self):
|
|
"""Should return only enabled providers."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"p1", ProviderSettings(name="p1", enabled=True)
|
|
)
|
|
manager.set_provider_settings(
|
|
"p2", ProviderSettings(name="p2", enabled=False)
|
|
)
|
|
manager.set_provider_settings(
|
|
"p3", ProviderSettings(name="p3", enabled=True)
|
|
)
|
|
|
|
enabled = manager.get_enabled_providers()
|
|
assert "p1" in enabled
|
|
assert "p2" not in enabled
|
|
assert "p3" in enabled
|
|
|
|
|
|
class TestProviderPriority:
|
|
"""Test provider priority management."""
|
|
|
|
def test_set_provider_priority(self):
|
|
"""Should set priority for a provider."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"test", ProviderSettings(name="test", priority=0)
|
|
)
|
|
|
|
result = manager.set_provider_priority("test", 5)
|
|
assert result is True
|
|
assert manager.get_provider_settings("test").priority == 5
|
|
|
|
def test_set_priority_unknown_returns_false(self):
|
|
"""Should return False for unknown provider."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.set_provider_priority("unknown", 1) is False
|
|
|
|
def test_get_providers_by_priority(self):
|
|
"""Should return providers sorted by priority."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"low", ProviderSettings(name="low", priority=10)
|
|
)
|
|
manager.set_provider_settings(
|
|
"high", ProviderSettings(name="high", priority=1)
|
|
)
|
|
manager.set_provider_settings(
|
|
"mid", ProviderSettings(name="mid", priority=5)
|
|
)
|
|
|
|
sorted_providers = manager.get_providers_by_priority()
|
|
assert sorted_providers == ["high", "mid", "low"]
|
|
|
|
|
|
class TestGlobalSettings:
|
|
"""Test global settings management."""
|
|
|
|
def test_get_global_setting(self):
|
|
"""Should retrieve global setting value."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.get_global_setting("default_timeout") == 30
|
|
|
|
def test_get_unknown_global_setting(self):
|
|
"""Should return None for unknown global setting."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.get_global_setting("nonexistent") is None
|
|
|
|
def test_set_global_setting(self):
|
|
"""Should set a global setting."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_global_setting("custom_key", "custom_value")
|
|
assert manager.get_global_setting("custom_key") == "custom_value"
|
|
|
|
def test_get_all_global_settings(self):
|
|
"""Should return all global settings."""
|
|
manager = ProviderConfigManager()
|
|
all_settings = manager.get_all_global_settings()
|
|
assert "default_timeout" in all_settings
|
|
assert "enable_failover" in all_settings
|
|
|
|
def test_get_all_global_returns_copy(self):
|
|
"""get_all_global_settings should return a copy."""
|
|
manager = ProviderConfigManager()
|
|
settings = manager.get_all_global_settings()
|
|
settings["new_key"] = "new_value"
|
|
assert manager.get_global_setting("new_key") is None
|
|
|
|
|
|
class TestConfigPersistence:
|
|
"""Test configuration save/load functionality."""
|
|
|
|
def test_save_and_load_config(self):
|
|
"""Should save and load configuration from file."""
|
|
with tempfile.NamedTemporaryFile(
|
|
suffix=".json", delete=False, mode="w"
|
|
) as f:
|
|
config_path = Path(f.name)
|
|
|
|
try:
|
|
manager = ProviderConfigManager(config_file=config_path)
|
|
manager.set_provider_settings(
|
|
"test_provider",
|
|
ProviderSettings(name="test_provider", priority=3),
|
|
)
|
|
manager.set_global_setting("custom_option", True)
|
|
|
|
assert manager.save_config() is True
|
|
|
|
# Verify the file was created and is valid JSON
|
|
assert config_path.exists()
|
|
with open(config_path, "r") as f:
|
|
data = json.load(f)
|
|
assert "providers" in data
|
|
assert "test_provider" in data["providers"]
|
|
assert data["providers"]["test_provider"]["priority"] == 3
|
|
assert data["global"]["custom_option"] is True
|
|
finally:
|
|
config_path.unlink(missing_ok=True)
|
|
|
|
def test_save_config_no_path(self):
|
|
"""Should return False when no path is specified."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.save_config() is False
|
|
|
|
def test_load_config_nonexistent_file(self):
|
|
"""Should return False when file doesn't exist."""
|
|
manager = ProviderConfigManager()
|
|
assert manager.load_config(Path("/nonexistent.json")) is False
|
|
|
|
def test_load_config_invalid_json(self):
|
|
"""Should return False for invalid JSON file."""
|
|
with tempfile.NamedTemporaryFile(
|
|
suffix=".json", delete=False, mode="w"
|
|
) as f:
|
|
f.write("not valid json{{{")
|
|
config_path = Path(f.name)
|
|
|
|
try:
|
|
manager = ProviderConfigManager()
|
|
assert manager.load_config(config_path) is False
|
|
finally:
|
|
config_path.unlink(missing_ok=True)
|
|
|
|
def test_save_creates_parent_directories(self):
|
|
"""save_config should create parent directories if needed."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
config_path = Path(tmpdir) / "subdir" / "config.json"
|
|
manager = ProviderConfigManager(config_file=config_path)
|
|
manager.set_provider_settings(
|
|
"p1", ProviderSettings(name="p1")
|
|
)
|
|
assert manager.save_config() is True
|
|
assert config_path.exists()
|
|
|
|
|
|
class TestResetToDefaults:
|
|
"""Test reset functionality."""
|
|
|
|
def test_reset_clears_providers(self):
|
|
"""reset_to_defaults should clear all provider settings."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_provider_settings(
|
|
"p1", ProviderSettings(name="p1")
|
|
)
|
|
manager.reset_to_defaults()
|
|
assert manager.get_all_provider_settings() == {}
|
|
|
|
def test_reset_restores_global_defaults(self):
|
|
"""reset_to_defaults should restore default global settings."""
|
|
manager = ProviderConfigManager()
|
|
manager.set_global_setting("default_timeout", 999)
|
|
manager.set_global_setting("custom_key", "value")
|
|
|
|
manager.reset_to_defaults()
|
|
|
|
assert manager.get_global_setting("default_timeout") == 30
|
|
assert manager.get_global_setting("custom_key") is None
|
|
|
|
|
|
class TestGetConfigManagerSingleton:
|
|
"""Test the get_config_manager singleton function."""
|
|
|
|
def test_returns_instance(self):
|
|
"""get_config_manager should return a ProviderConfigManager."""
|
|
# Reset global state for test
|
|
import src.core.providers.config_manager as cm
|
|
cm._config_manager = None
|
|
|
|
manager = get_config_manager()
|
|
assert isinstance(manager, ProviderConfigManager)
|
|
|
|
# Cleanup
|
|
cm._config_manager = None
|
|
|
|
def test_returns_same_instance(self):
|
|
"""get_config_manager should return same instance on repeated calls."""
|
|
import src.core.providers.config_manager as cm
|
|
cm._config_manager = None
|
|
|
|
first = get_config_manager()
|
|
second = get_config_manager()
|
|
assert first is second
|
|
|
|
# Cleanup
|
|
cm._config_manager = None
|