scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-21 00:21:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 117

test(integration): drive cost tracking from literal request/response data

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-19 04:04:40 +00:00
parent 57d2fefa8d
commit ed0c32cdb0
12 changed files with 26495 additions and 6081 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/integration/README.md
View file

@ -2,7 +2,7 @@
These tests exercise a running gateway, PostgreSQL and Redis with an owned local upstream. CircleCI owns this suite. Tests are grouped by behavior, with no automatic test retries or fallback to paid provider calls
The `cost` group runs the scripted-shape cost matrix through the shared integration upstream, which serves the test-owned cost map over loopback through `LITELLM_MODEL_COST_MAP_URL`; cost goldens are checked into the integration suite and must not be copied into the E2E coverage registry. The upstream renders a scenario in the shape LiteLLM's own provider config resolves to for the deployment, so a provider LiteLLM already parses with one of the five rendered families is a cost-map entry plus a `cases.json` `providers` row with its deployment parameters; a provider whose config class is none of those families fails at collection until `scripted_shapes.py` gains a renderer
The `cost` group is driven by `cost_tracking_cases.json`, which contains the cost map, literal requests, literal provider responses and expected accounting values. Each case has a name, contract ID, cost-map model, optional deployment overrides, request body, tagged response and exact or recount expectations. Request bodies use `$MODEL` for the registered proxy model, while responses use `$REQUEST_ID` for the per-run scenario ID. To add a case, add a cost-map entry when the model is new, add the request body and exact provider response data, add hand-computed expected values and register the node ID in `contracts.json`. The upstream serves each stored response for any path under `/<scenario_id>`, while the test-owned cost map is served over loopback through `LITELLM_MODEL_COST_MAP_URL`
Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions`, `sdk` or `cost` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate

1364
tests/integration/_support/scripted_shapes.py
View file

File diff suppressed because it is too large Load diff

172
tests/integration/_support/upstream.py
View file

@ -2,32 +2,34 @@ from __future__ import annotations
import argparse
from collections import deque
from collections.abc import Mapping
import json
from dataclasses import dataclass, field
import os
from pathlib import Path
from queue import SimpleQueue
import struct
from typing import Final, cast
import zlib
import httpx
import uvicorn
from pydantic import JsonValue, TypeAdapter, ValidationError
from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import Route
from _fake_openai_endpoint_server import chat_completions, completions, embeddings, health, moderations
from integration._support.scripted_shapes import (
RenderedResponse,
Scenario,
ScenarioDeleted,
ScenarioRegistered,
ScenarioStore,
render,
from integration.cost_calculation.cost_tracking_case import (
EventStreamResponse,
JsonResponse,
SseResponse,
StoredResponse,
)
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
CASES_FILE: Final = Path(__file__).resolve().parents[1] / "cost_calculation" / "cost_tracking_cases.json"
INTERNAL_FIELDS: Final = frozenset(
{
"litellm_params",
@ -56,6 +58,53 @@ class Observation:
body: dict[str, JsonValue]
class _ScenarioRegistration(BaseModel):
scenario_id: str
response: StoredResponse
def _aws_str_header(name: str, value: str) -> bytes:
name_bytes: Final = name.encode()
value_bytes: Final = value.encode()
return (
struct.pack("!B", len(name_bytes))
+ name_bytes
+ struct.pack("!B", 7)
+ struct.pack("!H", len(value_bytes))
+ value_bytes
)
def _aws_event_frame(event_type: str, payload: Mapping[str, JsonValue], scenario_id: str) -> bytes:
payload_bytes: Final = json.dumps(payload, separators=(",", ":")).replace(
"$REQUEST_ID", scenario_id
).encode()
headers_bytes: Final = (
_aws_str_header(":event-type", event_type)
+ _aws_str_header(":content-type", "application/json")
+ _aws_str_header(":message-type", "event")
)
total_length: Final = 12 + len(headers_bytes) + len(payload_bytes) + 4
prelude: Final = struct.pack("!II", total_length, len(headers_bytes))
prelude_crc: Final = struct.pack("!I", zlib.crc32(prelude) & 0xFFFFFFFF)
message: Final = prelude + prelude_crc + headers_bytes + payload_bytes
return message + struct.pack("!I", zlib.crc32(message) & 0xFFFFFFFF)
class ScenarioStore:
def __init__(self) -> None:
self._scenarios: dict[str, StoredResponse] = {}
def put(self, scenario_id: str, response: StoredResponse) -> None:
self._scenarios[scenario_id] = response
def drop(self, scenario_id: str) -> bool:
return self._scenarios.pop(scenario_id, None) is not None
def get(self, scenario_id: str) -> StoredResponse | None:
return self._scenarios.get(scenario_id)
@dataclass(frozen=True, slots=True)
class Provider:
observations: SimpleQueue[Observation] = field(default_factory=SimpleQueue)
@ -91,7 +140,7 @@ class Provider:
return await chat_completions(request)
async def script(self, request: Request) -> Response:
name: Final = request.path_params["model"]
name: Final = cast(str, request.path_params["model"])
if request.method in {"DELETE", "GET"} and name not in self.scripts:
return JSONResponse({"error": "Script not found"}, status_code=404)
if request.method == "GET":
@ -118,71 +167,60 @@ class Provider:
async def register_scenario(self, request: Request) -> Response:
try:
scenario: Final = Scenario.model_validate_json(await request.body())
registration: Final = _ScenarioRegistration.model_validate_json(await request.body())
except ValidationError as exc:
return self._render(
RenderedResponse(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
)
self.scenario_store.put(scenario)
return self._render(
RenderedResponse(
200,
"application/json",
json.dumps({"scenario_id": scenario.scenario_id}).encode("utf-8"),
)
)
return JSONResponse({"error": str(exc)}, status_code=400)
self.scenario_store.put(registration.scenario_id, registration.response)
return JSONResponse({"scenario_id": registration.scenario_id})
async def delete_scenario(self, request: Request) -> Response:
scenario_id: Final = cast(str, request.path_params["scenario_id"])
deleted: Final = self.scenario_store.drop(scenario_id)
return self._render(
RenderedResponse(
200 if deleted else 404,
"application/json",
json.dumps({"deleted": deleted}).encode("utf-8"),
)
)
return JSONResponse({"deleted": deleted}, status_code=200 if deleted else 404)
async def cost_map(self, _request: Request) -> Response:
return self._render(
RenderedResponse(
200,
"application/json",
(Path(__file__).resolve().parents[1] / "cost_calculation" / "cost_map.json").read_bytes(),
)
)
cases_file: Final = JSON_OBJECT.validate_json(CASES_FILE.read_bytes())
return JSONResponse(cases_file["cost_map"])
async def oauth_token(self, _request: Request) -> Response:
return self._render(
RenderedResponse(
200,
"application/json",
json.dumps(
{
"access_token": "scripted-token",
"token_type": "Bearer",
"expires_in": 3600,
}
).encode("utf-8"),
)
return JSONResponse(
{
"access_token": "scripted-token",
"token_type": "Bearer",
"expires_in": 3600,
}
)
async def scripted(self, request: Request) -> Response:
rendered: Final = render(
self.scenario_store,
request.method,
request.url.path,
await request.body(),
)
return self._render(rendered)
segments: Final = tuple(segment for segment in cast(str, request.path_params["path"]).split("/") if segment)
if not segments:
return JSONResponse({"error": "Unknown scenario"}, status_code=404)
scenario_id: Final = segments[0].split(":", 1)[0]
response: Final = self.scenario_store.get(scenario_id)
if response is None:
return JSONResponse({"error": "Unknown scenario"}, status_code=404)
return self._response(response, scenario_id)
@staticmethod
def _render(rendered: RenderedResponse) -> Response:
return Response(
content=rendered.body,
status_code=rendered.status_code,
media_type=rendered.content_type,
)
def _response(response: StoredResponse, scenario_id: str) -> Response:
match response:
case JsonResponse():
return Response(
content=json.dumps(response.body, separators=(",", ":")).replace(
"$REQUEST_ID", scenario_id
).encode(),
media_type=response.content_type,
)
case SseResponse():
stream_body: Final = ("\n\n".join(response.frames) + "\n\n").replace(
"$REQUEST_ID", scenario_id
)
return Response(content=stream_body.encode(), media_type=response.content_type)
case EventStreamResponse():
event_body: Final = b"".join(
_aws_event_frame(event.event_type, event.payload, scenario_id) for event in response.events
)
return Response(content=event_body, media_type=response.content_type)
def app(self) -> Starlette:
return Starlette(
@ -198,7 +236,7 @@ class Provider:
Route("/v1/completions", completions, methods=["POST"]),
Route("/v1/embeddings", embeddings, methods=["POST"]),
Route("/v1/moderations", moderations, methods=["POST"]),
Route("/{scenario_id}/{tail:path}", self.scripted, methods=["POST"]),
Route("/{path:path}", self.scripted, methods=["POST"]),
]
)
@ -215,17 +253,16 @@ class ScenarioHandle:
return f"{self.control_url}/{self.scenario_id}"
def register_scenario(scenario: Scenario) -> ScenarioHandle:
response: Final = httpx.post(
def register_scenario(scenario_id: str, response: StoredResponse) -> ScenarioHandle:
http_response: Final = httpx.post(
f"{CONTROL_URL}/__scenarios",
json=scenario.model_dump(mode="json"),
json={"scenario_id": scenario_id, "response": response.model_dump(mode="json")},
trust_env=False,
timeout=15,
)
response.raise_for_status()
result: Final = ScenarioRegistered.model_validate_json(response.content)
http_response.raise_for_status()
return ScenarioHandle(
scenario_id=result.scenario_id,
scenario_id=scenario_id,
control_url=CONTROL_URL,
)
@ -237,7 +274,6 @@ def delete_scenario(handle: ScenarioHandle) -> None:
timeout=15,
)
response.raise_for_status()
ScenarioDeleted.model_validate_json(response.content)
def main() -> None:

726
tests/integration/contracts.json
View file

File diff suppressed because it is too large Load diff

2958
tests/integration/cost_calculation/cases.json
View file

File diff suppressed because it is too large Load diff

28
tests/integration/cost_calculation/conftest.py
View file

@ -14,7 +14,7 @@ from pydantic import BaseModel, ConfigDict
from integration._support.client import JSON_OBJECT, Scenario, eventually, object_value, string_value
from integration._support.database import read_rows
from integration._support.upstream import delete_scenario, register_scenario
from integration.cost_calculation.cost_matrix import Case, FrontierModel
from integration.cost_calculation.cost_tracking_case import CostTrackingTestCase
class CostBreakdown(BaseModel):
@ -118,25 +118,23 @@ def _vertex_service_account_json(url: str) -> str:
def register_scenario_deployment(
scenario: Scenario,
model: FrontierModel,
case: Case,
case: CostTrackingTestCase,
marker: str,
key: str,
) -> str:
control_url: Final = os.environ["INTEGRATION_UPSTREAM_URL"].rstrip("/")
sidecar_scenario: Final = case.scenario(
scenario_id=f"sc-{marker}", model=model, text=f"scripted answer {marker}"
)
handle: Final = register_scenario(sidecar_scenario)
run_marker: Final = sha256(key.encode()).hexdigest()[:12]
handle: Final = register_scenario(f"sc-{marker}-{run_marker}", case.response)
scenario.cleanups.callback(delete_scenario, handle)
model_name: Final = f"{model.model_name}-{marker}"
model_name: Final = f"cost-{marker}-{run_marker}"
parameters: Final = {
"model": model.litellm_model,
"api_key": model.api_key,
"model": case.litellm_model,
"api_key": case.api_key,
"api_base": handle.api_base(),
**model.litellm_params,
**case.litellm_params,
**(
{"vertex_credentials": _vertex_service_account_json(control_url)}
if model.llm_provider == "vertex_ai"
if case.rates.litellm_provider == "vertex_ai-language-models"
else {}
),
}
@ -145,7 +143,11 @@ def register_scenario_deployment(
JSON_OBJECT.validate_python({
"model_name": model_name,
"litellm_params": parameters,
"model_info": {"base_model": model.base_model},
"model_info": (
{"base_model": case.base_model}
if case.base_model is not None
else {}
),
}),
)
identity: Final = string_value(object_value(created["model_info"])["id"])

411
tests/integration/cost_calculation/cost_map.json
View file

@ -1,411 +0,0 @@
{
"gpt-5.6": {
"cache_read_input_token_cost": 1.75e-07,
"input_cost_per_audio_token": 4e-05,
"input_cost_per_token": 1.75e-06,
"input_cost_per_token_flex": 8.75e-07,
"input_cost_per_token_priority": 3.5e-06,
"litellm_provider": "openai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 8e-05,
"output_cost_per_reasoning_token": 1.6e-05,
"output_cost_per_token": 1.4e-05,
"output_cost_per_token_flex": 7e-06,
"output_cost_per_token_priority": 2.8e-05,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"gpt-5.4-mini": {
"cache_read_input_token_cost": 3.5e-08,
"input_cost_per_audio_token": 1e-05,
"input_cost_per_token": 3.5e-07,
"input_cost_per_token_flex": 1.75e-07,
"input_cost_per_token_priority": 7e-07,
"litellm_provider": "openai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_reasoning_token": 3.2e-06,
"output_cost_per_token": 2.8e-06,
"output_cost_per_token_flex": 1.4e-06,
"output_cost_per_token_priority": 5.6e-06,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"azure/gpt-5.6": {
"cache_read_input_token_cost": 1.8e-07,
"input_cost_per_audio_token": 4.1e-05,
"input_cost_per_token": 1.8e-06,
"input_cost_per_token_flex": 9e-07,
"input_cost_per_token_priority": 3.6e-06,
"litellm_provider": "azure",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 8.2e-05,
"output_cost_per_reasoning_token": 1.65e-05,
"output_cost_per_token": 1.44e-05,
"output_cost_per_token_flex": 7.2e-06,
"output_cost_per_token_priority": 2.88e-05,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"azure/gpt-5.4-mini": {
"cache_read_input_token_cost": 3.6e-08,
"input_cost_per_audio_token": 1.05e-05,
"input_cost_per_token": 3.6e-07,
"input_cost_per_token_flex": 1.8e-07,
"input_cost_per_token_priority": 7.2e-07,
"litellm_provider": "azure",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 2.1e-05,
"output_cost_per_reasoning_token": 3.3e-06,
"output_cost_per_token": 2.88e-06,
"output_cost_per_token_flex": 1.44e-06,
"output_cost_per_token_priority": 5.76e-06,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"gpt-5.3-codex": {
"cache_read_input_token_cost": 1.5e-07,
"file_search_cost_per_1k_calls": 0.0025,
"input_cost_per_token": 1.5e-06,
"input_cost_per_token_flex": 7.5e-07,
"input_cost_per_token_priority": 3e-06,
"litellm_provider": "openai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "responses",
"output_cost_per_reasoning_token": 1.3e-05,
"output_cost_per_token": 1.2e-05,
"output_cost_per_token_flex": 6e-06,
"output_cost_per_token_priority": 2.4e-05,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"gpt-5.5-pro": {
"cache_read_input_token_cost": 1.5e-06,
"file_search_cost_per_1k_calls": 0.0025,
"input_cost_per_token": 1.5e-05,
"input_cost_per_token_flex": 7.5e-06,
"input_cost_per_token_priority": 3e-05,
"litellm_provider": "openai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "responses",
"output_cost_per_reasoning_token": 0.00013,
"output_cost_per_token": 0.00012,
"output_cost_per_token_flex": 6e-05,
"output_cost_per_token_priority": 0.00024,
"search_context_cost_per_query": {
"search_context_size_low": 0.01,
"search_context_size_medium": 0.0125,
"search_context_size_high": 0.015
},
"supports_function_calling": true
},
"claude-opus-5": {
"cache_creation_input_token_cost": 6.25e-06,
"cache_creation_input_token_cost_above_1hr": 1e-05,
"cache_creation_input_token_cost_above_200k_tokens": 1.25e-05,
"cache_read_input_token_cost": 5e-07,
"cache_read_input_token_cost_above_200k_tokens": 1e-06,
"input_cost_per_token": 5e-06,
"input_cost_per_token_above_200k_tokens": 1e-05,
"input_cost_per_token_priority": 6.25e-06,
"litellm_provider": "anthropic",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 2.5e-05,
"output_cost_per_token_above_200k_tokens": 3.75e-05,
"output_cost_per_token_priority": 3.125e-05,
"provider_specific_entry": {
"fast": 6.0,
"us": 1.1
},
"search_context_cost_per_query": {
"search_context_size_medium": 0.01
},
"supports_function_calling": true
},
"claude-sonnet-5": {
"cache_creation_input_token_cost": 3.75e-06,
"cache_creation_input_token_cost_above_1hr": 6e-06,
"cache_creation_input_token_cost_above_200k_tokens": 7.5e-06,
"cache_read_input_token_cost": 3e-07,
"cache_read_input_token_cost_above_200k_tokens": 6e-07,
"input_cost_per_token": 3e-06,
"input_cost_per_token_above_200k_tokens": 6e-06,
"input_cost_per_token_priority": 3.75e-06,
"litellm_provider": "anthropic",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 1.5e-05,
"output_cost_per_token_above_200k_tokens": 2.25e-05,
"output_cost_per_token_priority": 1.875e-05,
"provider_specific_entry": {
"us": 1.1
},
"search_context_cost_per_query": {
"search_context_size_medium": 0.01
},
"supports_function_calling": true
},
"claude-haiku-4-5": {
"cache_creation_input_token_cost": 1.25e-06,
"cache_creation_input_token_cost_above_1hr": 2e-06,
"cache_read_input_token_cost": 1e-07,
"input_cost_per_token": 1e-06,
"input_cost_per_token_priority": 1.25e-06,
"litellm_provider": "anthropic",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 5e-06,
"output_cost_per_token_priority": 6.25e-06,
"provider_specific_entry": {
"us": 1.1
},
"search_context_cost_per_query": {
"search_context_size_medium": 0.01
},
"supports_function_calling": true
},
"us.anthropic.claude-opus-5-v1:0": {
"cache_creation_input_token_cost": 6.875e-06,
"cache_creation_input_token_cost_above_1hr": 1.1e-05,
"cache_creation_input_token_cost_above_200k_tokens": 1.375e-05,
"cache_read_input_token_cost": 5.5e-07,
"cache_read_input_token_cost_above_200k_tokens": 1.1e-06,
"input_cost_per_token": 5.5e-06,
"input_cost_per_token_above_200k_tokens": 1.1e-05,
"input_cost_per_token_flex": 2.75e-06,
"input_cost_per_token_priority": 6.875e-06,
"litellm_provider": "bedrock_converse",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 2.75e-05,
"output_cost_per_token_above_200k_tokens": 4.125e-05,
"output_cost_per_token_flex": 1.375e-05,
"output_cost_per_token_priority": 3.4375e-05,
"supports_function_calling": true
},
"anthropic.claude-sonnet-5-v1:0": {
"cache_creation_input_token_cost": 4.125e-06,
"cache_creation_input_token_cost_above_1hr": 6.6e-06,
"cache_read_input_token_cost": 3.3e-07,
"input_cost_per_token": 3.3e-06,
"input_cost_per_token_flex": 1.65e-06,
"input_cost_per_token_priority": 4.125e-06,
"litellm_provider": "bedrock_converse",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 1.65e-05,
"output_cost_per_token_flex": 8.25e-06,
"output_cost_per_token_priority": 2.0625e-05,
"supports_function_calling": true
},
"meta.llama4-maverick-17b-instruct-v1:0": {
"input_cost_per_token": 2.4e-07,
"litellm_provider": "bedrock_converse",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 9.7e-07,
"supports_function_calling": true
},
"gemini/gemini-3.1-pro": {
"cache_read_input_token_cost": 2e-07,
"cache_read_input_token_cost_above_200k_tokens": 4e-07,
"google_maps_grounding_cost_per_query": 0.025,
"input_cost_per_audio_token": 2.6e-06,
"input_cost_per_image_token": 2.2e-06,
"input_cost_per_token": 2e-06,
"input_cost_per_token_above_200k_tokens": 4e-06,
"input_cost_per_token_flex": 1e-06,
"input_cost_per_token_priority": 2.5e-06,
"input_cost_per_video_token": 2.4e-06,
"litellm_provider": "gemini",
"max_input_tokens": 1048576,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_reasoning_token": 1.3e-05,
"output_cost_per_token": 1.2e-05,
"output_cost_per_token_above_200k_tokens": 1.8e-05,
"output_cost_per_token_flex": 6e-06,
"output_cost_per_token_priority": 1.5e-05,
"search_context_cost_per_query": {
"search_context_size_medium": 0.035
},
"supports_function_calling": true,
"web_search_billing_unit": "per_query"
},
"gemini/gemini-3.8-flash": {
"cache_read_input_token_cost": 5e-08,
"google_maps_grounding_cost_per_query": 0.025,
"input_cost_per_audio_token": 1e-06,
"input_cost_per_image_token": 5.5e-07,
"input_cost_per_token": 5e-07,
"input_cost_per_token_flex": 2.5e-07,
"input_cost_per_token_priority": 6.25e-07,
"input_cost_per_video_token": 6e-07,
"litellm_provider": "gemini",
"max_input_tokens": 1048576,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 6e-06,
"output_cost_per_reasoning_token": 3.5e-06,
"output_cost_per_token": 3e-06,
"output_cost_per_token_flex": 1.5e-06,
"output_cost_per_token_priority": 3.75e-06,
"search_context_cost_per_query": {
"search_context_size_medium": 0.035
},
"supports_function_calling": true,
"web_search_billing_unit": "per_prompt"
},
"gemini-3.1-pro": {
"cache_read_input_token_cost": 2.1e-07,
"cache_read_input_token_cost_above_200k_tokens": 4.2e-07,
"google_maps_grounding_cost_per_query": 0.025,
"input_cost_per_audio_token": 2.7e-06,
"input_cost_per_image_token": 2.3e-06,
"input_cost_per_token": 2.1e-06,
"input_cost_per_token_above_200k_tokens": 4.2e-06,
"input_cost_per_token_flex": 1.05e-06,
"input_cost_per_token_priority": 2.625e-06,
"litellm_provider": "vertex_ai-language-models",
"max_input_tokens": 1048576,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_reasoning_token": 1.35e-05,
"output_cost_per_token": 1.26e-05,
"output_cost_per_token_above_200k_tokens": 1.89e-05,
"output_cost_per_token_flex": 6.3e-06,
"output_cost_per_token_priority": 1.575e-05,
"search_context_cost_per_query": {
"search_context_size_medium": 0.035
},
"supports_function_calling": true,
"web_search_billing_unit": "per_query"
},
"gemini-3.8-flash": {
"cache_read_input_token_cost": 5.2e-08,
"google_maps_grounding_cost_per_query": 0.025,
"input_cost_per_audio_token": 1.04e-06,
"input_cost_per_token": 5.2e-07,
"input_cost_per_token_flex": 2.6e-07,
"input_cost_per_token_priority": 6.5e-07,
"input_cost_per_video_token": 6.2e-07,
"litellm_provider": "vertex_ai-language-models",
"max_input_tokens": 1048576,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_audio_token": 6.24e-06,
"output_cost_per_token": 3.12e-06,
"output_cost_per_token_flex": 1.56e-06,
"output_cost_per_token_priority": 3.9e-06,
"search_context_cost_per_query": {
"search_context_size_medium": 0.035
},
"supports_function_calling": true,
"web_search_billing_unit": "per_prompt"
},
"together_ai/moonshotai/Kimi-K3": {
"input_cost_per_token": 1.15e-06,
"litellm_provider": "together_ai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 3.45e-06,
"supports_function_calling": true
},
"together_ai/zai-org/GLM-5.3": {
"input_cost_per_token": 5.5e-07,
"litellm_provider": "together_ai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 2.2e-06,
"supports_function_calling": true
},
"fireworks_ai/accounts/fireworks/models/kimi-k3": {
"cache_read_input_token_cost": 6e-08,
"input_cost_per_token": 6e-07,
"litellm_provider": "fireworks_ai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 2.5e-06,
"supports_function_calling": true
},
"fireworks_ai/accounts/fireworks/models/deepseek-v4p1-flash": {
"input_cost_per_token": 1.5e-07,
"litellm_provider": "fireworks_ai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 6e-07,
"supports_function_calling": true
},
"fireworks_ai/accounts/fireworks/models/qwen3p8-max": {
"cache_read_input_token_cost": 9e-08,
"input_cost_per_token": 9e-07,
"litellm_provider": "fireworks_ai",
"max_input_tokens": 400000,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_token": 3.6e-06,
"supports_function_calling": true
}
}

658
tests/integration/cost_calculation/cost_matrix.py
View file

@ -1,658 +0,0 @@
"""The cost-calculation matrix: the model set derived from the test cost map,
the request/response cases from ``cases.json``, and the loaders both use.
Two data files drive the suite; nothing in Python lists models or cases:
- ``tests/integration/cost_calculation/cost_map.json`` is the proxy's ENTIRE model cost map
(LITELLM_MODEL_COST_MAP_URL); every entry becomes a deployment under test.
- ``tests/integration/cost_calculation/cases.json`` is the case list plus the reviewed
goldens: each exact-spend case carries an ``expected`` cell per map key it
runs against, each recount case carries its ``models`` list, so matrix
membership and expected values are literal data read side by side.
"""
from __future__ import annotations
import base64
import io
import json
import math
import random
import struct
import wave
import zlib
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal
from litellm import get_llm_provider
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.azure.chat.gpt_transformation import AzureOpenAIConfig
from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexGeminiConfig
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
from integration._support.scripted_shapes import (
Scenario,
Shape,
ScriptedOutput,
ScriptedToolCall,
ScriptedUsage,
)
COST_MAP_PATH: Final = Path(__file__).resolve().parent / "cost_map.json"
CASES_PATH: Final = Path(__file__).resolve().parent / "cases.json"
class SearchContextCostPerQuery(BaseModel):
model_config = ConfigDict(frozen=True)
search_context_size_low: float | None = None
search_context_size_medium: float | None = None
search_context_size_high: float | None = None
class ProviderSpecificEntry(BaseModel):
"""Provider-specific key rates, keyed by the named suffix litellm looks up
(``fast`` for Anthropic fast mode, ``us`` for US inference geography)."""
model_config = ConfigDict(frozen=True)
fast: float | None = None
us: float | None = None
class CostMapEntry(BaseModel):
"""The pricing fields of a cost-map entry the matrix reads. Shaped like a
``model_prices_and_context_window.json`` entry; the file is test-owned so
undeclared keys are forbidden rather than ignored."""
model_config = ConfigDict(frozen=True, extra="forbid")
litellm_provider: str
mode: str
max_tokens: int | None = None
max_input_tokens: int | None = None
max_output_tokens: int | None = None
supports_function_calling: bool | None = None
input_cost_per_token: float | None = None
output_cost_per_token: float | None = None
cache_read_input_token_cost: float | None = None
cache_creation_input_token_cost: float | None = None
cache_creation_input_token_cost_above_1hr: float | None = None
cache_read_input_token_cost_above_200k_tokens: float | None = None
cache_creation_input_token_cost_above_200k_tokens: float | None = None
output_cost_per_reasoning_token: float | None = None
input_cost_per_audio_token: float | None = None
output_cost_per_audio_token: float | None = None
input_cost_per_image_token: float | None = None
input_cost_per_video_token: float | None = None
input_cost_per_token_above_200k_tokens: float | None = None
output_cost_per_token_above_200k_tokens: float | None = None
input_cost_per_token_flex: float | None = None
output_cost_per_token_flex: float | None = None
input_cost_per_token_priority: float | None = None
output_cost_per_token_priority: float | None = None
search_context_cost_per_query: SearchContextCostPerQuery | None = None
web_search_billing_unit: str | None = None
google_maps_grounding_cost_per_query: float | None = None
file_search_cost_per_1k_calls: float | None = None
provider_specific_entry: ProviderSpecificEntry | None = None
_METADATA_FIELDS: Final = frozenset(
{
"litellm_provider",
"mode",
"max_tokens",
"max_input_tokens",
"max_output_tokens",
"supports_function_calling",
}
)
_CONTAINER_FIELDS: Final = frozenset({"search_context_cost_per_query", "provider_specific_entry"})
def _submodel_rate_keys(
field: str, sub: SearchContextCostPerQuery | ProviderSpecificEntry | None
) -> tuple[str, ...]:
if sub is None:
return ()
return tuple(
f"{field}.{name}"
for name in type(sub).model_fields
if getattr(sub, name) is not None
)
def _entry_rate_keys(entry: CostMapEntry) -> frozenset[str]:
"""Every cost key an entry carries, with container subfields expanded to
dotted names (``search_context_cost_per_query.search_context_size_low``).
``web_search_billing_unit`` counts as a rate key whenever present,
for both ``per_query`` and ``per_prompt`` values."""
plain: Final = frozenset(
name
for name in CostMapEntry.model_fields
if name not in _METADATA_FIELDS
and name not in _CONTAINER_FIELDS
and getattr(entry, name) is not None
)
return (
plain
| frozenset(
_submodel_rate_keys("search_context_cost_per_query", entry.search_context_cost_per_query)
)
| frozenset(_submodel_rate_keys("provider_specific_entry", entry.provider_specific_entry))
)
def _entry_has_rate_key(entry: CostMapEntry, rate_key: str) -> bool:
outer, _, inner = rate_key.partition(".")
if outer == "search_context_cost_per_query":
return f"{outer}.{inner}" in _submodel_rate_keys(outer, entry.search_context_cost_per_query)
if outer == "provider_specific_entry":
return f"{outer}.{inner}" in _submodel_rate_keys(outer, entry.provider_specific_entry)
value: Final[object] = getattr(entry, outer, None)
return value is not None
SERVICE_TIER_REQUEST_SHAPES: Final = frozenset(
{"openai_chat", "openai_responses", "bedrock_converse"}
)
COST_MAP_ADAPTER: Final = TypeAdapter(dict[str, CostMapEntry])
COST_MAP: Final[Mapping[str, CostMapEntry]] = MappingProxyType(
COST_MAP_ADAPTER.validate_python(json.loads(COST_MAP_PATH.read_text()))
)
TIER_THRESHOLD_TOKENS: Final = 200_000
class DeploymentSpec(BaseModel):
"""A deployment-level fact from cases.json: when a map key needs a
registered deployment name that is not its provider model (or a
model_info.base_model pin), the matrix uses these instead of the defaults."""
model_config = ConfigDict(frozen=True)
map_key: str
litellm_model: str | None = None
base_model: str | None = None
class ExpectedCell(BaseModel):
model_config = ConfigDict(frozen=True)
spend: float
input_cost: float
output_cost: float
prompt_tokens: int
completion_tokens: int
class Case(BaseModel):
"""One request/response shape from cases.json.
``family`` splits the matrix: ``pricing`` cases own cost keys (``owns``,
dotted subfield names allowed) or declare which keys they deliberately
leave absent (``fallback_for``) so every cost key in the map has exactly
one owning case; ``transport`` cases exercise counting/transport only and
run wherever they list membership. An exact-spend case names its models
implicitly by carrying one ``expected`` golden per map key; a recount
case (``exact_spend=False``) names them in ``models`` instead. The
feature flags drive request realism in ``_chat_body``."""
model_config = ConfigDict(frozen=True)
name: str
family: Literal["pricing", "transport"]
usage: ScriptedUsage
usage_by_model: Mapping[str, ScriptedUsage] = Field(default_factory=lambda: MappingProxyType({}))
stream: bool = False
stream_usage: Literal["final_chunk", "absent"] = "final_chunk"
service_tier: Literal["flex", "priority"] | None = None
speed: Literal["fast"] | None = None
inference_geo: Literal["us"] | None = None
response_model_override: bool = False
exact_spend: bool = True
tool_call: bool = False
image_input: bool = False
audio_input: bool = False
audio_output: bool = False
video_input: bool = False
reasoning: bool = False
web_search: Literal["low", "medium", "high"] | None = None
google_maps: bool = False
file_search: bool = False
terminal: Literal["completed", "incomplete", "unvalidated", "prompt_blocked"] = "completed"
owns: tuple[str, ...] = ()
fallback_for: tuple[str, ...] = ()
expected: Mapping[str, ExpectedCell] = Field(default_factory=lambda: MappingProxyType({}))
models: tuple[str, ...] = ()
def applies_to(self, model: FrontierModel) -> bool:
if self.exact_spend:
return model.map_key in self.expected
return model.map_key in self.models
def expected_for(self, model: FrontierModel) -> ExpectedCell:
return self.expected[model.map_key]
def usage_for(self, map_key: str) -> ScriptedUsage:
return self.usage_by_model.get(map_key, self.usage)
def scenario(self, scenario_id: str, model: FrontierModel, text: str) -> Scenario:
return Scenario(
scenario_id=scenario_id,
shape=model.shape,
usage=self.usage_for(model.map_key),
model=model.provider_model,
output=ScriptedOutput(
text=text,
response_model=model.override_model if self.response_model_override else None,
tool_call=ScriptedToolCall(name="get_weather", arguments=TOOL_CALL_ARGUMENTS)
if self.tool_call
else None,
terminal=self.terminal,
),
stream_usage=self.stream_usage,
service_tier=self.service_tier,
speed=self.speed,
inference_geo=self.inference_geo,
)
class _ProviderWiringRow(BaseModel):
model_config = ConfigDict(frozen=True)
litellm_provider: str
mode: str
model_prefix: str | None
litellm_params: Mapping[str, str]
class _CasesFile(BaseModel):
model_config = ConfigDict(frozen=True)
providers: tuple[_ProviderWiringRow, ...] = ()
deployments: tuple[DeploymentSpec, ...] = ()
cases: tuple[Case, ...] = ()
CASES_FILE: Final = _CasesFile.model_validate(json.loads(CASES_PATH.read_text()))
CASES: Final[tuple[Case, ...]] = CASES_FILE.cases
_DEPLOYMENTS: Final[Mapping[str, DeploymentSpec]] = MappingProxyType(
{spec.map_key: spec for spec in CASES_FILE.deployments}
)
@dataclass(frozen=True, slots=True)
class _DeploymentDefaults:
"""How a (litellm_provider, mode) pair maps to deployment defaults."""
model_prefix: str | None
litellm_params: Mapping[str, str]
def _deployment_defaults(
rows: tuple[_ProviderWiringRow, ...],
) -> Mapping[tuple[str, str], _DeploymentDefaults]:
return MappingProxyType(
{
(row.litellm_provider, row.mode): _DeploymentDefaults(
row.model_prefix,
MappingProxyType(dict(row.litellm_params)),
)
for row in rows
}
)
_DEPLOYMENT_DEFAULTS: Final[Mapping[tuple[str, str], _DeploymentDefaults]] = _deployment_defaults(
CASES_FILE.providers
)
@dataclass(frozen=True, slots=True)
class FrontierModel:
"""One deployment under test, derived from a cost-map entry: the model_name
the suite registers, the provider-prefixed litellm model string, the
response shape the scripted upstream speaks, and the sibling map model the
response_model override case reports."""
model_name: str
litellm_model: str
shape: Shape
llm_provider: str
map_key: str
override_model: str | None = None
override_map_key: str | None = None
# Registered as model_info.base_model; when set, the provider-reported
# model loses to it and every case bills at this deployment's own rates.
base_model: str | None = None
litellm_params: Mapping[str, str] = MappingProxyType({})
@property
def rates(self) -> CostMapEntry:
return COST_MAP[self.map_key]
@property
def override_rates(self) -> CostMapEntry:
# bedrock_converse responses carry no model field, so a reported-model
# override can never repoint pricing there, same as a base_model pin.
if (
self.base_model is not None
or self.shape == "bedrock_converse"
or self.override_map_key is None
):
return self.rates
return COST_MAP[self.override_map_key]
@property
def provider_model(self) -> str:
"""The bare provider-facing model name: litellm_model minus the provider
prefix and any routing segment (converse/, responses/)."""
return _provider_model(self.litellm_model)
@property
def provider(self) -> str:
return self.rates.litellm_provider
@property
def api_key(self) -> str:
# The scripted upstream ignores auth; a fixed bogus key proves the suite
# spends zero real provider calls.
return "sk-scripted-provider"
def _provider_model(litellm_model: str) -> str:
tail: Final = litellm_model.split("/")[1:]
return "/".join(tail[1:] if tail and tail[0] in ("converse", "responses") else tail)
def _litellm_model_for(map_key: str, defaults: _DeploymentDefaults) -> str:
if defaults.model_prefix is None:
return map_key
if map_key.startswith(f"{defaults.model_prefix}/"):
return map_key
return f"{defaults.model_prefix}/{map_key}"
def _resolve(litellm_model: str, mode: str) -> tuple[str, Shape]:
model, provider, _, _ = get_llm_provider(model=litellm_model)
llm_provider: Final = LlmProviders(provider)
if mode == "responses":
responses_config: Final = ProviderConfigManager.get_provider_responses_api_config(
model=model,
provider=llm_provider,
)
if isinstance(responses_config, OpenAIResponsesAPIConfig):
return provider, "openai_responses"
raise ValueError(f"no scripted renderer for {type(responses_config).__name__} ({litellm_model})")
config: Final = ProviderConfigManager.get_provider_chat_config(model=model, provider=llm_provider)
if isinstance(config, AmazonConverseConfig):
return provider, "bedrock_converse"
if isinstance(config, VertexGeminiConfig):
return provider, "gemini_generate"
if isinstance(config, AnthropicConfig):
return provider, "anthropic_messages"
if isinstance(config, (AzureOpenAIConfig, OpenAIGPTConfig)):
return provider, "openai_chat"
raise ValueError(f"no scripted renderer for {type(config).__name__} ({litellm_model})")
def _frontier() -> tuple[FrontierModel, ...]:
groups: Final[Mapping[tuple[str, str], tuple[str, ...]]] = MappingProxyType(
{
pair: tuple(sorted(k for k, e in COST_MAP.items() if (e.litellm_provider, e.mode) == pair))
for pair in {(e.litellm_provider, e.mode) for e in COST_MAP.values()}
}
)
models: list[FrontierModel] = [] # mutable-ok: accumulated once at import into a tuple
for map_key in sorted(COST_MAP):
entry = COST_MAP[map_key]
pair = (entry.litellm_provider, entry.mode)
defaults = _DEPLOYMENT_DEFAULTS.get(pair)
if defaults is None:
continue
siblings = groups[pair]
override_key = (
siblings[(siblings.index(map_key) + 1) % len(siblings)] if len(siblings) > 1 else None
)
override_litellm = (
_litellm_model_for(override_key, defaults) if override_key is not None else None
)
deployment = _DEPLOYMENTS.get(map_key)
litellm_model = (
deployment.litellm_model
if deployment is not None and deployment.litellm_model is not None
else _litellm_model_for(map_key, defaults)
)
llm_provider, shape = _resolve(litellm_model, entry.mode)
models.append(
FrontierModel(
model_name=f"cc-{map_key.replace('/', '-').replace(':', '-').replace('.', '-').lower()}",
litellm_model=litellm_model,
shape=shape,
llm_provider=llm_provider,
map_key=map_key,
override_model=(
_provider_model(override_litellm)
if override_litellm is not None
else None
),
override_map_key=override_key,
base_model=deployment.base_model if deployment is not None else None,
litellm_params=defaults.litellm_params,
)
)
return tuple(models)
FRONTIER_MODELS: Final[tuple[FrontierModel, ...]] = _frontier()
TOOL_CALL_ARGUMENTS: Final = json.dumps({
"city": "Berlin",
"days": 7,
"units": "metric",
"notes": "filler " * 30,
})
def cases_for(model: FrontierModel) -> tuple[Case, ...]:
return tuple(case for case in CASES if case.applies_to(model))
def recount_cost(
model: FrontierModel, case: Case, prompt_tokens: int, completion_tokens: int
) -> float:
"""What the proxy's own token recount should cost at the case's rates,
without pinning the tokenizer's exact counts."""
rates: Final = model.override_rates if case.response_model_override else model.rates
return prompt_tokens * (rates.input_cost_per_token or 0.0) + completion_tokens * (
rates.output_cost_per_token or 0.0
)
def _png_chunk(tag: bytes, payload: bytes) -> bytes:
return struct.pack(">I", len(payload)) + tag + payload + struct.pack(">I", zlib.crc32(tag + payload))
def audio_input_data_url() -> str:
"""A deterministic 0.5 s 16-bit PCM WAV (8 kHz, 220 Hz sine) as a data
URL, small enough to stay a fixture but real audio to the provider."""
frames: Final = b"".join(
struct.pack("<h", int(12000 * math.sin(2 * math.pi * 220 * i / 8000)))
for i in range(4000)
)
buffer: Final = io.BytesIO()
with wave.open(buffer, "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(8000)
wav.writeframes(frames)
return "data:audio/wav;base64," + base64.b64encode(buffer.getvalue()).decode()
def video_input_data_url() -> str:
"""A deterministic mp4-looking blob (ftyp box plus a fixed mdat payload)
as a data URL; only the media type and bytes matter to the response."""
ftyp: Final = struct.pack(">I4s4sI4s4s", 24, b"ftyp", b"isom", 0x200, b"isom", b"iso6")
mdat_payload: Final = bytes((i * 7 + 13) % 256 for i in range(4096))
mdat: Final = struct.pack(">I4s", 8 + len(mdat_payload), b"mdat") + mdat_payload
return "data:video/mp4;base64," + base64.b64encode(ftyp + mdat).decode()
def image_input_data_url() -> str:
"""A deterministic 256x256 RGB noise PNG as a data URL; noise compresses
poorly on purpose so the base64 payload stays well above 100 KB and would
blow up the prompt recount if the URL were ever tokenized as text."""
rng: Final = random.Random(0)
side: Final = 256
raw: Final = b"".join(
b"\x00" + rng.randbytes(side * 3) for _ in range(side)
)
png: Final = (
b"\x89PNG\r\n\x1a\n"
+ _png_chunk(b"IHDR", struct.pack(">IIBBBBB", side, side, 8, 2, 0, 0, 0))
+ _png_chunk(b"IDAT", zlib.compress(raw))
+ _png_chunk(b"IEND", b"")
)
return "data:image/png;base64," + base64.b64encode(png).decode()
IMAGE_INPUT_DATA_URL: Final = image_input_data_url()
AUDIO_INPUT_DATA_URL: Final = audio_input_data_url()
VIDEO_INPUT_DATA_URL: Final = video_input_data_url()
def matrix_data_errors() -> tuple[str, ...]:
"""Consistency findings for the data files, as human-readable strings.
Called at collection time by the integration suite, so a map key named by a case
but absent from cost_map.json fails the suite's collection loudly.
"""
unknown_deployments: Final = sorted(
spec.map_key for spec in CASES_FILE.deployments if spec.map_key not in COST_MAP
)
unknown_case_models: Final = sorted(
{
map_key
for case in CASES
for map_key in (*case.expected, *case.models)
if map_key not in COST_MAP
}
)
misshapen_cases: Final = sorted(
case.name
for case in CASES
if case.exact_spend == bool(case.models) or case.exact_spend != bool(case.expected)
)
all_pairs: Final = frozenset(
(map_key, key)
for map_key, entry in COST_MAP.items()
for key in _entry_rate_keys(entry)
)
owned_pairs: Final = tuple(
(map_key, key)
for case in CASES
if case.family == "pricing"
for map_key in case.expected
for key in case.owns
if map_key in COST_MAP and _entry_has_rate_key(COST_MAP[map_key], key)
)
unowned_pairs: Final = sorted(
f"{map_key}:{key}" for map_key, key in all_pairs - frozenset(owned_pairs)
)
duplicate_pairs: Final = sorted(
f"{map_key}:{key}"
for map_key, key in set(owned_pairs)
if owned_pairs.count((map_key, key)) > 1
)
owns_without_holder: Final = sorted(
f"{case.name}:{key}"
for case in CASES
for key in case.owns
if not any(
map_key in COST_MAP and _entry_has_rate_key(COST_MAP[map_key], key)
for map_key in case.expected
)
)
fallback_violations: Final = sorted(
f"{case.name}:{map_key}:{key}"
for case in CASES
for key in case.fallback_for
for map_key in (*case.expected, *case.models)
if map_key in COST_MAP and _entry_has_rate_key(COST_MAP[map_key], key)
)
family_violations: Final = sorted(
case.name
for case in CASES
if (case.family == "transport") != (not case.owns and not case.fallback_for)
)
missing_provider_rows: Final = sorted(
f"cost_map entry {map_key} has no providers row for "
f"(litellm_provider={entry.litellm_provider}, mode={entry.mode}); "
f"add a providers row in cases.json"
for map_key, entry in COST_MAP.items()
if (entry.litellm_provider, entry.mode) not in _DEPLOYMENT_DEFAULTS
)
input_rates: Final = tuple(entry.input_cost_per_token for entry in COST_MAP.values())
findings: Final = (
(
f"deployments entries name map keys absent from cost_map.json: {unknown_deployments}"
if unknown_deployments
else None
),
(
f"case expected/models name map keys absent from cost_map.json: {unknown_case_models}"
if unknown_case_models
else None
),
(
f"cases must carry expected xor models (exact_spend matches the field): {misshapen_cases}"
if misshapen_cases
else None
),
(
"two cost_map entries share input_cost_per_token; the suite relies on "
"distinct rates so a wrong-model bill can never coincidentally match"
if len(input_rates) != len(set(input_rates))
else None
),
(
f"(model, rate key) pairs with no owning case: {unowned_pairs}"
if unowned_pairs
else None
),
(
f"(model, rate key) pairs owned by more than one case: {duplicate_pairs}"
if duplicate_pairs
else None
),
(
f"owns keys absent on all of the case's expected models: {owns_without_holder}"
if owns_without_holder
else None
),
(
f"fallback_for keys a case's models actually carry: {fallback_violations}"
if fallback_violations
else None
),
(
f"cases with owns/fallback_for inconsistent with family: {family_violations}"
if family_violations
else None
),
(
f"cost_map entries without providers rows: {missing_provider_rows}"
if missing_provider_rows
else None
),
)
return tuple(finding for finding in findings if finding is not None)

253
tests/integration/cost_calculation/cost_tracking_case.py Normal file
View file

@ -0,0 +1,253 @@
from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from types import MappingProxyType
from typing import Annotated, Final, Literal, TypeAlias
from pydantic import BaseModel, ConfigDict, Field, JsonValue
CASES_PATH: Final = Path(__file__).resolve().parent / "cost_tracking_cases.json"
class SearchContextCostPerQuery(BaseModel):
model_config = ConfigDict(frozen=True)
search_context_size_low: float | None = None
search_context_size_medium: float | None = None
search_context_size_high: float | None = None
class ProviderSpecificEntry(BaseModel):
model_config = ConfigDict(frozen=True)
fast: float | None = None
us: float | None = None
class CostMapEntry(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
litellm_provider: str
mode: str
max_tokens: int | None = None
max_input_tokens: int | None = None
max_output_tokens: int | None = None
supports_function_calling: bool | None = None
input_cost_per_token: float | None = None
output_cost_per_token: float | None = None
cache_read_input_token_cost: float | None = None
cache_creation_input_token_cost: float | None = None
cache_creation_input_token_cost_above_1hr: float | None = None
cache_read_input_token_cost_above_200k_tokens: float | None = None
cache_creation_input_token_cost_above_200k_tokens: float | None = None
output_cost_per_reasoning_token: float | None = None
input_cost_per_audio_token: float | None = None
output_cost_per_audio_token: float | None = None
input_cost_per_image_token: float | None = None
input_cost_per_video_token: float | None = None
input_cost_per_token_above_200k_tokens: float | None = None
output_cost_per_token_above_200k_tokens: float | None = None
input_cost_per_token_flex: float | None = None
output_cost_per_token_flex: float | None = None
input_cost_per_token_priority: float | None = None
output_cost_per_token_priority: float | None = None
search_context_cost_per_query: SearchContextCostPerQuery | None = None
web_search_billing_unit: str | None = None
google_maps_grounding_cost_per_query: float | None = None
file_search_cost_per_1k_calls: float | None = None
provider_specific_entry: ProviderSpecificEntry | None = None
class Deployment(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
model: str | None = None
base_model: str | None = None
class JsonResponse(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
content_type: Literal["application/json"]
body: dict[str, JsonValue]
class SseResponse(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
content_type: Literal["text/event-stream"]
frames: tuple[str, ...]
class EventStreamEvent(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
event_type: str
payload: dict[str, JsonValue]
class EventStreamResponse(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
content_type: Literal["application/vnd.amazon.eventstream"]
events: tuple[EventStreamEvent, ...]
StoredResponse: TypeAlias = Annotated[
JsonResponse | SseResponse | EventStreamResponse,
Field(discriminator="content_type"),
]
class ExactExpected(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
spend: float
input_cost: float
output_cost: float
prompt_tokens: int
completion_tokens: int
class RecountRates(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
input_cost_per_token: float
output_cost_per_token: float
class RecountExpected(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
recount: RecountRates
Expected: TypeAlias = ExactExpected | RecountExpected
class CostTrackingTestCase(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
name: str
covers: str
model: str
deployment: Deployment | None = None
request: dict[str, JsonValue]
response: StoredResponse
expected: Expected
@property
def rates(self) -> CostMapEntry:
return COST_MAP[self.model]
@property
def litellm_model(self) -> str:
provider: Final = self.rates.litellm_provider
prefix: Final = (
"openai"
if provider == "openai" and self.rates.mode == "chat"
else "openai/responses"
if provider == "openai"
else _PROVIDER_PREFIXES.get(provider)
)
if prefix is None:
raise ValueError(f"unsupported cost-map provider {provider} for {self.model}")
return self.deployment.model if self.deployment and self.deployment.model is not None else (
self.model if prefix == "" else f"{prefix}/{self.model}"
)
@property
def litellm_params(self) -> Mapping[str, str]:
return _LITELLM_PARAMS[self.rates.litellm_provider]
@property
def api_key(self) -> str:
return "sk-scripted-provider"
@property
def base_model(self) -> str | None:
return self.deployment.base_model if self.deployment else None
class _CasesFile(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
cost_map: dict[str, CostMapEntry]
cases: tuple[CostTrackingTestCase, ...]
_PROVIDER_PREFIXES: Final[Mapping[str, str]] = MappingProxyType(
{
"anthropic": "anthropic",
"bedrock_converse": "bedrock/converse",
"vertex_ai-language-models": "vertex_ai",
"gemini": "",
"together_ai": "",
"fireworks_ai": "",
"azure": "",
}
)
_LITELLM_PARAMS: Final[Mapping[str, Mapping[str, str]]] = MappingProxyType(
{
"anthropic": MappingProxyType({}),
"bedrock_converse": MappingProxyType(
{
"aws_access_key_id": "AKIASCRIPTEDPROVIDER",
"aws_secret_access_key": "scripted-secret",
"aws_region_name": "us-east-1",
}
),
"vertex_ai-language-models": MappingProxyType(
{"vertex_project": "cc-scripted-project", "vertex_location": "us-central1"}
),
"gemini": MappingProxyType({}),
"together_ai": MappingProxyType({}),
"fireworks_ai": MappingProxyType({}),
"azure": MappingProxyType({"api_version": "2025-04-01-preview"}),
"openai": MappingProxyType({}),
}
)
_LOADED: Final = _CasesFile.model_validate_json(CASES_PATH.read_bytes())
COST_MAP: Final[Mapping[str, CostMapEntry]] = MappingProxyType(dict(_LOADED.cost_map))
CASES: Final[tuple[CostTrackingTestCase, ...]] = _LOADED.cases
_LITELLM_MODELS: Final = tuple(case.litellm_model for case in CASES)
def data_errors() -> tuple[str, ...]:
case_models: Final = frozenset(case.model for case in CASES)
unknown_models: Final = sorted(case.model for case in CASES if case.model not in COST_MAP)
missing_cases: Final = sorted(model for model in COST_MAP if model not in case_models)
duplicate_names: Final = sorted(
name for name in {case.name for case in CASES} if sum(case.name == name for case in CASES) > 1
)
input_rates: Final = tuple(
(entry.input_cost_per_token, model) for model, entry in COST_MAP.items()
)
shared_input_rates: Final = sorted(
f"{rate}: {tuple(model for value, model in input_rates if value == rate)}"
for rate in {value for value, _ in input_rates if value is not None}
if sum(value == rate for value, _ in input_rates) > 1
)
recount_mismatches: Final = sorted(
case.name
for case in CASES
if isinstance(case.expected, RecountExpected)
and case.model in COST_MAP
and (
case.expected.recount.input_cost_per_token != (COST_MAP[case.model].input_cost_per_token or 0.0)
or case.expected.recount.output_cost_per_token != (COST_MAP[case.model].output_cost_per_token or 0.0)
)
)
return tuple(
message
for message in (
f"case models absent from cost_map: {unknown_models}" if unknown_models else None,
f"cost-map entries without cases: {missing_cases}" if missing_cases else None,
f"duplicate case names: {duplicate_names}" if duplicate_names else None,
f"cost-map entries share input_cost_per_token: {shared_input_rates}" if shared_input_rates else None,
f"recount rates differ from cost-map rates: {recount_mismatches}" if recount_mismatches else None,
)
if message is not None
)

25658
tests/integration/cost_calculation/cost_tracking_cases.json Normal file
View file

File diff suppressed because it is too large Load diff

101
tests/integration/cost_calculation/test_cost_tracking.py Normal file
View file

@ -0,0 +1,101 @@
"""Cost tracking coverage for literal integration request and response data."""
from __future__ import annotations
from hashlib import sha256
from typing import Final, cast
import pytest
from integration._support.client import JSON_OBJECT, Gateway
from integration.cost_calculation.conftest import (
approx_equal,
assert_total_is_sum_of_components,
poll_cost_row,
register_scenario_deployment,
)
from integration.cost_calculation.cost_tracking_case import (
CASES,
CostTrackingTestCase,
ExactExpected,
RecountExpected,
data_errors,
)
if _data_errors := data_errors():
raise ValueError("\n".join(_data_errors))
_CASES: Final = tuple(
pytest.param(case, marks=pytest.mark.covers(case.covers), id=case.name)
for case in CASES
)
def _assert_stream_has_no_error(response_text: str) -> None:
for line in response_text.splitlines():
if not line.startswith("data:"):
continue
payload = line.removeprefix("data:").strip()
if payload == "[DONE]":
continue
parsed = JSON_OBJECT.validate_json(payload)
assert "error" not in parsed, f"stream carried an error event: {parsed}"
@pytest.mark.parametrize("case", _CASES)
def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase) -> None:
marker: Final = sha256(case.name.encode()).hexdigest()[:12]
with gateway.scenario() as scenario:
key: Final = scenario.key()
model_name: Final = register_scenario_deployment(scenario, case, marker, key)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
{**case.request, "model": model_name},
key=key,
)
assert response.is_success, f"{case.name}: proxy returned {response.status_code}: {response.text[:400]}"
if case.response.content_type == "text/event-stream":
_assert_stream_has_no_error(response.text)
row: Final = poll_cost_row(key)
if isinstance(case.expected, RecountExpected):
assert row.prompt_tokens is not None and row.prompt_tokens > 0, (
f"{case.name}: recount case counted no input tokens: prompt_tokens={row.prompt_tokens}"
)
assert row.completion_tokens is not None and row.completion_tokens > 0, (
f"{case.name}: recount case counted no output tokens: completion_tokens={row.completion_tokens}"
)
recount: Final = row.prompt_tokens * case.expected.recount.input_cost_per_token + (
row.completion_tokens * case.expected.recount.output_cost_per_token
)
assert row.spend is not None and approx_equal(row.spend, recount), (
f"{case.name}: spend {row.spend} != recount {recount} at map rates"
)
assert_total_is_sum_of_components(row, case.name)
return
expected: Final = case.expected
assert isinstance(expected, ExactExpected)
if case.response.content_type == "application/json":
header: Final = cast(str | None, response.headers.get("x-litellm-response-cost"))
assert header is not None and approx_equal(float(header), expected.spend), (
f"{case.name}: x-litellm-response-cost {header} != expected {expected.spend}"
)
assert row.spend is not None and approx_equal(row.spend, expected.spend), (
f"{case.name}: spend {row.spend} != expected {expected.spend} "
f"(breakdown {row.breakdown.model_dump()})"
)
breakdown: Final = row.breakdown
assert breakdown.input_cost is not None and approx_equal(breakdown.input_cost, expected.input_cost), (
f"{case.name}: input_cost {breakdown.input_cost} != expected {expected.input_cost}"
)
assert breakdown.output_cost is not None and approx_equal(breakdown.output_cost, expected.output_cost), (
f"{case.name}: output_cost {breakdown.output_cost} != expected {expected.output_cost}"
)
assert row.prompt_tokens == expected.prompt_tokens, (
f"{case.name}: prompt_tokens {row.prompt_tokens} != expected {expected.prompt_tokens}"
)
assert row.completion_tokens == expected.completion_tokens, (
f"{case.name}: completion_tokens {row.completion_tokens} != expected {expected.completion_tokens}"
)
assert_total_is_sum_of_components(row, case.name)

245
tests/integration/cost_calculation/test_token_pricing.py
View file

@ -1,245 +0,0 @@
"""Token pricing coverage for the integration scripted-shape cost shard."""
from __future__ import annotations
import uuid
from typing import Final, cast
import pytest
from pydantic import JsonValue
from integration._support.client import JSON_OBJECT, Gateway
from integration._support.scripted_shapes import ScriptedUsage, Shape
from integration.cost_calculation.conftest import (
approx_equal,
assert_total_is_sum_of_components,
poll_cost_row,
register_scenario_deployment,
)
from integration.cost_calculation.cost_matrix import (
AUDIO_INPUT_DATA_URL,
FRONTIER_MODELS,
IMAGE_INPUT_DATA_URL,
SERVICE_TIER_REQUEST_SHAPES,
VIDEO_INPUT_DATA_URL,
Case,
FrontierModel,
cases_for,
matrix_data_errors,
recount_cost,
)
if _data_errors := matrix_data_errors():
raise ValueError("\n".join(_data_errors))
def _case_id(param: tuple[FrontierModel, Case]) -> str:
model, case = param
return f"{model.map_key.replace('/', '-')}-{case.name}"
_MATRIX: Final = tuple(
pytest.param(
(model, case),
marks=pytest.mark.covers(
"quota_management.spend_tracking.scripted_wire.logs_cost"
if case.family == "transport"
else "quota_management.spend_tracking.cost_matrix.logs_cost"
),
id=_case_id((model, case)),
)
for model in FRONTIER_MODELS
for case in cases_for(model)
)
_CACHE_SHAPES: Final = frozenset({"anthropic_messages", "bedrock_converse"})
_WEB_SEARCH_OPTION_SHAPES: Final = frozenset({"openai_chat", "openai_responses"})
def _cache_control(usage: ScriptedUsage, shape: Shape) -> dict[str, JsonValue] | None:
if shape not in _CACHE_SHAPES:
return None
if not (usage.cache_read_tokens or usage.cache_write_5m_tokens or usage.cache_write_1h_tokens):
return None
return {"type": "ephemeral", **({"ttl": "1h"} if usage.cache_write_1h_tokens else {})}
def _chat_body(model: FrontierModel, case: Case, model_name: str, marker: str) -> dict[str, JsonValue]:
usage: Final = case.usage_for(model.map_key)
user_parts: Final = [
{"type": "text", "text": f"{marker} summarize the attached material in one line and name the city weather"},
*(
[{"type": "image_url", "image_url": {"url": IMAGE_INPUT_DATA_URL, "detail": "high"}}]
if case.image_input
else []
),
*(
[{"type": "input_audio", "input_audio": {"data": AUDIO_INPUT_DATA_URL.split(",", 1)[1], "format": "wav"}}]
if case.audio_input
else []
),
*(
[{"type": "file", "file": {"file_data": VIDEO_INPUT_DATA_URL, "format": "mp4"}}]
if case.video_input
else []
),
]
tools: Final[list[JsonValue]] = [
*(
[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather and a short forecast for a city.",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"},
"days": {"type": "integer", "description": "Forecast horizon in days"},
"units": {"type": "string", "enum": ["metric", "imperial"]},
},
"required": ["city"],
},
},
}
]
if case.tool_call
else []
),
*(
[{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}]
if case.web_search is not None and model.shape == "anthropic_messages"
else []
),
*(
[{"googleSearch": {}}]
if case.web_search is not None and model.shape == "gemini_generate"
else []
),
*([{"googleMaps": {}}] if case.google_maps else []),
*([{"type": "file_search", "vector_store_ids": ["vs_cost_calc_fixture"]}] if case.file_search else []),
]
cache_control: Final = _cache_control(usage, model.shape)
message: Final = {
"role": "system",
"content": [
{
"type": "text",
"text": "You are a deterministic pricing-harness assistant. Keep answers to a single short line.",
**({"cache_control": cache_control} if cache_control else {}),
}
],
}
return cast(dict[str, JsonValue], {
"model": model_name,
"messages": [message, {"role": "user", "content": user_parts}],
"stream": case.stream,
**({"stream_options": {"include_usage": True}} if case.stream else {}),
**(
{"service_tier": case.service_tier}
if case.service_tier is not None and model.shape in SERVICE_TIER_REQUEST_SHAPES
else {}
),
**({"reasoning_effort": "medium"} if case.reasoning else {}),
**(
{"modalities": ["text", "audio"] if case.audio_output else ["text"]}
if case.audio_input or case.audio_output
else {}
),
**({"audio": {"voice": "alloy", "format": "pcm16"}} if case.audio_output else {}),
**(
{"web_search_options": {"search_context_size": case.web_search}}
if case.web_search is not None and model.shape in _WEB_SEARCH_OPTION_SHAPES
else {}
),
**({"tools": tools} if tools else {}),
**({"tool_choice": "auto"} if case.tool_call and model.shape != "bedrock_converse" else {}),
"allowed_openai_params": [
name
for name, sent in (
("tool_choice", case.tool_call and model.shape != "bedrock_converse"),
("modalities", case.audio_input or case.audio_output),
("audio", case.audio_output),
("web_search_options", case.web_search is not None),
("reasoning_effort", case.reasoning),
)
if sent
],
})
def _assert_stream_has_no_error(response_text: str) -> None:
for line in response_text.splitlines():
if not line.startswith("data:"):
continue
payload = line.removeprefix("data:").strip()
if payload == "[DONE]":
continue
parsed = JSON_OBJECT.validate_json(payload)
assert "error" not in parsed, f"stream carried an error event: {parsed}"
@pytest.mark.parametrize("model_case", _MATRIX)
def test_scripted_usage_bills_at_map_rates(
gateway: Gateway,
model_case: tuple[FrontierModel, Case],
) -> None:
model, case = model_case
marker: Final = uuid.uuid4().hex[:12]
with gateway.scenario() as scenario:
key: Final = scenario.key()
model_name: Final = register_scenario_deployment(scenario, model, case, marker)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(model, case, model_name, marker),
key=key,
)
assert response.is_success, (
f"{model.map_key}/{case.name}: proxy returned {response.status_code}: {response.text[:400]}"
)
if case.stream:
_assert_stream_has_no_error(response.text)
row: Final = poll_cost_row(key)
context: Final = f"{model.map_key}/{case.name}"
if not case.exact_spend:
assert row.prompt_tokens is not None and row.prompt_tokens > 0, (
f"{context}: no-usage stream counted no input tokens: prompt_tokens={row.prompt_tokens}"
)
assert row.completion_tokens is not None and row.completion_tokens > 0, (
f"{context}: no-usage stream counted no output tokens: completion_tokens={row.completion_tokens}"
)
if case.image_input:
assert row.prompt_tokens < 4000, (
f"{context}: image data URL looks tokenized as text: prompt_tokens={row.prompt_tokens}"
)
recount: Final = recount_cost(model, case, row.prompt_tokens, row.completion_tokens)
assert row.spend is not None and approx_equal(
row.spend, recount
), f"{context}: no-usage stream spend {row.spend} != recount {recount} at map rates"
assert_total_is_sum_of_components(row, context)
return
golden: Final = case.expected_for(model)
if not case.stream:
header: Final = cast(str | None, response.headers.get("x-litellm-response-cost"))
assert header is not None and approx_equal(float(header), golden.spend), (
f"{context}: x-litellm-response-cost {header} != golden {golden.spend}"
)
assert row.spend is not None and approx_equal(row.spend, golden.spend), (
f"{context}: spend {row.spend} != golden {golden.spend} "
f"(breakdown {row.breakdown.model_dump()})"
)
breakdown: Final = row.breakdown
assert breakdown.input_cost is not None and approx_equal(breakdown.input_cost, golden.input_cost), (
f"{context}: gross input_cost {breakdown.input_cost} != golden {golden.input_cost}; "
"cached/written tokens billed at the input rate"
)
assert breakdown.output_cost is not None and approx_equal(breakdown.output_cost, golden.output_cost), (
f"{context}: output_cost {breakdown.output_cost} != golden {golden.output_cost}"
)
assert row.prompt_tokens == golden.prompt_tokens, (
f"{context}: prompt_tokens {row.prompt_tokens} != golden {golden.prompt_tokens}"
)
assert row.completion_tokens == golden.completion_tokens, (
f"{context}: completion_tokens {row.completion_tokens} != golden {golden.completion_tokens}"
)
assert_total_is_sum_of_components(row, context)