"""Load-time validation for ``RouterConfig`` and its nested models. This file covers core config shape tests that are not tied to the service endpoints in ``test_config_endpoints.py``. """ from __future__ import annotations import copy from pathlib import Path import pytest import yaml from pydantic import ValidationError from config import RouterConfig, _merge_overlay, load_config ROOT = Path(__file__).resolve().parent.parent @pytest.fixture def raw() -> dict: with open(ROOT / "config" / "config.yaml") as fh: return yaml.safe_load(fh) # --- routing profiles ------------------------------------------------------- def test_profiles_empty_dict_defaults(raw): """A missing ``profiles`` section defaults to an empty dict.""" cfg = copy.deepcopy(raw) cfg.pop("profiles", None) loaded = RouterConfig(**cfg) assert loaded.profiles == {} def test_a_valid_locality_profile_loads(raw): cfg = copy.deepcopy(raw) cfg["profiles"] = {"my_locality": {"provider": "ollama-local"}} loaded = RouterConfig(**cfg) assert loaded.profiles["my_locality"].provider == "ollama-local" def test_a_profile_with_all_fields_loads(raw): cfg = copy.deepcopy(raw) cfg["profiles"] = { "my_onlycheaps": { "min_tier": 1, "max_tier": 2, "latency_tolerance": "batch", "max_cost_per_1m_completion": 0.50, "allowed_model_ids": {"foo", "bar"}, } } loaded = RouterConfig(**cfg) profile = loaded.profiles["my_onlycheaps"] assert profile.min_tier == 1 assert profile.max_tier == 2 assert profile.latency_tolerance == "batch" assert profile.max_cost_per_1m_completion == 0.50 assert profile.allowed_model_ids == {"foo", "bar"} def test_profile_with_unknown_key_is_rejected(raw): cfg = copy.deepcopy(raw) cfg["profiles"] = {"locality": {"provider": "ollama-local", "quality_tolerance": 0.2}} with pytest.raises(ValueError, match="quality_tolerance"): RouterConfig(**cfg) @pytest.mark.parametrize( "tier_key,tier_value", [ ("min_tier", 0), ("min_tier", 4), ("max_tier", 0), ("max_tier", 4), ], ) def test_profile_tier_out_of_range_is_rejected(raw, tier_key, tier_value): cfg = copy.deepcopy(raw) cfg["profiles"] = {"bigboybritches": {tier_key: tier_value}} with pytest.raises(ValueError, match="tier"): RouterConfig(**cfg) def test_profile_min_tier_above_max_tier_is_rejected(raw): cfg = copy.deepcopy(raw) cfg["profiles"] = {"bad": {"min_tier": 3, "max_tier": 1}} with pytest.raises(ValueError, match="min_tier.*max_tier"): RouterConfig(**cfg) @pytest.mark.parametrize("latency", ["realtime", "", "BATCH"]) def test_profile_bad_latency_tolerance_is_rejected(raw, latency): cfg = copy.deepcopy(raw) cfg["profiles"] = {"locality": {"latency_tolerance": latency}} with pytest.raises(ValueError, match="latency_tolerance"): RouterConfig(**cfg) @pytest.mark.parametrize("cost", [0, -1.5]) def test_profile_nonpositive_cost_is_rejected(raw, cost): cfg = copy.deepcopy(raw) cfg["profiles"] = {"locality": {"max_cost_per_1m_completion": cost}} with pytest.raises(ValueError, match="max_cost_per_1m_completion"): RouterConfig(**cfg) def test_profile_empty_allowed_model_ids_is_rejected(raw): cfg = copy.deepcopy(raw) cfg["profiles"] = {"locality": {"allowed_model_ids": []}} with pytest.raises(ValueError, match="allowed_model_ids"): RouterConfig(**cfg) # --- default_profile + builtin-profile collisions -------------------------------- def test_default_profile_defaults_to_default(raw): """``routing.default_profile`` defaults to the literal string "default".""" cfg = copy.deepcopy(raw) loaded = RouterConfig(**cfg) assert loaded.routing.default_profile == "default" @pytest.mark.parametrize("value", ["", " "]) def test_default_profile_rejects_empty_string(raw, value): """A blank default_profile is a config error: bare `auto` needs somewhere to resolve.""" cfg = copy.deepcopy(raw) cfg.setdefault("routing", {})["default_profile"] = value with pytest.raises(ValueError, match="default_profile"): RouterConfig(**cfg) @pytest.mark.parametrize( "builtin_name", ["default", "batch", "locality", "bigboybritches", "onlycheaps"], ) def test_config_profile_name_collision_with_builtin_is_rejected(raw, builtin_name): """A configured profile may not shadow any built-in profile name.""" cfg = copy.deepcopy(raw) cfg["profiles"] = {builtin_name: {"latency_tolerance": "batch"}} with pytest.raises(ValueError, match=builtin_name) as exc: RouterConfig(**cfg) detail = str(exc.value) # The error must also list the reserved names so the operator does not have to guess. assert "default" in detail assert "batch" in detail def test_default_profile_rejects_unknown_name(raw): """routing.default_profile must name a builtin or configured profile.""" cfg = copy.deepcopy(raw) cfg.setdefault("routing", {})["default_profile"] = "nosuchprofile" with pytest.raises(ValueError, match="nosuchprofile") as exc: RouterConfig(**cfg) detail = str(exc.value) assert "default" in detail assert "batch" in detail def test_default_profile_accepts_configured_profile(raw): """A configured profile name is a valid default_profile.""" cfg = copy.deepcopy(raw) cfg["profiles"] = {"somesuch": {"min_tier": 1}} cfg.setdefault("routing", {})["default_profile"] = "somesuch" loaded = RouterConfig(**cfg) assert loaded.routing.default_profile == "somesuch" @pytest.mark.parametrize( "builtin_name", ["default", "batch", "locality", "bigboybritches", "onlycheaps"], ) def test_default_profile_accepts_builtin_name(raw, builtin_name): """Any built-in profile name is a valid default_profile.""" cfg = copy.deepcopy(raw) cfg.setdefault("routing", {})["default_profile"] = builtin_name loaded = RouterConfig(**cfg) assert loaded.routing.default_profile == builtin_name def _write_configs(tmp_path: Path, base: dict | None, overlay: dict | None): """Write *base* and *overlay* YAML files into *tmp_path* and return the base path.""" cfg_dir = tmp_path / "config" cfg_dir.mkdir(exist_ok=True) base_path = cfg_dir / "config.yaml" if base is not None: base_path.write_text(yaml.safe_dump(base)) if overlay is not None: (cfg_dir / "config.local.yaml").write_text(yaml.safe_dump(overlay)) return base_path def test_load_config_without_overlay_returns_base_only(tmp_path: Path, raw): """No overlay file ⇒ load_config returns the same as RouterConfig(**base_raw).""" base_path = _write_configs(tmp_path, raw, overlay=None) loaded = load_config(str(base_path), include_overlay=True) assert loaded.objective.quality_tolerance == 0.1 assert loaded.routing.default_profile == "default" assert loaded.routing.allowed_access_levels == ["public"] assert loaded.local_energy.enabled is False def test_load_config_overlay_deep_merge_preserves_sibling_keys(tmp_path: Path, raw): """Overlay sets local_energy.tariff_usd_per_kwh only; base keys remain intact.""" overlay = {"local_energy": {"tariff_usd_per_kwh": 0.999}} base_path = _write_configs(tmp_path, copy.deepcopy(raw), overlay) loaded = load_config(str(base_path), include_overlay=True) assert loaded.local_energy.tariff_usd_per_kwh == 0.999 assert loaded.local_energy.enabled is False assert loaded.local_energy.meter == "nvidia_smi" assert loaded.objective.quality_tolerance == 0.1 def test_load_config_overlay_list_replaces_wholesale(tmp_path: Path, raw): """An overlay list replaces the base list — it does NOT append.""" overlay = {"routing": {"allowed_access_levels": ["public", "canary"]}} base_path = _write_configs(tmp_path, copy.deepcopy(raw), overlay) loaded = load_config(str(base_path), include_overlay=True) assert loaded.routing.allowed_access_levels == ["public", "canary"] assert "private" not in loaded.routing.allowed_access_levels assert loaded.routing.default_profile == "default" def test_load_config_overlay_unknown_key_raises(tmp_path: Path): """Overlay containing a top-level unknown key triggers Pydantic validation.""" base = {"objective": {"quality_tolerance": 0.1}} overlay = {"objective": {"quality_tolerance": 0.1, "total_bullshit_key": 42}} base_path = _write_configs(tmp_path, base, overlay) with pytest.raises((ValueError, ValidationError)) as exc_info: load_config(str(base_path), include_overlay=True) assert "total_bullshit_key" in str(exc_info.value) def test_default_provider_unknown_is_rejected(raw): """dispatch_settings.default_provider must name a key in dispatch_providers.""" cfg = copy.deepcopy(raw) cfg["dispatch_settings"] = {"default_provider": "nosuchprovider"} with pytest.raises((ValueError, ValidationError), match="nosuchprovider") as exc: RouterConfig(**cfg) assert "dispatch_providers" in str(exc.value) def test_default_provider_defaults_to_neuralwatt_when_present(raw): """The default value must be valid against the default config.""" cfg = copy.deepcopy(raw) loaded = RouterConfig(**cfg) assert loaded.dispatch_settings.default_provider == "neuralwatt" def test_merge_overlay_helper_list_replacement(): """_merge_overlay replaces lists rather than appending.""" base = {"routing": {"allowed_access_levels": ["public"]}} overlay = {"routing": {"allowed_access_levels": ["private"]}} merged = _merge_overlay(base, overlay) assert merged["routing"]["allowed_access_levels"] == ["private"] assert len(merged["routing"]["allowed_access_levels"]) == 1 def test_overlay_enables_local_energy(tmp_path: Path, raw): """The overlay can flip local_energy.enabled without touching the tracked base file.""" overlay = {"local_energy": {"enabled": True, "tariff_usd_per_kwh": 0.12}} base_path = _write_configs(tmp_path, copy.deepcopy(raw), overlay) loaded = load_config(str(base_path), include_overlay=True) assert loaded.local_energy.enabled is True assert loaded.local_energy.tariff_usd_per_kwh == 0.12 def test_merge_overlay_helper_recursive_mapping_merge(): """_merge_overlay recursively merges nested mappings, preserving sibling keys.""" base = { "a": {"x": 1, "y": 2}, "b": "base", } overlay = { "a": {"y": 99, "z": 3}, "c": "new", } merged = _merge_overlay(base, overlay) assert merged["a"]["x"] == 1 assert merged["a"]["y"] == 99 assert merged["a"]["z"] == 3 assert merged["b"] == "base" assert merged["c"] == "new" # --- incumbent pricing knobs ------------------------------------------------- def test_incumbent_knobs_default_from_config_yaml(raw): """The four new incumbent knobs load with their documented defaults.""" cfg = RouterConfig(**raw) obj = cfg.objective assert obj.incumbent_cache_pricing is False # Accept that the model_validator resolves None → assumed_cache_rate assert obj.incumbent_challenger_cache_rate == obj.assumed_cache_rate assert obj.incumbent_rate_refresh_seconds == 300 assert obj.incumbent_rate_min_observations == 25 def test_explicit_challenger_cache_rate_zero(raw): """incumbent_challenger_cache_rate: 0.0 stays 0.0 after resolution.""" cfg = copy.deepcopy(raw) cfg["objective"]["incumbent_challenger_cache_rate"] = 0.0 loaded = RouterConfig(**cfg) assert loaded.objective.incumbent_challenger_cache_rate == 0.0 def test_explicit_challenger_cache_rate_nonzero(raw): """An explicit nonzero challenger rate round-trips.""" cfg = copy.deepcopy(raw) cfg["objective"]["incumbent_challenger_cache_rate"] = 0.5 loaded = RouterConfig(**cfg) assert loaded.objective.incumbent_challenger_cache_rate == 0.5 @pytest.mark.parametrize( "overrides,error_match", [ ({"incumbent_rate_refresh_seconds": 0}, "refresh_seconds must be > 0"), ({"incumbent_rate_refresh_seconds": -5}, "refresh_seconds must be > 0"), ({"incumbent_challenger_cache_rate": -0.1}, r"must be in \[0, 1\]"), ({"incumbent_challenger_cache_rate": 1.5}, r"must be in \[0, 1\]"), ({"incumbent_rate_min_observations": -1}, "must be >= 0"), ], ) def test_incumbent_knobs_bad_values_rejected(raw, overrides, error_match): """Each invalid value raises ValidationError with a descriptive message.""" cfg = copy.deepcopy(raw) for key, value in overrides.items(): cfg["objective"][key] = value with pytest.raises(ValidationError, match=error_match): RouterConfig(**cfg)