diff --git a/enterprise/pyproject.toml b/enterprise/pyproject.toml index 78862d4c942..4ffb7169f24 100644 --- a/enterprise/pyproject.toml +++ b/enterprise/pyproject.toml @@ -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==", diff --git a/litellm-proxy-extras/pyproject.toml b/litellm-proxy-extras/pyproject.toml index 68efeac71ae..61d56c6b301 100644 --- a/litellm-proxy-extras/pyproject.toml +++ b/litellm-proxy-extras/pyproject.toml @@ -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==", diff --git a/litellm/llms/openai/responses/transformation.py b/litellm/llms/openai/responses/transformation.py index ec4f0c04475..abd3d03b139 100644 --- a/litellm/llms/openai/responses/transformation.py +++ b/litellm/llms/openai/responses/transformation.py @@ -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 diff --git a/litellm/llms/openai/vector_store_files/transformation.py b/litellm/llms/openai/vector_store_files/transformation.py index 8519b5f4bc4..d5f15d639d6 100644 --- a/litellm/llms/openai/vector_store_files/transformation.py +++ b/litellm/llms/openai/vector_store_files/transformation.py @@ -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 diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 37675643733..8977c5dd88e 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -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, diff --git a/litellm/proxy/_experimental/mcp_server/tool_search.py b/litellm/proxy/_experimental/mcp_server/tool_search.py index a8aff2b2e4b..23a63c00bd0 100644 --- a/litellm/proxy/_experimental/mcp_server/tool_search.py +++ b/litellm/proxy/_experimental/mcp_server/tool_search.py @@ -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 ) diff --git a/litellm/proxy/management_endpoints/common_daily_activity.py b/litellm/proxy/management_endpoints/common_daily_activity.py index 28dfbb09eab..f948a49441a 100644 --- a/litellm/proxy/management_endpoints/common_daily_activity.py +++ b/litellm/proxy/management_endpoints/common_daily_activity.py @@ -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) diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py index 544bc471ce5..6a4c8ac2c9e 100644 --- a/litellm/proxy/management_endpoints/team_endpoints.py +++ b/litellm/proxy/management_endpoints/team_endpoints.py @@ -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 diff --git a/litellm/proxy/management_endpoints/usage_endpoints/ai_usage_chat.py b/litellm/proxy/management_endpoints/usage_endpoints/ai_usage_chat.py index 3c2300f14fd..7a45d0438de 100644 --- a/litellm/proxy/management_endpoints/usage_endpoints/ai_usage_chat.py +++ b/litellm/proxy/management_endpoints/usage_endpoints/ai_usage_chat.py @@ -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]], diff --git a/litellm/proxy/ocr_endpoints/endpoints.py b/litellm/proxy/ocr_endpoints/endpoints.py index 94fe9354386..62fde60dcac 100644 --- a/litellm/proxy/ocr_endpoints/endpoints.py +++ b/litellm/proxy/ocr_endpoints/endpoints.py @@ -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 " diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index d2593a0f66e..23b20f017cf 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -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 [] diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 37675643733..8977c5dd88e 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -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, diff --git a/pyproject.toml b/pyproject.toml index 0ed8b8757ac..0be8adb7337 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", ] diff --git a/tests/integration/_support/process.py b/tests/integration/_support/process.py index 5479629420a..f135a091cb3 100644 --- a/tests/integration/_support/process.py +++ b/tests/integration/_support/process.py @@ -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" diff --git a/tests/integration/_support/runtime.py b/tests/integration/_support/runtime.py index a638dcfa4bb..a657ce4dff0 100644 --- a/tests/integration/_support/runtime.py +++ b/tests/integration/_support/runtime.py @@ -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) diff --git a/tests/integration/compatibility/_status_code_audit.py b/tests/integration/compatibility/_status_code_audit.py new file mode 100644 index 00000000000..59513891ec0 --- /dev/null +++ b/tests/integration/compatibility/_status_code_audit.py @@ -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]" + ) diff --git a/tests/integration/compatibility/test_status_code_followups.py b/tests/integration/compatibility/test_status_code_followups.py new file mode 100644 index 00000000000..466eeeddde1 --- /dev/null +++ b/tests/integration/compatibility/test_status_code_followups.py @@ -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 diff --git a/tests/integration/compatibility/test_status_code_followups_chaos.py b/tests/integration/compatibility/test_status_code_followups_chaos.py new file mode 100644 index 00000000000..1cb7f45e681 --- /dev/null +++ b/tests/integration/compatibility/test_status_code_followups_chaos.py @@ -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")) diff --git a/tests/integration/compatibility/test_status_code_followups_defaults.py b/tests/integration/compatibility/test_status_code_followups_defaults.py new file mode 100644 index 00000000000..a0c73d12dec --- /dev/null +++ b/tests/integration/compatibility/test_status_code_followups_defaults.py @@ -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" diff --git a/tests/integration/compatibility/test_status_code_followups_requests.py b/tests/integration/compatibility/test_status_code_followups_requests.py new file mode 100644 index 00000000000..cffaf09ec55 --- /dev/null +++ b/tests/integration/compatibility/test_status_code_followups_requests.py @@ -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 diff --git a/tests/integration/compatibility/test_status_code_followups_usage_chat.py b/tests/integration/compatibility/test_status_code_followups_usage_chat.py new file mode 100644 index 00000000000..6d9d9507158 --- /dev/null +++ b/tests/integration/compatibility/test_status_code_followups_usage_chat.py @@ -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 diff --git a/tests/integration/management/test_team_update.py b/tests/integration/management/test_team_update.py index 561191c9722..b7b5fe34c87 100644 --- a/tests/integration/management/test_team_update.py +++ b/tests/integration/management/test_team_update.py @@ -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}"}, diff --git a/tests/integration/observability/test_langfuse_delivery.py b/tests/integration/observability/test_langfuse_delivery.py index bee754297f0..7fc2ced63e8 100644 --- a/tests/integration/observability/test_langfuse_delivery.py +++ b/tests/integration/observability/test_langfuse_delivery.py @@ -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 diff --git a/tests/integration/observability/test_otel_excluded_services.py b/tests/integration/observability/test_otel_excluded_services.py index a90d1836d24..b5d932c3648 100644 --- a/tests/integration/observability/test_otel_excluded_services.py +++ b/tests/integration/observability/test_otel_excluded_services.py @@ -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: diff --git a/tests/integration/providers/test_realtime_transcription_guardrail_session_update.py b/tests/integration/providers/test_realtime_transcription_guardrail_session_update.py index cda00b5f3fa..f12254b8d1d 100644 --- a/tests/integration/providers/test_realtime_transcription_guardrail_session_update.py +++ b/tests/integration/providers/test_realtime_transcription_guardrail_session_update.py @@ -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) diff --git a/tests/integration/routing/test_fallback_hop_foreign_encrypted_reasoning.py b/tests/integration/routing/test_fallback_hop_foreign_encrypted_reasoning.py index 32c7b99d5f1..490c26226bd 100644 --- a/tests/integration/routing/test_fallback_hop_foreign_encrypted_reasoning.py +++ b/tests/integration/routing/test_fallback_hop_foreign_encrypted_reasoning.py @@ -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: diff --git a/tests/unit/integration_support/test_process.py b/tests/unit/integration_support/test_process.py index b2271a1ed47..1a6257e9193 100644 --- a/tests/unit/integration_support/test_process.py +++ b/tests/unit/integration_support/test_process.py @@ -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: diff --git a/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py b/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py index be39e83f989..67a3e2ddd5c 100644 --- a/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py +++ b/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py @@ -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") diff --git a/tests/unit/llms/openai/responses/test_openai_responses_transformation.py b/tests/unit/llms/openai/responses/test_openai_responses_transformation.py index 73332728964..2121e04dcac 100644 --- a/tests/unit/llms/openai/responses/test_openai_responses_transformation.py +++ b/tests/unit/llms/openai/responses/test_openai_responses_transformation.py @@ -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): diff --git a/tests/unit/llms/openai/vector_store_files/test_openai_vector_store_files_transformation.py b/tests/unit/llms/openai/vector_store_files/test_openai_vector_store_files_transformation.py index aebd93d062e..089b092dfa6 100644 --- a/tests/unit/llms/openai/vector_store_files/test_openai_vector_store_files_transformation.py +++ b/tests/unit/llms/openai/vector_store_files/test_openai_vector_store_files_transformation.py @@ -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( diff --git a/tests/unit/proxy/_experimental/mcp_server/test_mcp_tool_search.py b/tests/unit/proxy/_experimental/mcp_server/test_mcp_tool_search.py index bd4962829c2..0d6b83b3275 100644 --- a/tests/unit/proxy/_experimental/mcp_server/test_mcp_tool_search.py +++ b/tests/unit/proxy/_experimental/mcp_server/test_mcp_tool_search.py @@ -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: diff --git a/tests/unit/proxy/management_endpoints/test_common_daily_activity.py b/tests/unit/proxy/management_endpoints/test_common_daily_activity.py index e6a6680d3e4..01d2f7007a1 100644 --- a/tests/unit/proxy/management_endpoints/test_common_daily_activity.py +++ b/tests/unit/proxy/management_endpoints/test_common_daily_activity.py @@ -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: diff --git a/tests/unit/proxy/management_endpoints/usage_endpoints/test_ai_usage_chat.py b/tests/unit/proxy/management_endpoints/usage_endpoints/test_ai_usage_chat.py index e5616a1975d..4daddb89350 100644 --- a/tests/unit/proxy/management_endpoints/usage_endpoints/test_ai_usage_chat.py +++ b/tests/unit/proxy/management_endpoints/usage_endpoints/test_ai_usage_chat.py @@ -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" + ] diff --git a/tests/unit/proxy/ocr_endpoints/test_endpoints.py b/tests/unit/proxy/ocr_endpoints/test_endpoints.py index 2b4f370ddee..6764e0cb434 100644 --- a/tests/unit/proxy/ocr_endpoints/test_endpoints.py +++ b/tests/unit/proxy/ocr_endpoints/test_endpoints.py @@ -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) diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index 1d7ccf90650..7526b6617c8 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -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" diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index a82067ef310..a298f57b3e2 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -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" diff --git a/uv.lock b/uv.lock index 471460a32b5..05331266b3e 100644 --- a/uv.lock +++ b/uv.lock @@ -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" },