338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""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)
|