test: restore runtime-derived tests dropped by mistake

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-15 21:44:39 +00:00
parent 62aa21e810
commit a428cc8d84
5 changed files with 277 additions and 1 deletions

View file

@ -4,6 +4,7 @@ from datetime import datetime, timezone
import pytest
import litellm
from litellm._internal_context import pinned_billing_time
from litellm.litellm_core_utils.llm_cost_calc.utils import (
BilledTokenRates,
CostCalculatorUtils,
@ -2086,6 +2087,95 @@ def test_cache_writing_cost_with_zero_creation_tokens_and_ephemeral_details():
assert round(result, 6) == round(expected, 6)
def test_a_pinned_billing_time_prices_the_totals_and_the_reported_rates_at_one_moment(monkeypatch):
monkeypatch.setitem(
litellm.model_cost,
"off-peak-model",
{
"input_cost_per_token": 3e-6,
"output_cost_per_token": 15e-6,
"off_peak_pricing": {
"hours_utc": "02:00-03:00",
"input_cost_per_token": 1e-6,
"output_cost_per_token": 5e-6,
},
"litellm_provider": "openai",
"mode": "chat",
},
)
usage = Usage(prompt_tokens=1000, completion_tokens=500, total_tokens=1500)
with pinned_billing_time(datetime(2026, 1, 1, 2, 30, tzinfo=timezone.utc)):
off_peak_prompt_cost, off_peak_completion_cost = generic_cost_per_token(
model="off-peak-model", usage=usage, custom_llm_provider="openai"
)
off_peak_rates = get_billed_token_rates(model="off-peak-model", custom_llm_provider="openai", usage=usage)
with pinned_billing_time(datetime(2026, 1, 1, 12, 30, tzinfo=timezone.utc)):
peak_prompt_cost, peak_completion_cost = generic_cost_per_token(
model="off-peak-model", usage=usage, custom_llm_provider="openai"
)
peak_rates = get_billed_token_rates(model="off-peak-model", custom_llm_provider="openai", usage=usage)
assert off_peak_rates.input_cost_per_token == pytest.approx(1e-6)
assert peak_rates.input_cost_per_token == pytest.approx(3e-6)
assert off_peak_prompt_cost == pytest.approx(1000 * off_peak_rates.input_cost_per_token)
assert off_peak_completion_cost == pytest.approx(500 * off_peak_rates.output_cost_per_token)
assert peak_prompt_cost == pytest.approx(1000 * peak_rates.input_cost_per_token)
assert peak_completion_cost == pytest.approx(500 * peak_rates.output_cost_per_token)
def test_token_type_cost_breakdown_applies_anthropic_geo_multiplier(_local_model_cost_map, monkeypatch):
from litellm.llms.anthropic.cost_calculation import cost_per_token as anthropic_cost_per_token
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
model = "claude-test-geo-breakdown-model"
litellm.register_model(
model_cost={
model: {
"input_cost_per_token": 5e-6,
"output_cost_per_token": 25e-6,
"cache_creation_input_token_cost": 6.25e-6,
"cache_read_input_token_cost": 0.5e-6,
"litellm_provider": "anthropic",
"max_tokens": 8192,
"provider_specific_entry": {"us": 1.1},
}
}
)
def make_usage() -> Usage:
return Usage(
prompt_tokens=10_000,
completion_tokens=500,
total_tokens=10_500,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=2_000,
cache_creation_tokens=6_000,
),
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=200, text_tokens=300),
)
base_usage = make_usage()
geo_usage = make_usage()
geo_usage.inference_geo = "us"
base = get_token_type_cost_breakdown(model=model, custom_llm_provider="anthropic", usage=base_usage)
geo = get_token_type_cost_breakdown(model=model, custom_llm_provider="anthropic", usage=geo_usage)
assert base.cache_read_cost == pytest.approx(2_000 * 0.5e-6)
assert base.cache_creation_cost == pytest.approx(6_000 * 6.25e-6)
assert geo.cache_read_cost == pytest.approx(base.cache_read_cost * 1.1)
assert geo.cache_creation_cost == pytest.approx(base.cache_creation_cost * 1.1)
assert geo.reasoning_cost == pytest.approx(base.reasoning_cost * 1.1)
prompt_cost, completion_cost = anthropic_cost_per_token(model=model, usage=geo_usage)
text_input_cost = 2_000 * 5e-6 * 1.1
text_output_cost = 300 * 25e-6 * 1.1
assert text_input_cost + geo.cache_read_cost + geo.cache_creation_cost == pytest.approx(prompt_cost)
assert text_output_cost + geo.reasoning_cost == pytest.approx(completion_cost)
def test_service_tier_ultrafast_pricing():
"""An ultrafast request bills the *_ultrafast rates for all token types.
@ -3251,6 +3341,58 @@ def test_token_type_cost_breakdown_applies_vertex_regional_uplift(_local_model_c
assert text_input_cost + regional.cache_read_cost == pytest.approx(prompt_cost)
@pytest.mark.parametrize("details_as_dict", [True, False])
def test_image_response_input_image_tokens_priced_at_image_rate(details_as_dict):
"""
Image input tokens must be priced at input_cost_per_image_token even when
input_tokens_details is a plain dict, as in OpenAI image edit responses.
Regression test: dict-shaped input_tokens_details was read with getattr(),
which returns None for dicts, so image input tokens silently fell back to
the text input rate (e.g. $5/M instead of $8/M for gpt-image-2).
"""
from unittest.mock import patch
from litellm.litellm_core_utils.llm_cost_calc.utils import (
calculate_image_response_cost_from_usage,
)
from litellm.types.utils import Usage
mock_model_info = {
"input_cost_per_token": 5e-6,
"input_cost_per_image_token": 8e-6,
"output_cost_per_image_token": 3e-5,
}
input_details = {"text_tokens": 19, "image_tokens": 512}
image_response = ImageResponse(data=[ImageObject(b64_json="x")])
# Mirror the usage shape of a real OpenAI images.edit response:
# a Usage object carrying input_tokens/output_tokens with detail dicts.
image_response.usage = Usage(
prompt_tokens=0,
completion_tokens=0,
total_tokens=689,
input_tokens=531,
input_tokens_details=(input_details if details_as_dict else ImageUsageInputTokensDetails(**input_details)),
output_tokens=158,
output_tokens_details={"image_tokens": 158, "text_tokens": 0},
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
cost = calculate_image_response_cost_from_usage(
model="gpt-image-2",
image_response=image_response,
custom_llm_provider="openai",
)
expected = 19 * 5e-6 + 512 * 8e-6 + 158 * 3e-5
assert cost is not None
assert round(cost, 12) == round(expected, 12)
GEMINI_DAY0_LAUNCH_PRICING = [
("gemini-3.6-flash", 7.5e-07, 3.75e-06, 7.5e-08),
("gemini/gemini-3.6-flash", 7.5e-07, 3.75e-06, 7.5e-08),

View file

@ -86,6 +86,21 @@ def test_validation_accepts_healthy_file_with_meta_keys():
)
def test_validation_rejects_significant_shrink_vs_backup():
# 600 real models vs a 2000-model backup is below the 50% shrink threshold.
shrunk = _make_models(600)
shrunk[FALLBACK_GENERALIZATIONS_KEY] = {"rules": []}
assert (
GetModelCostMap.validate_model_cost_map(
fetched_map=shrunk,
backup_model_count=2000,
min_model_count=50,
max_shrink_ratio=0.5,
)
is False
)
def test_finalize_pops_key_and_installs_rules():
previous = list(get_fallback_generalization_rules())
try:
@ -589,6 +604,41 @@ def test_boot_load_records_the_blob_id_of_the_bytes_served_and_the_fetch_etag():
assert source["loaded_at"] is not None
def test_boot_load_fallback_to_the_backup_reports_its_blob_id_and_drops_the_remote_etag():
remote, _ = _mock_client(
[httpx.Response(200, headers={"ETag": 'W/"boot"'}, content=_real_map_bytes())], client_cls=httpx.Client
)
get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=remote)
failing, _ = _mock_client([httpx.Response(404)], client_cls=httpx.Client)
get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=failing)
source = get_model_cost_map_source_info()
assert source["source"] == "local"
assert source["etag"] is None
assert source["source_revision"] == _bundled_blob_id()
def test_boot_load_that_fails_the_integrity_check_reports_the_backup_not_the_rejected_fetch():
remote, _ = _mock_client(
[httpx.Response(200, headers={"ETag": 'W/"boot"'}, content=_real_map_bytes())], client_cls=httpx.Client
)
get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=remote)
shrunk_body = b'{"gpt-5.4-mini": {"mode": "chat", "input_cost_per_token": 1e-06, "output_cost_per_token": 2e-06}}'
shrunk, _ = _mock_client(
[httpx.Response(200, headers={"ETag": 'W/"shrunk"'}, content=shrunk_body)], client_cls=httpx.Client
)
get_model_cost_map(url=_URL, sleep=_SyncSleepRecorder(), rng=random.Random(0), client=shrunk)
source = get_model_cost_map_source_info()
assert source["source"] == "local"
assert source["fallback_reason"] == "Remote data failed integrity validation"
assert source["etag"] is None
assert source["source_revision"] == _bundled_blob_id()
assert source["source_revision"] != git_blob_id(shrunk_body)
@pytest.mark.parametrize(
("argv0", "request_count"),
[

View file

@ -4,7 +4,7 @@ from typing import Final
import pytest
from pydantic import TypeAdapter
from litellm import completion_cost
from litellm import completion_cost, cost_per_token, get_model_info
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.types.utils import TranscriptionResponse
@ -52,6 +52,34 @@ def test_azure_ai_catalog_name_routes_to_azure_ai(catalog_name: str) -> None:
assert (routed_model, provider) == (catalog_name, "azure_ai")
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize("catalog_name", TOKEN_PRICED_NAMES)
def test_azure_ai_catalog_name_charges_its_own_entry_per_token(catalog_name: str) -> None:
entry: Final = get_model_info(f"azure_ai/{catalog_name}")
prompt_cost, completion_cost_usd = cost_per_token(
model=f"azure_ai/{catalog_name}", prompt_tokens=A_MILLION, completion_tokens=A_MILLION
)
assert prompt_cost > 0
assert prompt_cost == pytest.approx(A_MILLION * entry["input_cost_per_token"])
assert completion_cost_usd == pytest.approx(A_MILLION * entry["output_cost_per_token"])
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize("catalog_name", TOKEN_PRICED_NAMES)
def test_azure_ai_catalog_name_prices_the_same_in_any_casing(catalog_name: str) -> None:
lowercase_cost = cost_per_token(model=f"azure_ai/{catalog_name}", prompt_tokens=A_MILLION, completion_tokens=0)
upper_cost = cost_per_token(model=f"azure_ai/{catalog_name.upper()}", prompt_tokens=A_MILLION, completion_tokens=0)
assert upper_cost == lowercase_cost
@pytest.mark.usefixtures("local_model_cost_map")
def test_azure_ai_whisper_catalog_name_is_priced_per_second() -> None:
one_second_cost: Final = _whisper_transcription_cost(1)
one_hour_cost: Final = _whisper_transcription_cost(AN_HOUR_IN_SECONDS)
assert one_second_cost > 0
assert one_hour_cost == pytest.approx(AN_HOUR_IN_SECONDS * one_second_cost)
def test_azure_ai_model_router_spellings_share_one_entry() -> None:
underscore_entry = _cost_map_entry(MAIN_COST_MAP, "model_router")
hyphen_entry = _cost_map_entry(MAIN_COST_MAP, "model-router")

View file

@ -1,4 +1,6 @@
import copy
import json
import os
from unittest.mock import MagicMock, patch
from litellm.llms.vertex_ai.vertex_ai_partner_models.anthropic.experimental_pass_through.transformation import (
@ -542,3 +544,30 @@ class TestVertexAnthropicMidConversationSystem:
{"type": "text", "text": "You are terse."},
{"type": "text", "text": "Cite sources."},
]
def test_vertex_claude_4_8_plus_cost_map_entries_carry_mid_conversation_system_flag():
import re
import litellm
cost_map_path = os.path.join(os.path.dirname(litellm.__file__), "model_prices_and_context_window_backup.json")
with open(cost_map_path) as f:
cost_map = json.load(f)
rules = cost_map["fallback_generalizations"]["rules"]
rule_pattern = next(
(r["pattern"] for r in rules if r["name"] == "claude-mid-conversation-system"),
None,
)
assert rule_pattern is not None, "claude-mid-conversation-system rule not found in fallback_generalizations"
pattern = re.compile(rule_pattern, re.IGNORECASE)
missing = [
key
for key, info in cost_map.items()
if isinstance(info, dict)
and str(info.get("litellm_provider", "")).startswith("vertex_ai")
and "claude" in key
and pattern.search(key)
and info.get("supports_mid_conversation_system") is not True
]
assert missing == []

View file

@ -0,0 +1,27 @@
import pytest
import litellm
@pytest.fixture
def local_model_cost_map(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
litellm.get_model_info.cache_clear()
yield
litellm.get_model_info.cache_clear()
def test_entry_advertises_only_what_the_baseten_path_accepts(local_model_cost_map):
"""The Baseten path rejects unsupported request parameters."""
supported = litellm.get_supported_openai_params(model="zai-org/GLM-5.3", custom_llm_provider="baseten")
assert supported is not None
with pytest.raises(litellm.UnsupportedParamsError):
litellm.utils.get_optional_params(
model="zai-org/GLM-5.3",
custom_llm_provider="baseten",
parallel_tool_calls=True,
reasoning_effort="high",
drop_params=False,
)