142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""QA for DispatchProvider.balance_url + fail-closed validators."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from config import (
|
|
PROVIDERS_WITH_BALANCE_PARSERS,
|
|
DispatchProvider,
|
|
RouterConfig,
|
|
_merge_overlay,
|
|
load_config,
|
|
)
|
|
|
|
|
|
CONFIG_PATH = Path("config/config.yaml")
|
|
|
|
|
|
def _base_cfg_dict() -> dict:
|
|
"""Return the real base config dict so mutants start from a valid tree."""
|
|
raw = load_config(CONFIG_PATH).model_dump()
|
|
# The real config currently has only "neuralwatt" in dispatch_providers.
|
|
return raw
|
|
|
|
|
|
def _provider_dict(**overrides) -> dict:
|
|
base = {"base_url": "https://example.com/v1", "api_key_env": "EXAMPLE_KEY"}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
class TestDispatchProviderBalanceUrl:
|
|
def test_none_is_allowed(self):
|
|
p = DispatchProvider(base_url="https://x/v1", api_key_env="X")
|
|
assert p.balance_url is None
|
|
|
|
def test_https_url_is_allowed(self):
|
|
p = DispatchProvider(
|
|
base_url="https://x/v1",
|
|
api_key_env="X",
|
|
balance_url="https://x/credits",
|
|
)
|
|
assert p.balance_url == "https://x/credits"
|
|
|
|
def test_http_url_rejected(self):
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
DispatchProvider(
|
|
base_url="https://x/v1",
|
|
api_key_env="X",
|
|
balance_url="http://insecure.example/credits",
|
|
)
|
|
assert "https://" in str(exc_info.value)
|
|
assert "http://insecure.example/credits" in str(exc_info.value)
|
|
|
|
def test_unknown_provider_key_fails(self):
|
|
"""(a) balance_url on a provider without a parser is an error."""
|
|
raw = _base_cfg_dict()
|
|
raw["dispatch_providers"]["unknown_provider"] = _provider_dict(
|
|
balance_url="https://unknown.example/credits",
|
|
)
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
RouterConfig(**raw)
|
|
err = str(exc_info.value)
|
|
assert "unknown_provider" in err
|
|
assert "no balance parser implementation" in err
|
|
assert "neuralwatt" not in err
|
|
|
|
def test_telemetry_provider_with_balance_url_fails(self):
|
|
"""(b) balance_url + has_energy_telemetry is a double-source error."""
|
|
raw = _base_cfg_dict()
|
|
raw["dispatch_providers"]["neuralwatt"]["balance_url"] = (
|
|
"https://neuralwatt.example/credits"
|
|
)
|
|
raw["dispatch_providers"]["neuralwatt"]["has_energy_telemetry"] = True
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
RouterConfig(**raw)
|
|
err = str(exc_info.value)
|
|
assert "neuralwatt" in err
|
|
assert "has_energy_telemetry=true" in err
|
|
assert "Telemetry providers already report balance per completion" in err
|
|
|
|
def test_openrouter_balance_url_loads(self):
|
|
"""(c) openrouter with a balance URL loads fine."""
|
|
raw = _base_cfg_dict()
|
|
raw["dispatch_providers"]["openrouter"] = _provider_dict(
|
|
balance_url="https://openrouter.ai/api/v1/credits",
|
|
)
|
|
cfg = RouterConfig(**raw)
|
|
assert cfg.dispatch_providers["openrouter"].balance_url == (
|
|
"https://openrouter.ai/api/v1/credits"
|
|
)
|
|
|
|
def test_http_balance_url_rejected_at_provider_level(self):
|
|
"""(d) http:// balance URL fails the field validator."""
|
|
raw = _base_cfg_dict()
|
|
raw["dispatch_providers"]["openrouter"] = _provider_dict(
|
|
balance_url="http://openrouter.example/credits",
|
|
)
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
RouterConfig(**raw)
|
|
assert "https://" in str(exc_info.value)
|
|
|
|
|
|
class TestRegistryAndRealConfig:
|
|
def test_registry_is_openrouter_only(self):
|
|
assert PROVIDERS_WITH_BALANCE_PARSERS == frozenset({"openrouter"})
|
|
|
|
def test_real_config_still_loads(self):
|
|
cfg = load_config(CONFIG_PATH)
|
|
assert "neuralwatt" in cfg.dispatch_providers
|
|
assert cfg.dispatch_providers["neuralwatt"].balance_url is None
|
|
|
|
|
|
class TestAdminRoundTripPreservation:
|
|
"""(e2) _merge_overlay preserves base-only balance_url through provider edits."""
|
|
|
|
def test_overlay_preserves_balance_url(self):
|
|
base = {
|
|
"dispatch_providers": {
|
|
"openrouter": _provider_dict(
|
|
balance_url="https://openrouter.ai/api/v1/credits",
|
|
),
|
|
},
|
|
}
|
|
overlay = {
|
|
"dispatch_providers": {
|
|
"openrouter": {
|
|
"api_key_env": "OPENROUTER_API_KEY_EDITED",
|
|
"timeout_seconds": 30,
|
|
},
|
|
},
|
|
}
|
|
merged = _merge_overlay(base, overlay)
|
|
assert merged["dispatch_providers"]["openrouter"]["balance_url"] == (
|
|
"https://openrouter.ai/api/v1/credits"
|
|
)
|
|
assert (
|
|
merged["dispatch_providers"]["openrouter"]["api_key_env"]
|
|
== "OPENROUTER_API_KEY_EDITED"
|
|
)
|