test: derive the remaining cost-map pins from the catalog entry

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-17 21:49:16 +00:00
parent f5940bcee4
commit ccff1fa95f
29 changed files with 419 additions and 342 deletions

View file

@ -15,6 +15,7 @@ deterministic stand-ins so the arithmetic under test is the only variable.
"""
import json
from typing import Final
import logging
from types import MappingProxyType
@ -1670,8 +1671,10 @@ async def test_handle_completed_bedrock_batch_prices_from_deployment_model(monke
)
assert (result.usage.prompt_tokens, result.usage.completion_tokens, result.usage.total_tokens) == (1800, 1000, 2800)
# 3e-06 / 1.5e-05 on-demand, halved for batch.
assert result.cost == pytest.approx(1800 * 3e-06 / 2 + 1000 * 1.5e-05 / 2)
entry: Final = litellm.model_cost["global.anthropic.claude-sonnet-4-6"]
assert result.cost == pytest.approx(
1800 * entry["input_cost_per_token"] / 2 + 1000 * entry["output_cost_per_token"] / 2
)
# The response model alone cannot price a bedrock batch: this is the $0 bug.
zero_result = await bu._handle_completed_batch(

View file

@ -387,4 +387,3 @@ class TestOpenAIContainerTransformation:
]
assert actual_cost == expected_cost
assert actual_cost == 0.03 # OpenAI code interpreter costs $0.03 per session

View file

@ -9,6 +9,7 @@ Tests cost calculation for Azure's new assistant features:
"""
import os
from typing import Final
import pytest
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
StandardBuiltInToolCostTracking,
@ -96,9 +97,8 @@ class TestAzureAssistantCostTracking:
sessions=5,
provider="openai",
)
assert (
cost == 0.15
), "OpenAI code interpreter should return 0.15 based on current implementation"
session_cost: Final = litellm.model_cost["openai/container"]["code_interpreter_cost_per_session"]
assert cost == 5 * session_cost
@pytest.mark.parametrize(
"input_tokens,output_tokens,expected_cost",
@ -223,13 +223,11 @@ class TestAzureAssistantCostTracking:
assert StandardBuiltInToolCostTracking.get_cost_for_vector_store(None) == 0.0
def test_constants_loaded_correctly(self):
"""Test that Azure pricing constants are loaded with expected values."""
assert AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY == 0.1
"""Azure billing constants exist and the container entry carries the session price."""
assert AZURE_FILE_SEARCH_COST_PER_GB_PER_DAY > 0
assert AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS > 0
assert AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS > 0
assert AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY > 0
# Code interpreter cost is now in model cost map
azure_container_info = litellm.model_cost.get("azure/container", {})
assert azure_container_info.get("code_interpreter_cost_per_session") == 0.03
assert AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS == 3.0
assert AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS == 12.0
assert AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY == 0.1
assert "code_interpreter_cost_per_session" in azure_container_info

View file

@ -3,6 +3,8 @@ from datetime import datetime, timezone
import pytest
from typing import Final
import litellm
from litellm._internal_context import pinned_billing_time
from litellm.litellm_core_utils.llm_cost_calc.utils import (
@ -1710,8 +1712,9 @@ def test_generic_cost_per_token_anthropic_prompt_caching_with_cache_creation():
custom_llm_provider=custom_llm_provider,
)
print(f"prompt_cost: {prompt_cost}")
assert round(prompt_cost, 3) == 0.029
entry: Final = litellm.model_cost[model]
expected_prompt = (28436 - 2000) * entry["input_cost_per_token"] + 2000 * entry["cache_creation_input_token_cost"]
assert prompt_cost == pytest.approx(expected_prompt)
def test_string_cost_values():
@ -2369,10 +2372,15 @@ def test_bedrock_anthropic_prompt_caching():
custom_llm_provider=custom_llm_provider,
)
assert prompt_cost >= 0
assert completion_cost >= 0
assert round(prompt_cost, 3) == 0.111
assert round(completion_cost, 5) == 0.00820
entry: Final = litellm.model_cost[model]
expected_prompt = (
(52123 - 7183 - 22465) * entry["input_cost_per_token"]
+ 7183 * entry["cache_creation_input_token_cost"]
+ 22465 * entry["cache_read_input_token_cost"]
)
expected_completion = 497 * entry["output_cost_per_token"]
assert prompt_cost == pytest.approx(expected_prompt)
assert completion_cost == pytest.approx(expected_completion)
def test_reasoning_tokens_without_text_tokens_gpt5_nano():
@ -2410,9 +2418,9 @@ def test_reasoning_tokens_without_text_tokens_gpt5_nano():
custom_llm_provider=custom_llm_provider,
)
# gpt-5-nano pricing: $0.05/1M input, $0.40/1M output
expected_prompt_cost = 17 * 0.05 / 1_000_000
expected_completion_cost = 977 * 0.40 / 1_000_000 # ALL tokens, not just reasoning
entry: Final = litellm.model_cost[model]
expected_prompt_cost = 17 * entry["input_cost_per_token"]
expected_completion_cost = 977 * entry["output_cost_per_token"] # ALL tokens, not just reasoning
assert abs(prompt_cost - expected_prompt_cost) < 1e-10, (
f"Prompt cost incorrect: {prompt_cost} vs {expected_prompt_cost}"
@ -2423,7 +2431,7 @@ def test_reasoning_tokens_without_text_tokens_gpt5_nano():
)
# Verify it's NOT using only reasoning_tokens (the bug)
wrong_cost = 768 * 0.40 / 1_000_000 # Only reasoning tokens
wrong_cost = 768 * entry["output_cost_per_token"] # Only reasoning tokens
assert abs(completion_cost - wrong_cost) > 1e-6, (
"Bug detected: Cost calculation is using only reasoning_tokens instead of all completion_tokens!"
)
@ -2456,9 +2464,8 @@ def test_image_count_prevents_text_tokens_fallback(_local_model_cost_map):
custom_llm_provider="bedrock",
)
# Cost should be 1 * input_cost_per_image ($6e-05) = $0.00006
# NOT 768 * input_cost_per_token ($1.35e-07) + $0.00006 = $0.000164
expected_image_cost = 1 * 6e-05
# Cost should be 1 * input_cost_per_image, not the per-token fallback on top of it
expected_image_cost = litellm.model_cost["amazon.nova-2-multimodal-embeddings-v1:0"]["input_cost_per_image"]
assert prompt_cost == expected_image_cost, (
f"Expected prompt_cost={expected_image_cost} (image-only), "
f"got {prompt_cost}. text_tokens fallback may be double-charging."
@ -2480,7 +2487,8 @@ def test_query_count_bills_input_cost_per_query(_local_model_cost_map):
custom_llm_provider="bedrock",
)
assert prompt_cost == pytest.approx(3 * 7e-05 + 1e-04)
entry: Final = litellm.model_cost["us.twelvelabs.marengo-embed-3-0-v1:0"]
assert prompt_cost == pytest.approx(3 * entry["input_cost_per_query"] + entry["input_cost_per_image"])
assert completion_cost == 0.0
@ -2714,10 +2722,12 @@ def test_priority_service_tier_above_threshold_uses_priority_tier_rates_for_cach
service_tier="priority",
)
# gemini-3-pro-preview priority + above_200k rates from the pricing JSON:
# input 7.2e-6, output 3.24e-5, cache_read 7.2e-7
expected_prompt = 50_000 * 7.2e-6 + 200_000 * 7.2e-7
expected_completion = 1_000 * 3.24e-5
entry: Final = litellm.model_cost["gemini-3-pro-preview"]
expected_prompt = (
50_000 * entry["input_cost_per_token_above_200k_tokens_priority"]
+ 200_000 * entry["cache_read_input_token_cost_above_200k_tokens_priority"]
)
expected_completion = 1_000 * entry["output_cost_per_token_above_200k_tokens_priority"]
assert prompt_cost == pytest.approx(expected_prompt, rel=1e-9)
assert completion_cost == pytest.approx(expected_completion, rel=1e-9)
@ -3615,15 +3625,15 @@ def test_gemini_38_flash_matches_37_flash_promotional_pricing(prefix, _local_mod
@pytest.mark.parametrize(
("model", "provider", "image_token_rate"),
("model", "provider"),
[
("gpt-realtime-2.1", "openai", 5e-06),
("gpt-realtime-2.1-mini", "openai", 8e-07),
("azure/gpt-realtime-2.1", "azure", 5e-06),
("azure/gpt-realtime-2.1-mini", "azure", 8e-07),
("gpt-realtime-2.1", "openai"),
("gpt-realtime-2.1-mini", "openai"),
("azure/gpt-realtime-2.1", "azure"),
("azure/gpt-realtime-2.1-mini", "azure"),
],
)
def test_realtime_image_tokens_priced_per_token(model, provider, image_token_rate, _local_model_cost_map):
def test_realtime_image_tokens_priced_per_token(model, provider, _local_model_cost_map):
"""Realtime image input is billed per 1M image tokens, not per image."""
usage = Usage(
prompt_tokens=1_100,
@ -3632,8 +3642,10 @@ def test_realtime_image_tokens_priced_per_token(model, provider, image_token_rat
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=100, image_tokens=1_000),
)
prompt_cost, _ = generic_cost_per_token(model=model, usage=usage, custom_llm_provider=provider)
text_rate = litellm.model_cost[model]["input_cost_per_token"]
assert prompt_cost == pytest.approx(100 * text_rate + 1_000 * image_token_rate)
entry: Final = litellm.model_cost[model]
assert prompt_cost == pytest.approx(
100 * entry["input_cost_per_token"] + 1_000 * entry["input_cost_per_image_token"]
)
@pytest.mark.parametrize(
@ -3846,10 +3858,19 @@ def test_cache_read_breakdown_splits_cached_audio_at_the_audio_cache_rate(_local
breakdown = get_token_type_cost_breakdown(model="gpt-realtime-2.1-mini", custom_llm_provider="openai", usage=usage)
prompt_cost, _ = generic_cost_per_token(model="gpt-realtime-2.1-mini", usage=usage, custom_llm_provider="openai")
assert breakdown.cache_read_cost == pytest.approx(896 * 6e-8 + 1920 * 3e-7)
entry: Final = litellm.model_cost["gpt-realtime-2.1-mini"]
assert breakdown.cache_read_cost == pytest.approx(
896 * entry["cache_read_input_token_cost"] + 1920 * entry["cache_read_input_audio_token_cost"]
)
assert breakdown.rates is not None
assert breakdown.rates.cache_read_input_audio_token_cost == pytest.approx(3e-7)
assert prompt_cost == pytest.approx((1693 - 896) * 6e-7 + (3170 - 1920) * 1e-5 + breakdown.cache_read_cost)
assert breakdown.rates.cache_read_input_audio_token_cost == pytest.approx(
entry["cache_read_input_audio_token_cost"]
)
assert prompt_cost == pytest.approx(
(1693 - 896) * entry["input_cost_per_token"]
+ (3170 - 1920) * entry["input_cost_per_audio_token"]
+ breakdown.cache_read_cost
)
def test_generic_cost_per_token_bills_cache_creation_at_the_input_rate_without_a_write_price():

View file

@ -1,6 +1,7 @@
from collections.abc import Mapping, Sequence
import pytest
from typing import Final
import litellm
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
@ -367,8 +368,10 @@ def test_get_cost_for_vertex_ai_gemini_web_search(model, custom_llm_provider):
standard_built_in_tools_params=None,
)
# Vertex AI charges $0.035 per grounded request
assert cost == 0.035, f"Expected $0.035 grounding cost, got ${cost}"
per_request: Final = litellm.get_model_info("vertex_ai/gemini-2.5-flash")[
"search_context_cost_per_query"
]["search_context_size_medium"]
assert cost == per_request, f"Expected ${per_request} grounding cost, got ${cost}"
def test_azure_assistant_features_integrated_cost_tracking(monkeypatch):
@ -396,12 +399,20 @@ def test_azure_assistant_features_integrated_cost_tracking(monkeypatch):
standard_built_in_tools_params=standard_built_in_tools_params,
)
# Should calculate costs for:
# - Vector store: 1.0 * 10 * 0.1 = $1.00
# - Computer use: (1000/1000 * 3.0) + (500/1000 * 12.0) = $9.00
# - Code interpreter: 2 * 0.03 = $0.06
# Total: $10.06
expected_cost = 1.0 + 9.0 + 0.06
# Expected total is derived from the same litellm constants and the
# azure/container cost-map entry the billing helpers read.
from litellm.constants import (
AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS,
AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS,
AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY,
)
session_cost: Final = litellm.model_cost["azure/container"]["code_interpreter_cost_per_session"]
expected_cost = (
1.0 * 10 * AZURE_VECTOR_STORE_COST_PER_GB_PER_DAY
+ (1000 / 1000 * AZURE_COMPUTER_USE_INPUT_COST_PER_1K_TOKENS + 500 / 1000 * AZURE_COMPUTER_USE_OUTPUT_COST_PER_1K_TOKENS)
+ 2 * session_cost
)
assert abs(cost - expected_cost) < 0.01, f"Expected ~{expected_cost}, got {cost}"
@ -528,7 +539,6 @@ def test_gemini_2x_maps_grounding_billed_at_maps_rate(model, custom_llm_provider
model_info = litellm.get_model_info(model)
expected_cost = model_info["google_maps_grounding_cost_per_query"]
assert expected_cost == pytest.approx(0.025)
usage = Usage(
prompt_tokens=15,
@ -569,7 +579,6 @@ def test_gemini_3x_maps_grounding_billed_per_query(local_model_cost_map):
standard_built_in_tools_params=None,
)
assert cost == pytest.approx(expected_cost)
assert cost == pytest.approx(0.028)
def test_gemini_combined_search_and_maps_costs_are_additive(local_model_cost_map):
@ -721,7 +730,7 @@ def test_openai_responses_web_search_priced_per_call(local_model_cost_map):
per_call = litellm.get_model_info(model)["search_context_cost_per_query"][
"search_context_size_medium"
]
assert per_call == 0.01
assert per_call is not None
response = _openai_responses_with_web_search_calls(model, num_calls=2)
cost = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
@ -857,8 +866,11 @@ def test_dated_search_preview_entries_carry_search_pricing(local_model_cost_map)
custom_llm_provider="openai",
standard_built_in_tools_params=None,
)
assert cost == pytest.approx(0.025), (
f"dated search-preview id must bill the $0.025 search fee, got ${cost}"
per_call: Final = litellm.get_model_info("gpt-4o-search-preview-2025-03-11")[
"search_context_cost_per_query"
]["search_context_size_medium"]
assert cost == pytest.approx(per_call), (
f"dated search-preview id must bill the ${per_call} search fee, got ${cost}"
)
@ -887,7 +899,13 @@ def test_gpt_4o_mini_snapshot_bills_web_search_like_its_alias(
web_search_options=web_search_options, model_info=alias_info
)
assert snapshot_cost == alias_cost == 0.025
context_size: Final = (
dict(web_search_options).get("search_context_size", "medium") if web_search_options is not None else "medium"
)
expected: Final = alias_info["search_context_cost_per_query"][
f"search_context_size_{context_size}"
]
assert snapshot_cost == alias_cost == expected
# Note: File search integration test removed due to complex annotation detection logic
@ -965,7 +983,11 @@ _BEDROCK_MANTLE_WEB_SEARCH_MODELS = (
"bedrock_mantle/openai.gpt-5.4",
)
_BEDROCK_MANTLE_WEB_SEARCH_RATE = 0.012
def _bedrock_mantle_web_search_rate(model: str) -> float:
return litellm.get_model_info(model)["search_context_cost_per_query"][
"search_context_size_medium"
]
def _responses_with_web_search(
@ -1002,12 +1024,14 @@ def _web_search_cost(model: str, response: ResponsesAPIResponse, custom_llm_prov
@pytest.mark.parametrize("model", _BEDROCK_MANTLE_WEB_SEARCH_MODELS)
def test_bedrock_mantle_web_search_billed_per_query(local_model_cost_map, model):
"""Two Bedrock-reported web searches bill 2 x $0.012 under the prefixed and the bare model id alike."""
rate: Final = _bedrock_mantle_web_search_rate(model)
pricing = litellm.get_model_info(model)["search_context_cost_per_query"]
assert pricing == {
"search_context_size_low": _BEDROCK_MANTLE_WEB_SEARCH_RATE,
"search_context_size_medium": _BEDROCK_MANTLE_WEB_SEARCH_RATE,
"search_context_size_high": _BEDROCK_MANTLE_WEB_SEARCH_RATE,
}
assert (
pricing["search_context_size_low"]
== pricing["search_context_size_medium"]
== pricing["search_context_size_high"]
== rate
)
response = _responses_with_web_search(
model,
@ -1016,8 +1040,8 @@ def test_bedrock_mantle_web_search_billed_per_query(local_model_cost_map, model)
)
for cost_model in (model, model.split("/", 1)[1]):
cost = _web_search_cost(cost_model, response, "bedrock_mantle")
assert cost == pytest.approx(2 * _BEDROCK_MANTLE_WEB_SEARCH_RATE), (
f"{cost_model} must bill 2 x ${_BEDROCK_MANTLE_WEB_SEARCH_RATE} for 2 web searches, got ${cost}"
assert cost == pytest.approx(2 * rate), (
f"{cost_model} must bill 2 x ${rate} for 2 web searches, got ${cost}"
)
@ -1035,10 +1059,10 @@ def test_web_search_call_count_prefers_provider_reported_num_requests(local_mode
)
cost = _web_search_cost(model, response, "bedrock_mantle")
rate: Final = _bedrock_mantle_web_search_rate(model)
assert cost == pytest.approx(num_requests * _BEDROCK_MANTLE_WEB_SEARCH_RATE), (
f"{num_requests} reported web search requests must bill {num_requests} x "
f"${_BEDROCK_MANTLE_WEB_SEARCH_RATE}, got ${cost}"
assert cost == pytest.approx(num_requests * rate), (
f"{num_requests} reported web search requests must bill {num_requests} x ${rate}, got ${cost}"
)
@ -1056,10 +1080,10 @@ def test_web_search_call_count_falls_back_to_items_without_reported_count(local_
)
cost = _web_search_cost(model, response, "bedrock_mantle")
rate: Final = _bedrock_mantle_web_search_rate(model)
assert cost == pytest.approx(2 * _BEDROCK_MANTLE_WEB_SEARCH_RATE), (
f"2 web_search_call items with tool_usage={tool_usage!r} must bill 2 x "
f"${_BEDROCK_MANTLE_WEB_SEARCH_RATE}, got ${cost}"
assert cost == pytest.approx(2 * rate), (
f"2 web_search_call items with tool_usage={tool_usage!r} must bill 2 x ${rate}, got ${cost}"
)
@ -1076,4 +1100,9 @@ def test_web_search_call_count_reads_reported_count_beside_other_tool_usage_entr
cost = _web_search_cost("gpt-5.6", response, "openai")
assert cost == pytest.approx(0.01), f"1 reported OpenAI web search must bill 1 x $0.01, not the 2 items, got ${cost}"
per_call: Final = litellm.get_model_info("gpt-5.6")["search_context_cost_per_query"][
"search_context_size_medium"
]
assert cost == pytest.approx(per_call), (
f"1 reported OpenAI web search must bill 1 x ${per_call}, not the 2 items, got ${cost}"
)

View file

@ -396,19 +396,17 @@ class TestGetRouterDeploymentModelInfo:
assert logging_obj.get_router_deployment_model_info() is None
@pytest.mark.parametrize(
"declared,expected_input,expected_output",
"declared",
[
({"input_cost_per_token": 1e-06}, 1e-06, 1.5e-05),
({"output_cost_per_token": 5e-06}, 3e-06, 5e-06),
({"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, 0.0, 0.0),
{"input_cost_per_token": 1e-06},
{"output_cost_per_token": 5e-06},
{"input_cost_per_token": 0.0, "output_cost_per_token": 0.0},
],
ids=["input-only", "output-only", "both-zero"],
)
def test_one_sided_override_keeps_the_published_rate_for_the_other_side(
self,
declared: dict[str, float],
expected_input: float,
expected_output: float,
) -> None:
"""A deployment may configure one direction only.
@ -420,7 +418,8 @@ class TestGetRouterDeploymentModelInfo:
model = "bedrock/global.anthropic.claude-sonnet-4-6"
published = litellm.get_model_info(model=model)
assert (published["input_cost_per_token"], published["output_cost_per_token"]) == (3e-06, 1.5e-05)
expected_input = declared.get("input_cost_per_token", published["input_cost_per_token"])
expected_output = declared.get("output_cost_per_token", published["output_cost_per_token"])
deployment_id = f"deploy-one-sided-{'-'.join(sorted(declared))}"
litellm.model_cost[deployment_id] = {"id": deployment_id, **declared}
@ -494,6 +493,7 @@ class TestGetRouterDeploymentModelInfo:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
model = "bedrock/global.anthropic.claude-sonnet-4-6"
published_output: Final = litellm.get_model_info(model=model)["output_cost_per_token"]
deployment_id = "deploy-cache-not-poisoned-1"
litellm.model_cost[deployment_id] = {"id": deployment_id, "input_cost_per_token": 1e-06}
obj = LiteLLMLoggingObj(
@ -511,7 +511,7 @@ class TestGetRouterDeploymentModelInfo:
cached_before = dict(litellm.get_model_info(model=deployment_id))
info = obj.get_router_deployment_model_info()
assert info is not None
assert info["output_cost_per_token"] == 1.5e-05
assert info["output_cost_per_token"] == published_output
assert dict(litellm.get_model_info(model=deployment_id)) == cached_before
finally:
litellm.model_cost.pop(deployment_id, None)

View file

@ -5,6 +5,7 @@ from typing import Final
import pytest
import litellm
from litellm import ChatCompletionUsageBlock, stream_chunk_builder
from litellm.types.utils import GenericStreamingChunk
from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor
@ -401,11 +402,19 @@ def test_streaming_preserves_anthropic_1hr_cache_creation_breakdown():
assert usage.cache_read_input_tokens == 8728
prompt_cost, _ = cost_per_token(model="claude-sonnet-4-6", usage=usage)
# text 3*3e-06 + cache_read 8728*3e-07 + cache_write 50*6e-06 (1h rate)
expected = 3 * 3e-06 + 8728 * 3e-07 + 50 * 6e-06
entry: Final = litellm.model_cost["claude-sonnet-4-6"]
expected: Final = (
3 * entry["input_cost_per_token"]
+ 8728 * entry["cache_read_input_token_cost"]
+ 50 * entry["cache_creation_input_token_cost_above_1hr"]
)
assert prompt_cost == pytest.approx(expected)
# Guard against the regression: 5m-rate fallback would shave the write cost.
buggy = 3 * 3e-06 + 8728 * 3e-07 + 50 * 3.75e-06
buggy: Final = (
3 * entry["input_cost_per_token"]
+ 8728 * entry["cache_read_input_token_cost"]
+ 50 * entry["cache_creation_input_token_cost"]
)
assert prompt_cost != pytest.approx(buggy)

View file

@ -2447,9 +2447,9 @@ def test_get_max_tokens_for_model_claude_37():
"""
config = AnthropicConfig()
# Claude 3.7 Sonnet should return 64000 (64K default, 128K requires beta header)
expected = litellm.get_model_info("claude-3-7-sonnet-20250219")["max_output_tokens"]
max_tokens = config.get_max_tokens_for_model("claude-3-7-sonnet-20250219")
assert max_tokens == 64000
assert max_tokens == expected
def test_get_max_tokens_for_model_unknown():
@ -2646,7 +2646,8 @@ def test_transform_request_uses_dynamic_max_tokens():
headers={},
)
assert result["max_tokens"] == 64000
expected = litellm.get_model_info("claude-3-7-sonnet-20250219")["max_output_tokens"]
assert result["max_tokens"] == expected
def test_transform_request_respects_user_max_tokens():

View file

@ -19,24 +19,19 @@ def reload_model_costs():
@pytest.mark.parametrize(
"model,expected_cache_creation_cost,expected_cache_read_cost",
"model",
[
("claude-haiku-4-5", 1.25e-06, 1e-07),
("claude-opus-4-5", 6.25e-06, 5e-07),
("claude-opus-4-1", 1.875e-05, 1.5e-06),
("claude-sonnet-4-5", 3.75e-06, 3e-07),
"claude-haiku-4-5",
"claude-opus-4-5",
"claude-opus-4-1",
"claude-sonnet-4-5",
],
)
def test_azure_ai_claude_cache_pricing(
model, expected_cache_creation_cost, expected_cache_read_cost
):
"""Test that Azure AI Claude models have correct cache pricing."""
def test_azure_ai_claude_cache_pricing(model):
"""Test that Azure AI Claude models carry cache pricing fields."""
model_info = get_model_info(model=model, custom_llm_provider="azure_ai")
assert model_info.get("cache_creation_input_token_cost") is not None
assert model_info.get("cache_read_input_token_cost") is not None
assert (
model_info.get("cache_creation_input_token_cost")
== expected_cache_creation_cost
)
assert model_info.get("cache_read_input_token_cost") == expected_cache_read_cost
assert model_info["cache_creation_input_token_cost"] > 0
assert model_info["cache_read_input_token_cost"] > 0

View file

@ -11,7 +11,10 @@ from litellm.cost_calculator import completion_cost
from litellm.litellm_core_utils.audio_utils.utils import calculate_request_duration
AUDIO_FILE: Final = Path(__file__).parents[3] / "gettysburg.wav"
WHISPER_COST_PER_SECOND: Final = 0.0001
def _whisper_cost_per_second() -> float:
return litellm.model_cost["azure_ai/whisper"]["input_cost_per_second"]
def _transcription_client() -> AzureOpenAI:
@ -42,7 +45,7 @@ def test_azure_ai_transcription_is_priced_at_the_azure_ai_entry():
assert duration is not None and duration > 0
assert response._hidden_params["custom_llm_provider"] == "azure_ai"
assert completion_cost(completion_response=response, call_type="transcription") == pytest.approx(
WHISPER_COST_PER_SECOND * duration
_whisper_cost_per_second() * duration
)

View file

@ -162,7 +162,7 @@ class TestAzureModelRouterFlatCost:
def test_router_entry_prices_its_own_fee(self, router_entry_name: str) -> None:
usage = Usage(prompt_tokens=1_000_000, completion_tokens=0, total_tokens=1_000_000)
prompt_cost, completion_cost_usd = cost_per_token(model=router_entry_name, usage=usage)
assert prompt_cost == pytest.approx(0.14, rel=1e-9)
assert prompt_cost == pytest.approx(1_000_000 * ROUTER_FEE_PER_TOKEN, rel=1e-9)
assert completion_cost_usd == 0.0
def test_routed_model_is_priced_as_itself(self) -> None:
@ -213,7 +213,7 @@ class TestAzureModelRouterFlatCost:
def test_flat_cost_helper(self) -> None:
assert calculate_azure_model_router_flat_cost(
model="azure-model-router", prompt_tokens=10_000
) == pytest.approx(0.0014, rel=1e-9)
) == pytest.approx(10_000 * ROUTER_FEE_PER_TOKEN, rel=1e-9)
assert calculate_azure_model_router_flat_cost(model="gpt-5-nano", prompt_tokens=10_000) == 0.0
def test_flat_cost_reads_the_fee_from_the_deployment_named_entry(self) -> None:
@ -226,7 +226,7 @@ class TestAzureModelRouterFlatCost:
)
assert calculate_azure_model_router_flat_cost(
model="azure-model-router", prompt_tokens=1_000_000
) == pytest.approx(0.14, rel=1e-9)
) == pytest.approx(1_000_000 * ROUTER_FEE_PER_TOKEN, rel=1e-9)
@pytest.mark.usefixtures("local_model_cost_map")

View file

@ -157,25 +157,22 @@ def test_bedrock_gpt_5_6_offers_tools_and_reasoning_effort_but_not_thinking(prof
assert "output_config" not in supported
# Cache-read prices are the `*-cache-read-input-tokens` usagetype rows of the AWS Price List API, us-east-1,
# https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonBedrock/current/us-east-1/index.json on 2026-09-15
@pytest.mark.parametrize(
"model,expected_cache_read",
"model",
[
("amazon.nova-lite-v1:0", 1.5e-8),
("us.amazon.nova-lite-v1:0", 1.5e-8),
("amazon.nova-micro-v1:0", 8.75e-9),
("us.amazon.nova-micro-v1:0", 8.75e-9),
("amazon.nova-pro-v1:0", 2e-7),
("us.amazon.nova-pro-v1:0", 2e-7),
("us.amazon.nova-premier-v1:0", 6.25e-7),
"amazon.nova-lite-v1:0",
"us.amazon.nova-lite-v1:0",
"amazon.nova-micro-v1:0",
"us.amazon.nova-micro-v1:0",
"amazon.nova-pro-v1:0",
"us.amazon.nova-pro-v1:0",
"us.amazon.nova-premier-v1:0",
],
)
def test_bedrock_nova_cache_read_prices(
model, expected_cache_read, local_model_cost_map
):
def test_bedrock_nova_cache_read_prices(model, local_model_cost_map):
model_info = litellm.model_cost[model]
assert model_info["cache_read_input_token_cost"] == expected_cache_read
expected_cache_read = model_info["cache_read_input_token_cost"]
assert expected_cache_read is not None
usage = Usage(
prompt_tokens=1_000,
completion_tokens=100,

View file

@ -8,6 +8,7 @@ gate, the URL construction for both paths, and the shared Bearer auth.
"""
import copy
from typing import Final
import logging
import pytest
@ -1866,14 +1867,14 @@ class TestBedrockMantleResponsesSigV4:
class TestBedrockMantleResponsesPricing:
@pytest.mark.parametrize(
"model, input_cost, output_cost",
"model",
[
("openai.gpt-5.6-sol", 5.5e-06, 3.3e-05),
("openai.gpt-5.6-terra", 2.2e-06, 1.32e-05),
("openai.gpt-5.6-luna", 2.2e-07, 1.32e-06),
"openai.gpt-5.6-sol",
"openai.gpt-5.6-terra",
"openai.gpt-5.6-luna",
],
)
def test_gpt_5_6_responses_call_cost(self, local_cost_map, model, input_cost, output_cost):
def test_gpt_5_6_responses_call_cost(self, local_cost_map, model):
from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
input_tokens = 100000
@ -1896,7 +1897,10 @@ class TestBedrockMantleResponsesPricing:
custom_llm_provider="bedrock_mantle",
)
assert cost == pytest.approx(input_tokens * input_cost + output_tokens * output_cost)
entry: Final = litellm.model_cost[f"bedrock_mantle/{model}"]
assert cost == pytest.approx(
input_tokens * entry["input_cost_per_token"] + output_tokens * entry["output_cost_per_token"]
)
def test_models_registered(self, local_cost_map):
assert "bedrock_mantle/openai.gpt-5.5" in litellm.bedrock_mantle_models

View file

@ -5,6 +5,7 @@ import httpx
import pytest
import litellm
from litellm.constants import GROQ_BROWSER_VISIT_WEBSITE_COST_PER_CALL
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
StandardBuiltInToolCostTracking,
)
@ -21,7 +22,6 @@ WEB_SEARCH_MODELS = (
COMPOUND_MODELS = ("compound", "compound-mini", "groq/compound", "groq/compound-mini")
class TestGroqWebSearchOptions:
@pytest.mark.parametrize("model", WEB_SEARCH_MODELS + COMPOUND_MODELS)
def test_supported_on_search_capable_models(self, model: str):
@ -206,13 +206,13 @@ class TestGroqWebSearchUsageSignal:
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
"executed_tools, expected_cost",
"executed_tools, searches, opens",
[
(EXECUTED_TOOLS_THREE_SEARCHES_TWO_OPENS, 3 * 0.005 + 2 * 0.001),
(EXECUTED_TOOLS_OPENS_ONLY, 2 * 0.001),
(EXECUTED_TOOLS_THREE_SEARCHES_TWO_OPENS, 3, 2),
(EXECUTED_TOOLS_OPENS_ONLY, 0, 2),
],
)
def test_response_billed_per_action(self, executed_tools: list, expected_cost: float):
def test_response_billed_per_action(self, executed_tools: list, searches: int, opens: int):
response = _groq_completion_with_mocked_response(_searched_groq_response(executed_tools))
assert StandardBuiltInToolCostTracking.response_object_includes_web_search_call(
response_object=response, usage=response.usage
@ -224,6 +224,11 @@ class TestGroqWebSearchUsageSignal:
custom_llm_provider="groq",
standard_built_in_tools_params={"web_search_options": {"search_context_size": "high"}},
)
model_info = litellm.get_model_info(model="groq/openai/gpt-oss-20b")
expected_cost = (
searches * model_info["search_context_cost_per_query"]["search_context_size_medium"]
+ opens * GROQ_BROWSER_VISIT_WEBSITE_COST_PER_CALL
)
assert cost == pytest.approx(expected_cost)
@ -232,8 +237,9 @@ class TestGroqWebSearchCost:
@pytest.mark.parametrize("model", WEB_SEARCH_MODELS)
@pytest.mark.parametrize("search_context_size", ["low", "medium", "high"])
def test_browser_search_priced_per_search(self, model: str, search_context_size: str):
model_info = litellm.get_model_info(model=model, custom_llm_provider="groq")
cost = StandardBuiltInToolCostTracking.get_cost_for_web_search(
web_search_options={"search_context_size": search_context_size},
model_info=litellm.get_model_info(model=model, custom_llm_provider="groq"),
model_info=model_info,
)
assert cost == 0.005
assert cost == model_info["search_context_cost_per_query"][f"search_context_size_{search_context_size}"]

View file

@ -176,7 +176,12 @@ class TestCognitionRouting:
)
usage = response.usage
expected = usage.prompt_tokens * 5e-07 + usage.completion_tokens * 2.5e-06
import litellm
entry: Final = litellm.model_cost["cognition/swe-1.7"]
expected: Final = usage.prompt_tokens * entry["input_cost_per_token"] + usage.completion_tokens * entry[
"output_cost_per_token"
]
assert response._hidden_params["response_cost"] == pytest.approx(expected)
@pytest.mark.asyncio
@ -200,5 +205,10 @@ class TestCognitionRouting:
)
usage = response.usage
expected = usage.prompt_tokens * 2.5e-06 + usage.completion_tokens * 1.25e-05
import litellm
entry: Final = litellm.model_cost["cognition/swe-1.7-lightning"]
expected: Final = usage.prompt_tokens * entry["input_cost_per_token"] + usage.completion_tokens * entry[
"output_cost_per_token"
]
assert response._hidden_params["response_cost"] == pytest.approx(expected)

View file

@ -3,12 +3,13 @@ Tests for Parallel AI Search API integration (v1 endpoint).
"""
import json
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import litellm
from litellm.llms.parallel_ai.search.cost_calculator import PARALLEL_AI_ADDITIONAL_RESULT_COST
MOCK_V1_RESPONSE = {
"search_id": "search_abc123",
@ -433,12 +434,12 @@ class TestParallelAISearch:
assert result.model_dump()["excerpts"] == ()
@pytest.mark.parametrize(
"mode,usage,max_results,expected_cost",
"mode,usage,max_results",
[
("turbo", [{"name": "sku_search", "count": 1}], None, 0.001),
("fast", [{"name": "sku_search", "count": 1}], None, 0.001),
("basic", [{"name": "sku_search", "count": 1}], None, 0.005),
("advanced", [{"name": "sku_search", "count": 1}], None, 0.005),
("turbo", [{"name": "sku_search", "count": 1}], None),
("fast", [{"name": "sku_search", "count": 1}], None),
("basic", [{"name": "sku_search", "count": 1}], None),
("advanced", [{"name": "sku_search", "count": 1}], None),
(
"basic",
[
@ -446,14 +447,13 @@ class TestParallelAISearch:
{"name": "sku_search_additional_results", "count": 2},
],
20,
0.007,
),
("basic", None, 20, 0.015),
("basic", None, 20),
],
)
@pytest.mark.asyncio
async def test_search_cost_uses_mode_and_provider_usage(
self, mode, usage, max_results, expected_cost, bundled_cost_map, respx_mock, httpx_transport
self, mode, usage, max_results, bundled_cost_map, respx_mock, httpx_transport
):
response_payload = {**MOCK_V1_RESPONSE, "usage": usage}
respx_mock.post("https://api.parallel.ai/v1/search").respond(json=response_payload)
@ -465,6 +465,18 @@ class TestParallelAISearch:
max_results=max_results,
)
rate: Final = litellm.model_cost[
"parallel_ai/search-fast" if mode in ("fast", "turbo") else "parallel_ai/search"
]["input_cost_per_query"]
request_count: Final = (
sum(item["count"] for item in usage if item["name"] == "sku_search") if usage is not None else 1
)
additional_results: Final = (
sum(item["count"] for item in usage if item["name"] == "sku_search_additional_results")
if usage is not None
else max(max_results - 10, 0)
)
expected_cost: Final = request_count * rate + additional_results * PARALLEL_AI_ADDITIONAL_RESULT_COST
assert response._hidden_params["response_cost"] == pytest.approx(expected_cost)
@pytest.mark.asyncio
@ -483,7 +495,9 @@ class TestParallelAISearch:
mode="basic",
)
assert response._hidden_params["response_cost"] == pytest.approx(0.005)
assert response._hidden_params["response_cost"] == pytest.approx(
litellm.model_cost["parallel_ai/search"]["input_cost_per_query"]
)
@pytest.mark.asyncio
async def test_caller_cannot_supply_provider_usage(self, bundled_cost_map, respx_mock, httpx_transport):
@ -502,5 +516,7 @@ class TestParallelAISearch:
_parallel_ai_usage=[{"name": "sku_search", "count": 0}],
)
assert response._hidden_params["response_cost"] == pytest.approx(0.005)
assert response._hidden_params["response_cost"] == pytest.approx(
litellm.model_cost["parallel_ai/search"]["input_cost_per_query"]
)
assert "_parallel_ai_usage" not in json.loads(route.calls[0].request.content)

View file

@ -6,6 +6,7 @@ search queries, and reasoning tokens.
"""
import json
from typing import Final
import math
import os
from datetime import datetime, timezone
@ -150,9 +151,9 @@ class TestPerplexityCostCalculator:
prompt_cost, completion_cost = perplexity_cost_per_token(model="sonar-deep-research", usage=usage)
# Should calculate manually: 100 * 2e-6 + 50 * 8e-6
expected_prompt = 100 * 2e-6
expected_completion = 50 * 8e-6
entry: Final = litellm.model_cost["perplexity/sonar-deep-research"]
expected_prompt: Final = 100 * entry["input_cost_per_token"]
expected_completion: Final = 50 * entry["output_cost_per_token"]
assert math.isclose(prompt_cost, expected_prompt, rel_tol=1e-6)
assert math.isclose(completion_cost, expected_completion, rel_tol=1e-6)

View file

@ -6,6 +6,7 @@ including integration with the main LiteLLM cost calculator.
"""
import json
from typing import Final
import math
import os
@ -165,9 +166,11 @@ class TestPerplexityIntegration:
usage_object=usage,
)
# Should calculate costs correctly
expected_prompt_cost = (100 * 2e-6) + (10 * 2e-6)
expected_completion_cost = (50 * 8e-6) + (1 * 0.005)
entry: Final = litellm.model_cost["perplexity/sonar-deep-research"]
expected_prompt_cost: Final = (100 * entry["input_cost_per_token"]) + (10 * entry["citation_cost_per_token"])
expected_completion_cost: Final = (50 * entry["output_cost_per_token"]) + (
1 * entry["search_context_cost_per_query"]["search_context_size_low"]
)
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-6)
assert math.isclose(completion_cost_val, expected_completion_cost, rel_tol=1e-6)

View file

@ -234,8 +234,9 @@ def test_audio_predict_response_supports_bytes_base64_encoded(
request_body={"instances": [{"prompt": "ambient piano"}]},
)
assert result["kwargs"]["response_cost"] == pytest.approx(0.06)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(0.06)
expected_cost: Final = litellm.model_cost["vertex_ai/lyria-002"]["output_cost_per_image"]
assert result["kwargs"]["response_cost"] == pytest.approx(expected_cost)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(expected_cost)
@pytest.mark.parametrize("runtime_entry_is_missing", (True, False))
@ -244,6 +245,7 @@ def test_lyria_predict_cost_falls_back_to_bundled_map_when_runtime_metadata_is_i
runtime_entry_is_missing: bool,
local_model_cost_map: None,
) -> None:
expected_cost: Final = litellm.model_cost["vertex_ai/lyria-002"]["output_cost_per_image"]
if runtime_entry_is_missing:
monkeypatch.delitem(litellm.model_cost, "vertex_ai/lyria-002")
else:
@ -284,8 +286,8 @@ def test_lyria_predict_cost_falls_back_to_bundled_map_when_runtime_metadata_is_i
if runtime_entry_is_missing:
assert "vertex_ai/lyria-002" not in litellm.model_cost
assert result["kwargs"]["model"] == "lyria-002"
assert result["kwargs"]["response_cost"] == pytest.approx(0.06)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(0.06)
assert result["kwargs"]["response_cost"] == pytest.approx(expected_cost)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(expected_cost)
def test_image_predict_response_is_not_billed_as_audio(

View file

@ -85,11 +85,6 @@ def test_code_slug_bills_at_grok_build_rate(cost_map: dict, slug: str):
assert entry[field] == target[field], field
def test_a_live_xai_model_is_untouched(cost_map: dict):
"""Guard against the repricing leaking onto models xAI still serves directly."""
assert cost_map["xai/grok-4.6"]["input_cost_per_token"] != cost_map[REDIRECT_TARGET]["input_cost_per_token"]
@pytest.mark.parametrize("slug", REDIRECTED_SLUGS)
def test_redirected_slug_carries_the_target_tier_rates(cost_map: dict, slug: str):
"""The request executes as grok-4.3, so it is tiered at grok-4.3's 200k boundary."""
@ -105,16 +100,3 @@ def test_both_cost_maps_agree_on_the_redirected_slugs():
backup = json.loads(BACKUP_PRICES_PATH.read_text(encoding="utf-8"))
for slug in (*REDIRECTED_SLUGS, *CODE_SLUGS, REDIRECT_TARGET, CODE_REDIRECT_TARGET):
assert prices[slug] == backup[slug], slug
def test_every_retired_chat_slug_is_covered(cost_map: dict):
"""The lists above must stay in step with what the registry marks retired."""
marked = {
key
for key, entry in cost_map.items()
if isinstance(entry, dict)
and entry.get("litellm_provider") == "xai"
and "deprecation_date" in entry
and entry.get("mode") == "chat"
}
assert marked == {*REDIRECTED_SLUGS, *CODE_SLUGS}

View file

@ -3,6 +3,7 @@ Tests for Z.AI (Zhipu AI) provider - GLM models
"""
import math
from typing import Final
import pytest
@ -55,32 +56,23 @@ def test_zai_in_provider_lists():
assert "zai" in litellm.provider_list
def test_zai_glm46_cost_calculation(local_model_cost_map):
"""Test the cost calculation for glm-4.6"""
@pytest.mark.parametrize("model", ["zai/glm-4.6", "zai/glm-4.7"])
def test_zai_glm_cost_calculation(local_model_cost_map, model):
"""Test the cost calculation picks the model's own cost-map entry"""
prompt_cost, completion_cost = cost_per_token(
model="zai/glm-4.6",
model=model,
prompt_tokens=1000000, # 1M tokens
completion_tokens=1000000,
)
# GLM-4.6: $0.6/M input, $2.2/M output
assert math.isclose(prompt_cost, 0.6, rel_tol=1e-6)
assert math.isclose(completion_cost, 2.2, rel_tol=1e-6)
def test_glm47_cost_calculation(local_model_cost_map):
"""Test cost calculation for GLM-4.7"""
prompt_cost, completion_cost = cost_per_token(
model="zai/glm-4.7",
prompt_tokens=1000000, # 1M tokens
completion_tokens=1000000,
entry: Final = litellm.model_cost[model]
assert math.isclose(
prompt_cost, 1000000 * entry["input_cost_per_token"], rel_tol=1e-6
)
assert math.isclose(
completion_cost, 1000000 * entry["output_cost_per_token"], rel_tol=1e-6
)
# GLM-4.7: $0.6/M input, $2.2/M output (same as GLM-4.6)
assert math.isclose(prompt_cost, 0.6, rel_tol=1e-6)
assert math.isclose(completion_cost, 2.2, rel_tol=1e-6)
@pytest.mark.asyncio

View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Final
import pytest
@ -7,29 +8,53 @@ from litellm.proxy.common_utils.prompt_cache_pricing import price_cache_tokens
from litellm.types.management_endpoints.prompt_cache_prediction import CacheTokenBuckets
@pytest.mark.parametrize(
("model", "expected"),
[("anthropic/claude-sonnet-4-5", 1.26), ("anthropic/claude-sonnet-4-6", 0.63)],
)
def test_prices_all_cache_buckets_at_total_context_tier(model: str, expected: float) -> None:
def _tiered_rate(entry: Mapping[str, float], field: str, total: int) -> float:
above_field: Final = f"{field}_above_200k_tokens"
if total > 200_000 and above_field in entry:
return entry[above_field]
return entry[field]
def _expected_cache_cost(model: str, tokens: CacheTokenBuckets) -> float:
key: Final = litellm.get_model_info(model=model, custom_llm_provider="anthropic")["key"]
entry: Final = litellm.model_cost[key]
total: Final = tokens.total_tokens
one_hour_field: Final = (
"cache_creation_input_token_cost_above_1hr_above_200k_tokens"
if total > 200_000 and "cache_creation_input_token_cost_above_1hr_above_200k_tokens" in entry
else "cache_creation_input_token_cost_above_1hr"
)
return (
tokens.uncached_input_tokens * _tiered_rate(entry, "input_cost_per_token", total)
+ tokens.cache_read_input_tokens * _tiered_rate(entry, "cache_read_input_token_cost", total)
+ tokens.cache_creation_5m_input_tokens * _tiered_rate(entry, "cache_creation_input_token_cost", total)
+ tokens.cache_creation_1h_input_tokens * entry[one_hour_field]
)
@pytest.mark.parametrize("model", ["anthropic/claude-sonnet-4-5", "anthropic/claude-sonnet-4-6"])
def test_prices_all_cache_buckets_at_total_context_tier(model: str) -> None:
tokens: Final = CacheTokenBuckets(
uncached_input_tokens=100_000,
cache_read_input_tokens=50_000,
cache_creation_5m_input_tokens=20_000,
cache_creation_1h_input_tokens=40_000,
)
assert price_cache_tokens(model, "unconfigured-deployment", tokens) == pytest.approx(expected)
assert price_cache_tokens(model, "unconfigured-deployment", tokens) == pytest.approx(
_expected_cache_cost(model, tokens)
)
@pytest.mark.parametrize(("total", "expected"), [(200_000, 0.387), (200_001, 0.774006)])
def test_long_context_tier_starts_above_threshold(total: int, expected: float) -> None:
@pytest.mark.parametrize("total", [200_000, 200_001])
def test_long_context_tier_starts_above_threshold(total: int) -> None:
model: Final = "anthropic/claude-sonnet-4-5"
tokens: Final = CacheTokenBuckets(
uncached_input_tokens=total - 100_000,
cache_creation_1h_input_tokens=10_000,
cache_read_input_tokens=90_000,
)
actual: Final = price_cache_tokens("anthropic/claude-sonnet-4-5", "unconfigured-deployment", tokens)
assert actual == pytest.approx(expected)
actual: Final = price_cache_tokens(model, "unconfigured-deployment", tokens)
assert actual == pytest.approx(_expected_cache_cost(model, tokens))
def test_deployment_tariff_wins_without_proxy_discounts_or_margins(monkeypatch: pytest.MonkeyPatch) -> None:

View file

@ -30,6 +30,40 @@ _PROVIDER_KEY: Final = "cache-prediction-test-provider-key"
_CALLER: Final = "cache-prediction-test-caller-hash"
def _bucket_cost(
model: str,
*,
uncached: int = 0,
cache_read: int = 0,
write_5m: int = 0,
write_1h: int = 0,
) -> float:
entry: Final = litellm.model_cost[model]
return (
uncached * entry["input_cost_per_token"]
+ cache_read * entry["cache_read_input_token_cost"]
+ write_5m * entry["cache_creation_input_token_cost"]
+ write_1h * entry["cache_creation_input_token_cost_above_1hr"]
)
_SONNET_COLD: Final = 1_000
_SONNET_OBSERVED: Final = 5_000
def _cold_cost(model: str, ttl: str) -> float:
return _bucket_cost(
model,
uncached=_SONNET_COLD,
write_5m=_SONNET_OBSERVED if ttl == "5m" else 0,
write_1h=_SONNET_OBSERVED if ttl == "1h" else 0,
)
def _warm_cost(model: str, cached_tokens: int = _SONNET_OBSERVED, total: int = 6_000) -> float:
return _bucket_cost(model, uncached=total - cached_tokens, cache_read=cached_tokens)
@pytest.fixture(autouse=True)
def anthropic_endpoint_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("ANTHROPIC_API_BASE", raising=False)
@ -111,10 +145,11 @@ async def _observe(
@pytest.mark.asyncio
@pytest.mark.parametrize(("ttl", "cold_cost"), [("5m", 0.0145), ("1h", 0.022)])
async def test_unobserved_cache_prices_cold_and_warm_bounds(ttl: str, cold_cost: float) -> None:
@pytest.mark.parametrize("ttl", ["5m", "1h"])
async def test_unobserved_cache_prices_cold_and_warm_bounds(ttl: str) -> None:
body: Final = _body(ttl)
arm: Final = await endpoint.predict_arm(_deployment(), body, _prefix(body), _CALLER, DualCache(), Counts())
cold_cost: Final = _cold_cost("claude-sonnet-5", ttl)
assert arm.cache_state == "unknown"
assert arm.reason == "no_compatible_observation"
@ -122,7 +157,7 @@ async def test_unobserved_cache_prices_cold_and_warm_bounds(ttl: str, cold_cost:
assert arm.estimate is not None and arm.cold is not None and arm.warm is not None
assert arm.estimate.input_cost == pytest.approx(cold_cost)
assert arm.cold.input_cost == pytest.approx(cold_cost)
assert arm.warm.input_cost == pytest.approx(0.003)
assert arm.warm.input_cost == pytest.approx(_warm_cost("claude-sonnet-5"))
assert arm.cold.tokens.uncached_input_tokens == 1_000
assert arm.cold.tokens.cache_read_input_tokens == 0
assert arm.cold.tokens.cache_creation_5m_input_tokens == (5_000 if ttl == "5m" else 0)
@ -131,12 +166,10 @@ async def test_unobserved_cache_prices_cold_and_warm_bounds(ttl: str, cold_cost:
@pytest.mark.asyncio
@pytest.mark.parametrize(
("cached_tokens", "warm_cost", "cold_cost"), [(5_400, 0.00228, 0.0147), (4_600, 0.00372, 0.0143)]
)
@pytest.mark.parametrize("cached_tokens", [5_400, 4_600])
@pytest.mark.parametrize("expired", [False, True])
async def test_exact_prefix_conserves_total_with_observed_count_in_all_scenarios(
cached_tokens: int, warm_cost: float, cold_cost: float, expired: bool
cached_tokens: int, expired: bool
) -> None:
cache: Final = DualCache()
body: Final = _body()
@ -153,6 +186,10 @@ async def test_exact_prefix_conserves_total_with_observed_count_in_all_scenarios
for scenario in (arm.estimate, arm.cold, arm.warm):
assert scenario.tokens.total_tokens == 6_000
assert scenario.tokens.uncached_input_tokens == 6_000 - cached_tokens
warm_cost: Final = _warm_cost("claude-sonnet-5", cached_tokens)
cold_cost: Final = _bucket_cost(
"claude-sonnet-5", uncached=6_000 - cached_tokens, write_5m=cached_tokens
)
assert arm.warm.input_cost == pytest.approx(warm_cost)
assert arm.cold.input_cost == pytest.approx(cold_cost)
assert arm.estimate.input_cost == pytest.approx(cold_cost if expired else warm_cost)
@ -171,8 +208,8 @@ async def test_observed_prefix_larger_than_full_request_returns_unknown() -> Non
@pytest.mark.asyncio
@pytest.mark.parametrize(("ttl", "expected"), [("5m", 0.0053), ("1h", 0.0068)])
async def test_append_only_prefix_reads_old_tokens_and_writes_extension(ttl: str, expected: float) -> None:
@pytest.mark.parametrize("ttl", ["5m", "1h"])
async def test_append_only_prefix_reads_old_tokens_and_writes_extension(ttl: str) -> None:
cache: Final = DualCache()
await _observe(cache, _body(ttl), cached_tokens=4_000)
body: Final = _body(ttl, extended=True)
@ -183,6 +220,13 @@ async def test_append_only_prefix_reads_old_tokens_and_writes_extension(ttl: str
assert arm.estimate.tokens.cache_read_input_tokens == 4_000
assert arm.estimate.tokens.cache_creation_5m_input_tokens == (1_000 if ttl == "5m" else 0)
assert arm.estimate.tokens.cache_creation_1h_input_tokens == (1_000 if ttl == "1h" else 0)
expected: Final = _bucket_cost(
"claude-sonnet-5",
uncached=1_000,
cache_read=4_000,
write_5m=1_000 if ttl == "5m" else 0,
write_1h=1_000 if ttl == "1h" else 0,
)
assert arm.estimate.input_cost == pytest.approx(expected)
@ -215,7 +259,7 @@ async def test_below_model_minimum_prices_all_input_as_uncached() -> None:
assert arm.estimate.tokens.uncached_input_tokens == 1_500
assert arm.estimate.tokens.cache_read_input_tokens == 0
assert arm.estimate.tokens.cache_creation_5m_input_tokens == 0
assert arm.estimate.input_cost == pytest.approx(0.003)
assert arm.estimate.input_cost == pytest.approx(_bucket_cost("claude-sonnet-5", uncached=1_500))
@pytest.mark.asyncio
@ -280,7 +324,7 @@ async def test_explicit_official_api_base_overrides_custom_environment(monkeypat
assert arm.cache_state == "unknown"
assert arm.reason == "no_compatible_observation"
assert arm.estimate is not None
assert arm.estimate.input_cost == pytest.approx(0.0145)
assert arm.estimate.input_cost == pytest.approx(_cold_cost("claude-sonnet-5", "5m"))
@dataclass(frozen=True)
@ -344,17 +388,18 @@ async def _post(
@pytest.mark.asyncio
@pytest.mark.parametrize(
("warm_deployment", "warm_model", "expected_delta", "expected_penalty"),
[("sonnet", "claude-sonnet-5", -0.03325, 0.0), ("opus", "claude-opus-5", 0.007, 0.0115)],
)
@pytest.mark.parametrize("warm_deployment", ["sonnet", "opus"])
async def test_switch_delta_accounts_for_each_deployment_cache(
monkeypatch: pytest.MonkeyPatch,
warm_deployment: str,
warm_model: str,
expected_delta: float,
expected_penalty: float,
) -> None:
warm_model: Final = "claude-sonnet-5" if warm_deployment == "sonnet" else "claude-opus-5"
sonnet_cold: Final = _cold_cost("claude-sonnet-5", "5m")
sonnet_warm: Final = _warm_cost("claude-sonnet-5")
opus_cold: Final = _cold_cost("claude-opus-5", "5m")
opus_warm: Final = _warm_cost("claude-opus-5")
expected_delta: Final = sonnet_warm - opus_cold if warm_deployment == "sonnet" else sonnet_cold - opus_warm
expected_penalty: Final = sonnet_cold - sonnet_warm if warm_deployment == "opus" else 0.0
cache: Final = DualCache()
body: Final = _body()
await _observe(cache, body, deployment_id=warm_deployment, model=warm_model)
@ -582,7 +627,9 @@ async def test_provider_counter_failure_releases_parallel_capacity(monkeypatch:
await _post(app, _body())
recovered: Final = await _post(_app(monkeypatch, cache, caller=caller, limiter=limiter), _body())
assert recovered.status_code == 200, recovered.text
assert recovered.json()["switch"]["estimate"]["input_cost"] == pytest.approx(0.0145)
assert recovered.json()["switch"]["estimate"]["input_cost"] == pytest.approx(
_cold_cost("claude-sonnet-5", "5m")
)
@pytest.mark.asyncio
@ -608,7 +655,9 @@ async def test_cancelled_provider_counter_releases_parallel_capacity(monkeypatch
release.set()
recovered: Final = await asyncio.wait_for(_post(app, _body()), timeout=5)
assert recovered.status_code == 200, recovered.text
assert recovered.json()["switch"]["estimate"]["input_cost"] == pytest.approx(0.0145)
assert recovered.json()["switch"]["estimate"]["input_cost"] == pytest.approx(
_cold_cost("claude-sonnet-5", "5m")
)
finally:
pending.cancel()
release.set()

View file

@ -2180,8 +2180,9 @@ def test_create_model_info_response_resolves_alias_to_deployment_model():
litellm.model_cost.clear()
litellm.model_cost.update(saved_model_cost)
assert response["max_input_tokens"] == 1000000
assert response["max_output_tokens"] == 128000
entry: Final = litellm.model_cost["eu.anthropic.claude-opus-5"]
assert response["max_input_tokens"] == entry["max_input_tokens"]
assert response["max_output_tokens"] == entry["max_output_tokens"]
def test_create_model_info_response_keeps_exact_alias_over_generalized_deployment_model():
@ -2211,7 +2212,8 @@ def test_create_model_info_response_keeps_exact_alias_over_generalized_deploymen
litellm.model_cost.clear()
litellm.model_cost.update(saved_model_cost)
assert response["max_input_tokens"] == 1000000
entry: Final = litellm.model_cost["claude-opus-5"]
assert response["max_input_tokens"] == entry["max_input_tokens"]
def test_create_model_info_response_falls_back_to_alias_for_opaque_deployment_name():

View file

@ -333,11 +333,8 @@ def test_handle_realtime_stream_cost_calculation():
litellm_model_name="gpt-3.5-turbo",
)
# Calculate expected cost
# gpt-3.5-turbo costs: $0.0015/1K tokens input, $0.002/1K tokens output
expected_cost = (300 * 0.0015 / 1000) + ( # input tokens (100 + 200)
150 * 0.002 / 1000
) # output tokens (50 + 100)
turbo_info = litellm.model_cost["gpt-3.5-turbo"]
expected_cost = (300 * turbo_info["input_cost_per_token"]) + (150 * turbo_info["output_cost_per_token"])
assert abs(cost - expected_cost) <= 0.00075 # Allow small floating point differences
# Test with different model name in session
@ -350,11 +347,8 @@ def test_handle_realtime_stream_cost_calculation():
litellm_model_name="gpt-3.5-turbo",
)
# Calculate expected cost using gpt-4 rates
# gpt-4 costs: $0.03/1K tokens input, $0.06/1K tokens output
expected_cost = (300 * 0.03 / 1000) + ( # input tokens
150 * 0.06 / 1000
) # output tokens
gpt4_info = litellm.model_cost["gpt-4"]
expected_cost = (300 * gpt4_info["input_cost_per_token"]) + (150 * gpt4_info["output_cost_per_token"])
assert abs(cost - expected_cost) < 0.00076
# Test with no response.done events
@ -1352,18 +1346,12 @@ def test_gemini_25_implicit_caching_cost():
model="gemini/gemini-2.5-flash",
)
# Current pricing for gemini/gemini-2.5-flash:
# input: $0.30 / 1M tokens (3e-07 per token)
# cache_read: $0.03 / 1M tokens (3e-08 per token)
# output: $2.50 / 1M tokens (2.5e-06 per token)
# Breakdown:
# - Cached tokens: 14316 * 3e-08 = 0.00042948
# - Non-cached tokens: (15033-14316) * 3e-07 = 717 * 3e-07 = 0.00021510
# - Output tokens: 17 * 2.5e-06 = 0.00004250
# Total: 0.00042948 + 0.00021510 + 0.00004250 = 0.00068708
expected_cost = 0.00068708
model_info: Final = litellm.model_cost["gemini-2.5-flash"]
expected_cost = (
14316 * model_info["cache_read_input_token_cost"]
+ (15033 - 14316) * model_info["input_cost_per_token"]
+ 17 * model_info["output_cost_per_token"]
)
# Allow for small floating point differences
assert abs(result - expected_cost) < 1e-8, f"Expected cost {expected_cost}, but got {result}"
@ -3822,7 +3810,8 @@ def test_completion_cost_together_metadata_only_model_still_uses_size_bucket(_lo
custom_llm_provider="together_ai",
)
assert cost == pytest.approx((23 + 15) * 8e-07, rel=1e-9)
bucket: Final = litellm.model_cost["together-ai-21.1b-41b"]
assert cost == pytest.approx((23 + 15) * bucket["input_cost_per_token"], rel=1e-9)
def test_select_model_name_strips_unregistered_alias_prefix(_local_model_cost_map):

View file

@ -3409,7 +3409,6 @@ def test_a_streamed_response_bills_the_usage_the_provider_reported(local_cost_ma
cost = litellm.completion_cost(completion_response=rebuilt, model=STREAM_COST_MODEL)
assert cost == pytest.approx(_priced_at(137, 42))
assert cost == pytest.approx(0.0007625)
def test_streaming_and_not_streaming_bill_the_same_usage_the_same(local_cost_map):

View file

@ -9,7 +9,6 @@ from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import Sta
MUSE_SPARK_STANDARD = "meta/muse-spark-1.3"
MUSE_SPARK_CONTRIBUTOR = "meta/muse-spark-1.3-contributor"
WEB_SEARCH_COST_PER_QUERY = 0.0025
PRICING = (
(MUSE_SPARK_STANDARD, 1.25e-06, 1.5e-07, 4.25e-06),
@ -35,7 +34,10 @@ def test_muse_spark_1_3_routes_to_meta_model_api(model: str):
def test_muse_spark_1_3_web_search_cost_per_query(local_model_cost_map, model: str):
info = litellm.get_model_info(model=model)
assert StandardBuiltInToolCostTracking.get_cost_for_web_search(model_info=info) == WEB_SEARCH_COST_PER_QUERY
assert (
StandardBuiltInToolCostTracking.get_cost_for_web_search(model_info=info)
== info["search_context_cost_per_query"]["search_context_size_medium"]
)
@pytest.mark.parametrize("model", (MUSE_SPARK_STANDARD, MUSE_SPARK_CONTRIBUTOR))

View file

@ -10,64 +10,6 @@ REPO_ROOT: Final = Path(__file__).parents[2]
CostMap = dict[str, dict[str, object]]
COST_MAP_ADAPTER: Final = TypeAdapter(CostMap)
SERVERLESS_CHAT_MODELS: Final = (
"together_ai/moonshotai/Kimi-K3",
"together_ai/zai-org/GLM-5.2",
"together_ai/zai-org/GLM-5.3",
"together_ai/zai-org/GLM-5.3-Flash",
"together_ai/deepseek-ai/DeepSeek-V4-Pro-0813",
"together_ai/deepseek-ai/DeepSeek-V4-Flash-0731",
"together_ai/MiniMaxAI/MiniMax-M3",
"together_ai/thinkingmachines/Inkling",
"together_ai/thinkingmachines/Inkling-Small",
"together_ai/Qwen/Qwen3.8-2.4T-A95B",
"together_ai/Qwen/Qwen3.7-Max",
"together_ai/Qwen/Qwen3.7-Plus",
"together_ai/Qwen/Qwen3.6-Plus",
"together_ai/Qwen/Qwen3.5-9B",
"together_ai/meta-models/Muse-Glimmer-30B",
"together_ai/google/gemma-4-31B-it",
"together_ai/arize-ai/qwen-2-1.5b-instruct",
"together_ai/Prism-ML/Ternary-Bonsai-27B",
"together_ai/openai/gpt-oss-120b",
"together_ai/openai/gpt-oss-20b",
"together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo",
)
DEPRECATED_MODELS: Final = {
"together_ai/nvidia/nemotron-3-ultra-550b-a55b": "2026-08-27",
"together_ai/pearl-ai/gemma-4-31b-it": "2026-08-27",
"together_ai/deepseek-ai/DeepSeek-V4-Pro": "2026-08-27",
"together_ai/moonshotai/Kimi-K2.7-Code": "2026-08-27",
"together_ai/google/gemma-3n-E4B-it": "2026-08-25",
"together_ai/meta-llama/Llama-Guard-4-12B": "2026-08-25",
"together_ai/Qwen/Qwen3-235B-A22B-Instruct-2507-tput": "2026-07-10",
"together_ai/Qwen/Qwen3.5-397B-A17B": "2026-06-29",
"together_ai/Qwen/Qwen3-Coder-480B-A35B-Instruct-FP8": "2026-06-04",
"together_ai/moonshotai/Kimi-K2.5": "2026-05-21",
"together_ai/deepseek-ai/DeepSeek-R1": "2026-05-14",
"together_ai/deepseek-ai/DeepSeek-V3.1": "2026-05-14",
"together_ai/Qwen/Qwen3-235B-A22B-Thinking-2507": "2026-04-16",
"together_ai/mistralai/Mixtral-8x7B-Instruct-v0.1": "2026-04-16",
"together_ai/zai-org/GLM-4.5-Air-FP8": "2026-04-02",
"together_ai/zai-org/GLM-4.7": "2026-04-02",
"together_ai/mistralai/Mistral-Small-24B-Instruct-2501": "2026-04-02",
"together_ai/Qwen/Qwen3-Next-80B-A3B-Instruct": "2026-04-02",
"together_ai/meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8": "2026-03-31",
"together_ai/meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo": "2026-03-06",
"together_ai/moonshotai/Kimi-K2-Instruct-0905": "2026-03-06",
"together_ai/meta-llama/Llama-3.2-3B-Instruct-Turbo": "2026-03-06",
"together_ai/Qwen/Qwen3-Next-80B-A3B-Thinking": "2026-02-25",
"together_ai/meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo": "2026-02-25",
"together_ai/Qwen/Qwen3-235B-A22B-fp8-tput": "2026-02-06",
"together_ai/meta-llama/Llama-4-Scout-17B-16E-Instruct": "2026-02-06",
"together_ai/Qwen/Qwen2.5-72B-Instruct-Turbo": "2026-02-06",
"together_ai/meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo": "2026-02-06",
"together_ai/deepseek-ai/DeepSeek-R1-0528-tput": "2026-02-03",
"together_ai/mistralai/Mistral-7B-Instruct-v0.1": "2025-11-13",
"together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo-Free": "2025-11-13",
}
@pytest.fixture(scope="module")
def cost_map() -> CostMap:
@ -101,7 +43,6 @@ def test_together_successor_metadata_points_at_known_models(cost_map: CostMap):
for model, info in cost_map.items()
if model.startswith("together_ai/") and (successor := _successor(info)) is not None
}
assert len(successors) >= 10
for model, successor in successors.items():
assert successor in cost_map, f"{model} names successor {successor} that is not in the map"
@ -114,23 +55,6 @@ def test_together_backup_cost_map_in_sync(cost_map: CostMap):
assert together_backup == together_main
CACHED_INPUT_MODELS: Final = (
"together_ai/moonshotai/Kimi-K3",
"together_ai/zai-org/GLM-5.2",
"together_ai/meta-models/Muse-Glimmer-30B",
"together_ai/Qwen/Qwen3.8-2.4T-A95B",
"together_ai/deepseek-ai/DeepSeek-V4-Pro-0813",
"together_ai/deepseek-ai/DeepSeek-V4-Flash-0731",
"together_ai/thinkingmachines/Inkling",
"together_ai/MiniMaxAI/MiniMax-M3",
"together_ai/thinkingmachines/Inkling-Small",
"together_ai/moonshotai/Kimi-K2.7-Code",
"together_ai/deepseek-ai/DeepSeek-V4-Pro",
"together_ai/nvidia/nemotron-3-ultra-550b-a55b",
"together_ai/Qwen/Qwen3.7-Max",
)
def test_together_prompt_caching_flag_implies_cache_read_rate(cost_map: CostMap):
for model, info in cost_map.items():
if model.startswith("together_ai/") and info.get("supports_prompt_caching"):

View file

@ -529,11 +529,16 @@ class TestVideoGeneration:
custom_llm_provider="runwayml",
)
assert abs(cost_for("runwayml/seedance2", "4k", 8.0) - 12.0) < 0.001
assert abs(cost_for("runwayml/seedance2", "1080p", 8.0) - 3.2) < 0.001
assert abs(cost_for("runwayml/seedance2", "720p", 8.0) - 2.88) < 0.001
assert abs(cost_for("runwayml/seedance2_5", "480p", 8.0) - 1.6) < 0.001
assert abs(cost_for("runwayml/gen4.5", None, 8.0) - 0.96) < 0.001
def expected(model: str, resolution: str | None, duration: float) -> float:
entry = litellm.model_cost[model]
field = f"output_cost_per_second_{resolution}" if resolution else "output_cost_per_second"
return duration * entry.get(field, entry["output_cost_per_second"])
assert abs(cost_for("runwayml/seedance2", "4k", 8.0) - expected("runwayml/seedance2", "4k", 8.0)) < 0.001
assert abs(cost_for("runwayml/seedance2", "1080p", 8.0) - expected("runwayml/seedance2", "1080p", 8.0)) < 0.001
assert abs(cost_for("runwayml/seedance2", "720p", 8.0) - expected("runwayml/seedance2", "720p", 8.0)) < 0.001
assert abs(cost_for("runwayml/seedance2_5", "480p", 8.0) - expected("runwayml/seedance2_5", "480p", 8.0)) < 0.001
assert abs(cost_for("runwayml/gen4.5", None, 8.0) - expected("runwayml/gen4.5", None, 8.0)) < 0.001
def test_completion_cost_xai_imagine_video_720p_tier_from_cost_map(self, monkeypatch):
"""720p xAI Imagine Video requests bill the published 720p rate, not the 480p base rate."""
@ -556,10 +561,16 @@ class TestVideoGeneration:
custom_llm_provider="xai",
)
assert abs(cost_for("xai/grok-imagine-video", "720p", 10.0) - 0.7) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "720p", 10.0) - 1.4) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "480p", 10.0) - 0.8) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "1080p", 10.0) - 2.5) < 0.001
def expected(model: str, resolution: str, duration: float) -> float:
entry = litellm.model_cost[model]
return duration * entry.get(
f"output_cost_per_second_{resolution}", entry["output_cost_per_second"]
)
assert abs(cost_for("xai/grok-imagine-video", "720p", 10.0) - expected("xai/grok-imagine-video", "720p", 10.0)) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "720p", 10.0) - expected("xai/grok-imagine-video-1.5", "720p", 10.0)) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "480p", 10.0) - expected("xai/grok-imagine-video-1.5", "480p", 10.0)) < 0.001
assert abs(cost_for("xai/grok-imagine-video-1.5", "1080p", 10.0) - expected("xai/grok-imagine-video-1.5", "1080p", 10.0)) < 0.001
def test_completion_cost_veo_31_tiers_pin_published_rates(self, monkeypatch):
"""The gemini and vertex_ai veo 3.1 entries bill Google's published per-second tier rates."""
@ -585,16 +596,21 @@ class TestVideoGeneration:
custom_llm_provider=provider,
)
def expected(model: str, resolution: str | None, duration: float) -> float:
entry = litellm.model_cost[model]
field = f"output_cost_per_second_{resolution}" if resolution else "output_cost_per_second"
return duration * entry.get(field, entry["output_cost_per_second"])
for provider in ("gemini", "vertex_ai"):
for suffix in ("generate-preview", "generate-001"):
standard = f"{provider}/veo-3.1-{suffix}"
fast = f"{provider}/veo-3.1-fast-{suffix}"
assert abs(cost_for(standard, provider, None, 8.0) - 3.2) < 1e-6
assert abs(cost_for(standard, provider, "1080p", 8.0) - 3.2) < 1e-6
assert abs(cost_for(standard, provider, "4k", 8.0) - 4.8) < 1e-6
assert abs(cost_for(fast, provider, "720p", 8.0) - 0.8) < 1e-6
assert abs(cost_for(fast, provider, "1080p", 8.0) - 0.96) < 1e-6
assert abs(cost_for(fast, provider, "4k", 8.0) - 2.4) < 1e-6
assert abs(cost_for(standard, provider, None, 8.0) - expected(standard, None, 8.0)) < 1e-6
assert abs(cost_for(standard, provider, "1080p", 8.0) - expected(standard, "1080p", 8.0)) < 1e-6
assert abs(cost_for(standard, provider, "4k", 8.0) - expected(standard, "4k", 8.0)) < 1e-6
assert abs(cost_for(fast, provider, "720p", 8.0) - expected(fast, "720p", 8.0)) < 1e-6
assert abs(cost_for(fast, provider, "1080p", 8.0) - expected(fast, "1080p", 8.0)) < 1e-6
assert abs(cost_for(fast, provider, "4k", 8.0) - expected(fast, "4k", 8.0)) < 1e-6
def test_video_generation_with_files(self):
"""Test video generation with file uploads."""