265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""classifier.mode: local_llm / cloud_llm / local_encoder.
|
|
|
|
Every existing deployment is on "local_llm" implicitly today; these tests
|
|
pin that the new field defaults to it (so nothing changes for anyone who
|
|
never touches the new keys) and that the two cross-field validators reject
|
|
the ambiguous or incomplete cloud_llm/local_encoder configurations the same
|
|
way ``tool_use_category_is_a_real_category`` already rejects a dangling
|
|
category reference elsewhere in this file's config.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from pydantic import ValidationError
|
|
|
|
from config import RouterConfig
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
@pytest.fixture
|
|
def raw() -> dict:
|
|
with open(ROOT / "config" / "config.yaml") as fh:
|
|
return yaml.safe_load(fh)
|
|
|
|
|
|
def test_mode_defaults_to_local_llm(raw):
|
|
"""The shipped config.yaml writes it explicitly, but a deployment that
|
|
doesn't should still get today's behavior, unchanged."""
|
|
cfg = copy.deepcopy(raw)
|
|
del cfg["classifier"]["mode"]
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.mode == "local_llm"
|
|
|
|
|
|
def test_shipped_config_loads_with_mode_local_llm(raw):
|
|
loaded = RouterConfig(**raw)
|
|
assert loaded.classifier.mode == "local_llm"
|
|
assert loaded.classifier.cloud_primary is None
|
|
assert loaded.classifier.cloud_primary_auto is False
|
|
assert loaded.classifier.encoder is None
|
|
|
|
|
|
def test_unknown_mode_value_is_rejected(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "cloud_encoder_llm_hybrid" # not a real mode
|
|
with pytest.raises(ValidationError, match="literal_error"):
|
|
# match on the error TYPE, not just any ValidationError: on a tree
|
|
# that doesn't know about classifier.mode at all yet, setting it
|
|
# also raises ValidationError (extra_forbidden), which would let
|
|
# this test pass for the wrong reason on the pre-change code.
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
# --- cloud_llm: exactly one of cloud_primary / cloud_primary_auto ---------
|
|
|
|
|
|
def test_cloud_llm_with_neither_primary_set_is_rejected(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "cloud_llm"
|
|
with pytest.raises(ValueError, match="neither classifier.cloud_primary"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_cloud_llm_with_both_primaries_set_is_rejected(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "cloud_llm"
|
|
cfg["classifier"]["cloud_primary"] = {
|
|
"base_url": "https://api.neuralwatt.com/v1",
|
|
"model": "deepseek-v4-flash",
|
|
}
|
|
cfg["classifier"]["cloud_primary_auto"] = True
|
|
with pytest.raises(ValueError, match="BOTH classifier.cloud_primary"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_cloud_llm_with_pinned_primary_loads(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "cloud_llm"
|
|
cfg["classifier"]["cloud_primary"] = {
|
|
"base_url": "https://api.neuralwatt.com/v1",
|
|
"model": "deepseek-v4-flash",
|
|
"api_key_env": "NEURALWATT_API_KEY",
|
|
}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.cloud_primary.model == "deepseek-v4-flash"
|
|
assert loaded.classifier.cloud_primary_auto is False
|
|
|
|
|
|
def test_cloud_llm_with_auto_loads(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "cloud_llm"
|
|
cfg["classifier"]["cloud_primary_auto"] = True
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.cloud_primary_auto is True
|
|
assert loaded.classifier.cloud_primary is None
|
|
|
|
|
|
def test_local_llm_mode_does_not_require_either_primary(raw):
|
|
"""The validator is scoped to mode == cloud_llm -- it must not fire for
|
|
the default mode just because cloud_primary/auto are both unset."""
|
|
cfg = copy.deepcopy(raw)
|
|
assert cfg["classifier"]["mode"] == "local_llm"
|
|
RouterConfig(**cfg) # must not raise
|
|
|
|
|
|
# --- local_encoder: needs the encoder block ------------------------------
|
|
|
|
|
|
def test_local_encoder_without_encoder_block_is_rejected(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
with pytest.raises(ValueError, match="classifier.encoder is not"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_encoder_with_empty_encoder_block_loads_with_defaults(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.encoder.model == "BAAI/bge-large-en-v1.5"
|
|
assert loaded.classifier.encoder.device == "cpu"
|
|
assert loaded.classifier.encoder.confidence_min == 0.5
|
|
|
|
|
|
def test_local_encoder_rejects_unknown_device(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {"device": "tpu"}
|
|
with pytest.raises(ValidationError, match="literal_error"):
|
|
# Same reasoning as test_unknown_mode_value_is_rejected: on a tree
|
|
# without classifier.encoder at all, this also raises (extra_
|
|
# forbidden on the block itself), which would pass for the wrong
|
|
# reason before LocalEncoderConfig's device field exists.
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_encoder_rejects_percent_style_confidence_min(raw):
|
|
"""classify_zero_shot returns a 0.0-1.0 raw similarity score, so a
|
|
percent-style value (e.g. 80 meaning "80%") must be rejected -- otherwise
|
|
no real confidence score can ever clear the threshold and every
|
|
classification silently fails. Caught live 2026-09-06 via the admin UI
|
|
taking a raw number with no conversion. The old key name
|
|
confidence_threshold is still accepted via a deprecation alias."""
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {"confidence_threshold": 80}
|
|
with pytest.raises(ValueError, match=r"must be in \[0.0, 1.0\]"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_encoder_rejects_negative_confidence_min(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {"confidence_threshold": -0.1}
|
|
with pytest.raises(ValueError, match=r"must be in \[0.0, 1.0\]"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_encoder_accepts_confidence_min_at_bounds(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {"confidence_threshold": 0.0}
|
|
assert RouterConfig(**cfg).classifier.encoder.confidence_min == 0.0
|
|
cfg["classifier"]["encoder"] = {"confidence_threshold": 1.0}
|
|
assert RouterConfig(**cfg).classifier.encoder.confidence_min == 1.0
|
|
|
|
|
|
def test_local_encoder_unaffected_by_the_cloud_llm_validator(raw):
|
|
"""A local_encoder config leaving cloud_primary/auto both unset must not
|
|
trip the cloud_llm validator -- it's scoped to mode == cloud_llm."""
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {}
|
|
RouterConfig(**cfg) # must not raise
|
|
|
|
|
|
# --- local_decision: needs the decision block -----------------------------
|
|
|
|
|
|
def test_local_decision_config_loads_with_defaults(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
cfg["classifier"]["decision"] = {}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.decision.base_url == "http://localhost:11434"
|
|
assert loaded.classifier.decision.model == "qwen3.5:4b"
|
|
assert loaded.classifier.decision.num_ctx == 8192
|
|
assert loaded.classifier.decision.timeout_s == 10
|
|
assert loaded.classifier.decision.confidence_min == 0.5
|
|
assert loaded.classifier.decision.coverage_min == 0.3
|
|
assert loaded.classifier.decision.tier_enabled is False
|
|
|
|
|
|
def test_local_decision_config_rejects_unknown_fields(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
cfg["classifier"]["decision"] = {"bogus_field": 1}
|
|
with pytest.raises(ValidationError, match="extra_forbidden"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_decision_config_rejects_invalid_confidence_min(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
cfg["classifier"]["decision"] = {"confidence_min": 1.5}
|
|
with pytest.raises(ValueError, match=r"must be in \[0.0, 1.0\]"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_missing_decision_block_rejected_for_local_decision_mode(raw):
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
with pytest.raises(ValueError, match="classifier.decision is not"):
|
|
RouterConfig(**cfg)
|
|
|
|
|
|
def test_local_decision_is_valid_mode(raw):
|
|
"""local_decision must be accepted as a classifier.mode literal alongside
|
|
local_llm/cloud_llm/local_encoder -- the entrypoint for the new mode."""
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
cfg["classifier"]["decision"] = {}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded.classifier.mode == "local_decision"
|
|
|
|
|
|
# --- local_energy metering uses the right base_url per mode ---------------
|
|
|
|
|
|
def test_local_decision_metering_uses_decision_base_url(raw):
|
|
"""local_decision mode meters classifier.decision.base_url, not
|
|
classifier.base_url."""
|
|
# --- loopback → metered ---
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_decision"
|
|
cfg["classifier"]["decision"] = {"base_url": "http://localhost:11434"}
|
|
cfg["local_energy"] = {"enabled": True, "meter": "nvidia_smi", "tariff_usd_per_kwh": 0.10}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded._sites_cache["classify"] is True
|
|
|
|
# --- remote → not metered ---
|
|
cfg2 = copy.deepcopy(raw)
|
|
cfg2["classifier"]["mode"] = "local_decision"
|
|
cfg2["classifier"]["decision"] = {"base_url": "http://remote:11434"}
|
|
cfg2["local_energy"] = {"enabled": True, "meter": "nvidia_smi", "tariff_usd_per_kwh": 0.10}
|
|
loaded2 = RouterConfig(**cfg2)
|
|
assert loaded2._sites_cache["classify"] is False
|
|
|
|
|
|
def test_local_encoder_metering_unconditional(raw):
|
|
"""local_encoder runs in-process, so classify is always metered
|
|
regardless of base_url."""
|
|
cfg = copy.deepcopy(raw)
|
|
cfg["classifier"]["mode"] = "local_encoder"
|
|
cfg["classifier"]["encoder"] = {}
|
|
cfg["classifier"]["base_url"] = "http://remote:11434/v1"
|
|
cfg["local_energy"] = {"enabled": True, "meter": "nvidia_smi", "tariff_usd_per_kwh": 0.10}
|
|
loaded = RouterConfig(**cfg)
|
|
assert loaded._sites_cache["classify"] is True
|