Merge origin/main into litellm_embedding_only_supported_endpoints

Keeps both appended test blocks in tests/unit/proxy/test_route_llm_request.py.
Four pre-call tests in test_common_request_processing.py now pass a real
UserAPIKeyAuth, since the spec mock they used came from a module where the
name is Any and had no team_id
This commit is contained in:
mateo-berri 2026-10-13 12:54:23 -07:00
commit c9da3bfb15
37 changed files with 2587 additions and 144 deletions

View file

@ -1,6 +1,6 @@
[project]
name = "litellm-enterprise"
version = "0.1.76"
version = "0.1.77"
description = "Package for LiteLLM Enterprise features"
readme = "README.md"
requires-python = ">=3.9"
@ -26,7 +26,7 @@ required-version = ">=0.10.9"
module-root = ""
[tool.commitizen]
version = "0.1.76"
version = "0.1.77"
version_files = [
"pyproject.toml:^version",
"../pyproject.toml:litellm-enterprise==",

View file

@ -1,6 +1,6 @@
[project]
name = "litellm-proxy-extras"
version = "0.4.108"
version = "0.4.109"
description = "Additional files for the LiteLLM Proxy. Reduces the size of the main litellm package."
readme = "README.md"
requires-python = ">=3.9"
@ -30,7 +30,7 @@ required-version = ">=0.10.9"
module-root = ""
[tool.commitizen]
version = "0.4.108"
version = "0.4.109"
version_files = [
"pyproject.toml:^version",
"../pyproject.toml:litellm-proxy-extras==",

View file

@ -183,10 +183,10 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
OpenAI's Responses API rejects max_output_tokens below 16 for every model
(not gpt-5 specific), so a client like Claude Code that sends a max_tokens=1
warmup probe on model switch would otherwise 400. Values that are None or
already at/above the minimum are returned unchanged.
warmup probe on model switch would otherwise 400. Values that are None,
already at/above the minimum, or not positive are returned unchanged.
"""
if isinstance(max_output_tokens, int) and max_output_tokens < OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS:
if isinstance(max_output_tokens, int) and 0 < max_output_tokens < OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS:
return OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS
return max_output_tokens

View file

@ -20,13 +20,17 @@ from litellm.types.vector_store_files import (
VectorStoreFileObject,
VectorStoreFileUpdateRequest,
)
from litellm.utils import add_openai_metadata
def _clean_dict(source: Mapping[str, object]) -> dict[str, object]:
return {k: v for k, v in source.items() if v is not None}
def _visible_attributes(attributes: Mapping[str, object]) -> Mapping[str, str]:
"""The caller's string attributes, all of them, so the provider enforces its own key limit."""
return {str(k): v for k, v in attributes.items() if k != "hidden_params" and isinstance(v, str)}
class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
ASSISTANTS_HEADER_KEY = "OpenAI-Beta"
ASSISTANTS_HEADER_VALUE = "assistants=v2"
@ -106,11 +110,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
payload: Final[dict[str, object]] = _clean_dict(dict(create_request))
attributes: Final = payload.get("attributes")
if isinstance(attributes, dict):
filtered_attributes: Final = add_openai_metadata(attributes)
if filtered_attributes is not None:
payload["attributes"] = filtered_attributes
else:
payload.pop("attributes", None)
payload["attributes"] = _visible_attributes(attributes)
url: Final = api_base
return url, payload
@ -211,11 +211,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
payload: Final[dict[str, object]] = dict(update_request)
attributes: Final = payload.get("attributes")
if isinstance(attributes, dict):
filtered_attributes: Final = add_openai_metadata(attributes)
if filtered_attributes is not None:
payload["attributes"] = filtered_attributes
else:
payload.pop("attributes", None)
payload["attributes"] = _visible_attributes(attributes)
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base}/{encoded_file_id}", payload

View file

@ -72588,8 +72588,8 @@
"input_cost_per_token_priority": 3.6e-06,
"litellm_provider": "gemini",
"max_input_tokens": 1048576,
"max_output_tokens": 32768,
"max_tokens": 32768,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_reasoning_token": 1e-05,
"output_cost_per_token": 1e-05,
@ -72690,30 +72690,31 @@
"supports_web_search": true,
"web_search_billing_unit": "per_query"
},
"gemini-4-argon": {
"cache_read_input_token_cost": 1e-07,
"cache_read_input_token_cost_batches": 5e-08,
"cache_read_input_token_cost_flex": 5e-08,
"cache_read_input_token_cost_priority": 1.8e-07,
"vertex_ai/gemini-4-argon-cyber-permissive": {
"cache_read_input_token_cost": 2e-07,
"cache_read_input_token_cost_batches": 1e-07,
"cache_read_input_token_cost_flex": 1e-07,
"cache_read_input_token_cost_priority": 3.6e-07,
"google_maps_grounding_cost_per_query": 0.014,
"input_cost_per_audio_token": 2e-06,
"input_cost_per_audio_token_batches": 1e-06,
"input_cost_per_audio_token_priority": 3.6e-06,
"input_cost_per_token": 2e-06,
"input_cost_per_token_batches": 1e-06,
"input_cost_per_token_flex": 1e-06,
"input_cost_per_token_priority": 3.6e-06,
"litellm_provider": "vertex_ai-language-models",
"input_cost_per_audio_token": 4e-06,
"input_cost_per_audio_token_batches": 2e-06,
"input_cost_per_audio_token_priority": 7.2e-06,
"input_cost_per_token": 4e-06,
"input_cost_per_token_batches": 2e-06,
"input_cost_per_token_flex": 2e-06,
"input_cost_per_token_priority": 7.2e-06,
"litellm_provider": "vertex_ai",
"max_input_tokens": 1048576,
"max_output_tokens": 1048576,
"max_tokens": 1048576,
"mode": "chat",
"output_cost_per_reasoning_token": 1e-05,
"output_cost_per_token": 1e-05,
"output_cost_per_token_batches": 5e-06,
"output_cost_per_token_flex": 5e-06,
"output_cost_per_token_priority": 1.8e-05,
"output_cost_per_reasoning_token": 2e-05,
"output_cost_per_token": 2e-05,
"output_cost_per_token_batches": 1e-05,
"output_cost_per_token_flex": 1e-05,
"output_cost_per_token_priority": 3.6e-05,
"prompt_cache_min_tokens": 6144.0,
"regional_endpoint_uplift_multiplier": 1.1,
"search_context_cost_per_query": {"search_context_size_low":0.014,"search_context_size_medium":0.014,"search_context_size_high":0.014},
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
"supported_endpoints": [
@ -72731,7 +72732,6 @@
"text"
],
"supports_audio_input": true,
"supports_audio_output": false,
"supports_function_calling": true,
"supports_minimal_reasoning_effort": false,
"supports_native_streaming": true,

View file

@ -102,7 +102,9 @@ def _validate_tool_arguments(schema: Mapping[str, JsonValue], arguments: Mapping
async def _tool_argument_validation_error(
schema: Mapping[str, JsonValue], arguments: Mapping[str, JsonValue]
schema: Mapping[str, JsonValue],
arguments: Mapping[str, JsonValue],
timeout_seconds: float = _VALIDATION_TIMEOUT_SECONDS,
) -> str | None:
limit_error: Final = _validation_limit_error(schema, arguments)
if limit_error is not None:
@ -112,7 +114,7 @@ async def _tool_argument_validation_error(
if existing is None:
_SCHEMA_VALIDATION_LIMITER.set(limiter)
try:
with anyio.fail_after(_VALIDATION_TIMEOUT_SECONDS):
with anyio.fail_after(timeout_seconds):
return await to_process.run_sync(
_validate_tool_arguments, schema, arguments, cancellable=True, limiter=limiter
)

View file

@ -90,6 +90,8 @@ def parse_canonical_date_range(start_date: str | None, end_date: str | None) ->
end: Final = parse_canonical_date(end_date)
if start is None or end is None:
return InvalidDateRange(reason="start_date and end_date must be valid YYYY-MM-DD dates")
if end < start:
return InvalidDateRange(reason="end_date must be on or after start_date")
return CanonicalDateRange(start=start, end=end)

View file

@ -6833,8 +6833,6 @@ def aggregated_date_range_error(start_date: str | None, end_date: str | None) ->
date_range: Final = parse_canonical_date_range(start_date, end_date)
if isinstance(date_range, InvalidDateRange):
return date_range.reason
if date_range.end < date_range.start:
return "end_date must be on or after start_date"
if (date_range.end - date_range.start).days > _MAX_AGGREGATED_RANGE_DAYS:
return f"Date range must be at most {_MAX_AGGREGATED_RANGE_DAYS} days"
return None

View file

@ -497,12 +497,29 @@ async def _execute_tool_call(
user_id: str | None,
is_admin: bool,
) -> str:
"""Run a single tool and return the summarised result text."""
"""Run a single tool and return the summarised result text, or the reason the query
layer rejected its dates so the model can correct them."""
kwargs: Final = _resolve_fetch_kwargs(fn_name, fn_args, user_id, is_admin)
raw_data: Final = await handler["fetch"](**kwargs)
try:
raw_data: Final = await handler["fetch"](**kwargs)
except HTTPException as e:
reason: Final = _invalid_date_reason(e)
if reason is None:
raise
return f"Invalid date range: {reason}"
return handler["summarise"](raw_data)
def _invalid_date_reason(error: HTTPException) -> str | None:
"""The reason in the daily activity query layer's 400 for invalid or reversed dates."""
detail: Final = cast(object, error.detail) # cast-ok: starlette types detail as str; the query layer passes a dict
match detail:
case {"error": str() as reason} if error.status_code == 400:
return reason
case _:
return None
async def _process_tool_call(
tc: ChatCompletionMessageToolCall,
chat_messages: list[Mapping[str, object]],

View file

@ -17,6 +17,7 @@ from litellm.llms.base_llm.ocr.transformation import (
parse_ocr_request_format,
)
from litellm.proxy._types import *
from litellm.proxy._types import ProxyException
from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
@ -32,6 +33,15 @@ class _NamedUpload(io.BytesIO):
self.name = name
class OCRRequestError(ProxyException):
"""A malformed OCR request, answered with a 400 before any provider call."""
def __init__(self, message: str, param: str | None = None) -> None:
super().__init__( # pyright: ignore[reportUnknownMemberType] # ProxyException.__init__ takes an untyped provider_specific_fields dict
message=message, type="invalid_request_error", param=param, code=400
)
def _build_document_from_upload(
file_content: bytes,
filename: str | None,
@ -99,7 +109,7 @@ async def _parse_multipart_form(request: Request) -> dict[str, object]:
try:
form: Final = await request.form()
except Exception as e:
raise ValueError(
raise OCRRequestError(
f"Failed to parse multipart form data: {e}. "
"When using curl with --form/-F, do NOT set the Content-Type header "
"manually — curl will set it automatically with the required boundary."
@ -110,7 +120,7 @@ async def _parse_multipart_form(request: Request) -> dict[str, object]:
# depending on middleware; check both via isinstance (FastAPI's UploadFile
# is a subclass of Starlette's) and fall back to duck-type check.
if uploaded_file is None or (not isinstance(uploaded_file, UploadFile) and not hasattr(uploaded_file, "read")):
raise ValueError("Multipart OCR request must include a 'file' field with the document to process")
raise OCRRequestError("Multipart OCR request must include a 'file' field with the document to process", "file")
uploaded_file = cast(UploadFile, uploaded_file)
@ -118,9 +128,9 @@ async def _parse_multipart_form(request: Request) -> dict[str, object]:
await uploaded_file.seek(0)
file_content: Final = await uploaded_file.read(_MAX_FILE_BYTES + 1)
if not file_content:
raise ValueError("Uploaded file is empty")
raise OCRRequestError("Uploaded file is empty", "file")
if len(file_content) > _MAX_FILE_BYTES:
raise ValueError("OCR file exceeds the size limit")
raise OCRRequestError("OCR file exceeds the size limit", "file")
document: Final = _build_document_from_upload(
file_content=file_content,
@ -197,7 +207,7 @@ async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
)
return await _parse_multipart_form(request)
raise ValueError(
raise OCRRequestError(
"Empty request body. For file uploads, use multipart/form-data content type "
"with a file field. When using curl with --form/-F, do NOT set the Content-Type "
"header manually."
@ -206,7 +216,7 @@ async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
try:
data: Final = orjson.loads(body)
except orjson.JSONDecodeError as e:
raise ValueError(
raise OCRRequestError(
f"Invalid JSON in request body: {e}. "
"Ensure the request body is valid JSON with Content-Type: application/json, "
"or use multipart/form-data for file uploads."
@ -220,7 +230,7 @@ async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
# File uploads must go through multipart/form-data instead.
doc: Final = data.get("document") if isinstance(data, dict) else None
if isinstance(doc, dict) and doc.get("type") == "file":
raise ValueError(
raise OCRRequestError(
"document type 'file' is not supported through the JSON API. "
"To upload a local file, use multipart/form-data with a 'file' field. "
"For JSON requests, use 'document_url' or 'image_url' document types."
@ -237,7 +247,7 @@ async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
for url_field in ("document_url", "image_url"):
url_value = doc.get(url_field)
if isinstance(url_value, str) and url_value.startswith("reducto://"):
raise ValueError(
raise OCRRequestError(
"reducto:// file IDs are not accepted through the proxy "
"OCR API; upload the file in the same request via "
"multipart/form-data with a 'file' field, or pass an "

View file

@ -246,13 +246,18 @@ def _find_missing_required_body_param(
if not missing_present_params:
return None
candidate_litellm_params: Final = _candidate_deployment_litellm_params(data, llm_router)
router_default_litellm_params: Final = _router_default_litellm_params(route_type, data, llm_router)
router_default_litellm_params: Final = _router_default_litellm_params(
route_type, data, llm_router, model_listed=bool(candidate_litellm_params)
)
missing_param: Final = next(
(
param
for param in missing_present_params
if router_default_litellm_params.get(param) is None
and not any(deployment_params.get(param) is not None for deployment_params in candidate_litellm_params)
and (
route_type in _ROUTE_TYPES_WITH_POSITIONAL_REQUIRED_PARAM
or not any(deployment_params.get(param) is not None for deployment_params in candidate_litellm_params)
)
),
None,
)
@ -261,19 +266,34 @@ def _find_missing_required_body_param(
return MissingBodyParam(name=missing_param, model_deployments_loaded=bool(candidate_litellm_params))
_ROUTE_TYPES_WITHOUT_ROUTER_DEFAULTS_MERGE: Final[frozenset[str]] = frozenset(
{"asearch", "acreate_agent", "acreate_eval", "acreate_run"}
# The Router method takes the route's required param positionally, so it binds before
# any default or deployment param is merged in, and only the request body can supply it.
_ROUTE_TYPES_WITH_POSITIONAL_REQUIRED_PARAM: Final[frozenset[str]] = frozenset(
{"aimage_generation", "aspeech", "atext_completion", "atranscription"}
)
_ROUTE_TYPES_WITHOUT_ROUTER_DEFAULTS_MERGE: Final[frozenset[str]] = (
frozenset({"asearch", "acreate_agent", "acreate_eval", "acreate_run"}) | _ROUTE_TYPES_WITH_POSITIONAL_REQUIRED_PARAM
)
# The Router merges defaults on these only when `get_model_list` lists the model.
_ROUTE_TYPES_MERGING_ONLY_LISTED_MODELS: Final[frozenset[str]] = frozenset({"amoderation"})
def _router_default_litellm_params(
route_type: str,
data: Mapping[str, object],
llm_router: LitellmRouter | None,
*,
model_listed: bool,
) -> Mapping[str, object]:
# Mirror exactly the defaults the dispatching router will merge at dispatch time:
# user_config requests dispatch on their own throwaway Router, and the listed route
# types (plus model-less direct dispatch) never pass through the router's merge.
# the listed route types either never pass through the router's merge or take the
# param positionally, which binds before the merge; user_config requests dispatch on
# their own throwaway Router; model-less direct dispatch never merges; and moderation
# merges only for a model `get_model_list` lists.
if route_type in _ROUTE_TYPES_WITHOUT_ROUTER_DEFAULTS_MERGE:
return {}
user_config: Final[Mapping[str, object] | None] = (
data.get("user_config") if isinstance(data.get("user_config"), Mapping) else None
)
@ -281,7 +301,11 @@ def _router_default_litellm_params(
defaults: Final[Mapping[str, object] | None] = user_config.get("default_litellm_params")
return defaults if isinstance(defaults, Mapping) else {}
model_name: Final = data.get("model")
if route_type in _ROUTE_TYPES_WITHOUT_ROUTER_DEFAULTS_MERGE or not isinstance(model_name, str) or not model_name:
if (
not isinstance(model_name, str)
or not model_name
or (route_type in _ROUTE_TYPES_MERGING_ONLY_LISTED_MODELS and not model_listed)
):
return {}
router_defaults: Final[Mapping[str, object] | None] = getattr(llm_router, "default_litellm_params", None)
return router_defaults if isinstance(router_defaults, Mapping) else {}
@ -689,6 +713,8 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr
raise_if_mock_testing_params_disallowed(data, allowed=mock_testing_params_allowed())
data.pop("enable_tag_filtering", None)
if "user_config" in data and data["user_config"] is None:
data.pop("user_config") # pyright: ignore[reportUnknownMemberType] # route_request takes an untyped dict body
team_id: Final = get_team_id_from_data(data)
router_model_names: Final = llm_router.model_names if llm_router is not None else []

View file

@ -72588,8 +72588,8 @@
"input_cost_per_token_priority": 3.6e-06,
"litellm_provider": "gemini",
"max_input_tokens": 1048576,
"max_output_tokens": 32768,
"max_tokens": 32768,
"max_output_tokens": 65536,
"max_tokens": 65536,
"mode": "chat",
"output_cost_per_reasoning_token": 1e-05,
"output_cost_per_token": 1e-05,
@ -72690,30 +72690,31 @@
"supports_web_search": true,
"web_search_billing_unit": "per_query"
},
"gemini-4-argon": {
"cache_read_input_token_cost": 1e-07,
"cache_read_input_token_cost_batches": 5e-08,
"cache_read_input_token_cost_flex": 5e-08,
"cache_read_input_token_cost_priority": 1.8e-07,
"vertex_ai/gemini-4-argon-cyber-permissive": {
"cache_read_input_token_cost": 2e-07,
"cache_read_input_token_cost_batches": 1e-07,
"cache_read_input_token_cost_flex": 1e-07,
"cache_read_input_token_cost_priority": 3.6e-07,
"google_maps_grounding_cost_per_query": 0.014,
"input_cost_per_audio_token": 2e-06,
"input_cost_per_audio_token_batches": 1e-06,
"input_cost_per_audio_token_priority": 3.6e-06,
"input_cost_per_token": 2e-06,
"input_cost_per_token_batches": 1e-06,
"input_cost_per_token_flex": 1e-06,
"input_cost_per_token_priority": 3.6e-06,
"litellm_provider": "vertex_ai-language-models",
"input_cost_per_audio_token": 4e-06,
"input_cost_per_audio_token_batches": 2e-06,
"input_cost_per_audio_token_priority": 7.2e-06,
"input_cost_per_token": 4e-06,
"input_cost_per_token_batches": 2e-06,
"input_cost_per_token_flex": 2e-06,
"input_cost_per_token_priority": 7.2e-06,
"litellm_provider": "vertex_ai",
"max_input_tokens": 1048576,
"max_output_tokens": 1048576,
"max_tokens": 1048576,
"mode": "chat",
"output_cost_per_reasoning_token": 1e-05,
"output_cost_per_token": 1e-05,
"output_cost_per_token_batches": 5e-06,
"output_cost_per_token_flex": 5e-06,
"output_cost_per_token_priority": 1.8e-05,
"output_cost_per_reasoning_token": 2e-05,
"output_cost_per_token": 2e-05,
"output_cost_per_token_batches": 1e-05,
"output_cost_per_token_flex": 1e-05,
"output_cost_per_token_priority": 3.6e-05,
"prompt_cache_min_tokens": 6144.0,
"regional_endpoint_uplift_multiplier": 1.1,
"search_context_cost_per_query": {"search_context_size_low":0.014,"search_context_size_medium":0.014,"search_context_size_high":0.014},
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
"supported_endpoints": [
@ -72731,7 +72732,6 @@
"text"
],
"supports_audio_input": true,
"supports_audio_output": false,
"supports_function_calling": true,
"supports_minimal_reasoning_effort": false,
"supports_native_streaming": true,

View file

@ -1,6 +1,6 @@
[project]
name = "litellm"
version = "1.106.0"
version = "1.107.0"
description = "Library to easily interface with LLM API providers"
readme = "README.md"
requires-python = ">=3.10, <3.15"
@ -77,8 +77,8 @@ proxy = [
"mcp>=2.2.0,<3",
"httpx2>=2.5.0,<3",
"pydantic>=2.12.0,<3",
"litellm-proxy-extras==0.4.108",
"litellm-enterprise==0.1.76",
"litellm-proxy-extras==0.4.109",
"litellm-enterprise==0.1.77",
"RestrictedPython>=8.5,<9.0",
"rich>=13.9.4,<14.0",
"InquirerPy>=0.3.4,<1.0",
@ -372,7 +372,7 @@ litellm-enterprise = { workspace = true }
members = ["enterprise", "litellm-proxy-extras"]
[tool.commitizen]
version = "1.106.0"
version = "1.107.0"
version_files = [
"pyproject.toml:^version",
]

View file

@ -10,6 +10,7 @@ import sys
import time
import uuid
from collections.abc import Generator, Iterable, Iterator, Mapping
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from enum import Enum
@ -37,6 +38,7 @@ DB_PUSH: Final = ("--use_prisma_db_push",)
MIGRATE_DEPLOY: Final = ()
LEGACY_MIGRATE_DEPLOY: Final = ("--use_legacy_migration_resolver",)
CONFIGURATION: Final = TypeAdapter(dict[str, object])
_DEPLOYMENT_IDS: Final = TypeAdapter(dict[str, list[str]])
MCP_COVERAGE_CONFIG: Final = Path(__file__).resolve().parents[1] / "mcp_coverage.toml"
@ -208,6 +210,33 @@ class OwnedProxy:
identity: Final = response.headers.get("x-integration-worker-pid", "")
return int(identity) if response.status_code == 200 and identity.isdecimal() else None
def _deployment_ids_on_a_worker(self, model_name: str) -> tuple[int, frozenset[str]] | None:
try:
with httpx.Client(
base_url=str(self.gateway.client.base_url),
timeout=2,
trust_env=False,
limits=httpx.Limits(max_keepalive_connections=0),
) as client:
response: Final = client.get("/health/integration-deployment-ids", params={"model_name": model_name})
except httpx.TransportError:
return None
identity: Final = response.headers.get("x-integration-worker-pid", "")
if response.status_code != 200 or not identity.isdecimal():
return None
return int(identity), frozenset(_DEPLOYMENT_IDS.validate_python(response.json())["model_ids"])
def serves_only_on_every_worker(self, model_name: str, model_id: str) -> bool:
with ThreadPoolExecutor(max_workers=8) as pool:
answers: Final = tuple(pool.map(self._deployment_ids_on_a_worker, (model_name,) * 16))
samples: Final = tuple(answer for answer in answers if answer is not None)
workers: Final = self.worker_pids()
return (
bool(workers)
and workers <= frozenset(pid for pid, _ in samples)
and all(model_ids == {model_id} for _, model_ids in samples)
)
def _await_replacement(self, before: frozenset[int]) -> int:
budget: Final = readiness_seconds(1)
deadline: Final = time.monotonic() + budget
@ -494,6 +523,8 @@ def _prisma_running(launch: _Launch) -> bool:
def _answer(client: httpx.Client, launch: _Launch) -> tuple[bool, int | None]:
if not _owns_listener(launch):
return False, None
try:
response: Final = client.get("/health/integration-readiness", timeout=2)
except httpx.ConnectError:
@ -501,7 +532,7 @@ def _answer(client: httpx.Client, launch: _Launch) -> tuple[bool, int | None]:
except httpx.TransportError:
return True, None
identity: Final = response.headers.get("x-integration-worker-pid", "")
ready: Final = response.status_code == 200 and identity.isdecimal() and _owns_listener(launch)
ready: Final = response.status_code == 200 and identity.isdecimal()
return True, int(identity) if ready else None
@ -833,7 +864,7 @@ def refused_boot_log(
with httpx.Client(base_url=f"http://127.0.0.1:{launch.port}", timeout=15, trust_env=False) as client:
deadline: Final = time.monotonic() + readiness_seconds(1)
while launch.process.poll() is None:
assert not _is_ready(client), (
assert not (_owns_listener(launch) and _is_ready(client)), (
f"Proxy became ready instead of refusing to boot:\n{launch.log.read_text()}"
)
assert time.monotonic() < deadline, "Proxy neither exited nor became ready within the deadline"

View file

@ -26,7 +26,15 @@ async def worker_readiness() -> JSONResponse:
)
async def worker_deployment_ids(model_name: str) -> JSONResponse:
from litellm.proxy.proxy_server import llm_router
model_ids: Final = [] if llm_router is None else llm_router.get_model_ids(model_name=model_name)
return JSONResponse(content={"model_ids": model_ids}, headers={"x-integration-worker-pid": str(os.getpid())})
def configure_readiness() -> None:
from litellm.proxy.proxy_server import app
app.add_api_route("/health/integration-readiness", worker_readiness, include_in_schema=False)
app.add_api_route("/health/integration-deployment-ids", worker_deployment_ids, include_in_schema=False)

View file

@ -0,0 +1,280 @@
"""Shared payloads and wire helpers for the status-code follow-up audit cells.
Every cell drives a real proxy against the scripted upstream and reads the outbound provider body back from the
upstream's observation feed, so a cell asserts both what the caller got and what the provider received."""
from __future__ import annotations
import json
import os
import uuid
from collections.abc import Iterator, Mapping, Sequence
from contextlib import contextmanager
from pathlib import Path
from typing import Final
import httpx
from integration._support.client import JSON_OBJECT, Gateway, Scenario, gateway_from_environment, object_value
from integration._support.process import owned_proxy_process
from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
from pydantic import JsonValue
from tests.integration.cost_calculation.cost_tracking_case import (
JsonResponse,
SseResponse,
StoredResponse,
)
ROUTER_DEFAULTS: Final[dict[str, JsonValue]] = {
"prompt": "router default prompt",
"input": "router default input",
"max_tokens": 32,
}
CHAT: Final[dict[str, JsonValue]] = {
"id": "chatcmpl-$UNIQUE_ID",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "scripted"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
RESPONSE: Final[dict[str, JsonValue]] = {
"id": "resp_$UNIQUE_ID",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [
{
"type": "message",
"id": "msg_$UNIQUE_ID",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "scripted", "annotations": []}],
}
],
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
}
MESSAGE: Final[dict[str, JsonValue]] = {
"id": "msg_$UNIQUE_ID",
"type": "message",
"role": "assistant",
"model": "claude-haiku-4-5",
"content": [{"type": "text", "text": "scripted"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
}
MODERATION: Final[dict[str, JsonValue]] = {
"id": "modr-$UNIQUE_ID",
"model": "omni-moderation-latest",
"results": [{"flagged": False, "categories": {}, "category_scores": {}}],
}
CHAT_FRAMES: Final = (
'data: {"id":"chatcmpl-$UNIQUE_ID","object":"chat.completion.chunk","created":1,"model":"gpt-4o-mini",'
'"choices":[{"index":0,"delta":{"role":"assistant","content":"streamed "},"finish_reason":null}]}',
'data: {"id":"chatcmpl-$UNIQUE_ID","object":"chat.completion.chunk","created":1,"model":"gpt-4o-mini",'
'"choices":[{"index":0,"delta":{"content":"response"},"finish_reason":null}]}',
'data: {"id":"chatcmpl-$UNIQUE_ID","object":"chat.completion.chunk","created":1,"model":"gpt-4o-mini",'
'"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}',
"data: [DONE]",
)
RESPONSES_FRAMES: Final = (
"event: response.created\n"
'data: {"type":"response.created","sequence_number":0,"response":{"id":"resp_$UNIQUE_ID","object":"response",'
'"created_at":1,"status":"in_progress","model":"gpt-4o-mini","output":[],"usage":null}}',
"event: response.output_text.delta\n"
'data: {"type":"response.output_text.delta","sequence_number":1,"item_id":"msg_$UNIQUE_ID","output_index":0,'
'"content_index":0,"delta":"streamed response"}',
"event: response.completed\n"
'data: {"type":"response.completed","sequence_number":2,"response":{"id":"resp_$UNIQUE_ID","object":"response",'
'"created_at":1,"status":"completed","model":"gpt-4o-mini","output":[{"type":"message","id":"msg_$UNIQUE_ID",'
'"status":"completed","role":"assistant","content":[{"type":"output_text","text":"streamed response",'
'"annotations":[]}]}],"usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}}',
)
MESSAGE_FRAMES: Final = (
"event: message_start\n"
'data: {"type":"message_start","message":{"id":"msg_$UNIQUE_ID","type":"message","role":"assistant",'
'"model":"claude-haiku-4-5","content":[],"stop_reason":null,"stop_sequence":null,'
'"usage":{"input_tokens":1,"output_tokens":0}}}',
"event: content_block_start\n"
'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
"event: content_block_delta\n"
'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"streamed response"}}',
'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
"event: message_delta\n"
'data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
'event: message_stop\ndata: {"type":"message_stop"}',
)
UNSET: Final = object()
USER_MESSAGES: Final[list[JsonValue]] = [{"role": "user", "content": "Hello"}]
STREAM_TERMINALS: Final[Mapping[str, str]] = {
"chat": "data: [DONE]",
"responses": '"type":"response.completed"',
"messages": "event: message_stop",
}
ENDPOINT_PATHS: Final[Mapping[str, str]] = {
"chat": "/v1/chat/completions",
"responses": "/v1/responses",
"messages": "/v1/messages",
}
ENDPOINT_BODIES: Final[Mapping[str, dict[str, JsonValue]]] = {
"chat": {"messages": USER_MESSAGES},
"responses": {"input": "Hello", "store": False},
"messages": {"messages": USER_MESSAGES, "max_tokens": 16},
}
def json_response(body: dict[str, JsonValue], status: int = 200) -> JsonResponse:
return JsonResponse(content_type="application/json", body=body, status=status)
def sse_response(frames: tuple[str, ...]) -> SseResponse:
return SseResponse(content_type="text/event-stream", frames=frames)
def openai_error(message: str, param: str) -> JsonResponse:
return json_response(
{"error": {"message": message, "type": "invalid_request_error", "param": param, "code": None}}, 400
)
class Upstream:
"""Reads the scripted upstream's destructive observation feed and keeps every record it drained."""
def __init__(self, url: str) -> None:
self.url = url.rstrip("/")
self.items: tuple[dict[str, JsonValue], ...] = ()
def drain(self) -> None:
with httpx.Client(timeout=10, trust_env=False) as client:
payload: Final = JSON_OBJECT.validate_python(
client.get(f"{self.url}/__observations?include_method=true").json()
)
requests: Final = payload.get("requests")
assert isinstance(requests, list)
self.items = (*self.items, *(object_value(item) for item in requests if isinstance(item, dict)))
def calls(self, identity: str) -> tuple[dict[str, JsonValue], ...]:
self.drain()
return tuple(
item
for item in self.items
if f"/{identity}/" in str(item.get("path")) and not str(item.get("path", "")).endswith("/models")
)
def drain_rig_upstream() -> None:
"""Drops every observation earlier cells left on the shared upstream, so a cell reads only its own calls."""
Upstream(os.environ["INTEGRATION_UPSTREAM_URL"]).drain()
def assert_no_provider_call(gateway: Gateway, *identities: str) -> None:
upstream: Final = Upstream(gateway.upstream_url)
assert {identity: upstream.calls(identity) for identity in identities} == dict.fromkeys(identities, ())
def one_outbound(gateway: Gateway, identity: str) -> dict[str, JsonValue]:
calls: Final = Upstream(gateway.upstream_url).calls(identity)
assert len(calls) == 1, calls
return object_value(calls[0]["body"])
def post(
gateway: Gateway,
path: str,
body: Mapping[str, JsonValue],
*,
key: str | None = None,
headers: Mapping[str, str] | None = None,
) -> httpx.Response:
return gateway.client.post(
path,
json=dict(body),
headers={"Authorization": f"Bearer {gateway.key if key is None else key}", **(headers or {})},
timeout=30,
)
def stream_lines(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple[int, tuple[str, ...]]:
with gateway.client.stream(
"POST", path, json=dict(body), headers={"Authorization": f"Bearer {gateway.key}"}, timeout=30
) as response:
return response.status_code, tuple(response.iter_lines())
def error_body(response: httpx.Response) -> dict[str, JsonValue]:
"""The error object of an OpenAI-shaped or Anthropic-shaped error response."""
return object_value(JSON_OBJECT.validate_python(response.json())["error"])
def invalid_request(response: httpx.Response) -> dict[str, JsonValue]:
assert response.status_code == 400, response.text
error: Final = error_body(response)
assert error.get("type") == "invalid_request_error", response.text
return error
def register(scenario: Scenario, prefix: str, response: StoredResponse) -> tuple[str, ScenarioHandle]:
identity: Final = f"{prefix}-{uuid.uuid4().hex}"
handle: Final = register_scenario(identity, response)
scenario.cleanups.callback(delete_scenario, handle)
return identity, handle
def deployment(
scenario: Scenario, name: str, model: str, response: StoredResponse, **extra: JsonValue
) -> tuple[str, dict[str, JsonValue]]:
identity, handle = register(scenario, name, response)
return identity, {
"model_name": name,
"litellm_params": {"model": model, "api_base": handle.api_base(), "api_key": identity, **extra},
}
@contextmanager
def owned_gateway(
directory: Path,
config: Mapping[str, JsonValue],
*,
workers: int = 1,
environment: Mapping[str, str] | None = None,
) -> Iterator[Gateway]:
"""A proxy booted from this checkout with `config` and extra `environment`, sharing the rig's database, Redis
and upstream."""
path: Final = directory / "config.yaml"
path.write_text(json.dumps(dict(config)), encoding="utf-8")
with (
gateway_from_environment() as gateway,
owned_proxy_process(gateway, directory, environment or {}, config=path, workers=workers) as owned,
):
yield Gateway(owned.gateway.client, owned.gateway.key, gateway.upstream_url)
def stream_finished(endpoint: str, lines: Sequence[str]) -> bool:
"""Whether the stream reached the endpoint's terminal frame."""
return any(STREAM_TERMINALS[endpoint] in line for line in lines)
def _frame_text(frame: Mapping[str, JsonValue]) -> str:
"""The text one chat, responses or messages delta frame carries, or "" for any other frame."""
choices: Final = frame.get("choices")
first: Final = choices[0] if isinstance(choices, list) and choices else None
chat_delta: Final = first.get("delta") if isinstance(first, dict) else None
if isinstance(chat_delta, dict) and isinstance(chat_delta.get("content"), str):
return str(chat_delta["content"])
delta: Final = frame.get("delta")
if frame.get("type") == "response.output_text.delta" and isinstance(delta, str):
return delta
if frame.get("type") == "content_block_delta" and isinstance(delta, dict) and delta.get("type") == "text_delta":
return str(delta.get("text"))
return ""
def assembled_text(lines: Sequence[str]) -> str:
"""The text a chat, responses or messages stream carried, joined across its delta frames."""
return "".join(
_frame_text(JSON_OBJECT.validate_json(line.removeprefix("data: ")))
for line in lines
if line.startswith("data: ") and line != "data: [DONE]"
)

View file

@ -0,0 +1,374 @@
from __future__ import annotations
import json
import uuid
from collections.abc import Iterator
from pathlib import Path
from typing import Final
import httpx
import pytest
from integration._support.client import (
JSON_OBJECT,
Gateway,
Scenario,
eventually,
gateway_from_environment,
object_value,
)
from integration._support.process import owned_proxy_process
from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
from pydantic import JsonValue
from tests.integration.cost_calculation.cost_tracking_case import BinaryResponse, JsonResponse
_ROUTER_DEFAULTS: Final[dict[str, JsonValue]] = {
"prompt": "router default prompt",
"input": "router default input",
"max_tokens": 32,
}
_MESSAGE: Final[dict[str, JsonValue]] = {
"id": "msg-$UNIQUE_ID",
"type": "message",
"role": "assistant",
"model": "claude-haiku-4-5",
"content": [{"type": "text", "text": "scripted"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
}
_CHAT: Final[dict[str, JsonValue]] = {
"id": "chatcmpl-$UNIQUE_ID",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": "scripted"}, "finish_reason": "stop"},
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
_RESPONSE: Final[dict[str, JsonValue]] = {
"id": "resp_$UNIQUE_ID",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [],
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
}
_VECTOR_STORE_FILE: Final[dict[str, JsonValue]] = {
"id": "file-audit",
"object": "vector_store.file",
"created_at": 1,
"usage_bytes": 0,
"vector_store_id": "vs-audit",
"status": "completed",
"last_error": None,
"attributes": {},
}
_SEVENTEEN_ATTRIBUTES: Final[dict[str, JsonValue]] = {f"key_{index:02d}": "v" for index in range(17)}
def _openai_error(message: str, param: str) -> JsonResponse:
return JsonResponse(
content_type="application/json",
status=400,
body={"error": {"message": message, "type": "invalid_request_error", "param": param, "code": None}},
)
class _Observations:
def __init__(self, url: str) -> None:
self.url = url.rstrip("/")
self.items: tuple[dict[str, JsonValue], ...] = ()
def read(self) -> tuple[dict[str, JsonValue], ...]:
with httpx.Client(timeout=10, trust_env=False) as client:
payload: Final = JSON_OBJECT.validate_python(
client.get(f"{self.url}/__observations?include_method=true").json()
)
requests: Final = payload.get("requests")
assert isinstance(requests, list)
self.items = (*self.items, *(object_value(item) for item in requests if isinstance(item, dict)))
return self.items
def provider_calls(self, identity: str) -> tuple[dict[str, JsonValue], ...]:
return tuple(
item
for item in self.items
if f"/{identity}/" in str(item.get("path"))
and not (item.get("method") == "GET" and str(item.get("path", "")).endswith(("/v1/models", "/models")))
)
def _scripted(scenario: Scenario, prefix: str, response: JsonResponse | BinaryResponse) -> tuple[str, ScenarioHandle]:
identity: Final = f"{prefix}-{uuid.uuid4().hex}"
handle: Final = register_scenario(identity, response)
scenario.cleanups.callback(delete_scenario, handle)
return identity, handle
def _outbound(gateway: Gateway, identity: str) -> dict[str, JsonValue]:
observations: Final = _Observations(gateway.upstream_url)
eventually(observations.read, lambda _items: len(observations.provider_calls(identity)) == 1, seconds=20)
return object_value(observations.provider_calls(identity)[0]["body"])
def _assert_no_provider_call(gateway: Gateway, identity: str) -> None:
observations: Final = _Observations(gateway.upstream_url)
observations.read()
assert observations.provider_calls(identity) == ()
def _assert_no_call_to_deployment(gateway: Gateway, model_name: str) -> None:
observations: Final = _Observations(gateway.upstream_url)
observations.read()
calls: Final = tuple(
f"{item.get('method')} {item.get('path')}"
for item in observations.items
if f"/{model_name}-" in str(item.get("path")) and not str(item.get("path")).endswith("/models")
)
assert calls == (), calls
def _post(gateway: Gateway, path: str, body: dict[str, JsonValue]) -> httpx.Response:
return gateway.client.post(path, json=body, headers={"Authorization": f"Bearer {gateway.key}"})
def _invalid_request(response: httpx.Response) -> dict[str, JsonValue]:
assert response.status_code == 400, response.text
error: Final = object_value(JSON_OBJECT.validate_python(response.json())["error"])
assert error.get("type") == "invalid_request_error", response.text
return error
def _write_config(path: Path, config: dict[str, JsonValue]) -> Path:
path.write_text(json.dumps(config), encoding="utf-8")
return path
@pytest.fixture(scope="module")
def router_default_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]:
"""A proxy whose router-wide defaults name positional and route-required params,
with client-side `user_config` allowed so a null value can be sent."""
directory: Final = tmp_path_factory.mktemp("router-defaults")
with gateway_from_environment() as gateway, gateway.scenario() as scenario:
image_identity, image_handle = _scripted(
scenario, "followup-image", JsonResponse(content_type="application/json", body={"created": 1, "data": []})
)
text_identity, text_handle = _scripted(
scenario, "followup-text", JsonResponse(content_type="application/json", body={"choices": []})
)
speech_identity, speech_handle = _scripted(
scenario, "followup-speech", BinaryResponse(content_type="audio/mpeg", length=16)
)
message_identity, message_handle = _scripted(
scenario, "followup-message", JsonResponse(content_type="application/json", body=_MESSAGE)
)
config: Final = _write_config(
directory / "router-defaults.yaml",
{
"model_list": [
{
"model_name": "followup-image",
"litellm_params": {
"model": "openai/gpt-image-1",
"api_base": image_handle.api_base(),
"api_key": image_identity,
},
},
{
"model_name": "followup-text",
"litellm_params": {
"model": "openai/gpt-3.5-turbo-instruct",
"api_base": text_handle.api_base(),
"api_key": text_identity,
},
},
{
"model_name": "followup-speech",
"litellm_params": {
"model": "openai/gpt-4o-mini-tts",
"api_base": speech_handle.api_base(),
"api_key": speech_identity,
},
},
{
"model_name": "followup-message",
"litellm_params": {
"model": "anthropic/claude-haiku-4-5",
"api_base": message_handle.api_base(),
"api_key": message_identity,
},
},
],
"router_settings": {"default_litellm_params": _ROUTER_DEFAULTS},
"general_settings": {"allow_client_side_credentials": True},
},
)
with owned_proxy_process(gateway, directory, {}, config=config) as owned:
yield Gateway(owned.gateway.client, owned.gateway.key, gateway.upstream_url)
def test_ocr_multipart_without_file_returns_400(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
identity, handle = _scripted(
scenario, "followup-ocr", JsonResponse(content_type="application/json", body={"pages": []})
)
model: Final = scenario.model(model="mistral/mistral-ocr-latest", api_base=handle.api_base(), api_key=identity)
response: Final = gateway.client.post(
"/v1/ocr", files={"model": (None, model)}, headers={"Authorization": f"Bearer {gateway.key}"}
)
error: Final = _invalid_request(response)
assert error.get("param") == "file", response.text
_assert_no_provider_call(gateway, identity)
def test_ocr_empty_file_returns_400(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
identity, handle = _scripted(
scenario, "followup-ocr-empty", JsonResponse(content_type="application/json", body={"pages": []})
)
model: Final = scenario.model(model="mistral/mistral-ocr-latest", api_base=handle.api_base(), api_key=identity)
response: Final = gateway.request_multipart(
"/v1/ocr", {"model": model}, {"file": ("empty.pdf", b"", "application/pdf")}
)
error: Final = _invalid_request(response)
assert error.get("message") == "Uploaded file is empty", response.text
_assert_no_provider_call(gateway, identity)
@pytest.mark.parametrize("max_output_tokens", (-1, 0), ids=("negative", "zero"))
def test_responses_non_positive_max_output_tokens_returns_provider_400(
gateway: Gateway, max_output_tokens: int
) -> None:
message: Final = f"Invalid 'max_output_tokens': integer below minimum value. Expected a value >= 16, but got {max_output_tokens} instead."
with gateway.scenario() as scenario:
identity, handle = _scripted(scenario, "followup-max-output", _openai_error(message, "max_output_tokens"))
model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=handle.api_base(), api_key=identity)
response: Final = _post(
gateway,
"/v1/responses",
{"model": model, "input": "Say pong", "store": False, "max_output_tokens": max_output_tokens},
)
assert response.status_code == 400, response.text
assert message in response.text, response.text
assert _outbound(gateway, identity).get("max_output_tokens") == max_output_tokens
def test_responses_small_positive_max_output_tokens_is_still_raised_to_minimum(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
identity, handle = _scripted(
scenario, "followup-max-output-min", JsonResponse(content_type="application/json", body=_RESPONSE)
)
model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=handle.api_base(), api_key=identity)
response: Final = _post(
gateway, "/v1/responses", {"model": model, "input": "Say pong", "store": False, "max_output_tokens": 1}
)
assert response.status_code == 200, response.text
assert _outbound(gateway, identity).get("max_output_tokens") == 16
def test_team_daily_activity_end_before_start_returns_400(gateway: Gateway) -> None:
response: Final = gateway.request(
"GET",
"/team/daily/activity",
params={"start_date": "2026-10-01", "end_date": "2026-09-24", "page": "1"},
)
assert response.status_code == 400, response.text
assert response.json() == {"detail": {"error": "end_date must be on or after start_date"}}, response.text
def _registered_store(gateway: Gateway, scenario: Scenario, identity: str, handle: ScenarioHandle) -> str:
store: Final = f"vs-{uuid.uuid4().hex}"
model: Final = scenario.model(model="openai/text-embedding-3-small", api_base=handle.api_base(), api_key=identity)
gateway.post(
"/vector_store/new",
{
"vector_store_id": store,
"custom_llm_provider": "openai",
"litellm_params": {"model": model, "api_base": handle.api_base(), "api_key": identity},
},
)
scenario.cleanups.callback(gateway.post, "/vector_store/delete", {"vector_store_id": store})
return store
@pytest.mark.parametrize(
("path", "body"),
(
("/v1/vector_stores/{store}/files/file-audit", {"attributes": _SEVENTEEN_ATTRIBUTES}),
("/v1/vector_stores/{store}/files", {"file_id": "file-audit", "attributes": _SEVENTEEN_ATTRIBUTES}),
),
ids=("update", "create"),
)
def test_vector_store_file_attributes_over_provider_limit_reach_provider(
gateway: Gateway, path: str, body: dict[str, JsonValue]
) -> None:
message: Final = (
"Invalid 'attributes': too many properties. "
"Expected an object with at most 16 properties, but got an object with 17 properties instead."
)
with gateway.scenario() as scenario:
identity, handle = _scripted(scenario, "followup-attributes", _openai_error(message, "attributes"))
store: Final = _registered_store(gateway, scenario, identity, handle)
response: Final = _post(gateway, path.format(store=store), body)
assert response.status_code == 400, response.text
assert message in response.text, response.text
assert _outbound(gateway, identity).get("attributes") == _SEVENTEEN_ATTRIBUTES
@pytest.mark.parametrize(
("path", "body", "missing"),
(
("/v1/images/generations", {"model": "followup-image"}, "prompt"),
("/v1/completions", {"model": "followup-text"}, "prompt"),
("/v1/audio/speech", {"model": "followup-speech", "voice": "alloy"}, "input"),
),
ids=("image-prompt", "text-completion-prompt", "speech-input"),
)
def test_router_default_for_positional_param_returns_400(
router_default_proxy: Gateway, path: str, body: dict[str, JsonValue], missing: str
) -> None:
response: Final = _post(router_default_proxy, path, body)
error: Final = _invalid_request(response)
assert error.get("param") == missing, response.text
_assert_no_call_to_deployment(router_default_proxy, str(body["model"]))
def test_router_default_for_model_outside_router_returns_400(router_default_proxy: Gateway) -> None:
response: Final = _post(router_default_proxy, "/v1/moderations", {"model": "omni-moderation-2024-09-26"})
error: Final = _invalid_request(response)
assert error.get("param") == "input", response.text
@pytest.mark.parametrize(
("body", "expected_max_tokens"),
(
({"messages": [{"role": "user", "content": "router default"}]}, 32),
({"messages": [{"role": "user", "content": "explicit"}], "max_tokens": 8}, 8),
),
ids=("router-default", "explicit"),
)
def test_null_user_config_is_treated_as_absent(
router_default_proxy: Gateway, body: dict[str, JsonValue], expected_max_tokens: int
) -> None:
observations: Final = _Observations(router_default_proxy.upstream_url)
observations.read()
before: Final = len(observations.items)
response: Final = _post(
router_default_proxy, "/v1/messages", {"model": "followup-message", "user_config": None, **body}
)
assert response.status_code == 200, response.text
assert JSON_OBJECT.validate_python(response.json()).get("type") == "message", response.text
eventually(
observations.read,
lambda items: any("followup-message-" in str(item.get("path")) for item in items[before:]),
seconds=20,
)
outbound: Final = next(
object_value(item["body"])
for item in observations.items[before:]
if "followup-message-" in str(item.get("path"))
)
assert outbound.get("max_tokens") == expected_max_tokens

View file

@ -0,0 +1,283 @@
"""The provider goes down in the middle of a concurrent burst across /v1/chat/completions, /v1/messages and
/v1/responses, streaming and not, on a two-worker proxy: every failed call reaches the caller as an error status,
another key on another deployment keeps working, and once the provider is back every call serves and lands in the
spend log exactly once."""
from __future__ import annotations
import json
import os
import re
import signal
import uuid
from collections.abc import Iterator, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path
from typing import Final
import httpx
import pytest
from integration._support.client import Gateway, eventually, gateway_from_environment
from integration._support.database import read_rows
from integration._support.process import UpstreamSlot, owned_proxy_process, owned_upstream
from integration._support.upstream import register_scenario
from integration.compatibility._status_code_audit import (
CHAT,
CHAT_FRAMES,
ENDPOINT_BODIES,
ENDPOINT_PATHS,
MESSAGE,
MESSAGE_FRAMES,
RESPONSE,
RESPONSES_FRAMES,
Upstream,
json_response,
sse_response,
stream_finished,
)
from pydantic import JsonValue
from tests.integration.cost_calculation.cost_tracking_case import StoredResponse
pytestmark: Final = pytest.mark.timeout(480)
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
_MARKER_HEADER: Final = "x-litellm-spend-logs-metadata"
_KINDS: Final[dict[str, tuple[str, bool, str, str, StoredResponse]]] = {
"chat": ("chat", False, "chaos-chat", "openai/gpt-4o-mini", json_response(CHAT)),
"chatstream": ("chat", True, "chaos-chat-stream", "openai/gpt-4o-mini", sse_response(CHAT_FRAMES)),
"responses": ("responses", False, "chaos-responses", "openai/gpt-4o-mini", json_response(RESPONSE)),
"responsesstream": (
"responses",
True,
"chaos-responses-stream",
"openai/gpt-4o-mini",
sse_response(RESPONSES_FRAMES),
),
"messages": ("messages", False, "chaos-message", "anthropic/claude-haiku-4-5", json_response(MESSAGE)),
"messagesstream": (
"messages",
True,
"chaos-message-stream",
"anthropic/claude-haiku-4-5",
sse_response(MESSAGE_FRAMES),
),
}
_BURST: Final = 30
@dataclass(frozen=True, slots=True)
class _Chaos:
gateway: Gateway
log: Path
slot: UpstreamSlot
identities: Mapping[str, str]
stable_identity: str
burst_key: str
other_key: str
@dataclass(frozen=True, slots=True)
class _Outcome:
marker: str
kind: str
status: int | None
text: str
finished: bool = False
def _register_on_slot(slot: UpstreamSlot, identities: Mapping[str, str]) -> None:
for kind, identity in identities.items():
register_scenario(identity, _KINDS[kind][4], control_url=slot.url)
@pytest.fixture(scope="module")
def chaos(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Chaos]:
directory: Final = tmp_path_factory.mktemp("audit-chaos")
with (
owned_upstream(directory) as slot,
gateway_from_environment() as gateway,
gateway.scenario() as scenario,
):
identities: Final = {kind: f"chaos-{kind}-{uuid.uuid4().hex}" for kind in _KINDS}
_register_on_slot(slot, identities)
stable_identity: Final = f"chaos-stable-{uuid.uuid4().hex}"
stable_handle: Final = register_scenario(stable_identity, json_response(CHAT))
model_list: Final[list[JsonValue]] = [
*(
{
"model_name": _KINDS[kind][2],
"litellm_params": {
"model": _KINDS[kind][3],
"api_base": f"{slot.url}/{identity}",
"api_key": identity,
},
}
for kind, identity in identities.items()
),
{
"model_name": "chaos-stable",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_base": stable_handle.api_base(),
"api_key": stable_identity,
},
},
]
config: Final = directory / "chaos.yaml"
config.write_text(
json.dumps({"model_list": model_list, "router_settings": {"num_retries": 0, "disable_cooldowns": True}}),
encoding="utf-8",
)
burst_key: Final = scenario.key(key_alias=f"chaos-burst-{uuid.uuid4().hex}")
other_key: Final = scenario.key(key_alias=f"chaos-other-{uuid.uuid4().hex}", models=["chaos-stable"])
with owned_proxy_process(gateway, directory, {}, config=config, workers=2) as owned:
eventually(lambda: _worker_pids(owned.log), lambda pids: len(pids) == 2, seconds=120)
yield _Chaos(
Gateway(owned.gateway.client, owned.gateway.key, gateway.upstream_url),
owned.log,
slot,
identities,
stable_identity,
burst_key,
other_key,
)
def _worker_pids(log: Path) -> tuple[int, ...]:
return tuple(int(found[1]) for found in _STARTED_WORKER.finditer(log.read_text()))
def _call(chaos: _Chaos, kind: str, marker: str, *, key: str | None = None) -> _Outcome:
endpoint, stream, model_name, _target, _response = _KINDS[kind]
body: Final[dict[str, JsonValue]] = {
**ENDPOINT_BODIES[endpoint],
**({"stream": True} if stream else {}),
"model": model_name,
**({"input": marker} if endpoint == "responses" else {"messages": [{"role": "user", "content": marker}]}),
}
headers: Final = {
"Authorization": f"Bearer {chaos.burst_key if key is None else key}",
_MARKER_HEADER: json.dumps({"marker": marker}),
}
try:
with httpx.Client(base_url=str(chaos.gateway.client.base_url), trust_env=False, timeout=60) as client:
with client.stream("POST", ENDPOINT_PATHS[endpoint], json=body, headers=headers) as response:
lines: Final = tuple(response.iter_lines())
status: Final = response.status_code
except httpx.TransportError as error:
return _Outcome(marker, kind, None, repr(error))
return _Outcome(marker, kind, status, "\n".join(lines), not stream or stream_finished(endpoint, lines))
def _burst(chaos: _Chaos, phase: str) -> tuple[_Outcome, ...]:
kinds: Final = tuple(_KINDS)
plan: Final = tuple(
(kinds[index % len(kinds)], f"{phase}-{kinds[index % len(kinds)]}-{index}-{uuid.uuid4().hex}")
for index in range(_BURST)
)
with ThreadPoolExecutor(max_workers=12) as pool:
return tuple(pool.map(lambda item: _call(chaos, item[0], item[1]), plan))
def _upstream_hits(slot: UpstreamSlot, markers: frozenset[str]) -> Mapping[str, int]:
upstream: Final = Upstream(slot.url)
upstream.drain()
bodies: Final = tuple(json.dumps(item.get("body")) for item in upstream.items)
return {marker: sum(marker in body for body in bodies) for marker in markers}
def _spend_markers(key: str, markers: frozenset[str]) -> Mapping[str, tuple[str, ...]]:
rows: Final = read_rows(
"SELECT status, metadata->'spend_logs_metadata'->>'marker' AS marker "
'FROM "LiteLLM_SpendLogs" WHERE api_key = %s',
(sha256(key.encode()).hexdigest(),),
)
return {
marker: found
for marker in markers
if (found := tuple(str(row["status"]) for row in rows if str(row["marker"]) == marker))
}
def _assert_served(outcomes: tuple[_Outcome, ...]) -> None:
assert [(outcome.kind, outcome.status, outcome.finished) for outcome in outcomes] == [
(outcome.kind, 200, True) for outcome in outcomes
], [(outcome.kind, outcome.status, outcome.text[:300]) for outcome in outcomes if outcome.status != 200]
def _stable_call(chaos: _Chaos) -> httpx.Response:
return chaos.gateway.client.post(
"/v1/chat/completions",
json={"model": "chaos-stable", "messages": [{"role": "user", "content": "other key"}]},
headers={"Authorization": f"Bearer {chaos.other_key}"},
timeout=30,
)
def _settled(chaos: _Chaos) -> bool:
return all(_call(chaos, kind, f"settle-{kind}-{uuid.uuid4().hex}").status == 200 for kind in _KINDS)
def _stop_mid_burst(chaos: _Chaos) -> tuple[_Outcome, ...]:
with ThreadPoolExecutor(max_workers=1) as stopper:
stopping: Final = stopper.submit(chaos.slot.stop)
during: Final = _burst(chaos, "during")
stopping.result(timeout=120)
return during
def _restart(chaos: _Chaos) -> None:
chaos.slot.start()
_register_on_slot(chaos.slot, chaos.identities)
eventually(lambda: _settled(chaos), lambda ok: ok, seconds=90)
def test_provider_outage_mid_burst_errors_reach_callers_and_recovery_serves_once(chaos: _Chaos) -> None:
before: Final = _burst(chaos, "before")
_assert_served(before)
during: Final = _stop_mid_burst(chaos)
outage: Final = _burst(chaos, "outage")
for outcome in outage:
assert outcome.status is not None and outcome.status >= 500, (outcome.kind, outcome.status, outcome.text)
assert '"error"' in outcome.text, (outcome.kind, outcome.text)
for outcome in during:
assert (outcome.status == 200 and outcome.finished) or (
outcome.status is not None and outcome.status >= 500 and '"error"' in outcome.text
), (outcome.kind, outcome.status, outcome.text[:300])
assert chaos.gateway.request("GET", "/health/liveliness").status_code == 200
stable: Final = _stable_call(chaos)
assert stable.status_code == 200, stable.text
_restart(chaos)
after: Final = _burst(chaos, "after")
_assert_served(after)
after_markers: Final = frozenset(outcome.marker for outcome in after)
assert _upstream_hits(chaos.slot, after_markers) == dict.fromkeys(after_markers, 1)
served: Final = frozenset(outcome.marker for outcome in (*before, *after))
landed: Final = eventually(
lambda: _spend_markers(chaos.burst_key, served), lambda found: frozenset(found) == served, seconds=120
)
assert landed == dict.fromkeys(served, ("success",)), landed
assert len(_worker_pids(chaos.log)) >= 2
def _tolerant_status(chaos: _Chaos, kind: str) -> int | None:
return _call(chaos, kind, f"kill-probe-{kind}-{uuid.uuid4().hex}").status
def test_killed_worker_leaves_the_sibling_serving_and_the_respawn_serving(chaos: _Chaos) -> None:
workers: Final = _worker_pids(chaos.log)
assert len(workers) >= 2, workers
os.kill(workers[-1], signal.SIGKILL)
eventually(lambda: _tolerant_status(chaos, "chat"), lambda status: status == 200, seconds=60)
_assert_served(_burst(chaos, "killed"))
stable: Final = _stable_call(chaos)
assert stable.status_code == 200, stable.text
eventually(lambda: _worker_pids(chaos.log), lambda pids: len(pids) == len(workers) + 1, seconds=180)
eventually(lambda: _settled(chaos), lambda ok: ok, seconds=120)
_assert_served(_burst(chaos, "respawned"))

View file

@ -0,0 +1,391 @@
"""Router-wide, deployment and `user_config` defaults against the params the Router binds positionally, the
moderation merge, unknown and A2A models, and every `user_config` shape on the three LLM endpoints."""
from __future__ import annotations
import asyncio
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from typing import Final
import anthropic
import httpx
import openai
import pytest
from integration._support.client import Gateway, gateway_from_environment
from integration.compatibility._status_code_audit import (
UNSET,
CHAT,
CHAT_FRAMES,
ENDPOINT_BODIES,
ENDPOINT_PATHS,
MESSAGE,
MESSAGE_FRAMES,
MODERATION,
RESPONSE,
RESPONSES_FRAMES,
ROUTER_DEFAULTS,
USER_MESSAGES,
assembled_text,
assert_no_provider_call,
drain_rig_upstream,
deployment,
error_body,
invalid_request,
json_response,
one_outbound,
owned_gateway,
post,
sse_response,
stream_finished,
stream_lines,
)
from pydantic import JsonValue
from tests.integration.cost_calculation.cost_tracking_case import BinaryResponse, StoredResponse
pytestmark: Final = pytest.mark.timeout(180)
@pytest.fixture(autouse=True)
def _drained_upstream() -> None:
drain_rig_upstream()
@dataclass(frozen=True, slots=True)
class _Rig:
gateway: Gateway
identities: Mapping[str, str]
deployments: Mapping[str, dict[str, JsonValue]]
_SPECS: Final[tuple[tuple[str, str, StoredResponse, dict[str, JsonValue]], ...]] = (
("audit-image", "openai/gpt-image-1", json_response({"created": 1, "data": []}), {}),
("audit-image-dep", "openai/gpt-image-1", json_response({"created": 1, "data": []}), {"prompt": "deployment"}),
("audit-text", "openai/gpt-3.5-turbo-instruct", json_response({"choices": []}), {}),
("audit-text-dep", "openai/gpt-3.5-turbo-instruct", json_response({"choices": []}), {"prompt": "deployment"}),
("audit-speech", "openai/gpt-4o-mini-tts", BinaryResponse(content_type="audio/mpeg", length=16), {}),
(
"audit-speech-dep",
"openai/gpt-4o-mini-tts",
BinaryResponse(content_type="audio/mpeg", length=16),
{"input": "deployment"},
),
("audit-transcribe", "openai/whisper-1", json_response({"text": "scripted"}), {}),
("audit-moderation", "openai/omni-moderation-latest", json_response(MODERATION), {}),
("audit-chat", "openai/gpt-4o-mini", json_response(CHAT), {}),
("audit-chat-stream", "openai/gpt-4o-mini", sse_response(CHAT_FRAMES), {}),
("audit-responses", "openai/gpt-4o-mini", json_response(RESPONSE), {}),
("audit-responses-stream", "openai/gpt-4o-mini", sse_response(RESPONSES_FRAMES), {}),
("audit-message", "anthropic/claude-haiku-4-5", json_response(MESSAGE), {}),
("audit-message-stream", "anthropic/claude-haiku-4-5", sse_response(MESSAGE_FRAMES), {}),
)
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
"""A proxy whose router-wide defaults name every positional and route-required param, with client-side
`user_config` allowed, and one deployment per route (plus `-dep` copies that carry the param themselves)."""
with gateway_from_environment() as gateway, gateway.scenario() as scenario:
built: Final = tuple(
deployment(scenario, name, model, response, **extra) for name, model, response, extra in _SPECS
)
identities: Final = {name: identity for (name, *_), (identity, _) in zip(_SPECS, built, strict=True)}
deployments: Final = {name: entry for (name, *_), (_, entry) in zip(_SPECS, built, strict=True)}
config: Final[dict[str, JsonValue]] = {
"model_list": [entry for _, entry in built],
"router_settings": {"default_litellm_params": ROUTER_DEFAULTS, "num_retries": 0},
"general_settings": {"allow_client_side_credentials": True},
}
with owned_gateway(tmp_path_factory.mktemp("audit-defaults"), config) as owned:
yield _Rig(owned, identities, deployments)
_POSITIONAL: Final[dict[str, tuple[str, str, str, str, dict[str, JsonValue]]]] = {
"image": ("/v1/images/generations", "/image/generations", "audit-image", "prompt", {}),
"text-completion": ("/v1/completions", "/completions", "audit-text", "prompt", {}),
"speech": ("/v1/audio/speech", "/audio/speech", "audit-speech", "input", {"voice": "alloy"}),
}
def _user_config(rig: _Rig, name: str, defaults: Mapping[str, JsonValue]) -> dict[str, JsonValue]:
return {"model_list": [rig.deployments[name]], "default_litellm_params": dict(defaults)}
@pytest.mark.parametrize("route", tuple(_POSITIONAL))
@pytest.mark.parametrize("source", ("deployment-param", "user-config-default", "user-config-null"))
def test_positional_param_from_any_default_source_returns_400(rig: _Rig, route: str, source: str) -> None:
path, error_route, name, param, extra = _POSITIONAL[route]
user_config: Final[dict[str, JsonValue]] = (
{"user_config": _user_config(rig, name, {param: "user config default"})}
if source == "user-config-default"
else {"user_config": None}
if source == "user-config-null"
else {}
)
body: Final[dict[str, JsonValue]] = {
"model": f"{name}-dep" if source == "deployment-param" else name,
**extra,
**user_config,
}
response: Final = post(rig.gateway, path, body)
error: Final = invalid_request(response)
assert error.get("param") == param, response.text
assert error.get("message") == f"{error_route}: Missing required parameter: '{param}'.", response.text
assert_no_provider_call(rig.gateway, rig.identities[name], rig.identities[f"{name}-dep"])
@pytest.mark.parametrize("route", tuple(_POSITIONAL))
def test_positional_param_in_the_body_reaches_upstream(rig: _Rig, route: str) -> None:
pytest.skip(
"BUG: a router-wide default_litellm_params value replaces the caller's own positional param "
"(prompt/input) on its way to the provider; same on the merge base"
)
path, _error_route, name, param, extra = _POSITIONAL[route]
response: Final = post(rig.gateway, path, {"model": name, param: f"explicit {route}", **extra})
assert response.status_code == 200, response.text
assert one_outbound(rig.gateway, rig.identities[name]).get(param) == f"explicit {route}"
def test_transcription_without_file_is_rejected_by_the_form_before_routing(rig: _Rig) -> None:
response: Final = rig.gateway.client.post(
"/v1/audio/transcriptions",
data={"model": "audit-transcribe"},
files={"note": ("note.txt", b"not audio", "text/plain")},
headers={"Authorization": f"Bearer {rig.gateway.key}"},
)
assert response.status_code == 422, response.text
detail: Final = response.json()["detail"]
assert [entry["loc"] for entry in detail] == [["body", "file"]], response.text
assert_no_provider_call(rig.gateway, rig.identities["audit-transcribe"])
def test_transcription_with_file_reaches_upstream(rig: _Rig) -> None:
response: Final = rig.gateway.client.post(
"/v1/audio/transcriptions",
data={"model": "audit-transcribe"},
files={"file": ("clip.wav", b"RIFF0000WAVEfmt ", "audio/wav")},
headers={"Authorization": f"Bearer {rig.gateway.key}"},
)
assert response.status_code == 200, response.text
assert response.json()["text"] == "scripted"
outbound: Final = one_outbound(rig.gateway, rig.identities["audit-transcribe"])
assert outbound.get("model") == "whisper-1"
assert outbound.get("file") == {"filename": "clip.wav", "content_type": "audio/x-wav"}
@pytest.mark.parametrize(
("body", "expected_input"),
(({}, "router default input"), ({"input": "explicit input"}, "explicit input")),
ids=("router-default", "explicit"),
)
def test_listed_moderation_model_merges_the_router_default(
rig: _Rig, body: dict[str, JsonValue], expected_input: str
) -> None:
response: Final = post(rig.gateway, "/v1/moderations", {"model": "audit-moderation", **body})
assert response.status_code == 200, response.text
assert one_outbound(rig.gateway, rig.identities["audit-moderation"]).get("input") == expected_input
@pytest.mark.parametrize(
("path", "body", "message"),
(
pytest.param(
"/v1/images/generations",
{"model": "audit-unknown-model"},
"/image/generations: Missing required parameter: 'prompt'.",
id="image-unknown-model",
),
pytest.param(
"/v1/images/generations",
{"model": "a2a/audit-unknown-agent"},
"/image/generations: Missing required parameter: 'prompt'.",
id="image-a2a-model",
),
pytest.param(
"/v1/messages",
{"model": "a2a/audit-unknown-agent", "messages": USER_MESSAGES},
"anthropic_messages: Invalid model name passed in model=a2a/audit-unknown-agent. "
"Call `/v1/models` to view available models for your key.",
id="messages-a2a-model",
),
),
)
def test_model_outside_the_router_with_a_router_default_param(
rig: _Rig, path: str, body: dict[str, JsonValue], message: str
) -> None:
response: Final = post(rig.gateway, path, body)
assert response.status_code == 400, response.text
error: Final = error_body(response)
assert error.get("type") == "invalid_request_error", response.text
assert error.get("message") == message, response.text
@pytest.mark.parametrize(
("user_config", "message"),
(
pytest.param(None, None, id="null"),
pytest.param(UNSET, None, id="missing"),
pytest.param({}, "There are no healthy deployments for this model", id="empty-object"),
pytest.param("", "Invalid request format: 'str' object has no attribute 'items'", id="empty-string"),
pytest.param(7, "Invalid request format: 'int' object has no attribute 'items'", id="int"),
pytest.param([], "Invalid request format: 'list' object has no attribute 'items'", id="list"),
pytest.param("x" * 5000, "Invalid request format: 'str' object has no attribute 'items'", id="5kb-string"),
),
)
def test_chat_user_config_shapes(rig: _Rig, user_config: object, message: str | None) -> None:
body: Final[dict[str, JsonValue]] = {
"model": "audit-chat",
"messages": USER_MESSAGES,
**({} if user_config is UNSET else {"user_config": user_config}), # pyright: ignore[reportAssignmentType] # the sad shapes are deliberately not JSON objects
}
response: Final = post(rig.gateway, "/v1/chat/completions", body)
if message is None:
assert response.status_code == 200, response.text
assert one_outbound(rig.gateway, rig.identities["audit-chat"]).get("max_tokens") == 32
return
assert response.status_code == 400, response.text
assert message in str(error_body(response).get("message")), response.text
assert_no_provider_call(rig.gateway, rig.identities["audit-chat"])
def test_messages_empty_user_config_does_not_inherit_router_defaults(rig: _Rig) -> None:
response: Final = post(
rig.gateway, "/v1/messages", {"model": "audit-message", "messages": USER_MESSAGES, "user_config": {}}
)
assert response.status_code == 400, response.text
assert error_body(response).get("message") == "anthropic_messages: Missing required parameter: 'max_tokens'."
assert_no_provider_call(rig.gateway, rig.identities["audit-message"])
@pytest.mark.parametrize(
("endpoint", "model", "stream", "expected_max_tokens"),
(
pytest.param("chat", "audit-chat", False, 32, id="chat"),
pytest.param("chat", "audit-chat-stream", True, 32, id="chat-stream"),
pytest.param("responses", "audit-responses", False, None, id="responses"),
pytest.param("responses", "audit-responses-stream", True, None, id="responses-stream"),
pytest.param("messages", "audit-message-stream", True, 32, id="messages-stream"),
),
)
def test_null_user_config_serves_every_endpoint(
rig: _Rig, endpoint: str, model: str, stream: bool, expected_max_tokens: int | None
) -> None:
body: Final[dict[str, JsonValue]] = {
**{key: value for key, value in ENDPOINT_BODIES[endpoint].items() if key != "max_tokens"},
"model": model,
"user_config": None,
**({"stream": True} if stream else {}),
}
if stream:
status, lines = stream_lines(rig.gateway, ENDPOINT_PATHS[endpoint], body)
assert status == 200, lines
assert stream_finished(endpoint, lines), lines
assert assembled_text(lines) == "streamed response", lines
else:
response: Final = post(rig.gateway, ENDPOINT_PATHS[endpoint], body)
assert response.status_code == 200, response.text
assert "scripted" in response.text, response.text
outbound: Final = one_outbound(rig.gateway, rig.identities[model])
assert "user_config" not in outbound, outbound
assert outbound.get("max_tokens") == expected_max_tokens, outbound
def test_null_user_config_without_a_valid_key_is_401(rig: _Rig) -> None:
response: Final = post(
rig.gateway,
"/v1/chat/completions",
{"model": "audit-chat", "messages": USER_MESSAGES, "user_config": None},
key="sk-audit-not-a-key",
)
assert response.status_code == 401, response.text
assert_no_provider_call(rig.gateway, rig.identities["audit-chat"])
def _openai_chat(rig: _Rig, asynchronous: bool) -> str | None:
base_url: Final = f"{rig.gateway.client.base_url}/v1"
if not asynchronous:
with openai.OpenAI(base_url=base_url, api_key=rig.gateway.key, max_retries=0) as client:
completion: Final = client.chat.completions.create(
model="audit-chat",
messages=[{"role": "user", "content": "Hello"}],
extra_body={"user_config": None},
)
return completion.choices[0].message.content
async def call() -> str | None:
async with openai.AsyncOpenAI(base_url=base_url, api_key=rig.gateway.key, max_retries=0) as client:
completion: Final = await client.chat.completions.create(
model="audit-chat",
messages=[{"role": "user", "content": "Hello"}],
extra_body={"user_config": None},
)
return completion.choices[0].message.content
return asyncio.run(call())
def _anthropic_message(rig: _Rig, asynchronous: bool) -> str:
base_url: Final = str(rig.gateway.client.base_url)
if not asynchronous:
with anthropic.Anthropic(base_url=base_url, api_key=rig.gateway.key, max_retries=0) as client:
message: Final = client.messages.create(
model="audit-message",
max_tokens=8,
messages=[{"role": "user", "content": "Hello"}],
extra_body={"user_config": None},
)
block: Final = message.content[0]
assert isinstance(block, anthropic.types.TextBlock), message
return block.text
async def call() -> str:
async with anthropic.AsyncAnthropic(base_url=base_url, api_key=rig.gateway.key, max_retries=0) as client:
message: Final = await client.messages.create(
model="audit-message",
max_tokens=8,
messages=[{"role": "user", "content": "Hello"}],
extra_body={"user_config": None},
)
block: Final = message.content[0]
assert isinstance(block, anthropic.types.TextBlock), message
return block.text
return asyncio.run(call())
@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async"))
def test_null_user_config_through_the_openai_sdk(rig: _Rig, asynchronous: bool) -> None:
assert _openai_chat(rig, asynchronous) == "scripted"
assert one_outbound(rig.gateway, rig.identities["audit-chat"]).get("max_tokens") == 32
@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async"))
def test_null_user_config_through_the_anthropic_sdk(rig: _Rig, asynchronous: bool) -> None:
assert _anthropic_message(rig, asynchronous) == "scripted"
assert one_outbound(rig.gateway, rig.identities["audit-message"]).get("max_tokens") == 8
def test_null_user_config_without_the_client_side_opt_in_is_rejected(gateway: Gateway) -> None:
response: Final = gateway.client.post(
"/v1/chat/completions",
json={"model": "audit-not-routed", "messages": USER_MESSAGES, "user_config": None},
headers={"Authorization": f"Bearer {gateway.key}"},
)
assert response.status_code == 401, response.text
assert "user_config is not allowed in request body" in response.text, response.text
def test_raw_httpx_async_null_user_config(rig: _Rig) -> None:
async def call() -> httpx.Response:
async with httpx.AsyncClient(base_url=str(rig.gateway.client.base_url), trust_env=False) as client:
return await client.post(
"/v1/responses",
json={"model": "audit-responses", "input": "Hello", "store": False, "user_config": None},
headers={"Authorization": f"Bearer {rig.gateway.key}"},
timeout=30,
)
response: Final = asyncio.run(call())
assert response.status_code == 200, response.text
assert one_outbound(rig.gateway, rig.identities["audit-responses"]).get("input") == "Hello"

View file

@ -0,0 +1,425 @@
"""OCR parse errors, Responses `max_output_tokens`, vector store file attributes and the daily activity date range,
driven through the shared rig proxy with the provider body read back from the scripted upstream."""
from __future__ import annotations
import asyncio
import itertools
import uuid
from collections.abc import Callable
from typing import Final
import httpx
import openai
import pytest
from integration._support.client import Gateway, Scenario
from integration._support.upstream import ScenarioHandle
from integration.compatibility._status_code_audit import (
UNSET,
RESPONSE,
RESPONSES_FRAMES,
assembled_text,
assert_no_provider_call,
drain_rig_upstream,
invalid_request,
json_response,
one_outbound,
openai_error,
post,
register,
sse_response,
stream_finished,
stream_lines,
)
from pydantic import JsonValue
from tests.integration.cost_calculation.cost_tracking_case import StoredResponse
pytestmark: Final = pytest.mark.timeout(180)
@pytest.fixture(autouse=True)
def _drained_upstream() -> None:
drain_rig_upstream()
_OCR: Final[dict[str, JsonValue]] = {
"pages": [{"index": 0, "markdown": "scripted page", "images": [], "dimensions": None}],
"model": "mistral-ocr-latest",
"usage_info": {"pages_processed": 1},
}
_VECTOR_STORE_FILE: Final[dict[str, JsonValue]] = {
"id": "file-audit",
"object": "vector_store.file",
"created_at": 1,
"usage_bytes": 0,
"vector_store_id": "vs-audit",
"status": "completed",
"last_error": None,
"attributes": {},
}
def _ocr_model(scenario: Scenario) -> tuple[str, str]:
identity, handle = register(scenario, "audit-ocr", json_response(_OCR))
return scenario.model(model="mistral/mistral-ocr-latest", api_base=handle.api_base(), api_key=identity), identity
def _raw(gateway: Gateway, path: str, content: bytes, content_type: str, *, key: str | None = None) -> httpx.Response:
return gateway.client.post(
path,
content=content,
headers={"Authorization": f"Bearer {gateway.key if key is None else key}", "Content-Type": content_type},
timeout=60,
)
@pytest.mark.parametrize(
("path", "body", "message", "param"),
(
pytest.param(
"/v1/ocr",
{"document": {"type": "file", "file": "local.pdf"}},
"document type 'file' is not supported through the JSON API.",
None,
id="document-type-file",
),
pytest.param(
"/v1/ocr",
{"document": {"type": "document_url", "document_url": "reducto://file-123"}},
"reducto:// file IDs are not accepted through the proxy OCR API;",
None,
id="reducto-document-url",
),
pytest.param(
"/v1/ocr",
{"document": {"type": "image_url", "image_url": "reducto://file-123"}},
"reducto:// file IDs are not accepted through the proxy OCR API;",
None,
id="reducto-image-url",
),
pytest.param(
"/ocr",
{"document": {"type": "file", "file": "local.pdf"}},
"document type 'file' is not supported through the JSON API.",
None,
id="unversioned-route",
),
),
)
def test_ocr_unsupported_json_document_returns_400(
gateway: Gateway, path: str, body: dict[str, JsonValue], message: str, param: str | None
) -> None:
with gateway.scenario() as scenario:
model, identity = _ocr_model(scenario)
response: Final = post(gateway, path, {"model": model, **body})
error: Final = invalid_request(response)
assert str(error.get("message")).startswith(message), response.text
assert error.get("param") == param, response.text
assert_no_provider_call(gateway, identity)
@pytest.mark.parametrize(
("content", "content_type", "message", "param"),
(
pytest.param(b"", "application/json", "Empty request body.", None, id="empty-json-body"),
pytest.param(b"{not json", "application/json", "Invalid JSON payload", "request_body", id="invalid-json"),
pytest.param(b"hello", "text/plain", "Invalid JSON payload", "request_body", id="text-plain"),
pytest.param(
b"--x\r\n",
"multipart/form-data",
"Invalid form payload: 400: Missing boundary",
"request_body",
id="no-boundary",
),
),
)
def test_ocr_unparseable_body_returns_400(
gateway: Gateway, content: bytes, content_type: str, message: str, param: str | None
) -> None:
response: Final = _raw(gateway, "/v1/ocr", content, content_type)
error: Final = invalid_request(response)
assert str(error.get("message")).startswith(message), response.text
assert error.get("param") == param, response.text
def test_ocr_json_array_body_returns_400(gateway: Gateway) -> None:
pytest.skip("BUG: a JSON array body answers 500 on every route, from the auth pre-checks; same on base")
response: Final = _raw(gateway, "/v1/ocr", b"[1,2]", "application/json")
assert invalid_request(response).get("param") == "request_body", response.text
def test_ocr_file_over_the_size_limit_returns_400(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model, identity = _ocr_model(scenario)
response: Final = gateway.request_multipart(
"/v1/ocr", {"model": model}, {"file": ("big.pdf", b"%PDF" + b"0" * (50 * 1024 * 1024), "application/pdf")}
)
error: Final = invalid_request(response)
assert error.get("message") == "OCR file exceeds the size limit", response.text
assert error.get("param") == "file", response.text
assert_no_provider_call(gateway, identity)
def test_ocr_multipart_without_a_key_is_401(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model, identity = _ocr_model(scenario)
response: Final = gateway.client.post(
"/v1/ocr", files={"model": (None, model)}, headers={"Authorization": "Bearer sk-audit-not-a-key"}
)
assert response.status_code == 401, response.text
assert_no_provider_call(gateway, identity)
def test_ocr_document_url_reaches_the_provider(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model, identity = _ocr_model(scenario)
document: Final[dict[str, JsonValue]] = {"type": "document_url", "document_url": "https://example.com/a.pdf"}
response: Final = post(gateway, "/v1/ocr", {"model": model, "document": document})
assert response.status_code == 200, response.text
assert response.json()["pages"][0]["markdown"] == "scripted page"
assert one_outbound(gateway, identity).get("document") == document
def test_ocr_multipart_file_reaches_the_provider_as_a_data_uri(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model, identity = _ocr_model(scenario)
response: Final = gateway.request_multipart(
"/v1/ocr", {"model": model}, {"file": ("doc.pdf", b"%PDF-1.4 audit", "application/pdf")}
)
assert response.status_code == 200, response.text
document: Final = one_outbound(gateway, identity).get("document")
assert isinstance(document, dict), document
assert document.get("type") == "document_url", document
assert str(document.get("document_url")).startswith("data:application/pdf;base64,"), document
def _responses_model(scenario: Scenario, response: StoredResponse) -> tuple[str, str]:
identity, handle = register(scenario, "audit-max-output", response)
return scenario.model(model="openai/gpt-4o-mini", api_base=handle.api_base(), api_key=identity), identity
@pytest.mark.parametrize(
("sent", "forwarded"),
(
pytest.param(1, 16, id="1"),
pytest.param(15, 16, id="15"),
pytest.param(16, 16, id="16"),
pytest.param(4096, 4096, id="4096"),
pytest.param(True, 16, id="true"),
pytest.param(None, UNSET, id="null"),
pytest.param(UNSET, UNSET, id="absent"),
pytest.param("8", "8", id="string"),
pytest.param(0.5, 0.5, id="float"),
),
)
def test_responses_max_output_tokens_forwarding(gateway: Gateway, sent: object, forwarded: object) -> None:
with gateway.scenario() as scenario:
model, identity = _responses_model(scenario, json_response(RESPONSE))
body: Final[dict[str, JsonValue]] = {
"model": model,
"input": "Say pong",
"store": False,
**({} if sent is UNSET else {"max_output_tokens": sent}), # pyright: ignore[reportAssignmentType] # the parametrized shapes are JSON values
}
response: Final = post(gateway, "/v1/responses", body)
assert response.status_code == 200, response.text
outbound: Final = one_outbound(gateway, identity)
assert outbound.get("max_output_tokens", UNSET) == forwarded, outbound
_BELOW_MINIMUM: Final = (
"Invalid 'max_output_tokens': integer below minimum value. Expected a value >= 16, but got {} instead."
)
@pytest.mark.parametrize("sent", (0, -1), ids=("zero", "negative"))
def test_responses_non_positive_max_output_tokens_stream_returns_the_provider_400(gateway: Gateway, sent: int) -> None:
message: Final = _BELOW_MINIMUM.format(sent)
with gateway.scenario() as scenario:
model, identity = _responses_model(scenario, openai_error(message, "max_output_tokens"))
status, lines = stream_lines(
gateway,
"/v1/responses",
{"model": model, "input": "Say pong", "store": False, "stream": True, "max_output_tokens": sent},
)
assert status == 400, lines
assert any(message in line for line in lines), lines
outbound: Final = one_outbound(gateway, identity)
assert (outbound.get("max_output_tokens"), outbound.get("stream")) == (sent, True), outbound
def test_responses_small_max_output_tokens_stream_is_raised_to_the_minimum(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
model, identity = _responses_model(scenario, sse_response(RESPONSES_FRAMES))
status, lines = stream_lines(
gateway,
"/v1/responses",
{"model": model, "input": "Say pong", "store": False, "stream": True, "max_output_tokens": 1},
)
assert status == 200, lines
assert stream_finished("responses", lines), lines
assert assembled_text(lines) == "streamed response", lines
assert one_outbound(gateway, identity).get("max_output_tokens") == 16
def _sdk_responses(
gateway: Gateway, model: str, sent: int, *, asynchronous: bool, stream: bool
) -> Callable[[], object]:
base_url: Final = f"{gateway.client.base_url}/v1"
def sync_call() -> object:
with openai.OpenAI(base_url=base_url, api_key=gateway.key, max_retries=0) as client:
return client.responses.create(
model=model, input="Say pong", store=False, max_output_tokens=sent, stream=stream
)
async def async_call() -> object:
async with openai.AsyncOpenAI(base_url=base_url, api_key=gateway.key, max_retries=0) as client:
return await client.responses.create(
model=model, input="Say pong", store=False, max_output_tokens=sent, stream=stream
)
return (lambda: asyncio.run(async_call())) if asynchronous else sync_call
@pytest.mark.parametrize(
("sent", "asynchronous", "stream"),
((0, False, False), (0, True, False), (-1, False, True), (-1, True, True)),
ids=("zero-sync", "zero-async", "negative-sync-stream", "negative-async-stream"),
)
def test_responses_non_positive_max_output_tokens_through_the_openai_sdk(
gateway: Gateway, sent: int, asynchronous: bool, stream: bool
) -> None:
message: Final = _BELOW_MINIMUM.format(sent)
with gateway.scenario() as scenario:
model, identity = _responses_model(scenario, openai_error(message, "max_output_tokens"))
call: Final = _sdk_responses(gateway, model, sent, asynchronous=asynchronous, stream=stream)
with pytest.raises(openai.BadRequestError) as raised:
call()
assert raised.value.status_code == 400
assert message in raised.value.response.text
assert one_outbound(gateway, identity).get("max_output_tokens") == sent
def _registered_store(gateway: Gateway, scenario: Scenario, identity: str, handle: ScenarioHandle) -> str:
store: Final = f"vs-{uuid.uuid4().hex}"
model: Final = scenario.model(model="openai/text-embedding-3-small", api_base=handle.api_base(), api_key=identity)
gateway.post(
"/vector_store/new",
{
"vector_store_id": store,
"custom_llm_provider": "openai",
"litellm_params": {"model": model, "api_base": handle.api_base(), "api_key": identity},
},
)
scenario.cleanups.callback(gateway.post, "/vector_store/delete", {"vector_store_id": store})
return store
_SIXTEEN: Final[dict[str, JsonValue]] = {f"key_{index:02d}": f"value {index}" for index in range(16)}
_ATTRIBUTE_CASES: Final = (
pytest.param(_SIXTEEN, _SIXTEEN, id="sixteen"),
pytest.param({}, {}, id="empty"),
pytest.param({"team": "a", "hidden_params": "internal"}, {"team": "a"}, id="hidden-params"),
pytest.param({"long": "x" * 5000}, {"long": "x" * 5000}, id="5kb-value"),
)
@pytest.mark.parametrize("operation", ("create", "update"))
@pytest.mark.parametrize(("attributes", "forwarded"), _ATTRIBUTE_CASES)
def test_vector_store_file_attributes_reach_the_provider(
gateway: Gateway, operation: str, attributes: dict[str, JsonValue], forwarded: dict[str, JsonValue]
) -> None:
with gateway.scenario() as scenario:
identity, handle = register(scenario, "audit-attributes", json_response(_VECTOR_STORE_FILE))
store: Final = _registered_store(gateway, scenario, identity, handle)
path: Final = (
f"/v1/vector_stores/{store}/files"
if operation == "create"
else f"/v1/vector_stores/{store}/files/file-audit"
)
body: Final[dict[str, JsonValue]] = {
"attributes": attributes,
**({"file_id": "file-audit"} if operation == "create" else {}),
}
response: Final = post(gateway, path, body)
assert response.status_code == 200, response.text
assert one_outbound(gateway, identity).get("attributes") == forwarded
@pytest.mark.parametrize("operation", ("create", "update"))
def test_vector_store_file_number_and_boolean_attributes_reach_the_provider(gateway: Gateway, operation: str) -> None:
pytest.skip(
"BUG: number and boolean vector store file attributes (valid OpenAI attribute values) are dropped before "
"the provider call, so they are silently lost; same on base"
)
attributes: Final[dict[str, JsonValue]] = {"pages": 3, "public": True, "team": "a"}
with gateway.scenario() as scenario:
identity, handle = register(scenario, "audit-attributes-typed", json_response(_VECTOR_STORE_FILE))
store: Final = _registered_store(gateway, scenario, identity, handle)
path: Final = (
f"/v1/vector_stores/{store}/files"
if operation == "create"
else f"/v1/vector_stores/{store}/files/file-audit"
)
body: Final[dict[str, JsonValue]] = {
"attributes": attributes,
**({"file_id": "file-audit"} if operation == "create" else {}),
}
response: Final = post(gateway, path, body)
assert response.status_code == 200, response.text
assert one_outbound(gateway, identity).get("attributes") == attributes
_ENTITIES: Final = ("agent", "customer", "organization", "tag", "team", "user")
_SUFFIXES: Final = (
"",
"/aggregated",
"/aggregated/keys",
"/aggregated/model_top_keys",
"/aggregated/search",
"/export",
)
_DAILY_ROUTES: Final = tuple(
f"/{entity}/daily/activity{suffix}"
for entity, suffix in itertools.product(_ENTITIES, _SUFFIXES)
if f"/{entity}/daily/activity{suffix}" != "/organization/daily/activity"
) + ("/user/daily/activity/aggregated/cache_leakage_keys",)
_DATE_CASES: Final[dict[str, tuple[str, str, int, str | None]]] = {
"reversed": ("2026-10-01", "2026-09-24", 400, "end_date must be on or after start_date"),
"equal": ("2026-10-01", "2026-10-01", 200, None),
"bad-month": ("2026-13-01", "2026-10-01", 400, "start_date and end_date must be valid YYYY-MM-DD dates"),
}
def _route_params(route: str) -> dict[str, str]:
if route.endswith("/model_top_keys"):
return {"model_group": "audit-model"}
if route.endswith("/search"):
return {"search": "audit"}
return {}
@pytest.mark.parametrize("case", tuple(_DATE_CASES))
@pytest.mark.parametrize("route", _DAILY_ROUTES)
def test_daily_activity_date_range(gateway: Gateway, route: str, case: str) -> None:
start, end, status, error = _DATE_CASES[case]
response: Final = gateway.request(
"GET", route, params={"start_date": start, "end_date": end, **_route_params(route)}
)
assert response.status_code == status, response.text
if error is not None:
assert response.json() == {"detail": {"error": error}}, response.text
@pytest.mark.parametrize("case", ("reversed", "bad-month"))
def test_gateway_daily_activity_date_range(gateway: Gateway, case: str) -> None:
pytest.skip(
"BUG: /gateway/daily/activity answers 200 with empty totals for a reversed or invalid date range "
"instead of the 400 every other daily activity route gives; same on base"
)
start, end, status, error = _DATE_CASES[case]
response: Final = gateway.request("GET", "/gateway/daily/activity", params={"start_date": start, "end_date": end})
assert response.status_code == status, response.text
assert response.json() == {"detail": {"error": error}}, response.text

View file

@ -0,0 +1,125 @@
"""The /usage/ai/chat tools hand the model the reason the daily activity query rejected its dates, the same check
the daily activity routes answer 400 with, instead of a generic fetch error."""
from __future__ import annotations
import json
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Final
import pytest
from integration._support.client import JSON_OBJECT, Gateway, object_value
from integration._support.upstream import delete_scenario, register_scenario
from integration.compatibility._status_code_audit import Upstream, json_response, owned_gateway
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(180)
_TOOLS: Final = ("get_usage_data", "get_team_usage_data", "get_tag_usage_data")
@dataclass(frozen=True, slots=True)
class _UsageChat:
gateway: Gateway
identity: str
@pytest.fixture(scope="module")
def usage_chat(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_UsageChat]:
"""A proxy whose usage chat model (`openai/...`, called straight through litellm) is the scripted upstream."""
identity: Final = f"audit-usage-chat-{uuid.uuid4().hex}"
handle: Final = register_scenario(identity, json_response({}))
try:
with owned_gateway(
tmp_path_factory.mktemp("audit-usage-chat"),
{"model_list": [], "router_settings": {"num_retries": 0}},
environment={"OPENAI_API_BASE": handle.api_base(), "OPENAI_API_KEY": identity},
) as owned:
yield _UsageChat(owned, identity)
finally:
delete_scenario(handle)
def _tool_calls(start_date: str, end_date: str) -> dict[str, JsonValue]:
arguments: Final = json.dumps({"start_date": start_date, "end_date": end_date})
return {
"id": "chatcmpl-usage",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{"id": f"call-{tool}", "type": "function", "function": {"name": tool, "arguments": arguments}}
for tool in _TOOLS
],
},
"finish_reason": "tool_calls",
}
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
def _tool_results(usage_chat: _UsageChat, start_date: str, end_date: str) -> tuple[dict[str, str], tuple[str, ...]]:
"""Runs one chat turn whose model calls every tool with these dates; returns the tool messages the follow-up
model call carried, keyed by tool, and the tool statuses the SSE stream reported."""
register_scenario(usage_chat.identity, json_response(_tool_calls(start_date, end_date)))
Upstream(usage_chat.gateway.upstream_url).drain()
with usage_chat.gateway.client.stream(
"POST",
"/usage/ai/chat",
json={"model": "openai/gpt-4o-mini", "messages": [{"role": "user", "content": "How much did we spend?"}]},
headers={"Authorization": f"Bearer {usage_chat.gateway.key}"},
timeout=60,
) as response:
assert response.status_code == 200
events: Final = tuple(
object_value(JSON_OBJECT.validate_json(line.removeprefix("data: ")))
for line in response.iter_lines()
if line.startswith("data: ")
)
calls: Final = Upstream(usage_chat.gateway.upstream_url).calls(usage_chat.identity)
assert len(calls) == 2, calls
messages: Final = object_value(calls[1]["body"]).get("messages")
assert isinstance(messages, list), calls[1]
tool_messages: Final = {
str(message["tool_call_id"]).removeprefix("call-"): str(message["content"])
for message in messages
if isinstance(message, dict) and message.get("role") == "tool"
}
statuses: Final = tuple(str(event.get("status")) for event in events if event.get("type") == "tool_call")
return tool_messages, statuses
_REVERSED: Final = "Invalid date range: end_date must be on or after start_date"
_NOT_CANONICAL: Final = "Invalid date range: start_date and end_date must be valid YYYY-MM-DD dates"
_NO_TEAM_DATA: Final = "No Team usage data found for the given date range."
_NO_TAG_DATA: Final = "No Tag usage data found for the given date range."
@pytest.mark.parametrize(
("start_date", "end_date", "team_message", "tag_message"),
(
pytest.param("2026-10-09", "2026-10-01", _REVERSED, _REVERSED, id="reversed"),
pytest.param("2026-13-01", "2026-10-09", _NOT_CANONICAL, _NOT_CANONICAL, id="invalid-month"),
pytest.param("2001/01/01", "2001-01-09", _NOT_CANONICAL, _NOT_CANONICAL, id="slashes"),
pytest.param("2001-01-01", "2001-01-09", _NO_TEAM_DATA, _NO_TAG_DATA, id="valid"),
),
)
def test_usage_chat_tools_answer_with_the_query_layers_date_check(
usage_chat: _UsageChat, start_date: str, end_date: str, team_message: str, tag_message: str
) -> None:
"""The team and tag tools give the model the daily activity routes' 400 reason; the global tool, whose query
takes these dates without that check, still answers with its usage summary."""
tool_messages, statuses = _tool_results(usage_chat, start_date, end_date)
assert tool_messages["get_usage_data"].startswith("Total Spend: $"), tool_messages
assert tool_messages["get_team_usage_data"] == team_message, tool_messages
assert tool_messages["get_tag_usage_data"] == tag_message, tool_messages
assert statuses == ("running", "complete") * len(_TOOLS), statuses

View file

@ -2,7 +2,7 @@ import pytest
from litellm.proxy._types import LitellmUserRoles
from tests.integration._support.actors import Actor
from tests.integration._support.actors import Actor, SeededKey
from tests.integration._support.actors import create_scratch_actor, create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
@ -38,19 +38,39 @@ _MATRIX = [
]
async def _seed_target(db, actors, shape: str, team_id: str) -> str:
_TEAM_MEMBER_ACTORS = (
Actor.TEAM_ADMIN,
Actor.INTERNAL_USER,
Actor.OWNER,
Actor.UNRELATED_SAME_ORG,
Actor.SERVICE_ACCOUNT,
Actor.CROSS_ORG_USER,
)
async def _callers(db, actors, prefix: str) -> dict[Actor, SeededKey]:
team_members = {
actor: await create_scratch_actor(
db, prefix, user_role=LitellmUserRoles.INTERNAL_USER.value, suffix=actor.value
)
for actor in _TEAM_MEMBER_ACTORS
}
return {**actors.keys, **team_members}
async def _seed_target(db, actors, callers, shape: str, team_id: str) -> str:
"""Raw-seed the resources target team; returns its organization_id."""
if shape == "alpha":
await create_scratch_team(
db,
team_id,
organization_id=actors.org_a_id,
admin_user_ids=[actors.keys[Actor.TEAM_ADMIN].user_id],
admin_user_ids=[callers[Actor.TEAM_ADMIN].user_id],
member_user_ids=[
actors.keys[Actor.INTERNAL_USER].user_id,
actors.keys[Actor.OWNER].user_id,
actors.keys[Actor.UNRELATED_SAME_ORG].user_id,
actors.keys[Actor.SERVICE_ACCOUNT].user_id,
callers[Actor.INTERNAL_USER].user_id,
callers[Actor.OWNER].user_id,
callers[Actor.UNRELATED_SAME_ORG].user_id,
callers[Actor.SERVICE_ACCOUNT].user_id,
],
)
return actors.org_a_id
@ -59,7 +79,7 @@ async def _seed_target(db, actors, shape: str, team_id: str) -> str:
db,
team_id,
organization_id=actors.org_b_id,
member_user_ids=[actors.keys[Actor.CROSS_ORG_USER].user_id],
member_user_ids=[callers[Actor.CROSS_ORG_USER].user_id],
)
return actors.org_b_id
pytest.fail(f"unknown shape={shape}") # pragma: no cover
@ -79,8 +99,9 @@ async def test_team_update_authz_matrix(
resources,
actors,
):
org_id = await _seed_target(db, actors, shape, resources.prefix)
caller = actors.keys[actor]
callers = await _callers(db, actors, resources.prefix)
org_id = await _seed_target(db, actors, callers, shape, resources.prefix)
caller = callers[actor]
resp = await gateway_client.post(
"/team/update",
@ -113,7 +134,8 @@ async def test_team_update_org_admin_resolved_from_team_without_org_context(
(200), same as PROXY_ADMIN. A team admin of that same team reaches the
handler but is refused (403) until a proxy admin enables fields for team
admins, and the response says so."""
await _seed_target(db, actors, "alpha", resources.prefix)
callers = await _callers(db, actors, resources.prefix)
await _seed_target(db, actors, callers, "alpha", resources.prefix)
allowed_org_admin = await gateway_client.post(
"/team/update",
@ -131,7 +153,7 @@ async def test_team_update_org_admin_resolved_from_team_without_org_context(
denied_team_admin = await gateway_client.post(
"/team/update",
headers={"Authorization": f"Bearer {actors.keys[Actor.TEAM_ADMIN].cleartext}"},
headers={"Authorization": f"Bearer {callers[Actor.TEAM_ADMIN].cleartext}"},
json={"team_id": resources.prefix, "team_alias": MARKER_ALIAS},
)
assert denied_team_admin.status_code == 403, denied_team_admin.text
@ -169,8 +191,9 @@ async def test_team_update_org_relocation_gate(
resources,
actors,
):
await _seed_target(db, actors, "alpha", resources.prefix)
caller = actors.keys[actor]
callers = await _callers(db, actors, resources.prefix)
await _seed_target(db, actors, callers, "alpha", resources.prefix)
caller = callers[actor]
resp = await gateway_client.post(
"/team/update",
@ -199,7 +222,7 @@ async def test_team_update_org_relocation_gate(
async def test_team_update_org_b_admin_relocation_rejection_detail(
gateway_client, db, resources, actors
):
await _seed_target(db, actors, "alpha", resources.prefix)
await _seed_target(db, actors, await _callers(db, actors, resources.prefix), "alpha", resources.prefix)
resp = await gateway_client.post(
"/team/update",
headers={"Authorization": f"Bearer {actors.keys[Actor.ORG_B_ADMIN].cleartext}"},

View file

@ -720,19 +720,17 @@ def _audit_upstream(provider_secret: str, marker: str, targets: list[str]):
assert request.headers["authorization"] == f"Bearer {provider_secret}"
targets.append(request.target)
body: Final = TypeAdapter(dict[str, JsonValue]).validate_json(request.body)
index: Final = len(targets)
reply_id: Final = uuid.uuid4().hex
if request.target == "/v1/responses":
if body.get("stream"):
return Reply(
content_type="text/event-stream", chunks=_responses_stream_frames(f"resp-{marker}-{index}")
content_type="text/event-stream", chunks=_responses_stream_frames(f"resp-{marker}-{reply_id}")
)
return _responses_result(f"resp-{marker}-{index}")
return _responses_result(f"resp-{marker}-{reply_id}")
assert request.target == "/v1/chat/completions", request.target
if body.get("stream"):
return Reply(
content_type="text/event-stream", chunks=_chat_stream_frames(f"chatcmpl-{marker}-{index}")
)
return _completion(f"{marker}-{index}-answer")
return Reply(content_type="text/event-stream", chunks=_chat_stream_frames(f"chatcmpl-{marker}-{reply_id}"))
return _completion(f"{marker}-{reply_id}-answer")
return upstream

View file

@ -30,6 +30,7 @@ from integration._support.otlp_sink import (
spans_for_trace,
)
from integration._support.process import owned_proxy, owned_proxy_process
from integration._support.redis_process import owned_redis
from pydantic import JsonValue
DB_SYSTEM_KEYS: Final = frozenset({"db.system.name", "db.system"})
@ -42,6 +43,14 @@ def tenant_export_interval(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("OTEL_BSP_SCHEDULE_DELAY", str(TENANT_EXPORT_INTERVAL_MILLISECONDS))
@pytest.fixture(autouse=True)
def private_redis(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
with owned_redis(tmp_path / "redis") as cache:
monkeypatch.setenv("REDIS_HOST", cache.host)
monkeypatch.setenv("REDIS_PORT", str(cache.port))
yield
@pytest.fixture
def audit_sinks(tmp_path: Path) -> Iterator[SpanSinks]:
with owned_sinks(tmp_path / "sinks") as sinks:

View file

@ -362,7 +362,7 @@ def _muse_deployment(scenario: Scenario, scenario_id: str, api_base: str) -> str
return scenario.model(model=f"meta/{MUSE_MODEL}", api_key=scenario_id, api_base=api_base)
def _named_deployment(creator: Gateway, scenario: Scenario, name: str, model: str, scenario_id: str) -> str:
def _named_deployment_id(creator: Gateway, scenario: Scenario, name: str, model: str, scenario_id: str) -> str:
created: Final = creator.post(
"/model/new",
{
@ -371,10 +371,23 @@ def _named_deployment(creator: Gateway, scenario: Scenario, name: str, model: st
"model_info": {},
},
)
scenario.cleanups.callback(scenario.delete_model, string_value(object_value(created["model_info"])["id"]))
model_id: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(scenario.delete_model, model_id)
return model_id
def _named_deployment(creator: Gateway, scenario: Scenario, name: str, model: str, scenario_id: str) -> str:
_named_deployment_id(creator, scenario, name, model, scenario_id)
return name
def _whisper_default_on_every_worker(owned: OwnedProxy, scenario: Scenario, scenario_id: str) -> None:
model_id: Final = _named_deployment_id(
owned.gateway, scenario, WHISPER_DEFAULT, f"openai/{WHISPER_DEFAULT}", scenario_id
)
eventually(lambda: owned.serves_only_on_every_worker(WHISPER_DEFAULT, model_id), lambda served: served, seconds=70)
def _close_code(closed: ConnectionClosed) -> int:
return 1006 if closed.rcvd is None else closed.rcvd.code
@ -868,9 +881,7 @@ def test_intent_without_model_routes_to_the_whisper_default_without_an_injected_
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _transcription_scenario(CLEAN_TRANSCRIPT))
key: Final = scenario.key()
_named_deployment(
guardrail_proxy.gateway, scenario, WHISPER_DEFAULT, f"openai/{WHISPER_DEFAULT}", handle.scenario_id
)
_whisper_default_on_every_worker(guardrail_proxy, scenario, handle.scenario_id)
session: Final = _transcribe(_ws_base(_owned_url(guardrail_proxy)), TRANSCRIPTION_QUERY, key)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
forwarded: Final = _with_transcription_model(GA_TRANSCRIPTION_UPDATE, WHISPER_DEFAULT)
@ -1458,9 +1469,7 @@ def test_intent_without_model_blocked_transcript_reports_a_violation_on_the_whis
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _transcription_scenario(BLOCKED_TRANSCRIPT))
key: Final = scenario.key()
_named_deployment(
guardrail_proxy.gateway, scenario, WHISPER_DEFAULT, f"openai/{WHISPER_DEFAULT}", handle.scenario_id
)
_whisper_default_on_every_worker(guardrail_proxy, scenario, handle.scenario_id)
session: Final = _transcribe_blocked(_ws_base(_owned_url(guardrail_proxy)), TRANSCRIPTION_QUERY, key)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
forwarded: Final = _with_transcription_model(GA_TRANSCRIPTION_UPDATE, WHISPER_DEFAULT)

View file

@ -42,7 +42,7 @@ from integration._support.client import (
object_value,
string_value,
)
from integration._support.database import read_rows
from integration._support.database import read_rows, scratch_database
from integration._support.process import graceful_stop_seconds, owned_proxy_process
from integration._support.upstream import delete_scenario, register_scenario
from integration._support.wire import Reply, Request, wire_server
@ -860,34 +860,34 @@ def test_hop_inside_one_encryption_boundary_keeps_the_reasoning(gateway: Gateway
assert [body.get("input") for body in bodies] == [history(mark), history(mark)], bodies
def configured_fallback(scenario: Scenario, source: str, target: str) -> None:
gateway: Final = scenario.gateway
original: Final = router_setting(gateway, "fallbacks")
restored: Final = original if isinstance(original, list) else []
scenario.cleanups.callback(lambda: gateway.post("/config/update", {"router_settings": {"fallbacks": restored}}))
gateway.post("/config/update", {"router_settings": {"fallbacks": [*restored, {source: [target]}]}})
def assert_cross_group_hop_stripped(gateway: Gateway, source_scenario: str, target_scenario: str, mark: str) -> None:
records: Final = observed(gateway)
assert [body.get("input") for body in bodies_for(records, source_scenario)] == [history(mark)], records
assert [body.get("input") for body in bodies_for(records, target_scenario)] == [stripped_history(mark)], records
def test_configured_fallbacks_entry_hop_strips_the_source_groups_reasoning(gateway: Gateway) -> None:
with gateway.scenario() as scenario:
def test_configured_fallbacks_entry_hop_strips_the_source_groups_reasoning(gateway: Gateway, tmp_path: Path) -> None:
with (
scratch_database() as database,
owned_proxy_process(
gateway,
tmp_path,
{"DATABASE_URL": database},
remove_environment=("DIRECT_URL", "DATABASE_URL_READ_REPLICA"),
) as owned,
owned.gateway.scenario() as scenario,
):
source: Final = f"src-{uuid.uuid4().hex[:12]}"
target: Final = f"dst-{uuid.uuid4().hex[:12]}"
source_id: Final = deployment(scenario, source, scenario_api_base(scenario, source, failing()), order=None)
target_id: Final = deployment(scenario, target, scenario_api_base(scenario, target, healthy_json()), order=None)
configured_fallback(scenario, source, target)
owned.gateway.post("/config/update", {"router_settings": {"num_retries": 0, "fallbacks": [{source: [target]}]}})
mark: Final = marker()
answer: Final = responses_answer(
gateway, HTTPX, Target(source, frozenset({source_id, target_id})), history(mark)
owned.gateway, HTTPX, Target(source, frozenset({source_id, target_id})), history(mark)
)
assert_served_by(answer, target_id)
assert_cross_group_hop_stripped(gateway, source, target, mark)
assert source not in json.dumps(router_setting(gateway, "fallbacks")), router_setting(gateway, "fallbacks")
assert_cross_group_hop_stripped(owned.gateway, source, target, mark)
def test_client_fallbacks_entry_hop_strips_the_source_groups_reasoning(gateway: Gateway) -> None:

View file

@ -239,6 +239,35 @@ def test_only_live_readable_workers_can_satisfy_readiness(
process_module._wait_until_ready(launch)
def test_readiness_never_sends_a_request_to_a_port_the_launch_does_not_own(
process_module: ModuleType, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(psutil, "process_iter", lambda: iter(()))
parent: Final = MagicMock()
parent.pid = 99
parent.poll.return_value = None
launch: Final = process_module._Launch(parent, 8123, tmp_path / "unused.log", 0.0)
original_client: Final = httpx.Client
def foreign_listener(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"probed a listener another process owns: {request.url}")
def client(*, base_url: str, timeout: float, trust_env: bool, limits: httpx.Limits) -> httpx.Client:
return original_client(
base_url=base_url,
timeout=timeout,
trust_env=trust_env,
limits=limits,
transport=httpx.MockTransport(foreign_listener),
)
monkeypatch.setattr(httpx, "Client", client)
monkeypatch.setattr(time, "monotonic", lambda: 0.0)
monkeypatch.setenv("INTEGRATION_PROXY_READY_SECONDS", "-1")
with pytest.raises(AssertionError, match="readiness deadline"):
process_module._wait_until_ready(launch)
def test_boot_phases_mark_when_prisma_ran_the_port_bound_and_each_worker_answered(
process_module: ModuleType,
) -> None:

View file

@ -4027,3 +4027,29 @@ def test_token_type_cost_breakdown_bills_cache_tokens_at_the_provider_speed_mult
assert breakdown.cache_read_cost == pytest.approx(multiplier * 400 * 1e-7)
assert breakdown.cache_creation_cost == pytest.approx(multiplier * 300 * 1.25e-6)
GEMINI_4_ARGON_TOKEN_PRICE_KEYS: Final = (
"input_cost_per_token",
"input_cost_per_token_batches",
"input_cost_per_token_flex",
"input_cost_per_token_priority",
"output_cost_per_token",
"output_cost_per_token_batches",
"output_cost_per_token_flex",
"output_cost_per_token_priority",
"output_cost_per_reasoning_token",
"cache_read_input_token_cost",
"cache_read_input_token_cost_batches",
"cache_read_input_token_cost_flex",
"cache_read_input_token_cost_priority",
)
def test_vertex_gemini_4_argon_matches_gemini_api_global_pricing(_local_model_cost_map: None) -> None:
vertex_row: Final = litellm.model_cost["vertex_ai/gemini-4-argon"]
gemini_row: Final = litellm.model_cost["gemini/gemini-4-argon"]
assert {key: vertex_row[key] for key in GEMINI_4_ARGON_TOKEN_PRICE_KEYS} == {
key: gemini_row[key] for key in GEMINI_4_ARGON_TOKEN_PRICE_KEYS
}
assert vertex_row["litellm_provider"].startswith("vertex_ai")

View file

@ -145,11 +145,14 @@ class TestOpenAIResponsesAPIConfig:
(17, 17),
(256, 256),
(None, None),
(0, 0),
(-1, -1),
],
)
def test_enforce_min_max_output_tokens(self, value, expected):
"""Below the minimum clamps to 16; the boundary, larger values, and None
are returned unchanged so no previously-valid request regresses."""
"""Positive values below the minimum clamp to 16; the boundary, larger values,
None, and non-positive values are returned unchanged so the provider rejects
an invalid value instead of the proxy silently rewriting it."""
assert self.config._enforce_min_max_output_tokens(value) == expected
def validate_responses_api_request_params(self, params, expected_fields):

View file

@ -1,3 +1,5 @@
from typing import Final
import httpx
import pytest
@ -63,6 +65,30 @@ def test_transform_create_request(config: OpenAIVectorStoreFilesConfig):
assert payload["attributes"]["key"] == "value"
_SEVENTEEN_ATTRIBUTES: Final[dict[str, str]] = {f"key_{index}": f"value_{index}" for index in range(17)}
def test_transform_create_request_keeps_every_attribute(config: OpenAIVectorStoreFilesConfig):
_, payload = config.transform_create_vector_store_file_request(
vector_store_id="vs_123",
create_request={"file_id": "file-abc", "attributes": {**_SEVENTEEN_ATTRIBUTES, "hidden_params": "x"}},
api_base="https://api.example.com/v1/vector_stores/vs_123/files",
)
assert payload["attributes"] == _SEVENTEEN_ATTRIBUTES
def test_transform_update_request_keeps_every_attribute(config: OpenAIVectorStoreFilesConfig):
_, payload = config.transform_update_vector_store_file_request(
vector_store_id="vs_123",
file_id="file-abc",
update_request={"attributes": {**_SEVENTEEN_ATTRIBUTES, "hidden_params": "x"}},
api_base="https://api.example.com/v1/vector_stores/vs_123/files",
)
assert payload["attributes"] == _SEVENTEEN_ATTRIBUTES
def test_transform_list_request(config: OpenAIVectorStoreFilesConfig):
api_base = "https://api.example.com/v1/vector_stores/vs_123/files"
url, params = config.transform_list_vector_store_files_request(

View file

@ -1406,7 +1406,7 @@ async def test_cold_concurrent_schema_validation_accepts_valid_arguments() -> No
os.sched_setaffinity(0, {min(original_affinity)})
for _ in range(2):
results: Final = await asyncio.gather(
*(_tool_argument_validation_error(schema, {"amount": 0.75}) for _ in range(8))
*(_tool_argument_validation_error(schema, {"amount": 0.75}, timeout_seconds=90.0) for _ in range(8))
)
assert results == [None] * 8, "Cold and warm workers must accept valid concurrent tool arguments"
finally:

View file

@ -2513,6 +2513,15 @@ def test_parse_canonical_date_range_reports_missing_then_malformed_dates() -> No
)
def test_parse_canonical_date_range_rejects_end_before_start() -> None:
assert parse_canonical_date_range("2026-09-26", "2026-09-24") == InvalidDateRange(
reason="end_date must be on or after start_date"
)
assert parse_canonical_date_range("2026-09-24", "2026-09-24") == CanonicalDateRange(
start=date(2026, 9, 24), end=date(2026, 9, 24)
)
@pytest.mark.asyncio
@pytest.mark.parametrize("start_date", ("2026-9-24", "2026-09-24", "2026-09-4"))
async def test_get_daily_activity_rejects_non_canonical_dates_before_querying(start_date: str) -> None:

View file

@ -3,9 +3,12 @@ Tests for AI Usage Chat module.
"""
import json
from collections.abc import AsyncIterator
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import HTTPException
from litellm.proxy.management_endpoints.usage_endpoints.ai_usage_chat import (
TOOL_HANDLERS,
@ -317,6 +320,71 @@ class TestStreamUsageAiChat:
assert len(chunk_events) >= 1
assert "Engineering" in chunk_events[0]["content"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"reason",
[
"end_date must be on or after start_date",
"start_date and end_date must be valid YYYY-MM-DD dates",
],
ids=["reversed", "invalid-format"],
)
@pytest.mark.parametrize(
"tool_name", ["get_usage_data", "get_team_usage_data", "get_tag_usage_data"]
)
async def test_invalid_tool_dates_are_reported_to_the_model(
self, tool_name: str, reason: str
) -> None:
fetch: Final = AsyncMock(
side_effect=HTTPException(status_code=400, detail={"error": reason})
)
tool_message, statuses = await _run_one_tool(
tool_name, "2025-01-31", "2025-01-01", fetch
)
fetch.assert_awaited_once()
assert tool_message == f"Invalid date range: {reason}"
assert statuses == ["running", "complete"]
@pytest.mark.asyncio
async def test_tool_dates_the_query_layer_accepts_are_queried_as_sent(
self,
) -> None:
fetch: Final = AsyncMock(return_value=SAMPLE_AGGREGATED_RESPONSE)
tool_message, statuses = await _run_one_tool(
"get_usage_data", "2025/01/01", "2025-01-31T00:00:00", fetch
)
fetch.assert_awaited_once_with(
start_date="2025/01/01", end_date="2025-01-31T00:00:00"
)
assert tool_message.startswith("Total Spend: $")
assert statuses == ["running", "complete"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"error",
[
HTTPException(status_code=500, detail={"error": "db down"}),
HTTPException(status_code=400, detail="page and page_size must be >= 1"),
],
ids=["server-error", "other-400"],
)
async def test_other_tool_errors_stay_a_generic_fetch_error(
self, error: HTTPException
) -> None:
tool_message, statuses = await _run_one_tool(
"get_team_usage_data",
"2025-01-01",
"2025-01-31",
AsyncMock(side_effect=error),
)
assert tool_message == "Error fetching team usage data. Please try again."
assert statuses == ["running", "error"]
@pytest.mark.asyncio
async def test_stream_handles_error(self):
with patch(
@ -524,3 +592,66 @@ class TestUsageAiChatKeepalive:
assert b": ping\n\n" not in chunks
assert chunks[0].startswith(b'data: {"type": "status"')
assert chunks[-1] == b'data: {"type": "done"}\n\n'
async def _run_one_tool(
tool_name: str, start_date: str, end_date: str, fetch: AsyncMock
) -> tuple[str, list[object]]:
"""Runs one admin chat turn whose model calls `tool_name` with these dates; returns the
tool message the follow-up model call carried and the tool statuses streamed."""
arguments: Final = json.dumps({"start_date": start_date, "end_date": end_date})
mock_tool_call: Final = MagicMock()
mock_tool_call.id = "call_dates"
mock_tool_call.function.name = tool_name
mock_tool_call.function.arguments = arguments
mock_first_response: Final = MagicMock()
mock_first_response.choices = [MagicMock()]
mock_first_response.choices[0].message.tool_calls = [mock_tool_call]
mock_first_response.choices[0].message.model_dump.return_value = {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_dates",
"type": "function",
"function": {"name": tool_name, "arguments": arguments},
}
],
}
async def mock_stream() -> AsyncIterator[MagicMock]:
chunk: Final = MagicMock()
chunk.choices = [MagicMock()]
chunk.choices[0].delta.content = "Done."
yield chunk
with (
patch(
"litellm.proxy.management_endpoints.usage_endpoints.ai_usage_chat.litellm"
) as mock_litellm,
patch.dict(TOOL_HANDLERS[tool_name], {"fetch": fetch}),
):
mock_litellm.acompletion = AsyncMock(
side_effect=[mock_first_response, mock_stream()]
)
events: Final[tuple[dict[str, object], ...]] = tuple(
[
json.loads(event.replace("data: ", "").strip())
async for event in stream_usage_ai_chat(
messages=[{"role": "user", "content": "How much did we spend?"}],
model="gpt-4o-mini",
is_admin=True,
)
]
)
final_messages: Final = mock_litellm.acompletion.call_args_list[1].kwargs[
"messages"
]
tool_messages: Final = [m for m in final_messages if m["role"] == "tool"]
assert len(tool_messages) == 1, tool_messages
assert tool_messages[0]["tool_call_id"] == "call_dates"
return str(tool_messages[0]["content"]), [
e["status"] for e in events if e["type"] == "tool_call"
]

View file

@ -3,14 +3,16 @@ Tests for the proxy OCR endpoint helpers that select the response format
(`x-req-format: native | litellm`) and return the provider's native payload.
"""
import io
from typing import Final
from unittest.mock import AsyncMock, MagicMock
import orjson
import pytest
from fastapi import HTTPException
from fastapi import HTTPException, UploadFile
from litellm.llms.base_llm.ocr.transformation import OCRPage, OCRResponse
from litellm.proxy.ocr_endpoints.endpoints import _native_response, _parse_ocr_request
from litellm.proxy.ocr_endpoints.endpoints import OCRRequestError, _native_response, _parse_ocr_request
AZURE_NATIVE_OPERATION = {
"status": "succeeded",
@ -145,3 +147,83 @@ def test_upload_octet_stream_content_type_falls_back_to_filename_inference():
assert "mime_type" not in document
assert document["file"].name == "receipt.pdf"
def _multipart_request(form: dict[str, object]) -> MagicMock:
request: Final = MagicMock()
request.headers = {"content-type": "multipart/form-data; boundary=x"}
request.form = AsyncMock(return_value=form)
return request
@pytest.mark.asyncio
@pytest.mark.parametrize(
"form, message",
[
({"model": "mistral-ocr"}, "must include a 'file' field"),
(
{"model": "mistral-ocr", "file": UploadFile(file=io.BytesIO(b""), filename="empty.pdf")},
"Uploaded file is empty",
),
],
ids=["missing-file", "empty-file"],
)
async def test_malformed_multipart_upload_is_a_400_on_the_file_param(form: dict[str, object], message: str) -> None:
with pytest.raises(OCRRequestError) as exc_info:
await _parse_ocr_request(_multipart_request(form))
assert exc_info.value.code == "400"
assert exc_info.value.type == "invalid_request_error"
assert exc_info.value.param == "file"
assert message in exc_info.value.message
@pytest.mark.asyncio
async def test_unparseable_multipart_form_is_a_400() -> None:
request: Final = _multipart_request({})
request.form = AsyncMock(side_effect=ValueError("missing boundary"))
with pytest.raises(OCRRequestError) as exc_info:
await _parse_ocr_request(request)
assert exc_info.value.code == "400"
assert exc_info.value.type == "invalid_request_error"
assert "Failed to parse multipart form data: missing boundary" in exc_info.value.message
@pytest.mark.asyncio
async def test_multipart_file_over_the_size_limit_is_a_400_on_the_file_param(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("litellm.proxy.ocr_endpoints.endpoints._MAX_FILE_BYTES", 4)
form: Final = {"model": "mistral-ocr", "file": UploadFile(file=io.BytesIO(b"12345"), filename="big.pdf")}
with pytest.raises(OCRRequestError) as exc_info:
await _parse_ocr_request(_multipart_request(form))
assert exc_info.value.code == "400"
assert exc_info.value.param == "file"
assert exc_info.value.message == "OCR file exceeds the size limit"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"body, message",
[
(b"", "Empty request body."),
(b"{not json", "Invalid JSON in request body:"),
(
orjson.dumps({"model": "mistral-ocr", "document": {"type": "file", "file": "/etc/passwd"}}),
"document type 'file' is not supported through the JSON API.",
),
],
ids=["empty-body", "invalid-json", "document-type-file"],
)
async def test_malformed_json_ocr_request_is_a_400(body: bytes, message: str) -> None:
request: Final = _json_request({}, {})
request.body = AsyncMock(return_value=body)
with pytest.raises(OCRRequestError) as exc_info:
await _parse_ocr_request(request)
assert exc_info.value.code == "400"
assert exc_info.value.type == "invalid_request_error"
assert exc_info.value.message.startswith(message)

View file

@ -451,7 +451,7 @@ class TestProxyBaseLLMRequestProcessing:
mock_add_litellm_data_to_request,
)
mock_general_settings = {}
mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
mock_user_api_key_dict = ProxyUserAPIKeyAuth()
mock_proxy_config = MagicMock(spec=ProxyConfig)
route_type = "acompletion"
@ -573,7 +573,7 @@ class TestProxyBaseLLMRequestProcessing:
returned_data, logging_obj = await processing_obj.common_processing_pre_call_logic(
request=mock_request,
general_settings={},
user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
user_api_key_dict=ProxyUserAPIKeyAuth(),
proxy_logging_obj=mock_proxy_logging_obj,
proxy_config=MagicMock(spec=ProxyConfig),
route_type=route_type,
@ -1080,7 +1080,7 @@ class TestProxyBaseLLMRequestProcessing:
)
mock_general_settings = {}
mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
mock_user_api_key_dict = ProxyUserAPIKeyAuth()
mock_proxy_config = MagicMock(spec=ProxyConfig)
mock_router_settings = {
@ -2079,7 +2079,7 @@ class TestProxyBaseLLMRequestProcessing:
mock_add_litellm_data_to_request,
)
mock_general_settings = {}
mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
mock_user_api_key_dict = ProxyUserAPIKeyAuth()
mock_proxy_config = MagicMock(spec=ProxyConfig)
route_type = "acompletion"

View file

@ -1916,3 +1916,133 @@ def test_endpoint_unsupported_ignores_models_without_deployments(model):
def test_endpoint_unsupported_skips_the_check_without_a_router():
assert _unsupported("acompletion", "embed-only", None) is None
def _router_with_defaults(default_litellm_params: dict[str, object]):
import litellm
return litellm.Router(
model_list=[
{
"model_name": "served",
"litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "test-key"},
"model_info": {"id": "served-deployment"},
}
],
default_litellm_params=default_litellm_params,
)
@pytest.mark.parametrize(
"route_type, param",
[
("aimage_generation", "prompt"),
("aspeech", "input"),
("atext_completion", "prompt"),
("atranscription", "file"),
],
)
def test_router_default_for_positional_param_still_raises(route_type: str, param: str) -> None:
from litellm.proxy.route_llm_request import (
ProxyMissingRequiredParamError,
raise_if_required_body_param_missing,
)
with pytest.raises(ProxyMissingRequiredParamError) as exc_info:
raise_if_required_body_param_missing(
route_type=route_type,
data={"model": "served"},
llm_router=_router_with_defaults({param: "supplied-by-router-default"}),
)
assert exc_info.value.param == param
@pytest.mark.parametrize(
"route_type, param",
[
("aimage_generation", "prompt"),
("aspeech", "input"),
],
)
def test_user_config_default_for_positional_param_still_raises(route_type: str, param: str) -> None:
from litellm.proxy.route_llm_request import (
ProxyMissingRequiredParamError,
raise_if_required_body_param_missing,
)
with pytest.raises(ProxyMissingRequiredParamError) as exc_info:
raise_if_required_body_param_missing(
route_type=route_type,
data={"model": "img", "user_config": {"model_list": [], "default_litellm_params": {param: "x"}}},
llm_router=None,
)
assert exc_info.value.param == param
def test_deployment_param_for_positional_param_still_raises() -> None:
import litellm
from litellm.proxy.route_llm_request import (
ProxyMissingRequiredParamError,
raise_if_required_body_param_missing,
)
router: Final = litellm.Router(
model_list=[
{
"model_name": "img",
"litellm_params": {"model": "openai/gpt-image-1", "api_key": "test-key", "prompt": "a cat"},
}
]
)
with pytest.raises(ProxyMissingRequiredParamError) as exc_info:
raise_if_required_body_param_missing(route_type="aimage_generation", data={"model": "img"}, llm_router=router)
assert exc_info.value.param == "prompt"
@pytest.mark.parametrize(
"model",
[
pytest.param("omni-moderation-2024-09-26", id="unknown-model"),
pytest.param("openai/gpt-4o-mini", id="deployment-name"),
pytest.param("served-deployment", id="deployment-id"),
],
)
def test_router_default_ignored_for_moderation_model_the_router_does_not_list(model: str) -> None:
from litellm.proxy.route_llm_request import (
ProxyMissingRequiredParamError,
raise_if_required_body_param_missing,
)
with pytest.raises(ProxyMissingRequiredParamError) as exc_info:
raise_if_required_body_param_missing(
route_type="amoderation",
data={"model": model},
llm_router=_router_with_defaults({"input": "supplied"}),
)
assert exc_info.value.param == "input"
@pytest.mark.asyncio
async def test_null_user_config_is_treated_as_absent() -> None:
router: Final = _router_with_defaults({})
response: Final = await (
await route_request(
{
"model": "served",
"messages": [{"role": "user", "content": "hi"}],
"mock_response": "from the main router",
"user_config": None,
},
router,
None,
"acompletion",
)
)
assert response.choices[0].message.content == "from the main router"

6
uv.lock generated
View file

@ -4500,7 +4500,7 @@ wheels = [
[[package]]
name = "litellm"
version = "1.106.0"
version = "1.107.0"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@ -4996,12 +4996,12 @@ provides-extras = ["test"]
[[package]]
name = "litellm-enterprise"
version = "0.1.76"
version = "0.1.77"
source = { editable = "enterprise" }
[[package]]
name = "litellm-proxy-extras"
version = "0.4.108"
version = "0.4.109"
source = { editable = "litellm-proxy-extras" }
dependencies = [
{ name = "psycopg" },