mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
04a25083a6
Co-authored-by: yassin <yassin@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2363 lines
93 KiB
Python
2363 lines
93 KiB
Python
"""
|
|
Test that per-deployment custom pricing does not pollute the shared backend
|
|
model key in litellm.model_cost.
|
|
|
|
When two deployments share the same backend model (e.g. vertex_ai/gemini-2.5-flash)
|
|
and one has explicit zero-cost pricing in model_info, the other deployment
|
|
should still use the built-in pricing.
|
|
"""
|
|
|
|
import copy
|
|
import logging
|
|
import os
|
|
import re
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.litellm_core_utils.ptu_pricing import ptu_config_error
|
|
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
|
|
from litellm.utils import (
|
|
_invalidate_model_cost_lowercase_map,
|
|
reapply_runtime_model_cost_registrations,
|
|
)
|
|
|
|
|
|
def _simulate_price_data_reload(fetched_catalog):
|
|
"""Drive what a price data reload does to this process's litellm state.
|
|
|
|
Mirrors `litellm.proxy.proxy_server._swap_in_model_cost_map`, which is the
|
|
one place both reload paths adopt a freshly fetched catalog; that wiring is
|
|
covered in the proxy's own tests, so these exercise the replay itself
|
|
without dragging the proxy in. The provider model sets that helper also
|
|
repopulates are left alone, since nothing here reads them and rebuilding
|
|
them from a two-entry catalog would outlive the test.
|
|
"""
|
|
litellm.model_cost = fetched_catalog
|
|
_invalidate_model_cost_lowercase_map()
|
|
reapply_runtime_model_cost_registrations()
|
|
|
|
|
|
def _nested_container_ids(value: object) -> frozenset[int]:
|
|
"""Identities of every dict/list reachable from `value`, so two structures can be
|
|
checked for shared mutable state without writing into either one."""
|
|
if isinstance(value, dict):
|
|
return frozenset({id(value)} | {i for v in value.values() for i in _nested_container_ids(v)})
|
|
if isinstance(value, list):
|
|
return frozenset({id(value)} | {i for v in value for i in _nested_container_ids(v)})
|
|
return frozenset()
|
|
|
|
|
|
def _restore_model_cost_entries(original_entries):
|
|
for key, value in original_entries.items():
|
|
if value is None:
|
|
litellm.model_cost.pop(key, None)
|
|
else:
|
|
litellm.model_cost[key] = value
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_should_not_pollute_shared_key_with_zero_cost_pricing():
|
|
"""
|
|
When deployment A has input_cost_per_token=0 and deployment B has no
|
|
custom pricing, deployment B should still report the built-in pricing
|
|
(not zero).
|
|
"""
|
|
backend_model = "vertex_ai/gemini-2.5-flash"
|
|
|
|
# Grab built-in pricing before creating any router
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_input_cost = builtin_info["input_cost_per_token"]
|
|
builtin_output_cost = builtin_info["output_cost_per_token"]
|
|
|
|
# Sanity: built-in pricing should be non-zero for this model
|
|
assert builtin_input_cost > 0, "Test requires a model with non-zero built-in pricing"
|
|
assert builtin_output_cost > 0, "Test requires a model with non-zero built-in pricing"
|
|
|
|
router = Router(
|
|
model_list=[
|
|
# Deployment A: explicit zero-cost pricing
|
|
{
|
|
"model_name": "custom-zero-cost-model",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-1",
|
|
},
|
|
"model_info": {
|
|
"id": "deployment-a-zero-cost",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
},
|
|
},
|
|
# Deployment B: no custom pricing, relies on built-in
|
|
{
|
|
"model_name": "standard-cost-model",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-2",
|
|
},
|
|
"model_info": {
|
|
"id": "deployment-b-builtin-cost",
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
# Deployment A: should report zero pricing via its unique model_id
|
|
info_a = router.get_deployment_model_info(
|
|
model_id="deployment-a-zero-cost",
|
|
model_name=backend_model,
|
|
)
|
|
assert info_a is not None
|
|
assert info_a["input_cost_per_token"] == 0.0
|
|
assert info_a["output_cost_per_token"] == 0.0
|
|
|
|
# Deployment B: should report built-in pricing, NOT zero
|
|
info_b = router.get_deployment_model_info(
|
|
model_id="deployment-b-builtin-cost",
|
|
model_name=backend_model,
|
|
)
|
|
assert info_b is not None
|
|
assert info_b["input_cost_per_token"] == builtin_input_cost, (
|
|
f"Deployment B should use built-in input cost {builtin_input_cost}, got {info_b['input_cost_per_token']}"
|
|
)
|
|
assert info_b["output_cost_per_token"] == builtin_output_cost, (
|
|
f"Deployment B should use built-in output cost {builtin_output_cost}, got {info_b['output_cost_per_token']}"
|
|
)
|
|
|
|
|
|
def test_should_not_pollute_shared_key_with_custom_nonzero_pricing():
|
|
"""
|
|
A deployment with custom (non-zero) pricing should not overwrite
|
|
the shared backend key's built-in pricing.
|
|
"""
|
|
backend_model = "vertex_ai/gemini-2.5-flash"
|
|
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_input_cost = builtin_info["input_cost_per_token"]
|
|
|
|
router = Router(
|
|
model_list=[
|
|
# Deployment with custom high pricing
|
|
{
|
|
"model_name": "expensive-model",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-3",
|
|
},
|
|
"model_info": {
|
|
"id": "deployment-expensive",
|
|
"input_cost_per_token": 0.99,
|
|
"output_cost_per_token": 0.99,
|
|
},
|
|
},
|
|
# Deployment relying on built-in pricing
|
|
{
|
|
"model_name": "standard-model",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-4",
|
|
},
|
|
"model_info": {
|
|
"id": "deployment-standard",
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
# Custom pricing deployment should see its custom values
|
|
info_expensive = router.get_deployment_model_info(
|
|
model_id="deployment-expensive",
|
|
model_name=backend_model,
|
|
)
|
|
assert info_expensive is not None
|
|
assert info_expensive["input_cost_per_token"] == 0.99
|
|
assert info_expensive["output_cost_per_token"] == 0.99
|
|
|
|
# Standard deployment should still see built-in pricing
|
|
info_standard = router.get_deployment_model_info(
|
|
model_id="deployment-standard",
|
|
model_name=backend_model,
|
|
)
|
|
assert info_standard is not None
|
|
assert info_standard["input_cost_per_token"] == builtin_input_cost, (
|
|
f"Standard deployment should use built-in pricing {builtin_input_cost}, "
|
|
f"got {info_standard['input_cost_per_token']}"
|
|
)
|
|
|
|
|
|
def test_should_store_full_pricing_under_deployment_model_id():
|
|
"""
|
|
Per-deployment pricing (including zero) should be stored and
|
|
retrievable via the unique model_id key in litellm.model_cost.
|
|
"""
|
|
backend_model = "vertex_ai/gemini-2.5-flash"
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "zero-cost-model",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-5",
|
|
},
|
|
"model_info": {
|
|
"id": "deployment-zero-check",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
# The model_id entry should exist and have the zero pricing
|
|
entry = litellm.model_cost.get("deployment-zero-check")
|
|
assert entry is not None, "Deployment should be registered by model_id"
|
|
assert entry["input_cost_per_token"] == 0.0
|
|
assert entry["output_cost_per_token"] == 0.0
|
|
|
|
|
|
def test_should_preserve_builtin_pricing_regardless_of_deployment_order():
|
|
"""
|
|
The built-in pricing should be preserved no matter which deployment
|
|
is processed first (zero-cost first, or standard first).
|
|
"""
|
|
backend_model = "vertex_ai/gemini-2.5-flash"
|
|
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_input_cost = builtin_info["input_cost_per_token"]
|
|
builtin_output_cost = builtin_info["output_cost_per_token"]
|
|
|
|
# Order 1: standard first, then zero-cost
|
|
router1 = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "standard-first",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-6",
|
|
},
|
|
"model_info": {"id": "order1-standard"},
|
|
},
|
|
{
|
|
"model_name": "zero-cost-second",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-7",
|
|
},
|
|
"model_info": {
|
|
"id": "order1-zero",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
info_std_1 = router1.get_deployment_model_info(model_id="order1-standard", model_name=backend_model)
|
|
assert info_std_1["input_cost_per_token"] == builtin_input_cost
|
|
assert info_std_1["output_cost_per_token"] == builtin_output_cost
|
|
|
|
# Order 2: zero-cost first, then standard
|
|
router2 = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "zero-cost-first",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-8",
|
|
},
|
|
"model_info": {
|
|
"id": "order2-zero",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
},
|
|
},
|
|
{
|
|
"model_name": "standard-second",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-9",
|
|
},
|
|
"model_info": {"id": "order2-standard"},
|
|
},
|
|
],
|
|
)
|
|
|
|
info_std_2 = router2.get_deployment_model_info(model_id="order2-standard", model_name=backend_model)
|
|
assert info_std_2["input_cost_per_token"] == builtin_input_cost, (
|
|
f"Order should not matter. Expected {builtin_input_cost}, got {info_std_2['input_cost_per_token']}"
|
|
)
|
|
assert info_std_2["output_cost_per_token"] == builtin_output_cost, (
|
|
f"Order should not matter. Expected {builtin_output_cost}, got {info_std_2['output_cost_per_token']}"
|
|
)
|
|
|
|
|
|
def test_responses_prefix_stripped_alias_registered_for_model_list():
|
|
"""
|
|
Register ``litellm.model_cost`` under the backend key with ``responses/`` and
|
|
under the stripped key (``responses_api_bridge_check`` removes that segment).
|
|
"""
|
|
uid = "responses-strip-alias-test-a1b2c3d4"
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-responses-strip-test",
|
|
"litellm_params": {
|
|
"model": "responses/gpt-strip-test-a1b2c3d4",
|
|
"custom_llm_provider": "azure",
|
|
"api_key": "fake-key-strip",
|
|
},
|
|
"model_info": {
|
|
"id": uid,
|
|
"supports_native_streaming": True,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
assert "azure/responses/gpt-strip-test-a1b2c3d4" in litellm.model_cost
|
|
assert "azure/gpt-strip-test-a1b2c3d4" in litellm.model_cost
|
|
assert litellm.model_cost["azure/gpt-strip-test-a1b2c3d4"].get("supports_native_streaming") is True
|
|
|
|
|
|
def test_responses_prefix_stripped_alias_registered_for_add_deployment():
|
|
"""Dynamic ``add_deployment`` must mirror ``_create_deployment`` registration."""
|
|
uid = "add-dep-responses-strip-e5f6a7b8"
|
|
router = Router(model_list=[])
|
|
deployment = Deployment(
|
|
model_name="dyn-responses-strip",
|
|
litellm_params=LiteLLM_Params(
|
|
model="responses/gpt-add-strip-e5f6a7b8",
|
|
custom_llm_provider="azure",
|
|
api_key="fake-key-add",
|
|
),
|
|
model_info=ModelInfo(id=uid, supports_native_streaming=True),
|
|
)
|
|
router.add_deployment(deployment=deployment)
|
|
assert "azure/responses/gpt-add-strip-e5f6a7b8" in litellm.model_cost
|
|
assert "azure/gpt-add-strip-e5f6a7b8" in litellm.model_cost
|
|
assert litellm.model_cost["azure/gpt-add-strip-e5f6a7b8"].get("supports_native_streaming") is True
|
|
|
|
|
|
def test_should_not_downgrade_chatgpt_shared_key_mode_with_alias_override():
|
|
"""
|
|
ChatGPT aliases that share the same backend model should not be able to
|
|
downgrade the shared backend key from responses -> chat during router setup.
|
|
"""
|
|
from litellm.main import responses_api_bridge_check
|
|
|
|
backend_model = "chatgpt/gpt-5.4"
|
|
model_keys = {
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
"chatgpt-shared-mode-base": copy.deepcopy(litellm.model_cost.get("chatgpt-shared-mode-base")),
|
|
"chatgpt-shared-mode-alias": copy.deepcopy(litellm.model_cost.get("chatgpt-shared-mode-alias")),
|
|
}
|
|
|
|
try:
|
|
backend_entry = copy.deepcopy(model_keys[backend_model]) or {}
|
|
backend_entry["litellm_provider"] = "chatgpt"
|
|
backend_entry["mode"] = "responses"
|
|
litellm.model_cost[backend_model] = backend_entry
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
router = Router(model_list=[])
|
|
with patch.object(Router, "_add_deployment", lambda self, deployment: deployment):
|
|
router._create_deployment(
|
|
deployment_info={},
|
|
_model_name="chatgpt/gpt-5.4",
|
|
_litellm_params={
|
|
"model": "gpt-5.4",
|
|
"custom_llm_provider": "chatgpt",
|
|
},
|
|
_model_info={
|
|
"id": "chatgpt-shared-mode-base",
|
|
"mode": "responses",
|
|
},
|
|
)
|
|
router._create_deployment(
|
|
deployment_info={},
|
|
_model_name="chatgpt/gpt-5.4-medium",
|
|
_litellm_params={
|
|
"model": "gpt-5.4",
|
|
"custom_llm_provider": "chatgpt",
|
|
},
|
|
_model_info={
|
|
"id": "chatgpt-shared-mode-alias",
|
|
"mode": "chat",
|
|
},
|
|
)
|
|
|
|
assert litellm.model_cost[backend_model]["mode"] == "responses"
|
|
assert "mode" in litellm.model_cost[backend_model]
|
|
|
|
bridge_model_info, bridge_model = responses_api_bridge_check(
|
|
model="gpt-5.4",
|
|
custom_llm_provider="chatgpt",
|
|
)
|
|
assert bridge_model == "gpt-5.4"
|
|
assert bridge_model_info["mode"] == "responses"
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_partial_custom_pricing_inherits_builtin_cache_pricing():
|
|
"""A deployment that overrides only input/output cost on a cache-supporting
|
|
model must still bill cache_read and cache_creation tokens. Before the
|
|
fix the deploy-id entry was registered with the user's two fields and
|
|
nothing else, so the cost calculator silently billed cache tokens at 0.
|
|
Regression for the prompt-caching cost dropout reported by the customer.
|
|
"""
|
|
backend_model = "anthropic/claude-sonnet-4-5-20250929"
|
|
deploy_id = "claude-deploy-partial-pricing"
|
|
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_cache_create = builtin_info["cache_creation_input_token_cost"]
|
|
builtin_cache_read = builtin_info["cache_read_input_token_cost"]
|
|
assert builtin_cache_create is not None and builtin_cache_create > 0
|
|
assert builtin_cache_read is not None and builtin_cache_read > 0
|
|
|
|
model_keys = {
|
|
deploy_id: litellm.model_cost.get(deploy_id),
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
}
|
|
try:
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "claude-custom",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key",
|
|
},
|
|
"model_info": {
|
|
"id": deploy_id,
|
|
"input_cost_per_token": 0.000003,
|
|
"output_cost_per_token": 0.000015,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
entry = litellm.model_cost[deploy_id]
|
|
assert entry["input_cost_per_token"] == 0.000003
|
|
assert entry["output_cost_per_token"] == 0.000015
|
|
assert entry.get("cache_creation_input_token_cost") == builtin_cache_create
|
|
assert entry.get("cache_read_input_token_cost") == builtin_cache_read
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_partial_pricing_does_not_overwrite_explicit_cache_fields():
|
|
"""When the user explicitly sets cache_*_input_token_cost on a deployment,
|
|
those values must not be replaced by the built-in fallback.
|
|
"""
|
|
backend_model = "anthropic/claude-sonnet-4-5-20250929"
|
|
deploy_id = "claude-deploy-explicit-cache"
|
|
|
|
explicit_cache_create = 0.00001
|
|
explicit_cache_read = 0.0000005
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
assert builtin_info["cache_creation_input_token_cost"] != explicit_cache_create
|
|
assert builtin_info["cache_read_input_token_cost"] != explicit_cache_read
|
|
|
|
model_keys = {
|
|
deploy_id: litellm.model_cost.get(deploy_id),
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
}
|
|
try:
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "claude-custom-explicit",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key",
|
|
},
|
|
"model_info": {
|
|
"id": deploy_id,
|
|
"input_cost_per_token": 0.000003,
|
|
"output_cost_per_token": 0.000015,
|
|
"cache_creation_input_token_cost": explicit_cache_create,
|
|
"cache_read_input_token_cost": explicit_cache_read,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
entry = litellm.model_cost[deploy_id]
|
|
assert entry.get("cache_creation_input_token_cost") == explicit_cache_create
|
|
assert entry.get("cache_read_input_token_cost") == explicit_cache_read
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_inherit_builtin_cache_pricing_fills_only_missing_fields():
|
|
"""Direct unit test of the helper: missing cache fields are filled from the
|
|
backend model's built-in entry, while an explicitly set cache field and the
|
|
user's input/output pricing are left untouched.
|
|
"""
|
|
backend_model = "anthropic/claude-sonnet-4-5-20250929"
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_cache_create = builtin_info["cache_creation_input_token_cost"]
|
|
builtin_cache_read = builtin_info["cache_read_input_token_cost"]
|
|
assert builtin_cache_create is not None and builtin_cache_create > 0
|
|
assert builtin_cache_read is not None and builtin_cache_read > 0
|
|
|
|
explicit_cache_read = builtin_cache_read + 1
|
|
model_info = {
|
|
"input_cost_per_token": 0.000003,
|
|
"cache_read_input_token_cost": explicit_cache_read,
|
|
}
|
|
|
|
Router._inherit_builtin_cache_pricing(
|
|
model_info=model_info,
|
|
backend_model=backend_model,
|
|
custom_llm_provider="anthropic",
|
|
)
|
|
|
|
assert model_info["input_cost_per_token"] == 0.000003
|
|
assert model_info["cache_read_input_token_cost"] == explicit_cache_read
|
|
assert model_info["cache_creation_input_token_cost"] == builtin_cache_create
|
|
|
|
|
|
def test_inherit_builtin_cache_pricing_noop_for_unknown_backend():
|
|
"""No canonical entry for the backend model means the helper leaves the
|
|
passed-in dict unchanged rather than raising.
|
|
"""
|
|
model_info = {"input_cost_per_token": 0.000003}
|
|
|
|
Router._inherit_builtin_cache_pricing(
|
|
model_info=model_info,
|
|
backend_model="this-backend-model-does-not-exist-x9y8z7",
|
|
custom_llm_provider=None,
|
|
)
|
|
|
|
assert model_info == {"input_cost_per_token": 0.000003}
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_fills_missing_rates():
|
|
"""Direct unit test of the helper: an entry carrying only an
|
|
off_peak_pricing block inherits the backend model's built-in base token
|
|
rates, so cost lookup via the deployment id can bill standard rates
|
|
outside the windows.
|
|
"""
|
|
backend_model = "gpt-4o-mini"
|
|
builtin_info = litellm.get_model_info(model=backend_model, custom_llm_provider="openai")
|
|
off_peak_block = {
|
|
"hours_utc": "00:00-00:00",
|
|
"input_cost_per_token": 5e-07,
|
|
"output_cost_per_token": 1e-06,
|
|
}
|
|
model_info = {"off_peak_pricing": off_peak_block}
|
|
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=model_info,
|
|
backend_model=backend_model,
|
|
custom_llm_provider="openai",
|
|
)
|
|
|
|
assert model_info["input_cost_per_token"] == builtin_info["input_cost_per_token"]
|
|
assert model_info["output_cost_per_token"] == builtin_info["output_cost_per_token"]
|
|
assert model_info["off_peak_pricing"] == off_peak_block
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_carries_threshold_rates():
|
|
"""A backend with above-threshold pricing hands the whole rate structure to
|
|
the deployment entry, so peak-hour billing of large prompts through that
|
|
entry matches the shared backend entry instead of flattening to the base
|
|
rate.
|
|
"""
|
|
backend_model = "gemini/gemini-2.5-pro"
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
assert builtin_info["input_cost_per_token_above_200k_tokens"] is not None
|
|
|
|
model_info = {
|
|
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
|
|
}
|
|
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=model_info,
|
|
backend_model=backend_model,
|
|
custom_llm_provider="gemini",
|
|
)
|
|
|
|
assert model_info["input_cost_per_token"] == builtin_info["input_cost_per_token"]
|
|
assert (
|
|
model_info["input_cost_per_token_above_200k_tokens"]
|
|
== builtin_info["input_cost_per_token_above_200k_tokens"]
|
|
)
|
|
assert (
|
|
model_info["output_cost_per_token_above_200k_tokens"]
|
|
== builtin_info["output_cost_per_token_above_200k_tokens"]
|
|
)
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_carries_companion_billing_fields():
|
|
"""Billing rules that are not literal cost rates, like the web search
|
|
billing unit, must ride along, or grounding and regional uplifts would
|
|
bill differently through the deployment entry than through the shared
|
|
backend entry.
|
|
"""
|
|
backend_model = "gemini-3-pro-image"
|
|
raw_entry = litellm.model_cost[backend_model]
|
|
assert raw_entry.get("web_search_billing_unit") is not None
|
|
|
|
model_info = {
|
|
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
|
|
}
|
|
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=model_info,
|
|
backend_model=backend_model,
|
|
custom_llm_provider=None,
|
|
)
|
|
|
|
assert model_info["web_search_billing_unit"] == raw_entry["web_search_billing_unit"]
|
|
assert model_info["input_cost_per_token"] == raw_entry["input_cost_per_token"]
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_tiered_only_backend_stores_no_zero():
|
|
"""A tiered-only backend has no flat token rates; get_model_info synthesizes
|
|
zeros for them, and storing those would mark the deployment explicitly
|
|
priced free. The tier table itself must carry over as an isolated copy so
|
|
mutating the deployment entry never touches the shared cost map.
|
|
"""
|
|
backend_model = "dashscope/qwen-flash"
|
|
raw_tiers = litellm.model_cost[backend_model]["tiered_pricing"]
|
|
|
|
model_info = {
|
|
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
|
|
}
|
|
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=model_info,
|
|
backend_model=backend_model,
|
|
custom_llm_provider="dashscope",
|
|
)
|
|
|
|
assert model_info.get("input_cost_per_token") != 0
|
|
assert model_info.get("output_cost_per_token") != 0
|
|
assert model_info["tiered_pricing"] == raw_tiers
|
|
assert model_info["tiered_pricing"] is not raw_tiers
|
|
assert model_info["tiered_pricing"][0] is not raw_tiers[0]
|
|
|
|
original_first_tier = copy.deepcopy(raw_tiers[0])
|
|
model_info["tiered_pricing"][0]["input_cost_per_token"] = 123.0
|
|
assert raw_tiers[0] == original_first_tier
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_leaves_explicit_rates_alone():
|
|
"""An entry that sets its own base rate beside the block already counts as
|
|
a full custom pricing entry; the helper must not mix builtin rates into it.
|
|
"""
|
|
model_info = {
|
|
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
|
|
"input_cost_per_token": 3e-06,
|
|
}
|
|
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=model_info,
|
|
backend_model="gpt-4o-mini",
|
|
custom_llm_provider="openai",
|
|
)
|
|
|
|
assert model_info["input_cost_per_token"] == 3e-06
|
|
assert "output_cost_per_token" not in model_info
|
|
|
|
|
|
def test_inherit_builtin_base_rates_for_off_peak_noop_without_block_or_backend():
|
|
"""Nothing happens without an off_peak_pricing block, and an unmapped
|
|
backend model leaves the entry unchanged rather than raising.
|
|
"""
|
|
plain_info = {"id": "dep-1"}
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=plain_info,
|
|
backend_model="gpt-4o-mini",
|
|
custom_llm_provider="openai",
|
|
)
|
|
assert plain_info == {"id": "dep-1"}
|
|
|
|
off_peak_info = {"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07}}
|
|
Router._inherit_builtin_base_rates_for_off_peak(
|
|
model_info=off_peak_info,
|
|
backend_model="this-backend-model-does-not-exist-x9y8z7",
|
|
custom_llm_provider=None,
|
|
)
|
|
assert "input_cost_per_token" not in off_peak_info
|
|
|
|
|
|
def test_custom_pricing_field_denylist_covers_all_builtin_pricing_fields():
|
|
"""The shared-backend-key stripping in Router relies on
|
|
CustomPricingLiteLLMParams enumerating every per-deployment pricing field.
|
|
If a new pricing field is added to ModelInfoBase but not mirrored here, a
|
|
deployment override on that field leaks into the shared backend key and
|
|
every sibling deployment reads the wrong rate (LIT-3897). This guard fails
|
|
fast when the two drift apart.
|
|
"""
|
|
import typing
|
|
|
|
from litellm.types.utils import CustomPricingLiteLLMParams, ModelInfoBase
|
|
|
|
pricing_markers = ("cost", "price", "uplift", "vector_size", "tiered_pricing")
|
|
builtin_pricing_fields = {
|
|
name for name in typing.get_type_hints(ModelInfoBase) if any(marker in name for marker in pricing_markers)
|
|
}
|
|
denylisted_fields = set(CustomPricingLiteLLMParams.model_fields.keys())
|
|
|
|
uncovered = sorted(builtin_pricing_fields - denylisted_fields)
|
|
assert not uncovered, (
|
|
"ModelInfoBase pricing fields missing from CustomPricingLiteLLMParams; "
|
|
f"these would leak into shared backend keys: {uncovered}"
|
|
)
|
|
|
|
|
|
def test_tiered_pricing_override_isolated_from_sibling_via_model_info_lookup():
|
|
"""LIT-3897: a deployment that overrides a tiered pricing field
|
|
(input_cost_per_token_above_272k_tokens) must not pollute the shared
|
|
backend key, so a sibling sharing the same backend resolves its pricing
|
|
via litellm.get_model_info (the path /model/info uses) without seeing the
|
|
override.
|
|
"""
|
|
backend_model = "gemini/gemini-2.5-flash"
|
|
override = 0.000999
|
|
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
assert builtin_info.get("input_cost_per_token_above_272k_tokens") != override
|
|
|
|
model_keys = {
|
|
"lit3897-tiered-custom": litellm.model_cost.get("lit3897-tiered-custom"),
|
|
"lit3897-tiered-sibling": litellm.model_cost.get("lit3897-tiered-sibling"),
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
}
|
|
try:
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "custom-priced-flash",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-tiered-1",
|
|
},
|
|
"model_info": {
|
|
"id": "lit3897-tiered-custom",
|
|
"input_cost_per_token_above_272k_tokens": override,
|
|
"cache_read_input_token_cost_above_272k_tokens": override,
|
|
},
|
|
},
|
|
{
|
|
"model_name": "gemini-2.5-flash",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-tiered-2",
|
|
},
|
|
"model_info": {"id": "lit3897-tiered-sibling"},
|
|
},
|
|
],
|
|
)
|
|
|
|
shared = litellm.get_model_info(model=backend_model)
|
|
assert shared.get("input_cost_per_token_above_272k_tokens") != override, (
|
|
"Tiered override leaked into the shared backend key; siblings read the wrong rate via /model/info"
|
|
)
|
|
assert shared.get("cache_read_input_token_cost_above_272k_tokens") != override
|
|
|
|
custom_entry = litellm.model_cost["lit3897-tiered-custom"]
|
|
assert custom_entry["input_cost_per_token_above_272k_tokens"] == override
|
|
assert custom_entry["cache_read_input_token_cost_above_272k_tokens"] == override
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_custom_pricing_isolated_from_sibling_via_proxy_model_info_path():
|
|
"""LIT-3897 end to end through the proxy resolution helper: the override
|
|
deployment reports its custom input rate while the sibling keeps the
|
|
canonical gemini rate when /model/info resolves each deployment. Mirrors the
|
|
ticket config where the override is set on litellm_params.
|
|
"""
|
|
from litellm.proxy.proxy_server import _get_proxy_model_info
|
|
|
|
backend_model = "gemini/gemini-2.5-flash"
|
|
override_input = 5e-05
|
|
override_output = 1e-04
|
|
|
|
builtin_info = litellm.get_model_info(model=backend_model)
|
|
builtin_input = builtin_info["input_cost_per_token"]
|
|
assert builtin_input != override_input
|
|
|
|
model_keys = {
|
|
"lit3897-proxy-custom": litellm.model_cost.get("lit3897-proxy-custom"),
|
|
"lit3897-proxy-sibling": litellm.model_cost.get("lit3897-proxy-sibling"),
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
}
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "custom-priced-flash",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-proxy-1",
|
|
"input_cost_per_token": override_input,
|
|
"output_cost_per_token": override_output,
|
|
},
|
|
"model_info": {"id": "lit3897-proxy-custom"},
|
|
},
|
|
{
|
|
"model_name": "gemini-2.5-flash",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-proxy-2",
|
|
},
|
|
"model_info": {"id": "lit3897-proxy-sibling"},
|
|
},
|
|
],
|
|
)
|
|
|
|
resolved = {
|
|
m["model_name"]: _get_proxy_model_info(model=copy.deepcopy(m))["model_info"]["input_cost_per_token"]
|
|
for m in router.model_list
|
|
}
|
|
|
|
assert resolved["custom-priced-flash"] == override_input
|
|
assert resolved["gemini-2.5-flash"] == builtin_input
|
|
assert resolved["gemini-2.5-flash"] != resolved["custom-priced-flash"]
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_custom_model_info_metadata_not_leaked_to_shared_backend_key():
|
|
"""LIT-4544: two deployments share the same backend model but carry
|
|
different custom model_info (arbitrary keys, access_via_team_ids, ids).
|
|
None of that per-deployment metadata may land on the shared backend key in
|
|
litellm.model_cost (served raw by /public/litellm_model_cost_map);
|
|
before the fix it was merged last-write-wins so values flipped randomly.
|
|
"""
|
|
backend_model = "openai/gpt-4o-mini"
|
|
shared_keys = ("gpt-4o-mini", backend_model)
|
|
leak_fields = ("id", "additionalProp1", "access_via_team_ids", "db_model")
|
|
|
|
model_keys = {
|
|
key: copy.deepcopy(litellm.model_cost.get(key))
|
|
for key in (*shared_keys, "lit4544-deploy-a", "lit4544-deploy-b")
|
|
}
|
|
try:
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "alias-unrestricted",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-a",
|
|
},
|
|
"model_info": {
|
|
"id": "lit4544-deploy-a",
|
|
"additionalProp1": {"restricted": False, "model_location": "EU"},
|
|
},
|
|
},
|
|
{
|
|
"model_name": "alias-restricted",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key-b",
|
|
},
|
|
"model_info": {
|
|
"id": "lit4544-deploy-b",
|
|
"additionalProp1": {"restricted": True, "model_location": "US"},
|
|
"access_via_team_ids": ["team-b-only"],
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
for shared_key in shared_keys:
|
|
shared_entry = litellm.model_cost.get(shared_key) or {}
|
|
leaked = [field for field in leak_fields if field in shared_entry]
|
|
assert not leaked, f"per-deployment metadata {leaked} leaked onto shared key {shared_key}: {shared_entry}"
|
|
|
|
entry_a = litellm.model_cost["lit4544-deploy-a"]
|
|
assert entry_a["additionalProp1"] == {"restricted": False, "model_location": "EU"}
|
|
entry_b = litellm.model_cost["lit4544-deploy-b"]
|
|
assert entry_b["additionalProp1"] == {"restricted": True, "model_location": "US"}
|
|
assert entry_b["access_via_team_ids"] == ["team-b-only"]
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_add_deployment_does_not_leak_custom_metadata_to_shared_backend_key():
|
|
"""LIT-4544 dynamic path: deployments added at runtime (e.g. loaded from
|
|
the DB every scheduler cycle) must not re-pollute the shared backend key
|
|
with per-deployment metadata either.
|
|
"""
|
|
backend_model = "openai/gpt-4o-mini"
|
|
shared_keys = ("gpt-4o-mini", backend_model)
|
|
deploy_id = "lit4544-add-deployment"
|
|
|
|
model_keys = {key: copy.deepcopy(litellm.model_cost.get(key)) for key in (*shared_keys, deploy_id)}
|
|
try:
|
|
router = Router(model_list=[])
|
|
router.add_deployment(
|
|
deployment=Deployment(
|
|
model_name="alias-dynamic",
|
|
litellm_params=LiteLLM_Params(
|
|
model=backend_model,
|
|
api_key="fake-key-dynamic",
|
|
),
|
|
model_info=ModelInfo(
|
|
id=deploy_id,
|
|
additionalProp1={"restricted": True},
|
|
access_via_team_ids=["team-dynamic"],
|
|
),
|
|
)
|
|
)
|
|
|
|
for shared_key in shared_keys:
|
|
shared_entry = litellm.model_cost.get(shared_key) or {}
|
|
leaked = [
|
|
field for field in ("id", "additionalProp1", "access_via_team_ids", "db_model") if field in shared_entry
|
|
]
|
|
assert not leaked, f"per-deployment metadata {leaked} leaked onto shared key {shared_key}: {shared_entry}"
|
|
|
|
assert litellm.model_cost[deploy_id]["access_via_team_ids"] == ["team-dynamic"]
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_shared_backend_model_info_keeps_schema_fields_and_drops_the_rest():
|
|
"""Unit test of the whitelist helper: cost-map schema fields survive,
|
|
custom pricing overrides and per-deployment metadata do not.
|
|
"""
|
|
from litellm.types.utils import shared_backend_model_info
|
|
|
|
filtered = shared_backend_model_info(
|
|
{
|
|
"mode": "chat",
|
|
"litellm_provider": "openai",
|
|
"max_tokens": 128000,
|
|
"supports_vision": True,
|
|
"supported_endpoints": ["/v1/responses"],
|
|
"use_openai_responses_path": True,
|
|
"input_cost_per_token": 0.99,
|
|
"output_cost_per_token": 0.99,
|
|
"id": "deploy-a",
|
|
"db_model": False,
|
|
"access_via_team_ids": ["team-a"],
|
|
"additionalProp1": {"restricted": True},
|
|
"base_model": "gpt-4o-mini",
|
|
}
|
|
)
|
|
|
|
assert filtered == {
|
|
"mode": "chat",
|
|
"litellm_provider": "openai",
|
|
"max_tokens": 128000,
|
|
"supports_vision": True,
|
|
"supported_endpoints": ["/v1/responses"],
|
|
"use_openai_responses_path": True,
|
|
}
|
|
|
|
|
|
def test_capability_flags_propagate_from_deployment_model_info_to_shared_key():
|
|
"""Backend-model capability facts (supported_endpoints,
|
|
use_openai_responses_path) declared in a deployment's model_info must reach
|
|
the shared backend key: the Bedrock Mantle routing gates read them raw off
|
|
litellm.model_cost and document proxy model_info as an override path for
|
|
models missing from the built-in cost map.
|
|
"""
|
|
from litellm.llms.bedrock_mantle.common_utils import (
|
|
mantle_base_segment,
|
|
mantle_supports_responses,
|
|
)
|
|
|
|
bare_model = "somelab.lit4544-unmapped-model"
|
|
backend_model = f"bedrock_mantle/{bare_model}"
|
|
deploy_id = "lit4544-mantle-deploy"
|
|
|
|
model_keys = {key: copy.deepcopy(litellm.model_cost.get(key)) for key in (bare_model, backend_model, deploy_id)}
|
|
try:
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "mantle-alias",
|
|
"litellm_params": {
|
|
"model": backend_model,
|
|
"api_key": "fake-key",
|
|
},
|
|
"model_info": {
|
|
"id": deploy_id,
|
|
"supported_endpoints": ["/v1/responses"],
|
|
"use_openai_responses_path": True,
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
shared_entry = litellm.model_cost.get(backend_model) or {}
|
|
assert shared_entry.get("supported_endpoints") == ["/v1/responses"]
|
|
assert shared_entry.get("use_openai_responses_path") is True
|
|
assert "id" not in shared_entry
|
|
assert mantle_supports_responses(bare_model, litellm.model_cost) is True
|
|
assert mantle_base_segment(bare_model, litellm.model_cost) == "openai/v1"
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_wildcard_zero_cost_request_does_not_poison_named_deployment_pricing():
|
|
"""LIT-3991 end to end: a proxy has a named text-embedding-3-small
|
|
deployment relying on built-in pricing plus an ``openai/*`` wildcard with
|
|
explicit zero pricing. One embedding call routed through the wildcard must
|
|
not clobber the shared ``openai/text-embedding-3-small`` pricing; requests
|
|
to the named deployment afterwards must still cost non-zero.
|
|
"""
|
|
shared_key = "openai/text-embedding-3-small"
|
|
model_keys = {
|
|
shared_key: copy.deepcopy(litellm.model_cost.get(shared_key)),
|
|
"text-embedding-3-small": copy.deepcopy(litellm.model_cost.get("text-embedding-3-small")),
|
|
"openai/*": copy.deepcopy(litellm.model_cost.get("openai/*")),
|
|
"lit3991-named": litellm.model_cost.get("lit3991-named"),
|
|
"lit3991-wildcard": litellm.model_cost.get("lit3991-wildcard"),
|
|
}
|
|
builtin_input_cost = litellm.get_model_info(model=shared_key)["input_cost_per_token"]
|
|
assert builtin_input_cost > 0
|
|
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "text-embedding-3-small",
|
|
"litellm_params": {
|
|
"model": "openai/text-embedding-3-small",
|
|
"api_key": "fake-key-named",
|
|
},
|
|
"model_info": {"id": "lit3991-named"},
|
|
},
|
|
{
|
|
"model_name": "openai/*",
|
|
"litellm_params": {
|
|
"model": "openai/*",
|
|
"api_key": "fake-key-wildcard",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
},
|
|
"model_info": {"id": "lit3991-wildcard"},
|
|
},
|
|
],
|
|
)
|
|
|
|
router.embedding(
|
|
model="openai/text-embedding-3-small",
|
|
input=["hello"],
|
|
mock_response=[0.1, 0.2],
|
|
)
|
|
|
|
assert litellm.get_model_info(model=shared_key)["input_cost_per_token"] == builtin_input_cost, (
|
|
f"one call through the zero-cost wildcard poisoned the shared {shared_key} pricing for the named deployment"
|
|
)
|
|
|
|
named_response = router.embedding(
|
|
model="text-embedding-3-small",
|
|
input=["hello"],
|
|
mock_response=[0.1, 0.2],
|
|
)
|
|
named_cost = litellm.completion_cost(completion_response=named_response, call_type="embedding")
|
|
assert named_cost == pytest.approx(10 * builtin_input_cost)
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_price_data_reload_preserves_router_registered_model_info(monkeypatch):
|
|
"""
|
|
A price-data reload replaces litellm.model_cost wholesale. Deployment
|
|
model_info registered by the Router is not in the fetched catalog, so
|
|
without a replay of runtime registrations the reload silently strips
|
|
max_input_tokens / max_output_tokens from every custom model group and
|
|
/model_group/info starts reporting nulls.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "custom-alias",
|
|
"litellm_params": {"model": "hosted_vllm/not-in-the-catalog"},
|
|
"model_info": {
|
|
"id": "custom-alias-id",
|
|
"max_input_tokens": 128000,
|
|
"max_output_tokens": 16384,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
before = router.get_model_group_info(model_group="custom-alias")
|
|
assert before is not None
|
|
assert before.max_input_tokens == 128000
|
|
assert before.max_output_tokens == 16384
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
_simulate_price_data_reload(
|
|
{"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}},
|
|
)
|
|
|
|
after = router.get_model_group_info(model_group="custom-alias")
|
|
assert after is not None
|
|
assert after.max_input_tokens == 128000
|
|
assert after.max_output_tokens == 16384
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_price_data_reload_preserves_custom_override_of_a_catalog_model(monkeypatch):
|
|
"""
|
|
A deployment whose backend model IS in the catalog is the quieter half of
|
|
the same bug: the reload does not blank the metadata, it reverts the
|
|
operator's model_info override to the upstream catalog values.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "capped-gpt-4o",
|
|
"litellm_params": {"model": "openai/gpt-4o"},
|
|
"model_info": {
|
|
"id": "capped-gpt-4o-id",
|
|
"max_input_tokens": 12345,
|
|
"max_output_tokens": 678,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
_simulate_price_data_reload(
|
|
{
|
|
"openai/gpt-4o": {
|
|
"litellm_provider": "openai",
|
|
"mode": "chat",
|
|
"max_input_tokens": 999999,
|
|
"max_output_tokens": 888888,
|
|
}
|
|
},
|
|
)
|
|
|
|
after = router.get_model_group_info(model_group="capped-gpt-4o")
|
|
assert after is not None
|
|
assert after.max_input_tokens == 12345
|
|
assert after.max_output_tokens == 678
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_deleted_deployments_are_not_replayed_onto_later_reloads(monkeypatch):
|
|
"""
|
|
Runtime registrations are replayed onto every price data reload, so a
|
|
deleted deployment has to be withdrawn or it is re-asserted for the life of
|
|
the process and the registry grows with every create/delete cycle. A backend
|
|
key that another live deployment still points at must survive the same
|
|
deletion.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "doomed",
|
|
"litellm_params": {"model": "hosted_vllm/shared-backend"},
|
|
"model_info": {"id": "doomed-id", "max_input_tokens": 111},
|
|
},
|
|
{
|
|
"model_name": "kept",
|
|
"litellm_params": {"model": "hosted_vllm/shared-backend"},
|
|
"model_info": {"id": "kept-id", "max_input_tokens": 222},
|
|
},
|
|
{
|
|
"model_name": "solo",
|
|
"litellm_params": {"model": "hosted_vllm/solo-backend"},
|
|
"model_info": {"id": "solo-id", "max_input_tokens": 333},
|
|
},
|
|
],
|
|
)
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
assert router.delete_deployment(id="doomed-id") is not None
|
|
assert router.delete_deployment(id="solo-id") is not None
|
|
|
|
_simulate_price_data_reload(
|
|
{"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}},
|
|
)
|
|
|
|
assert "doomed-id" not in litellm.model_cost
|
|
assert "solo-id" not in litellm.model_cost
|
|
assert "hosted_vllm/solo-backend" not in litellm.model_cost
|
|
|
|
surviving = litellm.model_cost["kept-id"]
|
|
assert surviving["max_input_tokens"] == 222
|
|
assert "hosted_vllm/shared-backend" in litellm.model_cost
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_deleting_a_deployment_leaves_catalog_pricing_for_its_backend_model(monkeypatch):
|
|
"""
|
|
A backend key is shared with the fetched catalog, so withdrawing the entries
|
|
a deleted deployment owns must not take real upstream pricing down with it.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
backend_model = "gemini/gemini-2.5-pro"
|
|
catalog_entry = litellm.get_model_info(model=backend_model)
|
|
catalog_input_cost = catalog_entry["input_cost_per_token"]
|
|
assert catalog_input_cost > 0, "Test requires a catalog model with non-zero pricing"
|
|
|
|
saved_catalog = litellm.model_cost
|
|
fetched_catalog = copy.deepcopy(litellm.model_cost)
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "doomed-gemini",
|
|
"litellm_params": {"model": backend_model, "api_key": "sk-fake"},
|
|
"model_info": {"id": "doomed-gemini-id"},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert router.delete_deployment(id="doomed-gemini-id") is not None
|
|
|
|
_simulate_price_data_reload(
|
|
copy.deepcopy(fetched_catalog),
|
|
)
|
|
|
|
assert "doomed-gemini-id" not in litellm.model_cost
|
|
assert litellm.model_cost[backend_model]["input_cost_per_token"] == catalog_input_cost
|
|
finally:
|
|
litellm.model_cost = saved_catalog
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_repointing_a_deployment_drops_its_previous_backend_key(monkeypatch):
|
|
"""
|
|
An update that moves a deployment onto a different backend model leaves the
|
|
old backend key behind, and a replayed registry would re-assert it onto every
|
|
later catalog for the life of the process.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "moving-target",
|
|
"litellm_params": {"model": "hosted_vllm/old-backend"},
|
|
"model_info": {"id": "moving-target-id"},
|
|
}
|
|
],
|
|
)
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
router.upsert_deployment(
|
|
deployment=Deployment(
|
|
model_name="moving-target",
|
|
litellm_params=LiteLLM_Params(model="hosted_vllm/new-backend"),
|
|
model_info=ModelInfo(id="moving-target-id"),
|
|
)
|
|
)
|
|
|
|
_simulate_price_data_reload(
|
|
{"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}},
|
|
)
|
|
|
|
assert "hosted_vllm/old-backend" not in litellm.model_cost
|
|
assert "hosted_vllm/new-backend" in litellm.model_cost
|
|
assert "moving-target-id" in litellm.model_cost
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"model, custom_llm_provider, expected",
|
|
[
|
|
("gpt-4o", None, ("gpt-4o",)),
|
|
("gpt-4o", "openai", ("openai/gpt-4o",)),
|
|
("openai/gpt-4o", None, ("openai/gpt-4o",)),
|
|
("responses/gpt-4o", "openai", ("openai/responses/gpt-4o", "openai/gpt-4o")),
|
|
("responses/gpt-4o", None, ("responses/gpt-4o", "gpt-4o")),
|
|
],
|
|
)
|
|
def test_backend_cost_map_keys_matches_what_registration_writes(model, custom_llm_provider, expected):
|
|
"""
|
|
The withdrawal path drops exactly the keys the registration wrote, so the two
|
|
have to agree on the provider prefix and on the responses/ alias. The first
|
|
key is also the one the registration uses as the shared backend key, so its
|
|
position is load-bearing rather than incidental.
|
|
"""
|
|
keys = Router._backend_cost_map_keys(model=model, custom_llm_provider=custom_llm_provider)
|
|
assert keys == expected
|
|
assert keys[0] == (model if custom_llm_provider is None else f"{custom_llm_provider}/{model}")
|
|
|
|
|
|
def test_a_discarded_router_stops_contributing_to_later_reloads(monkeypatch):
|
|
"""
|
|
`_route_user_config_request` builds a Router per request from caller-supplied
|
|
config and discards it. Nothing can withdraw entries on its behalf afterwards,
|
|
so a rebuild driven off live routers is what keeps a caller from growing the
|
|
cost map one request at a time.
|
|
"""
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
kept = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "kept",
|
|
"litellm_params": {"model": "hosted_vllm/kept-backend"},
|
|
"model_info": {"id": "kept-router-id", "max_input_tokens": 4242},
|
|
}
|
|
],
|
|
)
|
|
throwaway = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "throwaway",
|
|
"litellm_params": {"model": "hosted_vllm/throwaway-backend"},
|
|
"model_info": {"id": "throwaway-router-id", "max_input_tokens": 111},
|
|
}
|
|
],
|
|
)
|
|
throwaway.discard()
|
|
|
|
_simulate_price_data_reload(
|
|
{"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}},
|
|
)
|
|
|
|
assert "throwaway-router-id" not in litellm.model_cost
|
|
assert "hosted_vllm/throwaway-backend" not in litellm.model_cost
|
|
assert litellm.model_cost["kept-router-id"]["max_input_tokens"] == 4242
|
|
assert "hosted_vllm/kept-backend" in litellm.model_cost
|
|
assert kept.model_list # keep the live router referenced for the duration
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_a_reload_rebuilds_exactly_what_a_fresh_boot_registered():
|
|
"""
|
|
The rebuild is only correct if it reproduces the entries the original
|
|
registration wrote, including the pieces that are derived rather than stored:
|
|
custom pricing carried on litellm_params, and the cache pricing inherited from
|
|
the built-in cost map.
|
|
"""
|
|
saved_catalog = litellm.model_cost
|
|
fetched_catalog = copy.deepcopy(litellm.model_cost)
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "priced",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o",
|
|
"api_key": "sk-fake",
|
|
"input_cost_per_token": 0.000123,
|
|
"output_cost_per_token": 0.000456,
|
|
},
|
|
"model_info": {"id": "priced-id", "max_input_tokens": 4242},
|
|
}
|
|
],
|
|
)
|
|
at_boot = copy.deepcopy(litellm.model_cost["priced-id"])
|
|
assert at_boot["input_cost_per_token"] == 0.000123
|
|
assert at_boot["cache_read_input_token_cost"] is not None
|
|
|
|
_simulate_price_data_reload(
|
|
copy.deepcopy(fetched_catalog),
|
|
)
|
|
|
|
rebuilt = litellm.model_cost["priced-id"]
|
|
assert at_boot.items() <= rebuilt.items(), (
|
|
f"the rebuild changed or dropped a field the boot registration wrote: "
|
|
f"{ {k: (v, rebuilt.get(k)) for k, v in at_boot.items() if rebuilt.get(k) != v} }"
|
|
)
|
|
# The rebuild goes through the deployment stored in model_list, which also
|
|
# carries the router's own db_model flag; add_deployment already registers it.
|
|
assert set(rebuilt) - set(at_boot) <= {"db_model"}
|
|
assert router.model_list
|
|
finally:
|
|
litellm.model_cost = saved_catalog
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_replay_model_cost_registrations_survives_a_malformed_deployment():
|
|
"""
|
|
The rebuild reads whatever dicts are sitting in model_list, so one entry that
|
|
cannot be rebuilt into a Deployment must not stop the rest being restored.
|
|
"""
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "healthy",
|
|
"litellm_params": {"model": "hosted_vllm/healthy-backend"},
|
|
"model_info": {"id": "healthy-id", "max_input_tokens": 777},
|
|
}
|
|
],
|
|
)
|
|
router.model_list.insert(0, {"litellm_params": {}})
|
|
|
|
litellm.model_cost = {"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}}
|
|
_invalidate_model_cost_lowercase_map()
|
|
router._replay_model_cost_registrations()
|
|
|
|
assert litellm.model_cost["healthy-id"]["max_input_tokens"] == 777
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_deployment_model_cost_payload_folds_in_litellm_params_pricing():
|
|
"""
|
|
Custom pricing is configured on litellm_params but has to land in the
|
|
cost-map entry, and setting it pulls in the built-in cache pricing for the
|
|
backend model. Both are what make the entry reproducible from a deployment.
|
|
"""
|
|
payload = Router._deployment_model_cost_payload(
|
|
deployment=Deployment(
|
|
model_name="priced",
|
|
litellm_params=LiteLLM_Params(
|
|
model="gemini/gemini-2.5-pro",
|
|
input_cost_per_token=0.000123,
|
|
),
|
|
model_info=ModelInfo(id="payload-id", max_input_tokens=4242),
|
|
)
|
|
)
|
|
|
|
assert payload["id"] == "payload-id"
|
|
assert payload["max_input_tokens"] == 4242
|
|
assert payload["input_cost_per_token"] == 0.000123
|
|
assert payload["cache_read_input_token_cost"] > 0
|
|
|
|
|
|
def test_register_deployment_in_model_cost_writes_both_key_families():
|
|
"""
|
|
A deployment contributes its full model_info under its unique id and the
|
|
cost-map subset under the shared backend key, and the shared key must not
|
|
pick up the deployment's private metadata.
|
|
"""
|
|
model_keys = {
|
|
"both-families-id": copy.deepcopy(litellm.model_cost.get("both-families-id")),
|
|
"hosted_vllm/both-families-backend": copy.deepcopy(litellm.model_cost.get("hosted_vllm/both-families-backend")),
|
|
}
|
|
try:
|
|
Router._register_deployment_in_model_cost(
|
|
model_id="both-families-id",
|
|
model_info={"id": "both-families-id", "max_input_tokens": 999, "litellm_provider": "hosted_vllm"},
|
|
model="hosted_vllm/both-families-backend",
|
|
custom_llm_provider=None,
|
|
)
|
|
|
|
assert litellm.model_cost["both-families-id"]["max_input_tokens"] == 999
|
|
shared = litellm.model_cost["hosted_vllm/both-families-backend"]
|
|
assert shared["max_input_tokens"] == 999
|
|
assert "id" not in shared
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
|
|
|
|
def test_reload_keeps_custom_pricing_configured_on_litellm_params_for_a_db_model():
|
|
"""
|
|
A deployment added at runtime, which is what /model/new does, configures its
|
|
custom pricing on litellm_params rather than on model_info. A price data
|
|
reload must not revert that to the catalog's pricing.
|
|
"""
|
|
saved_catalog = litellm.model_cost
|
|
fetched_catalog = copy.deepcopy(litellm.model_cost)
|
|
try:
|
|
router = Router(model_list=[])
|
|
router.add_deployment(
|
|
deployment=Deployment(
|
|
model_name="db-priced",
|
|
litellm_params=LiteLLM_Params(
|
|
model="openai/gpt-4o",
|
|
api_key="sk-fake",
|
|
input_cost_per_token=0.000123,
|
|
output_cost_per_token=0.000456,
|
|
),
|
|
model_info=ModelInfo(id="db-priced-id"),
|
|
)
|
|
)
|
|
|
|
assert litellm.model_cost["db-priced-id"]["input_cost_per_token"] == 0.000123
|
|
|
|
_simulate_price_data_reload(
|
|
copy.deepcopy(fetched_catalog),
|
|
)
|
|
|
|
assert litellm.model_cost["db-priced-id"]["input_cost_per_token"] == 0.000123
|
|
assert litellm.model_cost["db-priced-id"]["output_cost_per_token"] == 0.000456
|
|
finally:
|
|
litellm.model_cost = saved_catalog
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_replay_live_router_model_cost_rebuilds_every_live_router():
|
|
"""
|
|
A process can hold more than one Router, so the rebuild has to fan out across
|
|
all of them rather than restoring whichever one happens to be reachable.
|
|
"""
|
|
from litellm.router import _replay_live_router_model_cost
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
first = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "first",
|
|
"litellm_params": {"model": "hosted_vllm/first-backend"},
|
|
"model_info": {"id": "first-id", "max_input_tokens": 111},
|
|
}
|
|
],
|
|
)
|
|
second = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "second",
|
|
"litellm_params": {"model": "hosted_vllm/second-backend"},
|
|
"model_info": {"id": "second-id", "max_input_tokens": 222},
|
|
}
|
|
],
|
|
)
|
|
|
|
litellm.model_cost = {"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}}
|
|
_invalidate_model_cost_lowercase_map()
|
|
_replay_live_router_model_cost()
|
|
|
|
assert litellm.model_cost["first-id"]["max_input_tokens"] == 111
|
|
assert litellm.model_cost["second-id"]["max_input_tokens"] == 222
|
|
assert first.model_list and second.model_list
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_strategy_router_alias_pricing_never_enters_model_cost(monkeypatch):
|
|
"""
|
|
A strategy-router alias is never the deployment actually called or billed,
|
|
so custom pricing configured on it must not be registered under its
|
|
model_id - an explicit zero there makes the budget check treat the alias
|
|
as a genuinely free model while requests bill as a real deployment. The
|
|
strip must also survive a price-data reload, which rebuilds entries by
|
|
walking the live routers.
|
|
"""
|
|
from litellm import utils as litellm_utils
|
|
|
|
monkeypatch.setattr(
|
|
litellm_utils,
|
|
"_runtime_registered_model_cost",
|
|
dict(litellm_utils._runtime_registered_model_cost),
|
|
)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "smart-router",
|
|
"litellm_params": {
|
|
"model": "auto_router/complexity_router/smart-router",
|
|
"complexity_router_default_model": "paid-model",
|
|
"input_cost_per_token": 0.0,
|
|
"output_cost_per_token": 0.0,
|
|
"complexity_router_config": {"tiers": {"simple": "paid-model"}},
|
|
},
|
|
"model_info": {"id": "strategy-alias-id", "max_input_tokens": 128000},
|
|
},
|
|
{
|
|
"model_name": "paid-model",
|
|
"litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-fake"},
|
|
"model_info": {"id": "strategy-alias-paid-id"},
|
|
},
|
|
],
|
|
)
|
|
|
|
def _assert_alias_unpriced():
|
|
entry = litellm.model_cost.get("strategy-alias-id")
|
|
assert entry is not None, "Alias metadata should still be registered"
|
|
assert entry["max_input_tokens"] == 128000
|
|
assert "input_cost_per_token" not in entry
|
|
assert "output_cost_per_token" not in entry
|
|
|
|
_assert_alias_unpriced()
|
|
|
|
saved_model_cost = litellm.model_cost
|
|
try:
|
|
_simulate_price_data_reload(
|
|
{"gpt-4o": {"litellm_provider": "openai", "mode": "chat"}},
|
|
)
|
|
_assert_alias_unpriced()
|
|
assert router.model_list
|
|
finally:
|
|
litellm.model_cost = saved_model_cost
|
|
_invalidate_model_cost_lowercase_map()
|
|
|
|
|
|
def test_inherit_builtin_tiered_output_rate_fills_the_backend_flat_rate():
|
|
"""
|
|
A deployment entry whose custom tiers publish only input rates would bill
|
|
completions at 0, so the backend model's flat output rate is copied in at
|
|
registration.
|
|
"""
|
|
model_info = {"tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}]}
|
|
|
|
Router._inherit_builtin_tiered_output_rate(
|
|
model_info=model_info,
|
|
backend_model="claude-haiku-4-5",
|
|
custom_llm_provider="anthropic",
|
|
)
|
|
|
|
backend_rate = litellm.get_model_info(model="claude-haiku-4-5", custom_llm_provider="anthropic")[
|
|
"output_cost_per_token"
|
|
]
|
|
assert backend_rate > 0
|
|
assert model_info["output_cost_per_token"] == backend_rate
|
|
|
|
|
|
def test_inherit_builtin_tiered_output_rate_never_stores_a_synthesized_zero():
|
|
"""
|
|
Regression: get_model_info reports output_cost_per_token 0 for a backend that
|
|
only publishes tiered rates (e.g. dashscope/qwen-flash), and storing that zero
|
|
would mark the deployment as explicitly priced free.
|
|
"""
|
|
backend_info = litellm.get_model_info(model="qwen-flash", custom_llm_provider="dashscope")
|
|
assert backend_info["output_cost_per_token"] == 0
|
|
|
|
model_info = {"tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}]}
|
|
Router._inherit_builtin_tiered_output_rate(
|
|
model_info=model_info,
|
|
backend_model="qwen-flash",
|
|
custom_llm_provider="dashscope",
|
|
)
|
|
|
|
assert "output_cost_per_token" not in model_info
|
|
|
|
|
|
def test_inherit_builtin_tiered_output_rate_leaves_a_user_rate_alone():
|
|
model_info = {
|
|
"tiered_pricing": [{"range": [0, 3000], "input_cost_per_token": 3.25e-07}],
|
|
"output_cost_per_token": 9e-07,
|
|
}
|
|
|
|
Router._inherit_builtin_tiered_output_rate(
|
|
model_info=model_info,
|
|
backend_model="claude-haiku-4-5",
|
|
custom_llm_provider="anthropic",
|
|
)
|
|
|
|
assert model_info["output_cost_per_token"] == 9e-07
|
|
|
|
|
|
# --- a config.yaml PTU deployment must not also bill per token ------------------
|
|
|
|
_PTU_MODEL_INFO = {
|
|
"id": "ptu-alpha-eastus",
|
|
"team_id": "team-alpha",
|
|
"ptu_count": 100,
|
|
"cost_per_ptu_per_hour": 0.02,
|
|
"ptu_effective_from": "2026-01-01T00:00:00Z",
|
|
}
|
|
|
|
|
|
def _ptu_router(model_info=None, litellm_params=None, ptu_enabled=True):
|
|
"""A router built the way loading config.yaml builds one."""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True" if ptu_enabled else ""}, clear=False):
|
|
return Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-ptu",
|
|
"litellm_params": {
|
|
"model": "anthropic/claude-sonnet-4-5-20250929",
|
|
"api_key": "sk-not-used",
|
|
**(litellm_params or {}),
|
|
},
|
|
"model_info": dict(_PTU_MODEL_INFO if model_info is None else model_info),
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def test_a_config_ptu_deployment_bills_nothing_per_token():
|
|
"""Reserved capacity is already billed by the hour, so charging its traffic bills the
|
|
same tokens twice. Left unset the rate falls back to the public cost map, which makes
|
|
the double charge the default rather than an opt-in."""
|
|
router = _ptu_router(litellm_params={"input_cost_per_token": 5e-06, "output_cost_per_token": 1.5e-05})
|
|
entry = router.model_list[0]
|
|
|
|
assert entry["litellm_params"]["input_cost_per_token"] == 0.0
|
|
assert entry["litellm_params"]["output_cost_per_token"] == 0.0
|
|
assert entry["model_info"]["input_cost_per_token"] == 0.0
|
|
assert litellm.model_cost[entry["model_info"]["id"]]["input_cost_per_token"] == 0.0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"backend",
|
|
["anthropic/claude-sonnet-4-5-20250929", "azure/gpt-4o", "gemini/gemini-2.5-flash"],
|
|
)
|
|
def test_a_config_ptu_deployment_imports_no_cache_rate_from_its_backend(backend):
|
|
"""The cache back-fill runs whenever input_cost_per_token is set, and 0.0 is set, so a
|
|
partially zeroed deployment would silently inherit the backend model's real cache rates.
|
|
Every backend here publishes non-zero ones, which is what makes the assertion mean
|
|
something."""
|
|
cache_fields = (
|
|
"cache_creation_input_token_cost",
|
|
"cache_creation_input_token_cost_above_1hr",
|
|
"cache_creation_input_token_cost_above_200k_tokens",
|
|
"cache_read_input_token_cost",
|
|
"cache_read_input_token_cost_above_200k_tokens",
|
|
)
|
|
builtin = litellm.get_model_info(model=backend)
|
|
assert any(builtin.get(field) for field in cache_fields), "backend publishes no cache pricing to leak"
|
|
|
|
router = _ptu_router(litellm_params={"model": backend})
|
|
priced = litellm.model_cost[router.model_list[0]["model_info"]["id"]]
|
|
|
|
assert [field for field in cache_fields if priced.get(field)] == []
|
|
|
|
|
|
def test_zeroing_a_ptu_deployment_leaves_its_backend_model_priced():
|
|
"""A sibling deployment on the same backend must keep billing normally."""
|
|
backend = "anthropic/claude-sonnet-4-5-20250929"
|
|
builtin = litellm.get_model_info(model=backend)["input_cost_per_token"]
|
|
assert builtin > 0
|
|
|
|
_ptu_router(litellm_params={"model": backend})
|
|
|
|
assert litellm.get_model_info(model=backend)["input_cost_per_token"] == builtin
|
|
|
|
|
|
def test_the_registered_id_is_the_one_the_operator_declared():
|
|
"""Registration must key the deployment by the declared id, not by a hash of params that
|
|
zeroing has just rewritten. The id keys cooldowns, budgets and every spend row already
|
|
written, so minting one here would move all of them.
|
|
|
|
A derived id is no longer reachable for a reservation: zeroing requires PTU terms and
|
|
PTU terms now require a declared id, so the two never combine."""
|
|
params = {"input_cost_per_token": 5e-06}
|
|
priced = _ptu_router(litellm_params=params, ptu_enabled=False).model_list[0]["model_info"]["id"]
|
|
zeroed = _ptu_router(litellm_params=params).model_list[0]["model_info"]["id"]
|
|
|
|
assert priced == zeroed == "ptu-alpha-eastus"
|
|
|
|
|
|
def test_a_database_backed_deployment_is_left_alone():
|
|
"""The write endpoints already zero those, and they answer 400 rather than silently
|
|
rewriting a rate the caller sent."""
|
|
entry = _ptu_router(model_info={**_PTU_MODEL_INFO, "db_model": True}).model_list[0]
|
|
|
|
assert entry["litellm_params"].get("input_cost_per_token") is None
|
|
|
|
|
|
def test_nothing_is_zeroed_while_the_feature_is_off():
|
|
"""No flat cost accrues with the flag off, so zeroing would serve the traffic free."""
|
|
entry = _ptu_router(litellm_params={"input_cost_per_token": 5e-06}, ptu_enabled=False).model_list[0]
|
|
|
|
assert entry["litellm_params"]["input_cost_per_token"] == 5e-06
|
|
|
|
|
|
@pytest.mark.parametrize("dropped", ["team_id", "ptu_effective_from"], ids=["no team_id", "no ptu_effective_from"])
|
|
def test_an_incomplete_reservation_is_refused_rather_than_served(dropped):
|
|
"""POST /model/new answers 400 for exactly this config, so config.yaml must not quietly
|
|
accept it. Serving it would bill per token while accruing no flat cost, which is the
|
|
state the operator was trying to leave."""
|
|
incomplete = {k: v for k, v in _PTU_MODEL_INFO.items() if k != dropped}
|
|
|
|
with pytest.raises(ValueError, match="PTU configuration on model 'gpt") as raised:
|
|
_ptu_router(model_info=incomplete, litellm_params={"input_cost_per_token": 5e-06})
|
|
|
|
assert "gpt-4o-ptu" in str(raised.value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"dropped, expected",
|
|
[
|
|
("team_id", "team_id is required when PTU fields are set (one model maps to one team)"),
|
|
("cost_per_ptu_per_hour", "ptu_count and cost_per_ptu_per_hour must be set together"),
|
|
],
|
|
ids=["no team_id", "count without rate"],
|
|
)
|
|
def test_the_refusal_reason_is_the_one_the_model_endpoint_answers_with(dropped, expected):
|
|
"""One rule, stated once. If these drift, an operator gets contradictory guidance
|
|
depending on which path they used."""
|
|
incomplete = {k: v for k, v in _PTU_MODEL_INFO.items() if k != dropped}
|
|
|
|
assert ptu_config_error(incomplete) == expected
|
|
with pytest.raises(ValueError, match="PTU configuration on model 'gpt") as raised:
|
|
_ptu_router(model_info=incomplete)
|
|
|
|
assert expected in str(raised.value)
|
|
|
|
|
|
@pytest.mark.parametrize("dropped", ["team_id", "ptu_effective_from"], ids=["no team_id", "no ptu_effective_from"])
|
|
def test_an_incomplete_reservation_is_left_alone_while_the_feature_is_off(dropped):
|
|
"""Nothing accrues with the flag off, so refusing a deployment there would take a
|
|
serving model away from an operator who never opted in."""
|
|
incomplete = {k: v for k, v in _PTU_MODEL_INFO.items() if k != dropped}
|
|
entry = _ptu_router(
|
|
model_info=incomplete, litellm_params={"input_cost_per_token": 5e-06}, ptu_enabled=False
|
|
).model_list[0]
|
|
|
|
assert entry["litellm_params"]["input_cost_per_token"] == 5e-06
|
|
|
|
|
|
@pytest.mark.parametrize("dropped", ["team_id", "ptu_effective_from"], ids=["no team_id", "no ptu_effective_from"])
|
|
def test_the_proxy_drops_the_deployment_rather_than_failing_to_boot(dropped):
|
|
"""The proxy builds its router with ignore_invalid_deployments, so one bad entry must
|
|
cost that entry and not the whole config."""
|
|
incomplete = {k: v for k, v in _PTU_MODEL_INFO.items() if k != dropped}
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True"}, clear=False):
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-ptu",
|
|
"litellm_params": {"model": "anthropic/claude-sonnet-4-5-20250929", "api_key": "sk-not-used"},
|
|
"model_info": dict(incomplete),
|
|
},
|
|
{
|
|
"model_name": "plain-sibling",
|
|
"litellm_params": {"model": "anthropic/claude-sonnet-4-5-20250929", "api_key": "sk-not-used"},
|
|
},
|
|
],
|
|
ignore_invalid_deployments=True,
|
|
)
|
|
|
|
assert [entry["model_name"] for entry in router.model_list] == ["plain-sibling"]
|
|
|
|
|
|
def test_a_complete_reservation_still_registers():
|
|
"""The refusal must be scoped to a broken reservation, not to PTU configuration."""
|
|
entry = _ptu_router().model_list[0]
|
|
|
|
assert entry["model_name"] == "gpt-4o-ptu"
|
|
assert entry["litellm_params"]["input_cost_per_token"] == 0.0
|
|
|
|
|
|
def test_nested_custom_model_info_does_not_pollute_shared_backend():
|
|
backend_model = "gpt-4o-search-preview"
|
|
custom_id = "lit5471-search-custom"
|
|
sibling_id = "lit5471-search-sibling"
|
|
builtin_info = copy.deepcopy(litellm.get_model_info(model=backend_model))
|
|
expected_nested = copy.deepcopy(builtin_info["search_context_cost_per_query"])
|
|
model_keys = {
|
|
backend_model: copy.deepcopy(litellm.model_cost.get(backend_model)),
|
|
custom_id: copy.deepcopy(litellm.model_cost.get(custom_id)),
|
|
sibling_id: copy.deepcopy(litellm.model_cost.get(sibling_id)),
|
|
}
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "search-custom",
|
|
"litellm_params": {"model": backend_model, "api_key": "fake-key"},
|
|
"model_info": {
|
|
"id": custom_id,
|
|
"search_context_cost_per_query": {
|
|
"search_context_size_low": 0.123,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"model_name": "search-sibling",
|
|
"litellm_params": {"model": backend_model, "api_key": "fake-key"},
|
|
"model_info": {"id": sibling_id},
|
|
},
|
|
],
|
|
)
|
|
|
|
custom_info = router.get_deployment_model_info(model_id=custom_id, model_name=backend_model)
|
|
sibling_info = router.get_deployment_model_info(model_id=sibling_id, model_name=backend_model)
|
|
|
|
assert custom_info is not None
|
|
assert custom_info["search_context_cost_per_query"]["search_context_size_low"] == 0.123
|
|
assert litellm.model_cost[backend_model]["search_context_cost_per_query"] == expected_nested
|
|
assert sibling_info is not None
|
|
assert sibling_info["search_context_cost_per_query"] == expected_nested
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
litellm.get_model_info.cache_clear()
|
|
|
|
|
|
def test_base_model_custom_info_does_not_pollute_cached_base_model():
|
|
base_model = "azure/gpt-4o"
|
|
deployment_id = "lit5471-base-model"
|
|
base_model_info = copy.deepcopy(litellm.get_model_info(model=base_model))
|
|
model_keys = {
|
|
"azure/gpt-4o": copy.deepcopy(litellm.model_cost.get("azure/gpt-4o")),
|
|
deployment_id: copy.deepcopy(litellm.model_cost.get(deployment_id)),
|
|
}
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-custom",
|
|
"litellm_params": {
|
|
"model": "gpt-4o",
|
|
"custom_llm_provider": "azure",
|
|
"api_key": "fake-key",
|
|
},
|
|
"model_info": {
|
|
"id": deployment_id,
|
|
"base_model": base_model,
|
|
"input_cost_per_token": 0.777,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
info = router.get_deployment_model_info(model_id=deployment_id, model_name=base_model)
|
|
|
|
assert info is not None
|
|
assert info["input_cost_per_token"] == 0.777
|
|
assert litellm.get_model_info(model=base_model) == base_model_info
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
litellm.get_model_info.cache_clear()
|
|
|
|
|
|
def test_builtin_only_deployment_info_is_not_the_cached_object():
|
|
backend_model = "gpt-4o-search-preview"
|
|
deployment_id = "lit5471-builtin-only"
|
|
litellm.get_model_info.cache_clear()
|
|
model_keys = {deployment_id: copy.deepcopy(litellm.model_cost.get(deployment_id))}
|
|
try:
|
|
cached_info = litellm.get_model_info(model=backend_model)
|
|
assert cached_info["search_context_cost_per_query"]
|
|
|
|
info = Router(model_list=[]).get_deployment_model_info(model_id=deployment_id, model_name=backend_model)
|
|
|
|
assert info is not None
|
|
assert info["search_context_cost_per_query"] == cached_info["search_context_cost_per_query"]
|
|
assert _nested_container_ids(info).isdisjoint(_nested_container_ids(cached_info))
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
litellm.get_model_info.cache_clear()
|
|
|
|
|
|
def test_custom_only_deployment_info_is_not_the_registry_entry():
|
|
unknown_backend = "openai/lit5471-unknown-backend"
|
|
deployment_id = "lit5471-custom-only"
|
|
nested_pricing = {"search_context_size_low": 0.123}
|
|
model_keys = {
|
|
unknown_backend: copy.deepcopy(litellm.model_cost.get(unknown_backend)),
|
|
deployment_id: copy.deepcopy(litellm.model_cost.get(deployment_id)),
|
|
}
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "custom-only",
|
|
"litellm_params": {"model": unknown_backend, "api_key": "fake-key"},
|
|
"model_info": {"id": deployment_id, "search_context_cost_per_query": dict(nested_pricing)},
|
|
}
|
|
],
|
|
)
|
|
registry_entry = litellm.model_cost[deployment_id]
|
|
|
|
info = router.get_deployment_model_info(model_id=deployment_id, model_name=unknown_backend)
|
|
|
|
assert info is not None
|
|
assert info["search_context_cost_per_query"] == nested_pricing
|
|
assert _nested_container_ids(info).isdisjoint(_nested_container_ids(registry_entry))
|
|
finally:
|
|
_restore_model_cost_entries(model_keys)
|
|
litellm.get_model_info.cache_clear()
|
|
|
|
|
|
def test_router_model_info_deep_copies_nested_cached_metadata():
|
|
model = "openai/gpt-4o-search-preview"
|
|
litellm.get_model_info.cache_clear()
|
|
try:
|
|
cached_info = litellm.get_model_info(model=model)
|
|
assert cached_info is not None
|
|
expected_nested = copy.deepcopy(cached_info["search_context_cost_per_query"])
|
|
assert expected_nested
|
|
|
|
router = Router(model_list=[])
|
|
merged_info = router.get_router_model_info(
|
|
deployment={
|
|
"model_name": "search",
|
|
"litellm_params": {"model": "gpt-4o-search-preview"},
|
|
"model_info": {"id": "lit5471-router-model-info"},
|
|
},
|
|
received_model_name="search",
|
|
)
|
|
|
|
assert merged_info["search_context_cost_per_query"] == expected_nested
|
|
assert _nested_container_ids(merged_info).isdisjoint(_nested_container_ids(cached_info))
|
|
assert litellm.get_model_info(model=model)["search_context_cost_per_query"] == expected_nested
|
|
finally:
|
|
litellm.get_model_info.cache_clear()
|
|
|
|
|
|
# --- a config.yaml reservation must carry an id its operator owns --------------------
|
|
|
|
|
|
def test_a_reservation_without_a_declared_id_is_refused():
|
|
"""Left underived the id is a hash of the resolved litellm_params, so rotating the
|
|
credential mints a second identity and the catch-up bills the window again under it.
|
|
The flat cost is keyed by that id and a written charge is never retracted, so the
|
|
duplicate is permanent."""
|
|
anonymous = {k: v for k, v in _PTU_MODEL_INFO.items() if k != "id"}
|
|
|
|
with pytest.raises(ValueError, match=re.escape("model_info.id is required")):
|
|
_ptu_router(model_info=anonymous)
|
|
|
|
|
|
def test_the_id_rule_does_not_reach_a_deployment_without_ptu_config():
|
|
"""An ordinary deployment keeps deriving its id, which is most of every config.yaml."""
|
|
entry = _ptu_router(model_info={"team_id": "team-alpha"}).model_list[0]
|
|
|
|
assert entry["model_info"]["id"]
|
|
|
|
|
|
def test_a_reservation_is_left_alone_while_the_feature_is_off():
|
|
anonymous = {k: v for k, v in _PTU_MODEL_INFO.items() if k != "id"}
|
|
entry = _ptu_router(model_info=anonymous, ptu_enabled=False).model_list[0]
|
|
|
|
assert entry["model_info"]["id"]
|
|
|
|
|
|
def test_two_reservations_cannot_share_one_id():
|
|
"""Both would key the same sentinel row, so the second upsert overwrites the first and
|
|
one reservation is billed at the other's rate."""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True"}, clear=False):
|
|
with pytest.raises(ValueError, match="declared on more than one deployment"):
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-ptu",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://e.azure.com"},
|
|
"model_info": dict(_PTU_MODEL_INFO),
|
|
},
|
|
{
|
|
"model_name": "azure-ptu-west",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://w.azure.com"},
|
|
"model_info": dict(_PTU_MODEL_INFO),
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def test_two_reservations_with_distinct_ids_both_register():
|
|
"""The refusal must be scoped to a collision, not to a team running two regions."""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True"}, clear=False):
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-ptu",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://e.azure.com"},
|
|
"model_info": dict(_PTU_MODEL_INFO),
|
|
},
|
|
{
|
|
"model_name": "azure-ptu-west",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://w.azure.com"},
|
|
"model_info": {**_PTU_MODEL_INFO, "id": "ptu-alpha-westus"},
|
|
},
|
|
]
|
|
)
|
|
|
|
assert sorted(m["model_info"]["id"] for m in router.model_list) == ["ptu-alpha-eastus", "ptu-alpha-westus"]
|
|
|
|
|
|
@pytest.mark.parametrize("declared", ["dup-id", 12345], ids=["string id", "numeric id"])
|
|
def test_a_duplicate_id_is_caught_whatever_yaml_parsed_it_as(declared):
|
|
"""An unquoted id in config.yaml arrives as an int, and ModelInfo stores it as a string,
|
|
so both deployments would still key one flat-cost row."""
|
|
|
|
def entry(name, region):
|
|
return {
|
|
"model_name": name,
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": f"https://{region}.azure.com"},
|
|
"model_info": {**_PTU_MODEL_INFO, "id": declared},
|
|
}
|
|
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True"}, clear=False):
|
|
with pytest.raises(ValueError, match="declared on more than one deployment"):
|
|
Router(model_list=[entry("a", "eastus"), entry("b", "westus")])
|
|
|
|
|
|
def test_a_bare_yaml_date_bound_does_not_escape_the_id_rule():
|
|
"""`ptu_effective_to: 2027-01-01` unquoted loads as a date. While that failed to parse,
|
|
the reservation was invisible to PTU entirely: no id rule, no zeroing, no flat cost."""
|
|
import datetime as _dt
|
|
|
|
windowed = {k: v for k, v in _PTU_MODEL_INFO.items() if k != "id"}
|
|
|
|
with pytest.raises(ValueError, match=re.escape("model_info.id is required")):
|
|
_ptu_router(model_info={**windowed, "ptu_effective_to": _dt.date(2027, 1, 1)})
|
|
|
|
|
|
def test_a_reservation_declaring_id_zero_registers():
|
|
"""0 is stable and unique, so reading it as absent refused a correct config."""
|
|
entry = _ptu_router(model_info={**_PTU_MODEL_INFO, "id": 0}).model_list[0]
|
|
|
|
assert entry["model_info"]["id"] == "0"
|
|
|
|
|
|
def test_a_falsy_id_is_still_scanned_for_collisions():
|
|
"""The duplicate scan skipped falsy ids, so a reservation on '0' could share its key with
|
|
an ordinary deployment and the id index would keep only the last one registered."""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": "True"}, clear=False):
|
|
with pytest.raises(ValueError, match="declared on more than one deployment"):
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-ptu",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://e.azure.com"},
|
|
"model_info": {**_PTU_MODEL_INFO, "id": "0"},
|
|
},
|
|
{
|
|
"model_name": "plain-sibling",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://w.azure.com"},
|
|
"model_info": {"id": 0},
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
# --- a reservation declared while the feature is off says so ------------------------
|
|
|
|
|
|
def _ptu_warnings(caplog):
|
|
return tuple(
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if record.name == "LiteLLM Router" and record.levelno == logging.WARNING and "PTU" in record.getMessage()
|
|
)
|
|
|
|
|
|
def test_a_reservation_declared_while_the_feature_is_off_is_warned_about(caplog):
|
|
"""The deployment serves and bills per token, so without this the operator believes they
|
|
reserved capacity and sees no signal anywhere that nothing accrues."""
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
_ptu_router(ptu_enabled=False)
|
|
|
|
warnings = _ptu_warnings(caplog)
|
|
|
|
assert len(warnings) == 1
|
|
assert "gpt-4o-ptu" in warnings[0]
|
|
assert "LITELLM_ENABLE_PTU_COST_ATTRIBUTION" in warnings[0]
|
|
|
|
|
|
def test_a_reservation_is_not_warned_about_while_the_feature_is_on(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
_ptu_router()
|
|
|
|
assert _ptu_warnings(caplog) == ()
|
|
|
|
|
|
def test_a_deployment_carrying_no_ptu_field_is_not_warned_about(caplog):
|
|
"""Most of every config.yaml, so warning here would fire on proxies that never asked."""
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
_ptu_router(model_info={"team_id": "team-alpha"}, ptu_enabled=False)
|
|
|
|
assert _ptu_warnings(caplog) == ()
|
|
|
|
|
|
def test_a_half_written_reservation_is_warned_about(caplog):
|
|
"""A count with no rate is not a chargeable reservation, but the operator still meant to
|
|
declare one, so what they wrote is what decides whether they hear about it."""
|
|
half_written = {k: v for k, v in _PTU_MODEL_INFO.items() if k != "cost_per_ptu_per_hour"}
|
|
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
_ptu_router(model_info=half_written, ptu_enabled=False)
|
|
|
|
assert len(_ptu_warnings(caplog)) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"typo",
|
|
[
|
|
{"ptu_count": 0},
|
|
{"ptu_count": 0, "cost_per_ptu_per_hour": 0, "ptu_effective_from": None},
|
|
],
|
|
ids=["count out of range", "every value still a zero placeholder"],
|
|
)
|
|
def test_a_reservation_dropped_by_a_typo_is_warned_about(caplog, typo):
|
|
"""An out-of-range value fails ModelInfo before the flag is ever consulted, so the
|
|
deployment stops serving on a proxy that never enabled PTU. The warning is what tells the
|
|
operator which feature the entry that vanished belonged to.
|
|
|
|
Built the way proxy_server builds it, since dropping rather than raising is what
|
|
``ignore_invalid_deployments`` does and config.yaml is loaded with it on.
|
|
"""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": ""}, clear=False):
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
router = Router(
|
|
ignore_invalid_deployments=True,
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-ptu",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://e.azure.com"},
|
|
"model_info": {**_PTU_MODEL_INFO, **typo},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert router.model_list == []
|
|
assert len(_ptu_warnings(caplog)) == 1
|
|
|
|
|
|
def test_a_db_backed_reservation_is_not_warned_about(caplog):
|
|
"""/model/new already answered the caller with a 400, so repeating it on every reload
|
|
would report the operator's own rejected write back to them as a standing problem."""
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
_ptu_router(model_info={**_PTU_MODEL_INFO, "db_model": True}, ptu_enabled=False)
|
|
|
|
assert _ptu_warnings(caplog) == ()
|
|
|
|
|
|
def test_every_declaring_deployment_is_named(caplog):
|
|
"""One line naming all of them, so a reload does not bury the config in repeats."""
|
|
with patch.dict(os.environ, {"LITELLM_ENABLE_PTU_COST_ATTRIBUTION": ""}, clear=False):
|
|
with caplog.at_level(logging.WARNING, logger="LiteLLM Router"):
|
|
Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "azure-ptu-east",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://e.azure.com"},
|
|
"model_info": dict(_PTU_MODEL_INFO),
|
|
},
|
|
{
|
|
"model_name": "azure-ptu-west",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://w.azure.com"},
|
|
"model_info": {**_PTU_MODEL_INFO, "id": "ptu-alpha-westus"},
|
|
},
|
|
{
|
|
"model_name": "plain-gpt-4o",
|
|
"litellm_params": {"model": "azure/gpt-4o", "api_key": "k", "api_base": "https://p.azure.com"},
|
|
"model_info": {"id": "plain"},
|
|
},
|
|
]
|
|
)
|
|
|
|
warnings = _ptu_warnings(caplog)
|
|
|
|
assert len(warnings) == 1
|
|
assert "azure-ptu-east" in warnings[0]
|
|
assert "azure-ptu-west" in warnings[0]
|
|
assert "plain-gpt-4o" not in warnings[0]
|
|
|
|
|
|
def _simulate_price_data_reload_with_provider_sets(monkeypatch, fetched_catalog):
|
|
"""Like `_simulate_price_data_reload`, plus the provider model-set refresh the proxy's
|
|
`_swap_in_model_cost_map` does before replaying, so bare names in the new catalog resolve."""
|
|
monkeypatch.setattr(litellm, "model_cost", fetched_catalog)
|
|
_invalidate_model_cost_lowercase_map()
|
|
litellm.add_known_models(model_cost_map=fetched_catalog)
|
|
reapply_runtime_model_cost_registrations()
|
|
|
|
|
|
def test_a_config_deployment_dropped_by_a_stale_cost_map_comes_back_on_reload(monkeypatch):
|
|
"""
|
|
Booting on the bundled backup, a bare model that only the remote catalog knows
|
|
cannot be provider-resolved, so the proxy router (ignore_invalid_deployments) drops
|
|
it. Once a reload brings in a catalog that knows the model, the deployment must be
|
|
served again with its access groups, and exactly once however many reloads follow.
|
|
"""
|
|
backend = "lit-5766-only-in-remote-catalog"
|
|
try:
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "new-model",
|
|
"litellm_params": {"model": backend, "api_key": "k"},
|
|
"model_info": {"id": "new-id", "access_groups": ["team-models"]},
|
|
},
|
|
{
|
|
"model_name": "control-model",
|
|
"litellm_params": {"model": "hosted_vllm/control-backend", "api_key": "k"},
|
|
"model_info": {"id": "control-id", "access_groups": ["team-models"]},
|
|
},
|
|
],
|
|
ignore_invalid_deployments=True,
|
|
)
|
|
assert router.get_model_names() == ["control-model"]
|
|
assert router.get_model_access_groups(model_name="new-model") == {}
|
|
|
|
fresh_catalog = {**litellm.model_cost, backend: {"litellm_provider": "openai", "mode": "chat"}}
|
|
_simulate_price_data_reload_with_provider_sets(monkeypatch, fresh_catalog)
|
|
_simulate_price_data_reload_with_provider_sets(monkeypatch, fresh_catalog)
|
|
|
|
assert sorted(router.get_model_names()) == ["control-model", "new-model"]
|
|
assert router.get_model_access_groups(model_name="new-model") == {"team-models": ["new-model"]}
|
|
assert [d["model_info"]["id"] for d in router.model_list] == ["control-id", "new-id"]
|
|
assert "new-id" in litellm.model_cost
|
|
finally:
|
|
litellm.open_ai_chat_completion_models.discard(backend)
|
|
litellm.models_by_provider["openai"].discard(backend)
|
|
|
|
|
|
def test_a_config_deployment_dropped_for_a_permanent_reason_is_not_retried_on_reload(monkeypatch):
|
|
"""
|
|
Only provider-resolution drops can be healed by a fresh catalog. A deployment that
|
|
fails after its provider resolved (here a pass-through vertex entry with no project)
|
|
has already touched router state, so replaying it on every reload would leak into
|
|
`deployment_names` each time.
|
|
"""
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "vertex-passthrough",
|
|
"litellm_params": {"model": "vertex_ai/gemini-2.5-flash", "use_in_pass_through": True},
|
|
"model_info": {"id": "vertex-id"},
|
|
},
|
|
{
|
|
"model_name": "control-model",
|
|
"litellm_params": {"model": "hosted_vllm/control-backend", "api_key": "k"},
|
|
"model_info": {"id": "control-id"},
|
|
},
|
|
],
|
|
ignore_invalid_deployments=True,
|
|
)
|
|
assert router.get_model_names() == ["control-model"]
|
|
names_after_boot = list(router.deployment_names)
|
|
|
|
_simulate_price_data_reload_with_provider_sets(monkeypatch, dict(litellm.model_cost))
|
|
|
|
assert router.get_model_names() == ["control-model"]
|
|
assert router.deployment_names == names_after_boot
|