scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-25 01:02:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

test(integration): tighten batch and realtime cost assertions

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-20 01:29:46 +00:00
parent 807541291d
commit 2e16cd76c1
5 changed files with 46 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
tests/integration/contracts.json
View file

@ -247,8 +247,8 @@
"tests/integration/cost_calculation/test_batch_realtime_cost.py::test_realtime_costs[gpt-realtime-mini-2025-12-15-realtime-priced_from_session_created_model]": [
"quota_management.spend_tracking.realtime_costs.session_model"
],
"tests/integration/cost_calculation/test_batch_realtime_cost.py::test_realtime_no_turn_probe": [
"quota_management.spend_tracking.realtime_costs.no_turn_probe"
"tests/integration/cost_calculation/test_batch_realtime_cost.py::test_realtime_costs[gpt-realtime-mini-2025-12-15-realtime-session_without_turns_zero_spend]": [
"quota_management.spend_tracking.realtime_costs.session_without_turns"
],
"tests/integration/cost_calculation/test_cost_tracking.py::test_case_bills_expected_cost[anthropic.claude-sonnet-5-v1:0-cache_read]": [
"quota_management.spend_tracking.cost_matrix.logs_cost"

10
tests/integration/cost_calculation/assertions.py
View file

@ -15,8 +15,10 @@ def assert_breakdown(
response_content_type: str,
expected: ExactExpected,
breakdown: CostBreakdown,
response: httpx.Response,
response: httpx.Response | None,
) -> None:
if response is None:
assert not expected.cost_header, f"{case_name}: cost headers require an HTTP response"
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}"
)
@ -55,12 +57,12 @@ def assert_breakdown(
assert (actual_component is None and omitted_component_allowed) or (
actual_component is not None and approx_equal(actual_component, expected_component)
), f"{case_name}: {field} {actual_component} != expected {expected_component}"
if expected.cost_header and response_content_type == "application/json":
if response is not None and expected.cost_header and response_content_type == "application/json":
header: str | None = response.headers.get(header_name)
assert (header is None and omitted_component_allowed) or (
header is not None and approx_equal(float(header), expected_component)
), f"{case_name}: {header_name} {header} != expected {expected_component}"
if expected.cost_header and response_content_type == "application/json" and any(
if response is not None and expected.cost_header and response_content_type == "application/json" and any(
component is not None
for component in (
expected.cache_read_cost,
@ -87,7 +89,7 @@ def assert_exact(
response_content_type: str,
expected: ExactExpected,
row: CostRow,
response: httpx.Response,
response: httpx.Response | None,
) -> None:
assert row.spend is not None and approx_equal(row.spend, expected.spend), (
f"{case_name}: spend {row.spend} != expected {expected.spend} "

21
tests/integration/cost_calculation/cost_tracking_case.py
View file

@ -324,6 +324,12 @@ class BatchOutputLine(BaseModel):
raise ValueError("status_code must be 200 or a 4xx status")
return value
@model_validator(mode="after")
def validate_success_tokens(self) -> BatchOutputLine:
if self.status_code == 200 and (self.prompt_tokens is None or self.completion_tokens is None):
raise ValueError("successful batch output lines require prompt and completion tokens")
return self
def render(self, index: int, model: str, request_id: str) -> dict[str, JsonValue]:
if self.status_code != 200:
return {
@ -332,15 +338,18 @@ class BatchOutputLine(BaseModel):
"response": None,
"error": {"code": "bad_request", "message": "failed"},
}
assert self.prompt_tokens is not None
assert self.completion_tokens is not None
usage: dict[str, JsonValue] = {
if self.prompt_tokens is None or self.completion_tokens is None:
raise ValueError("successful batch output lines require prompt and completion tokens")
usage: Final = {
"prompt_tokens": self.prompt_tokens,
"completion_tokens": self.completion_tokens,
"total_tokens": self.prompt_tokens + self.completion_tokens,
**(
{"prompt_tokens_details": {"cached_tokens": self.cached_tokens}}
if self.cached_tokens is not None
else {}
),
}
if self.cached_tokens is not None:
usage["prompt_tokens_details"] = {"cached_tokens": self.cached_tokens}
return {
"id": f"batch_req_{index}",
"custom_id": f"r{index}",
@ -447,7 +456,7 @@ class RealtimeCostCase(BaseModel):
covers: str
model: str
litellm_model: str
turns: tuple[RealtimeTurn, ...] = Field(min_length=1)
turns: tuple[RealtimeTurn, ...] = Field(min_length=0)
session_model: str | None = None
expected: ExactExpected

16
tests/integration/cost_calculation/cost_tracking_cases.json
View file

@ -30237,6 +30237,22 @@
"completion_tokens": 100,
"cost_header": false
}
},
{
"name": "gpt-realtime-mini-2025-12-15-realtime-session_without_turns_zero_spend",
"covers": "quota_management.spend_tracking.realtime_costs.session_without_turns",
"model": "gpt-realtime-mini-2025-12-15",
"litellm_model": "openai/gpt-realtime-mini-2025-12-15",
"turns": [],
"expected": {
"spend": 0.0,
"input_cost": 0.0,
"output_cost": 0.0,
"prompt_tokens": 0,
"completion_tokens": 0,
"breakdown_persisted": false,
"cost_header": false
}
}
]
}

52
tests/integration/cost_calculation/test_batch_realtime_cost.py
View file

@ -2,19 +2,17 @@ from __future__ import annotations
import asyncio
import json
import logging
import os
import time
from hashlib import sha256
from typing import Final
import httpx
import pytest
import websockets
from integration._support.client import JSON_OBJECT, Gateway, Scenario, object_value, string_value
from integration._support.upstream import delete_scenario, register_scenario
from integration.cost_calculation.assertions import assert_exact
from integration.cost_calculation.conftest import CostRow, poll_rows, read_rows_now
from integration.cost_calculation.conftest import poll_rows, read_rows_now
from integration.cost_calculation.cost_tracking_case import (
BATCH_CASES,
REALTIME_CASES,
@ -69,7 +67,6 @@ def _batch_response(case: BatchCostCase) -> JsonResponse | RoutedResponse:
"completed": case.completed_count,
"failed": case.failed_count,
}
completed: Final = len(case.output_lines) > 0
batch: Final = {
"id": "batch-$REQUEST_ID",
"object": "batch",
@ -77,9 +74,9 @@ def _batch_response(case: BatchCostCase) -> JsonResponse | RoutedResponse:
"errors": None,
"input_file_id": "file-in-$REQUEST_ID",
"completion_window": "24h",
"status": "completed" if completed else "completed",
"output_file_id": "file-out-$REQUEST_ID" if completed else None,
"error_file_id": None if completed else "file-err-$REQUEST_ID",
"status": "completed",
"output_file_id": "file-out-$REQUEST_ID" if case.output_lines else None,
"error_file_id": None if case.output_lines else "file-err-$REQUEST_ID",
"created_at": 1,
"in_progress_at": 1,
"completed_at": 1,
@ -168,10 +165,6 @@ def test_batch_costs(gateway: Gateway, case: BatchCostCase) -> None:
assert file_response.is_success, file_response.text
file_body: Final = JSON_OBJECT.validate_json(file_response.content)
time.sleep(2)
file_rows: Final = read_rows_now(key)
if file_rows:
assert all(row.spend == 0.0 for row in file_rows)
logging.info("file creation rows: %s", file_rows)
batch_response: Final = gateway.request(
"POST",
"/v1/batches",
@ -190,17 +183,10 @@ def test_batch_costs(gateway: Gateway, case: BatchCostCase) -> None:
second_retrieval: Final = gateway.request("GET", f"/v1/batches/{batch_id}", key=key)
assert first_retrieval.is_success, first_retrieval.text
assert second_retrieval.is_success, second_retrieval.text
rows: tuple[CostRow, ...]
if case.output_lines:
rows = poll_rows(key, 1)
else:
time.sleep(5)
rows = read_rows_now(key)
if not rows:
logging.info("%s: completed failed batch produced no SpendLogs row", case.name)
return
rows: Final = poll_rows(key, 1)
retrieval_rows: Final = tuple(row for row in rows if row.call_type == "aretrieve_batch")
assert len(retrieval_rows) == 1
assert all(row.spend == 0.0 for row in rows if row.call_type != "aretrieve_batch")
row: Final = retrieval_rows[0]
assert row.status == "success"
assert row.call_type == "aretrieve_batch"
@ -261,28 +247,4 @@ def test_realtime_costs(gateway: Gateway, case: RealtimeCostCase) -> None:
assert row.status == "success"
assert row.call_type == "_arealtime"
assert row.model_id == identity
assert_exact(case.name, "application/json", case.expected, row, httpx.Response(200))
@pytest.mark.covers("quota_management.spend_tracking.realtime_costs.no_turn_probe")
def test_realtime_no_turn_probe(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
key: Final = scenario.key()
model_name, _identity = _register_deployment(
scenario,
"openai/gpt-realtime-mini-2025-12-15",
RealtimeResponse(content_type="application/x-realtime", events=()),
"realtime-no-turn",
realtime=True,
)
asyncio.run(
_run_realtime(
os.environ["INTEGRATION_PROXY_URL"].rstrip("/"),
key,
model_name,
0,
)
)
time.sleep(3)
rows: Final = read_rows_now(key)
logging.info("realtime no-turn probe rows=%s spend=%s", len(rows), rows[0].spend if rows else None)
assert_exact(case.name, "application/json", case.expected, row, None)