# retoor <retoor@molodetz.nl>
import pytest
from pydantic import ValidationError
from devplacepy.services.openai_gateway import routing as r
from devplacepy.services.openai_gateway.usage import pricing_from_cfg
def _cleanup(providers, models):
for name in providers:
r.provider_store.remove(name)
for source in models:
r.model_store.remove(source)
def test_no_routes_pass_through(local_db):
assert r.chat_overlay("ghost-model-xyz", {}) is None
assert r.embed_overlay("ghost-embed-xyz", {}) is None
assert r.image_overlay("ghost-image-xyz", {}) is None
def test_chat_route_overlay_and_economy(local_db):
r.provider_store.set(
r.ProviderIn(
name="utor",
base_url="https://up.example/v1/chat/completions",
api_key="k1",
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="ut-chat",
provider="utor",
target_model="vendor/big",
kind="chat",
context_window=64000,
price_cache_hit_per_m=1.0,
price_cache_miss_per_m=2.0,
price_output_per_m=8.0,
)
)
try:
overlay = r.chat_overlay("ut-chat", {"gateway_model_context_map": "{}"})
assert overlay["gateway_force_model"] is True
assert overlay["gateway_model"] == "vendor/big"
assert overlay["gateway_upstream_url"] == "https://up.example/v1/chat/completions"
assert overlay["gateway_api_key"] == "k1"
assert overlay["gateway_price_output_per_m"] == 8.0
assert overlay["gateway_model_context_map"]["vendor/big"] == 64000
pricing = pricing_from_cfg(overlay)
assert pricing.chat_output_per_m == 8.0
assert pricing.chat_cache_miss_per_m == 2.0
finally:
_cleanup(["utor"], ["ut-chat"])
def test_vision_merge_overlay(local_db):
r.provider_store.set(
r.ProviderIn(
name="utvis",
base_url="https://vis.example/v1/chat/completions",
api_key="vk",
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="ut-vision",
provider="utvis",
target_model="vendor/vlm",
kind="chat",
vision_model="vendor/describe",
price_input_per_m=0.5,
)
)
try:
overlay = r.chat_overlay("ut-vision", {})
assert overlay["gateway_vision_enabled"] is True
assert overlay["gateway_vision_model"] == "vendor/describe"
assert overlay["gateway_vision_url"] == "https://vis.example/v1/chat/completions"
assert overlay["gateway_vision_key"] == "vk"
assert overlay["gateway_vision_price_input_per_m"] == 0.5
finally:
_cleanup(["utvis"], ["ut-vision"])
def test_image_route_and_kind_isolation(local_db):
r.provider_store.set(
r.ProviderIn(
name="utimg",
base_url="https://img.example/v1/chat/completions",
api_key="ik",
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="ut-image",
provider="utimg",
target_model="vendor/flux",
kind="image",
price_input_per_m=0.05,
)
)
try:
overlay = r.image_overlay("ut-image", {})
assert overlay["gateway_image_model"] == "vendor/flux"
assert overlay["gateway_image_url"] == "https://img.example/v1/images"
assert overlay["gateway_image_key"] == "ik"
assert overlay["gateway_image_price_per_call"] == 0.05
assert r.chat_overlay("ut-image", {}) is None
assert r.embed_overlay("ut-image", {}) is None
pricing = pricing_from_cfg(overlay)
assert pricing.image_per_call == 0.05
finally:
_cleanup(["utimg"], ["ut-image"])
def test_embed_route_and_kind_isolation(local_db):
r.provider_store.set(
r.ProviderIn(
name="utemb",
base_url="https://emb.example/v1/chat/completions",
api_key="ek",
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="ut-embed",
provider="utemb",
target_model="vendor/embed",
kind="embed",
price_input_per_m=0.13,
)
)
try:
overlay = r.embed_overlay("ut-embed", {})
assert overlay["gateway_embed_model"] == "vendor/embed"
assert overlay["gateway_embed_url"] == "https://emb.example/v1/embeddings"
assert overlay["gateway_embed_key"] == "ek"
assert overlay["gateway_embed_price_input_per_m"] == 0.13
assert r.chat_overlay("ut-embed", {}) is None
finally:
_cleanup(["utemb"], ["ut-embed"])
def test_inactive_route_ignored(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="ut-off",
target_model="vendor/x",
kind="chat",
is_active=False,
)
)
try:
assert r.chat_overlay("ut-off", {}) is None
finally:
_cleanup([], ["ut-off"])
def test_blank_provider_uses_default_upstream(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="ut-default",
target_model="vendor/def",
kind="chat",
price_output_per_m=3.0,
)
)
try:
overlay = r.chat_overlay("ut-default", {})
assert overlay["gateway_model"] == "vendor/def"
assert "gateway_upstream_url" not in overlay
assert "gateway_api_key" not in overlay
assert overlay["gateway_price_output_per_m"] == 3.0
finally:
_cleanup([], ["ut-default"])
def test_tier2_and_off_peak_fields_propagate_through_overlay(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="ut-tiered",
target_model="vendor/tiered",
kind="chat",
price_cache_hit_per_m=0.0028,
price_cache_miss_per_m=0.14,
price_output_per_m=0.28,
context_tier_threshold_tokens=128_000,
price_output_per_m_tier2=0.56,
off_peak_start_minute=990,
off_peak_end_minute=30,
off_peak_discount_pct=25.0,
)
)
try:
overlay = r.chat_overlay("ut-tiered", {})
assert overlay["gateway_context_tier_threshold_tokens"] == 128_000
assert overlay["gateway_price_output_per_m_tier2"] == 0.56
assert overlay["gateway_price_cache_hit_per_m_tier2"] is None
assert overlay["gateway_off_peak_start_minute"] == 990
assert overlay["gateway_off_peak_end_minute"] == 30
assert overlay["gateway_off_peak_discount_pct"] == 25.0
pricing = pricing_from_cfg(overlay)
assert pricing.context_tier_threshold_tokens == 128_000
assert pricing.chat_output_per_m_tier2 == 0.56
assert pricing.chat_cache_hit_per_m_tier2 is None
assert pricing.off_peak_start_minute == 990
finally:
_cleanup([], ["ut-tiered"])
def test_unrouted_request_never_gains_tier_or_off_peak_keys(local_db):
assert r.chat_overlay("ghost-model-untiered", {}) is None
pricing = pricing_from_cfg({})
assert pricing.context_tier_threshold_tokens == 0
assert pricing.off_peak_start_minute is None
assert pricing.off_peak_end_minute is None
assert pricing.chat_output_per_m_tier2 is None
def test_off_peak_window_requires_both_start_and_end():
with pytest.raises(ValidationError):
r.ModelRouteIn(
source_model="ut-bad-window",
target_model="vendor/x",
off_peak_start_minute=60,
)
with pytest.raises(ValidationError):
r.ModelRouteIn(
source_model="ut-bad-window",
target_model="vendor/x",
off_peak_end_minute=120,
)
def test_seed_default_image_routes_is_idempotent(local_db):
r.model_store.remove(r.IMAGE_SOURCE_MODEL)
r.seed_default_image_routes()
route = r.model_store.get(r.IMAGE_SOURCE_MODEL)
assert route is not None
assert route.kind == "image"
assert route.target_model == r.FLUX_TARGET_MODEL
r.seed_default_image_routes()
assert r.model_store.get(r.IMAGE_SOURCE_MODEL).target_model == r.FLUX_TARGET_MODEL
def test_seed_default_deepseek_routes_is_idempotent_and_preserves_customization(
local_db,
):
r.seed_default_deepseek_routes()
flash = r.model_store.get("deepseek-v4-flash")
pro = r.model_store.get("deepseek-v4-pro")
assert flash is not None and pro is not None
assert flash.price_cache_hit_per_m == 0.0028
assert flash.price_cache_miss_per_m == 0.14
assert flash.price_output_per_m == 0.28
assert flash.context_window == 1_048_576
assert pro.price_cache_hit_per_m == 0.003625
assert pro.price_cache_miss_per_m == 0.435
assert pro.price_output_per_m == 0.87
r.model_store.set(
r.ModelRouteIn(
source_model="deepseek-v4-pro",
target_model="deepseek-v4-pro",
price_output_per_m=1.23,
)
)
try:
r.seed_default_deepseek_routes()
assert r.model_store.get("deepseek-v4-pro").price_output_per_m == 1.23
finally:
r.model_store.set(
r.ModelRouteIn(
source_model="deepseek-v4-pro",
target_model="deepseek-v4-pro",
price_cache_hit_per_m=0.003625,
price_cache_miss_per_m=0.435,
price_output_per_m=0.87,
context_window=1_048_576,
)
)
Reference in New Issue View Git Blame Copy Permalink