diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index e52f11641b0..4d54b58f73b 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -120,6 +120,7 @@ from litellm.proxy.dd_span_tagger import DDSpanTagger from litellm.proxy.guardrails.auto_router_compression import arm_pre_call as _arm_auto_router_compression from litellm.proxy.native_compaction import with_proxy_compaction_executor from litellm.proxy.route_llm_request import ( + endpoint_unsupported_by_model, route_request, ) from litellm.proxy.telemetry.request_context import bind_call_id as bind_telemetry_call_id @@ -132,6 +133,7 @@ from litellm.router import Router from litellm.router_utils.add_retry_fallback_headers import get_hidden_params_dict from litellm.router_utils.auto_router_model_naming import AUTO_ROUTER_MODEL_PREFIX from litellm.router_utils.common_utils import resolve_model_group_alias, resolve_served_model +from litellm.router_utils.endpoint_eligibility import UnsupportedEndpointError from litellm.types.guardrails import GuardrailEventHooks from litellm.types.router import RouterRateLimitError from litellm.types.router_weights import validate_router_weights @@ -2557,6 +2559,17 @@ class ProxyBaseLLMRequestProcessing: original_model, fallback_models, ) + raw_router_settings: Final = _DECODED_JSON.validate_python( + _REQUEST_FIELDS.validate_python(rate_limited_data).get("router_settings_override") + ) + group_alias: Final = _DECODED_JSON.validate_python( + _REQUEST_FIELDS.validate_python(raw_router_settings).get("model_group_alias") + if isinstance(raw_router_settings, Mapping) + else None + ) + key_aliases: Final[object] = cast( # cast-ok: auth aliases are dynamically typed at runtime + object, user_api_key_dict.aliases + ) try: for fallback_model in fallback_models: if fallback_model == original_model: @@ -2572,6 +2585,20 @@ class ProxyBaseLLMRequestProcessing: self.data = independent_snapshot(pristine) self.data["model"] = fallback_model try: + probe_data = { + **_REQUEST_FIELDS.validate_python(pristine), + "model": resolve_served_model( + fallback_model, + _DECODED_JSON.validate_python(litellm.model_alias_map), + key_aliases, + group_alias, + ), + } + if ( + endpoint_unsupported_by_model(route_type, probe_data, llm_router, user_api_key_dict.team_id) + is not None + ): + continue return await self.common_processing_pre_call_logic( request=request, general_settings=general_settings, @@ -2589,6 +2616,8 @@ class ProxyBaseLLMRequestProcessing: llm_router=llm_router, rate_limited_model=original_model, ) + except UnsupportedEndpointError: + continue except ProxyRateLimitError as fallback_exc: if is_hard_per_model_limit(fallback_exc): raise diff --git a/litellm/proxy/common_utils/callback_utils.py b/litellm/proxy/common_utils/callback_utils.py index 4fa7709e5c7..b95633f25fe 100644 --- a/litellm/proxy/common_utils/callback_utils.py +++ b/litellm/proxy/common_utils/callback_utils.py @@ -429,10 +429,7 @@ def get_model_group_from_litellm_kwargs(kwargs: dict) -> str | None: def get_model_group_from_request_data(data: dict) -> str | None: _metadata: Final = data.get("metadata", None) or {} _model_group: Final = _metadata.get("model_group", None) - if _model_group is not None: - return _model_group - - return None + return _model_group if isinstance(_model_group, str) else None def get_remaining_tokens_and_requests_from_request_data(data: dict) -> dict[str, str]: diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index aebcab75601..2111602a8f8 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -72,6 +72,7 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attributio from litellm.proxy.route_llm_request import raise_if_required_body_param_missing from litellm.responses.streaming_iterator import GuardedResponsesWsFrame, ResponsesWsFrameGuard from litellm.router import Router +from litellm.router_utils.endpoint_eligibility import UnsupportedEndpointError from litellm.types.llms.base import LiteLLMBaseModel from litellm.types.llms.openai import ( REASONING_EFFORT, @@ -1767,6 +1768,13 @@ def _routing_hints_from_first_ws_frame(first_message: str) -> Mapping[str, objec def _responses_ws_failure_frame(failure: Exception) -> str: + if isinstance(failure, UnsupportedEndpointError): + return json.dumps( + { + "type": "error", + "error": {"type": "invalid_request_error", "message": failure.message.split("\n\n", 1)[0]}, + } + ) raw_status: Final = getattr(failure, "status_code", None) status: Final = raw_status if isinstance(raw_status, int) and not isinstance(raw_status, bool) else 500 error_type: Final = ( @@ -2030,6 +2038,7 @@ async def _responses_websocket_session( user_api_base=user_api_base, model=resolved_model, route_type="_aresponses_websocket", + llm_router=llm_router, skip_guardrails=True, ) except Exception as e: @@ -2081,7 +2090,10 @@ async def _responses_websocket_session( original_exception=e, request_data=routed_data, ) - await websocket.close(code=1011, reason="Internal server error") + await websocket.close( + code=1008 if isinstance(e, UnsupportedEndpointError) else 1011, + reason="Pre-call error" if isinstance(e, UnsupportedEndpointError) else "Internal server error", + ) finally: await proxy_logging_obj.arelease_responses_websocket_slot(user_api_key_dict) diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 8bab7215b17..c0e4afae477 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -356,6 +356,26 @@ def raise_if_required_body_param_missing( ) +def endpoint_unsupported_by_model( + route_type: str, + data: Mapping[str, object], + llm_router: LitellmRouter | None, + team_id: str | None, +) -> litellm.BadRequestError | None: + model: Final = data.get("model") + if not isinstance(model, str) or llm_router is None: + return None + metadata: Final = data.get("metadata") + request_kwargs: Final = { + **data, + "metadata": { + **(metadata if isinstance(metadata, Mapping) else {}), + "user_api_key_team_id": team_id, + }, + } + return llm_router.endpoint_error_for_request(model, route_type, request_kwargs) + + class MockTestingParamsDisabledError(HTTPException): def __init__(self, params: tuple[str, ...]): super().__init__( diff --git a/litellm/proxy/spend_tracking/spend_tracking_utils.py b/litellm/proxy/spend_tracking/spend_tracking_utils.py index d1303b012c8..34a7d0a64af 100644 --- a/litellm/proxy/spend_tracking/spend_tracking_utils.py +++ b/litellm/proxy/spend_tracking/spend_tracking_utils.py @@ -433,7 +433,11 @@ def _sl_attribution_fallback( ) -> str: if standard_logging_payload is None: return "" - return standard_logging_payload.get(field) or "" + return _string_or_empty(standard_logging_payload.get(field)) + + +def _string_or_empty(value: object) -> str: + return value if isinstance(value, str) else "" def _deployment_provider(deployment: DeploymentTypedDict) -> str | None: @@ -634,7 +638,7 @@ def get_logging_payload( _model_id: Final = metadata.get("model_info", {}).get("id", "") or _sl_attribution_fallback( standard_logging_payload, "model_id" ) - _model_group: Final = metadata.get("model_group", "") or _sl_attribution_fallback( + _model_group: Final = _string_or_empty(metadata.get("model_group")) or _sl_attribution_fallback( standard_logging_payload, "model_group" ) _api_base: Final = litellm_params.get("api_base", "") or _sl_attribution_fallback( diff --git a/litellm/router.py b/litellm/router.py index 8d77fd0dac5..6d5a71f0218 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -226,6 +226,11 @@ from litellm.router_utils.cooldown_handlers import ( is_caller_timeout_408, set_cooldown_deployments, ) +from litellm.router_utils.endpoint_eligibility import ( + UnsupportedEndpointError, + eligible_deployments, + endpoint_selection, +) from litellm.router_utils.fallback_event_handlers import ( MID_STREAM_FALLBACK_CONTROLS_KEY, AttemptedFallbackTargets, @@ -1272,7 +1277,7 @@ class Router: self.assistants_config = assistants_config self.search_tools = search_tools or [] self.guardrail_list = guardrail_list or [] - self.deployment_names: list = [] # names of models under litellm_params. ex. azure/chatgpt-v-2 + self.deployment_names: list[str] = [] # names of models under litellm_params. ex. azure/chatgpt-v-2 self.deployment_latency_map = {} ### CACHING ### cache_type: Literal["local", "redis", "redis-semantic", "s3", "disk"] = "local" # default to an in-memory cache @@ -5529,12 +5534,13 @@ class Router: kwargs.setdefault("metadata", {}).update({"model_group": model}) # pick the one that is available (lowest TPM/RPM) - deployment: Final = self.get_available_deployment( - model=model, - messages=messages, - specific_deployment=kwargs.pop("specific_deployment", None), - request_kwargs=kwargs, - ) + with endpoint_selection("text_completion", model=model): + deployment: Final = self.get_available_deployment( + model=model, + messages=messages, + specific_deployment=kwargs.pop("specific_deployment", None), + request_kwargs=kwargs, + ) data: Final = deployment["litellm_params"].copy() for k, v in self.default_litellm_params.items(): @@ -8689,9 +8695,18 @@ class Router: Handler for making a call to the .completion()/.embeddings()/etc. functions. """ model_group: Final = kwargs.get("model") - response = original_function(*args, **kwargs) - if coroutine_checker.is_async_callable(response) or inspect.isawaitable(response): - response = await response + generic_function: Final = _OPAQUE_VALUE_ADAPTER.validate_python(kwargs.get("original_generic_function")) + endpoint_function: Final = ( + generic_function if callable(generic_function) else _OPAQUE_VALUE_ADAPTER.validate_python(original_function) + ) + with endpoint_selection( + getattr(endpoint_function, "__name__", None), + getattr(endpoint_function, "__module__", None), + model=_OPAQUE_VALUE_ADAPTER.validate_python(model_group), + ): + response = original_function(*args, **kwargs) + if coroutine_checker.is_async_callable(response) or inspect.isawaitable(response): + response = await response await self.increment_deployment_usage_for_response(response=response, request_kwargs=kwargs) ## PROCESS RESPONSE HEADERS response = await self.set_response_headers(response=response, model_group=model_group, request_kwargs=kwargs) @@ -13892,6 +13907,16 @@ class Router: deployment_model: Final = litellm_params.get("model") return isinstance(deployment_model, str) and classify_strategy_router_model(deployment_model) is not None + def endpoint_error_for_request( + self, model: str, call_type: str, request_kwargs: Mapping[str, object] + ) -> litellm.BadRequestError | None: + try: + with endpoint_selection(call_type, model=model): + self._common_checks_available_deployment(model=model, request_kwargs=request_kwargs) + except litellm.BadRequestError as error: + return error + return None + def _common_checks_available_deployment( self, model: str, @@ -14076,7 +14101,10 @@ class Router: model=model, llm_provider="", ) - return selectable + eligible: Final = eligible_deployments(model, selectable) + if isinstance(eligible, UnsupportedEndpointError): + raise eligible + return list(eligible) def _filter_reserved_deployments( self, diff --git a/litellm/router_utils/endpoint_eligibility.py b/litellm/router_utils/endpoint_eligibility.py new file mode 100644 index 00000000000..112ccaf535a --- /dev/null +++ b/litellm/router_utils/endpoint_eligibility.py @@ -0,0 +1,102 @@ +from collections.abc import Generator, Mapping, Sequence +from contextlib import contextmanager +from contextvars import ContextVar +from types import MappingProxyType +from typing import Final, TypeVar + +from pydantic import BaseModel, ConfigDict, ValidationError + +from litellm.exceptions import BadRequestError + +CHAT_ENDPOINT_BY_CALL: Final[Mapping[str, str]] = MappingProxyType( + { + "completion": "/v1/chat/completions", + "acompletion": "/v1/chat/completions", + "_completion": "/v1/chat/completions", + "_acompletion": "/v1/chat/completions", + "text_completion": "/v1/completions", + "atext_completion": "/v1/completions", + "_atext_completion": "/v1/completions", + "anthropic_messages": "/v1/messages", + "aresponses": "/v1/responses", + "agenerate_content": "generateContent", + "agenerate_content_stream": "streamGenerateContent", + "acreate_interaction": "/v1beta/interactions", + "litellm.interactions.main.acreate": "/v1beta/interactions", + "litellm.interactions.main.create": "/v1beta/interactions", + "_aresponses_websocket": "/v1/responses", + "acompact_responses": "/v1/responses/compact", + } +) + + +def _endpoint_path(endpoint: str) -> str: + return endpoint.removeprefix("/openai").removeprefix("/v1beta").removeprefix("/v1") + + +CHAT_ENDPOINTS: Final = frozenset(_endpoint_path(endpoint) for endpoint in CHAT_ENDPOINT_BY_CALL.values()) +_Deployment: Final = TypeVar("_Deployment", bound=Mapping[str, object]) +_REQUEST_SELECTION: Final[ContextVar[tuple[str | None, str | None]]] = ContextVar( + "router_request_endpoint", default=(None, None) +) + + +class _EndpointDeclaration(BaseModel): + model_config = ConfigDict(frozen=True) + + enforce_supported_endpoints: bool = False + supported_endpoints: tuple[str, ...] | None = None + + +class UnsupportedEndpointError(BadRequestError): + def __init__(self, message: str, model: str, llm_provider: str = "") -> None: + super().__init__( # pyright: ignore[reportUnknownMemberType] # inherited public exception constructor is untyped + message=message, model=model, llm_provider=llm_provider + ) + self.message = message + self.type = "invalid_request_error" + self.param = "model" + self.code = "400" + + +@contextmanager +def endpoint_selection(call_type: object, module: object = None, model: object = None) -> Generator[None]: + qualified_call: Final = f"{module}.{call_type}" if module == "litellm.interactions.main" else call_type + endpoint: Final = CHAT_ENDPOINT_BY_CALL.get(qualified_call) if isinstance(qualified_call, str) else None + token: Final = _REQUEST_SELECTION.set((endpoint, model if isinstance(model, str) else None)) + try: + yield + finally: + _REQUEST_SELECTION.reset(token) + + +def non_chat_endpoints(deployment: Mapping[str, object]) -> frozenset[str] | None: + try: + declaration: Final = _EndpointDeclaration.model_validate(deployment.get("model_info")) + except ValidationError: + return None + if not declaration.enforce_supported_endpoints or declaration.supported_endpoints is None: + return None + endpoints: Final = frozenset(declaration.supported_endpoints) + normalized: Final = frozenset(_endpoint_path(endpoint) for endpoint in endpoints) + return endpoints if normalized.isdisjoint(CHAT_ENDPOINTS) else None + + +def eligible_deployments( + model: str, deployments: Sequence[_Deployment] +) -> tuple[_Deployment, ...] | UnsupportedEndpointError: + endpoint, requested_model = _REQUEST_SELECTION.get() + if endpoint is None or not deployments: + return tuple(deployments) + eligible: Final = tuple(deployment for deployment in deployments if non_chat_endpoints(deployment) is None) + if eligible: + return eligible + requested: Final = requested_model or model + supported: Final = ", ".join( + sorted(frozenset[str]().union(*(non_chat_endpoints(deployment) or frozenset() for deployment in deployments))) + ) + return UnsupportedEndpointError( + message=f"{endpoint} is not supported by model={requested}. Supported endpoints: {supported or 'none'}", + model=model, + llm_provider="", + ) diff --git a/litellm/types/router.py b/litellm/types/router.py index 96add8bb2a1..54bacfc84a3 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -255,6 +255,7 @@ def _whole_float_as_int(value: object) -> object: class ModelInfo(MirroredPricingParams): + enforce_supported_endpoints: bool = False id: str | None # Allow id to be optional on input, but it will always be present as a str in the model instance db_model: bool = False # used for proxy - to separate models which are stored in the db vs. config. updated_at: datetime.datetime | None = None diff --git a/tests/e2e/coverage_registry/llm_nonconversational.yaml b/tests/e2e/coverage_registry/llm_nonconversational.yaml index 330ddf88fc5..a88ccf06c44 100644 --- a/tests/e2e/coverage_registry/llm_nonconversational.yaml +++ b/tests/e2e/coverage_registry/llm_nonconversational.yaml @@ -2,6 +2,7 @@ - {id: llm.completions.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: completions, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_completions_endpoint_e2e.py", rationale: "Legacy text /completions endpoint, second-highest production request volume"} - {id: llm.embeddings.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_embeddings_endpoint_e2e.py:23", rationale: "Core endpoint, live vector response"} - {id: llm.embeddings.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy ยง9.3 / LIT-4778", rationale: "Missing model/input on /embeddings return client errors"} +- {id: llm.embeddings.openai.supported_endpoints_gate.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], fail_before_fix: proven, source: "test_embedding_only_supported_endpoints_e2e.py", rationale: "Explicit endpoint restrictions preserve real embeddings and route chat through a live fallback; rejection and zero-provider-call contracts live in integration/routing/test_non_chat_model_endpoint_gate.py"} - {id: llm.embeddings.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "SPEND_TRACKING_COVERAGE_MATRIX.md:34", rationale: "Cost tracking on embeddings"} - {id: llm.embeddings.openai.passthrough.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "test_passthrough_e2e.py", rationale: "POST /openai_passthrough/v1/embeddings is costed; the route wrote no spend row at all, so budgets never saw traffic OpenAI was billing for (GitHub issue #36646)"} - {id: llm.embeddings.azure_openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/azure/azure.py", rationale: "Azure embeddings via translation"} diff --git a/tests/e2e/llm_translation/test_embedding_only_supported_endpoints_e2e.py b/tests/e2e/llm_translation/test_embedding_only_supported_endpoints_e2e.py new file mode 100644 index 00000000000..899b421b218 --- /dev/null +++ b/tests/e2e/llm_translation/test_embedding_only_supported_endpoints_e2e.py @@ -0,0 +1,104 @@ +"""Live provider checks for embeddings and chat fallback from an explicitly restricted model.""" + +from __future__ import annotations + +import dataclasses +from typing import Final + +import pytest +from e2e_config import unique_marker +from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta +from lifecycle import ResourceManager +from models import LiteLLMParamsBody, ModelInfoBody, ModelMode, ModelNewBody +from openai import OpenAI +from proxy_client import ProxyClient +from sdk_clients import NO_PROXY_CACHE, SdkClients + +pytestmark = pytest.mark.e2e + +EMBEDDING_BACKEND: Final = "openai/text-embedding-3-small" +CHAT_FALLBACK_BACKEND: Final = "openai/gpt-5.4-nano" +EMBEDDINGS_ONLY: Final = ("/v1/embeddings",) +PROMPT: Final = "Say this is a test!" + + +def _subject(route: Route) -> Subject: + return Subject( + domain=Domain.LLM_TRANSLATION, + route=route, + providers=(Provider.OPENAI,), + models=(EMBEDDING_BACKEND,), + mode=Mode.NONSTREAM, + ) + + +def _register( + proxy: ProxyClient, + resources: ResourceManager, + prefix: str, + supported_endpoints: tuple[str, ...] | None, + backend: str = EMBEDDING_BACKEND, + mode: ModelMode = "embedding", +) -> str: + model: Final = f"{prefix}-{unique_marker()}" + model_id: Final = proxy.register_model( + ModelNewBody( + model_name=model, + litellm_params=LiteLLMParamsBody(model=backend, api_key="os.environ/OPENAI_API_KEY"), + model_info=ModelInfoBody( + enforce_supported_endpoints=supported_endpoints is not None, + mode=mode, + supported_endpoints=None if supported_endpoints is None else list(supported_endpoints), + ), + ) + ) + resources.defer(lambda: proxy.delete_model(model_id)) + return model + + +def _openai_at(sdk: SdkClients, api_prefix: str, key: str) -> OpenAI: + return dataclasses.replace(sdk, base_url=f"{sdk.base_url}{api_prefix}").openai(key) + + +class TestEmbeddingOnlySupportedEndpoints: + @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works") + @meta(_subject(Route.EMBEDDINGS)) + @pytest.mark.parametrize( + ("prefix", "supported_endpoints"), + ( + pytest.param("e2e-embed-only", EMBEDDINGS_ONLY, id="embed-only"), + pytest.param("e2e-embed-control", None, id="control"), + ), + ) + def test_embeddings_return_a_vector( + self, + proxy: ProxyClient, + resources: ResourceManager, + sdk: SdkClients, + prefix: str, + supported_endpoints: tuple[str, ...] | None, + ) -> None: + model: Final = _register(proxy, resources, prefix, supported_endpoints) + embeddings: Final = sdk.openai(resources.key()).embeddings.create( + model=model, input=PROMPT, extra_body=NO_PROXY_CACHE + ) + assert embeddings.data, f"/embeddings returned no data for {model}: {embeddings!r}" + assert any(component != 0.0 for component in embeddings.data[0].embedding), ( + f"/embeddings returned an empty or all-zero vector for {model}" + ) + + @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works") + @meta(dataclasses.replace(_subject(Route.CHAT_COMPLETIONS), models=(EMBEDDING_BACKEND, CHAT_FALLBACK_BACKEND))) + def test_chat_completions_with_a_fallback_are_answered_by_the_fallback( + self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients + ) -> None: + model: Final = _register(proxy, resources, "e2e-embed-only", EMBEDDINGS_ONLY) + fallback: Final = _register(proxy, resources, "e2e-chat-fallback", None, CHAT_FALLBACK_BACKEND, "chat") + completion: Final = _openai_at(sdk, "/v1", resources.key()).chat.completions.create( + model=model, + messages=[{"role": "user", "content": PROMPT}], + extra_body={**NO_PROXY_CACHE, "fallbacks": [fallback]}, + ) + assert completion.choices and completion.choices[0].message.content, ( + f"the fallback {fallback} should answer a chat call to {model}, got {completion!r}" + ) diff --git a/tests/e2e/models.py b/tests/e2e/models.py index d7a10c6da1f..700afa4641c 100644 --- a/tests/e2e/models.py +++ b/tests/e2e/models.py @@ -1423,10 +1423,11 @@ class LiteLLMParamsBody(BaseModel): order: int | None = None -ModelMode = Literal["chat", "batch", "realtime", "image_generation"] +ModelMode = Literal["chat", "embedding", "batch", "realtime", "image_generation"] class ModelInfoBody(BaseModel): + enforce_supported_endpoints: bool | None = None # id is left unset so the proxy assigns a unique model_id per deployment. # Pinning it to the model_name made re-registrations of a fixed-name model # (e.g. the batch suite's openai-batch) collide on the model_id unique diff --git a/tests/integration/routing/test_non_chat_model_endpoint_gate.py b/tests/integration/routing/test_non_chat_model_endpoint_gate.py new file mode 100644 index 00000000000..346cc60bc26 --- /dev/null +++ b/tests/integration/routing/test_non_chat_model_endpoint_gate.py @@ -0,0 +1,1479 @@ +import asyncio +import contextlib +import json +import uuid +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import httpx +import openai +import pytest +import websockets +import yaml +from integration._support.client import ( + JSON_OBJECT, + Gateway, + Scenario, + eventually, + gateway_from_environment, + held, + list_value, + object_value, + string_value, +) +from integration._support import responses_ws_peer +from integration._support.database import read_rows +from integration._support.openai_wire import answering_model_discovery, chat_reply, responses_reply +from integration._support.process import graceful_stop_seconds, owned_proxy_process +from integration._support.tls import write_self_signed_cert +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue, TypeAdapter +from websockets.exceptions import ConnectionClosed + +_PROVIDER_KEY: Final = "integration-provider-key" +_EMBEDDING_MODEL: Final = "text-embedding-3-small" +_CHAT_MODEL: Final = "gpt-4o-mini" +_CHAT_BACKEND_PREFIX: Final = "gate-chat-backend-" +_SERVED_TEXT: Final = "served by the chat upstream" +_VECTOR: Final = (0.25, 0.5, 0.75) +_COUNTED_TOKENS: Final = 17 +_CLOSE: Final = MappingProxyType({"connection": "close"}) +_RELOAD_SECONDS: Final = 5 +_EMBED_ONLY: Final[Mapping[str, JsonValue]] = MappingProxyType( + {"mode": "embedding", "supported_endpoints": ["/v1/embeddings"], "enforce_supported_endpoints": True} +) + + +def _embedding_reply() -> Reply: + return Reply( + body=json.dumps( + { + "object": "list", + "data": [{"object": "embedding", "index": 0, "embedding": list(_VECTOR)}], + "model": _EMBEDDING_MODEL, + "usage": {"prompt_tokens": 3, "total_tokens": 3}, + } + ).encode() + ) + + +def _not_a_chat_model() -> Reply: + return Reply( + status=404, + body=json.dumps( + {"error": {"message": "scripted: not a chat model", "type": "invalid_request_error", "code": None}} + ).encode(), + ) + + +def _text_completion_reply() -> Reply: + return Reply( + body=json.dumps( + { + "id": f"cmpl-{uuid.uuid4().hex}", + "object": "text_completion", + "created": 1, + "model": _CHAT_MODEL, + "choices": [{"index": 0, "text": _SERVED_TEXT, "finish_reason": "stop", "logprobs": None}], + "usage": {"prompt_tokens": 3, "completion_tokens": 3, "total_tokens": 6}, + } + ).encode() + ) + + +def _serves_chat(model: JsonValue) -> bool: + return model == _CHAT_MODEL or (isinstance(model, str) and model.startswith(_CHAT_BACKEND_PREFIX)) + + +def _respond(request: Request) -> Reply: + path: Final = request.target.split("?")[0] + body: Final = JSON_OBJECT.validate_json(request.body or b"{}") + streamed: Final = body.get("stream") is True + if path == "/v1/embeddings": + return _embedding_reply() + if path == "/v1/responses/input_tokens": + return Reply(body=json.dumps({"object": "response.input_tokens", "input_tokens": _COUNTED_TOKENS}).encode()) + if not _serves_chat(body.get("model")): + return _not_a_chat_model() + if path == "/v1/chat/completions": + return chat_reply(f"chatcmpl-{uuid.uuid4().hex}", _CHAT_MODEL, _SERVED_TEXT, stream=streamed) + if path in ("/v1/responses", "/v1/responses/compact"): + return responses_reply(f"resp_{uuid.uuid4().hex}", _CHAT_MODEL, _SERVED_TEXT, stream=streamed) + if path == "/v1/completions": + return _text_completion_reply() + return _not_a_chat_model() + + +class _Upstream: + def __init__(self, wire: Wire) -> None: + self.wire = wire + self.seen: tuple[Request, ...] = () + + def calls(self, marker: str) -> tuple[Request, ...]: + self.seen = (*self.seen, *self.wire.drain()) + return tuple(request for request in self.seen if marker.encode() in request.body or marker in request.target) + + def targets(self, marker: str) -> tuple[str, ...]: + return tuple(request.target for request in self.calls(marker)) + + +@dataclass(frozen=True, slots=True) +class _Deployment: + litellm_model: str + declarations: tuple[Mapping[str, JsonValue], ...] + + +def _single(declaration: Mapping[str, JsonValue], litellm_model: str = f"openai/{_CHAT_MODEL}") -> _Deployment: + return _Deployment(litellm_model, (declaration,)) + + +_SHARED: Final[Mapping[str, _Deployment]] = MappingProxyType( + { + "embedding_only": _single(_EMBED_ONLY, f"openai/{_EMBEDDING_MODEL}"), + "embedding_only_second": _single(_EMBED_ONLY, f"openai/{_EMBEDDING_MODEL}"), + "image_generation": _single( + {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/images/generations"]} + ), + "chat_target": _single({"mode": "chat"}), + "chat_rate_limited": _single({"mode": "chat"}), + "mode_not_a_string": _single( + {"mode": 7, "supported_endpoints": ["/v1/embeddings"]}, "openai/gate-mode-not-a-string" + ), + } +) + + +@dataclass(frozen=True, slots=True) +class _Rig: + gateway: Gateway + upstream: _Upstream + names: Mapping[str, str] + api_base: str + + +def _create(gateway: Gateway, name: str, litellm_params: Mapping[str, JsonValue], info: Mapping[str, JsonValue]) -> str: + created: Final = gateway.post( + "/model/new", + { + "model_name": name, + "litellm_params": {"api_key": _PROVIDER_KEY, **litellm_params}, + "model_info": dict(info), + }, + ) + return string_value(object_value(created["model_info"])["id"]) + + +def _deploy(scenario: Scenario, api_base: str, label: str, deployment: _Deployment) -> tuple[str, tuple[str, ...]]: + name: Final = f"gate-{label}-{uuid.uuid4().hex[:12]}" + identities: Final = tuple( + _create(scenario.gateway, name, {"model": deployment.litellm_model, "api_base": api_base}, declaration) + for declaration in deployment.declarations + ) + tuple(scenario.cleanups.callback(scenario.delete_model, identity) for identity in identities) + return name, identities + + +def _listed_ids(gateway: Gateway) -> frozenset[str]: + listed: Final = gateway.request("GET", "/model/info", headers=_CLOSE) + assert listed.status_code == 200, listed.text + return frozenset( + string_value(object_value(object_value(entry)["model_info"])["id"]) + for entry in list_value(JSON_OBJECT.validate_json(listed.content)["data"]) + ) + + +def _served_by_every_worker(gateway: Gateway, identities: frozenset[str]) -> None: + held( + lambda: identities <= _listed_ids(gateway), + lambda served: served, + holding=_RELOAD_SECONDS * 2, + seconds=_RELOAD_SECONDS * 12, + ) + + +@pytest.fixture(scope="module") +def rig() -> Iterator[_Rig]: + with ( + wire_server(answering_model_discovery(_respond)) as wire, + gateway_from_environment() as gateway, + gateway.scenario() as scenario, + ): + api_base: Final = wire.url + "/v1" + deployed: Final = {label: _deploy(scenario, api_base, label, spec) for label, spec in _SHARED.items()} + _served_by_every_worker( + gateway, frozenset(identity for _name, identities in deployed.values() for identity in identities) + ) + yield _Rig( + gateway, + _Upstream(wire), + MappingProxyType({label: name for label, (name, _ids) in deployed.items()}), + api_base, + ) + + +def _marker() -> str: + return f"gate-{uuid.uuid4().hex}" + + +def _openai_error(endpoint: str, model: str, supported: str) -> dict[str, JsonValue]: + return { + "error": { + "message": f"{endpoint} is not supported by model={model}. Supported endpoints: {supported}", + "type": "invalid_request_error", + "param": "model", + "code": "400", + } + } + + +def _anthropic_error(model: str, supported: str) -> dict[str, JsonValue]: + return { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": f"/v1/messages is not supported by model={model}. Supported endpoints: {supported}", + }, + } + + +def _chat_body(model: str, marker: str, *, stream: bool = False) -> dict[str, JsonValue]: + return {"model": model, "messages": [{"role": "user", "content": marker}], **({"stream": True} if stream else {})} + + +def _failure_row(call_id: str) -> dict[str, JsonValue]: + rows: Final = eventually( + lambda: read_rows('SELECT request_id, status FROM "LiteLLM_SpendLogs" WHERE request_id = %s', (call_id,)), + lambda found: len(found) >= 1, + seconds=70, + ) + assert len(rows) == 1, rows + return rows[0] + + +def _assert_error(response: httpx.Response, expected: dict[str, JsonValue]) -> None: + assert response.status_code == 400, response.text + body: Final = JSON_OBJECT.validate_json(response.content) + error: Final = object_value(body["error"]) + message: Final = string_value(error["message"]).split("\n\n", 1)[0] + assert {**body, "error": {**error, "message": message}} == expected, response.text + + +def _assert_rejected_before( + upstream: _Upstream, response: httpx.Response, marker: str, expected: dict[str, JsonValue] +) -> None: + _assert_error(response, expected) + assert upstream.calls(marker) == (), upstream.targets(marker) + assert _failure_row(response.headers["x-litellm-call-id"])["status"] == "failure" + + +def _assert_rejected(rig: _Rig, response: httpx.Response, marker: str, expected: dict[str, JsonValue]) -> None: + _assert_rejected_before(rig.upstream, response, marker, expected) + + +@dataclass(frozen=True, slots=True) +class _Route: + path: str + endpoint: str + anthropic_shape: bool + stream: bool + model_in_path: bool + shape: str + + +_ROUTES: Final[Mapping[str, _Route]] = MappingProxyType( + { + "interactions": _Route("/v1beta/interactions", "/v1beta/interactions", False, False, False, "interactions"), + "chat_v1": _Route("/v1/chat/completions", "/v1/chat/completions", False, False, False, "chat"), + "chat_v1_stream": _Route("/v1/chat/completions", "/v1/chat/completions", False, True, False, "chat"), + "chat_bare": _Route("/chat/completions", "/v1/chat/completions", False, False, False, "chat"), + "chat_engines": _Route("/engines/{model}/chat/completions", "/v1/chat/completions", False, False, True, "chat"), + "chat_azure_deployments": _Route( + "/openai/deployments/{model}/chat/completions", "/v1/chat/completions", False, False, True, "chat" + ), + "messages": _Route("/v1/messages", "/v1/messages", True, False, False, "messages"), + "messages_stream": _Route("/v1/messages", "/v1/messages", True, True, False, "messages"), + "responses_v1": _Route("/v1/responses", "/v1/responses", False, False, False, "responses"), + "responses_v1_stream": _Route("/v1/responses", "/v1/responses", False, True, False, "responses"), + "responses_bare": _Route("/responses", "/v1/responses", False, False, False, "responses"), + "cursor_responses_body": _Route("/cursor/chat/completions", "/v1/responses", False, False, False, "cursor"), + "cursor_chat_body": _Route("/cursor/chat/completions", "/v1/chat/completions", False, False, False, "chat"), + } +) + + +def _route_body(route: _Route, model: str, marker: str) -> dict[str, JsonValue]: + shaped: Final[Mapping[str, dict[str, JsonValue]]] = { + "chat": {"messages": [{"role": "user", "content": marker}]}, + "messages": {"max_tokens": 16, "messages": [{"role": "user", "content": marker}]}, + "responses": {"input": marker}, + "interactions": {"input": marker}, + "cursor": {"input": [{"role": "user", "content": marker}]}, + } + return { + **({} if route.model_in_path else {"model": model}), + **shaped[route.shape], + **({"stream": True} if route.stream else {}), + } + + +def _send(gateway: Gateway, route: _Route, model: str, marker: str) -> httpx.Response: + path: Final = route.path.format(model=model) + body: Final = _route_body(route, model, marker) + if not route.stream: + return gateway.request("POST", path, body) + with gateway.client.stream("POST", path, json=body, headers={"Authorization": f"Bearer {gateway.key}"}) as streamed: + streamed.read() + return streamed + + +@pytest.mark.parametrize("route", tuple(_ROUTES)) +def test_chat_shaped_route_rejects_an_embedding_only_model_before_the_upstream(rig: _Rig, route: str) -> None: + spec: Final = _ROUTES[route] + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = _send(rig.gateway, spec, model, marker) + expected: Final = ( + _anthropic_error(model, "/v1/embeddings") + if spec.anthropic_shape + else _openai_error(spec.endpoint, model, "/v1/embeddings") + ) + _assert_rejected(rig, response, marker, expected) + + +def _sdk_embedding(rig: _Rig, model: str, marker: str) -> tuple[tuple[float, ...], str]: + async def call() -> tuple[tuple[float, ...], str]: + async with openai.AsyncOpenAI( + base_url=f"{rig.gateway.client.base_url}/v1", api_key=rig.gateway.key, max_retries=0 + ) as sdk: + raw: Final = await sdk.embeddings.with_raw_response.create( + model=model, input=marker, encoding_format="float" + ) + parsed: Final = raw.parse() + return tuple(parsed.data[0].embedding), raw.headers["x-litellm-call-id"] + + return asyncio.run(call()) + + +def _httpx_embedding(rig: _Rig, model: str, marker: str) -> tuple[tuple[float, ...], str]: + response: Final = rig.gateway.request( + "POST", "/v1/embeddings", {"model": model, "input": marker, "encoding_format": "float"} + ) + assert response.status_code == 200, response.text + data: Final = list_value(JSON_OBJECT.validate_json(response.content)["data"]) + vector: Final = list_value(object_value(data[0])["embedding"]) + return tuple(float(str(value)) for value in vector), response.headers["x-litellm-call-id"] + + +@pytest.mark.parametrize("client", ("httpx_sync", "openai_async_sdk")) +def test_embeddings_still_reach_the_embedding_only_model(rig: _Rig, client: str) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + vector, call_id = ( + _httpx_embedding(rig, model, marker) if client == "httpx_sync" else _sdk_embedding(rig, model, marker) + ) + assert vector == _VECTOR + assert rig.upstream.targets(marker) == ("/v1/embeddings",) + row: Final = eventually( + lambda: read_rows( + 'SELECT status, call_type FROM "LiteLLM_SpendLogs" WHERE request_id = %s OR litellm_call_id = %s', + (call_id, call_id), + ), + lambda found: len(found) >= 1, + seconds=70, + ) + assert [(entry["status"], entry["call_type"]) for entry in row] == [("success", "aembedding")], row + + +def _assert_served(rig: _Rig, response: httpx.Response, marker: str, target: str) -> None: + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert rig.upstream.targets(marker) == (target,) + + +def test_key_alias_to_an_embedding_only_model_is_rejected_under_the_resolved_name(rig: _Rig) -> None: + target: Final = rig.names["embedding_only"] + alias: Final = f"gate-alias-{uuid.uuid4().hex[:12]}" + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(aliases={alias: target}) + response: Final = rig.gateway.request("POST", "/v1/chat/completions", _chat_body(alias, marker), key=key) + _assert_rejected(rig, response, marker, _openai_error("/v1/chat/completions", target, "/v1/embeddings")) + + +async def _websocket_first_frame_and_close( + gateway: Gateway, key: str, model: str, marker: str +) -> tuple[dict[str, JsonValue], int | None, str | None]: + url: Final = str(gateway.client.base_url).rstrip("/").replace("http://", "ws://", 1) + "/v1/responses" + async with websockets.connect(url, additional_headers={"Authorization": f"Bearer {key}"}, open_timeout=10) as ws: + await ws.send(json.dumps({"type": "response.create", "model": model, "input": marker})) + first: Final = JSON_OBJECT.validate_json(await asyncio.wait_for(ws.recv(), 30)) + with contextlib.suppress(ConnectionClosed): + await asyncio.wait_for(ws.recv(), 30) + return first, ws.close_code, ws.close_reason + + +def _websocket_rejection(model: str) -> tuple[dict[str, JsonValue], int | None, str | None]: + return ( + { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": f"/v1/responses is not supported by model={model}. Supported endpoints: /v1/embeddings", + }, + }, + 1008, + "Pre-call error", + ) + + +@pytest.mark.parametrize("aliased", [False, True]) +def test_responses_websocket_closes_on_an_embedding_only_model_before_the_upstream(rig: _Rig, aliased: bool) -> None: + target: Final = rig.names["embedding_only"] + alias: Final = f"gate-ws-alias-{uuid.uuid4().hex[:12]}" + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(aliases={alias: target}) + closed: Final = asyncio.run( + _websocket_first_frame_and_close(rig.gateway, key, alias if aliased else target, marker) + ) + assert closed == _websocket_rejection(target) + assert rig.upstream.calls(marker) == () + + +_RESERVING_KEYS: Final[Mapping[str, Mapping[str, JsonValue]]] = MappingProxyType( + { + "max_parallel_requests": {"max_parallel_requests": 1}, + "tpm_limit": {"tpm_limit": 100, "default_estimated_output_tokens": 80}, + } +) + + +@pytest.mark.parametrize("limit", tuple(_RESERVING_KEYS)) +def test_responses_websocket_rejections_leave_the_key_reservation_free(rig: _Rig, limit: str) -> None: + target: Final = rig.names["embedding_only"] + markers: Final = tuple(_marker() for _ in range(3)) + served_marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(**_RESERVING_KEYS[limit]) + closed: Final = tuple( + asyncio.run(_websocket_first_frame_and_close(rig.gateway, key, target, marker)) for marker in markers + ) + assert closed == (_websocket_rejection(target),) * 3 + served: Final = rig.gateway.request( + "POST", "/v1/chat/completions", _chat_body(rig.names["chat_target"], served_marker), key=key + ) + _assert_served(rig, served, served_marker, "/v1/chat/completions") + assert [rig.upstream.calls(marker) for marker in markers] == [()] * 3 + + +def test_team_scoped_embedding_only_model_is_rejected_under_its_public_name(rig: _Rig) -> None: + public: Final = f"gate-team-{uuid.uuid4().hex[:12]}" + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + team: Final = scenario.team() + identity: Final = _create( + rig.gateway, + public, + {"model": f"openai/{_EMBEDDING_MODEL}", "api_base": rig.api_base}, + {**_EMBED_ONLY, "team_id": team}, + ) + scenario.cleanups.callback(scenario.delete_model, identity) + key: Final = scenario.key(team_id=team) + _served_by_every_worker(rig.gateway, frozenset({identity})) + embedded: Final = rig.gateway.request( + "POST", "/v1/embeddings", {"model": public, "input": marker, "encoding_format": "float"}, key=key + ) + assert embedded.status_code == 200, embedded.text + assert rig.upstream.targets(marker) == ("/v1/embeddings",) + rejected_marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", _chat_body(public, rejected_marker), key=key + ) + _assert_rejected( + rig, response, rejected_marker, _openai_error("/v1/chat/completions", public, "/v1/embeddings") + ) + + +def test_wildcard_embedding_only_deployment_rejects_chat_for_any_matched_model(rig: _Rig) -> None: + prefix: Final = f"gate-wild-{uuid.uuid4().hex[:12]}" + requested: Final = f"{prefix}/{_EMBEDDING_MODEL}" + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + identity: Final = _create( + rig.gateway, f"{prefix}/*", {"model": "openai/*", "api_base": rig.api_base}, _EMBED_ONLY + ) + scenario.cleanups.callback(scenario.delete_model, identity) + _served_by_every_worker(rig.gateway, frozenset({identity})) + response: Final = rig.gateway.request("POST", "/v1/chat/completions", _chat_body(requested, marker)) + _assert_rejected(rig, response, marker, _openai_error("/v1/chat/completions", requested, "/v1/embeddings")) + + +def test_key_without_access_gets_the_access_error_and_learns_nothing_about_endpoints(rig: _Rig) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(models=[rig.names["chat_target"]]) + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", _chat_body(rig.names["embedding_only"], marker), key=key + ) + assert response.status_code == 403, response.text + assert object_value(JSON_OBJECT.validate_json(response.content)["error"])["type"] == "key_model_access_denied", ( + response.text + ) + assert "Supported endpoints" not in response.text, response.text + assert rig.upstream.calls(marker) == () + + +def test_unknown_model_keeps_the_invalid_model_error(rig: _Rig) -> None: + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", _chat_body(f"gate-missing-{uuid.uuid4().hex}", marker) + ) + assert response.status_code == 400, response.text + assert "Invalid model name" in response.text and "Supported endpoints" not in response.text, response.text + assert rig.upstream.calls(marker) == () + + +def test_request_fallbacks_answer_for_an_embedding_only_primary(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", {**_chat_body(model, marker), "fallbacks": [rig.names["chat_target"]]} + ) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert rig.upstream.targets(marker) == ("/v1/chat/completions",) + + +def test_an_empty_request_fallback_list_keeps_the_rejection(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", {**_chat_body(model, marker), "fallbacks": []} + ) + _assert_rejected(rig, response, marker, _openai_error("/v1/chat/completions", model, "/v1/embeddings")) + + +def test_fallbacks_that_cannot_answer_chat_keep_the_rejection(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + fallbacks: Final = [rig.names["image_generation"], model] + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", {**_chat_body(model, marker), "fallbacks": fallbacks} + ) + _assert_rejected(rig, response, marker, _openai_error("/v1/chat/completions", model, "/v1/embeddings")) + + +def test_request_fallbacks_do_not_rescue_an_unknown_primary(rig: _Rig) -> None: + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {**_chat_body(f"gate-missing-{uuid.uuid4().hex}", marker), "fallbacks": [rig.names["chat_target"]]}, + ) + assert response.status_code == 400, response.text + assert "Invalid model name" in response.text, response.text + assert rig.upstream.calls(marker) == () + + +def _chat(rig: _Rig, model: str, marker: str, *, key: str | None = None, **extra: JsonValue) -> httpx.Response: + return rig.gateway.request("POST", "/v1/chat/completions", {**_chat_body(model, marker), **extra}, key=key) + + +def _assert_rejected_chat(rig: _Rig, response: httpx.Response, marker: str, model: str) -> None: + _assert_rejected(rig, response, marker, _openai_error("/v1/chat/completions", model, "/v1/embeddings")) + + +def test_a_fallback_chain_that_reaches_chat_two_hops_away_answers(rig: _Rig) -> None: + first, second, chat = rig.names["embedding_only"], rig.names["embedding_only_second"], rig.names["chat_target"] + marker: Final = _marker() + response: Final = _chat(rig, first, marker, fallbacks=[{first: [second]}, {second: [chat]}]) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert rig.upstream.targets(marker) == ("/v1/chat/completions",) + + +def test_a_fallback_chain_that_loops_between_embedding_only_groups_keeps_the_rejection(rig: _Rig) -> None: + first, second = rig.names["embedding_only"], rig.names["embedding_only_second"] + marker: Final = _marker() + response: Final = _chat(rig, first, marker, fallbacks=[{first: [second]}, {second: [first]}]) + _assert_rejected_chat(rig, response, marker, first) + + +def test_a_specific_chain_that_loops_beats_a_catch_all_that_could_answer(rig: _Rig) -> None: + first, second = rig.names["embedding_only"], rig.names["embedding_only_second"] + marker: Final = _marker() + response: Final = _chat( + rig, first, marker, fallbacks=[{"*": [rig.names["chat_target"]]}, {first: [second]}, {second: [first]}] + ) + _assert_rejected_chat(rig, response, marker, first) + + +def test_disable_fallbacks_keeps_the_rejection_over_request_fallbacks(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = _chat(rig, model, marker, fallbacks=[rig.names["chat_target"]], disable_fallbacks=True) + _assert_rejected_chat(rig, response, marker, model) + + +def _assert_fallback_answered( + upstream: _Upstream, response: httpx.Response, marker: str, target: str = "/v1/chat/completions" +) -> None: + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert upstream.targets(marker) == (target,), response.text + + +def _through_a_group_with_no_deployments(rig: _Rig) -> Mapping[str, JsonValue]: + ghost: Final = f"gate-ghost-{uuid.uuid4().hex}" + return {"fallbacks": [{rig.names["embedding_only"]: [ghost]}, {ghost: [rig.names["chat_target"]]}]} + + +def test_request_fallbacks_naming_a_chat_backend_name_answer_for_an_embedding_only_primary(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + backend: Final = f"openai/{_CHAT_BACKEND_PREFIX}{uuid.uuid4().hex[:12]}" + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + identity: Final = _create( + rig.gateway, f"gate-backend-{uuid.uuid4().hex[:12]}", {"model": backend, "api_base": rig.api_base}, {} + ) + scenario.cleanups.callback(scenario.delete_model, identity) + _served_by_every_worker(rig.gateway, frozenset({identity})) + response: Final = _chat(rig, model, marker, fallbacks=[{model: [backend]}]) + _assert_fallback_answered(rig.upstream, response, marker) + + +@pytest.mark.parametrize("level", ("key", "team")) +def test_router_settings_fallbacks_through_a_group_with_no_deployments_answer(rig: _Rig, level: str) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + settings: Final = {"router_settings": dict(_through_a_group_with_no_deployments(rig))} + key: Final = scenario.key(**settings) if level == "key" else scenario.key(team_id=scenario.team(**settings)) + response: Final = _chat(rig, rig.names["embedding_only"], marker, key=key) + _assert_fallback_answered(rig.upstream, response, marker) + + +def test_team_router_settings_fallbacks_answer_for_an_embedding_only_primary(rig: _Rig) -> None: + model, chat = rig.names["embedding_only"], rig.names["chat_target"] + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + team: Final = scenario.team(router_settings={"fallbacks": [{model: [chat]}]}) + response: Final = _chat(rig, model, marker, key=scenario.key(team_id=team)) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert rig.upstream.targets(marker) == ("/v1/chat/completions",) + + +def test_key_router_settings_fallbacks_answer_for_an_embedding_only_primary(rig: _Rig) -> None: + model, chat = rig.names["embedding_only"], rig.names["chat_target"] + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(router_settings={"fallbacks": [{model: [chat]}]}) + response: Final = _chat(rig, model, marker, key=key) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert rig.upstream.targets(marker) == ("/v1/chat/completions",) + + +def test_key_router_settings_that_cannot_answer_override_team_fallbacks_that_could(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + team: Final = scenario.team(router_settings={"fallbacks": [{model: [rig.names["chat_target"]]}]}) + key: Final = scenario.key( + team_id=team, router_settings={"fallbacks": [{model: [rig.names["embedding_only_second"]]}]} + ) + response: Final = _chat(rig, model, marker, key=key) + _assert_rejected_chat(rig, response, marker, model) + + +def test_an_empty_request_fallback_list_overrides_key_router_settings_fallbacks(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(router_settings={"fallbacks": [{model: [rig.names["chat_target"]]}]}) + response: Final = _chat(rig, model, marker, key=key, fallbacks=[]) + _assert_rejected_chat(rig, response, marker, model) + + +def test_local_rate_limit_fallback_skips_an_embedding_only_hop(rig: _Rig) -> None: + limited, chat, model = rig.names["chat_rate_limited"], rig.names["chat_target"], rig.names["embedding_only"] + first, second = _marker(), _marker() + with rig.gateway.scenario() as scenario: + key: Final = scenario.key( + model_rpm_limit={limited: 1}, router_settings={"fallbacks": [{limited: [model, chat]}]} + ) + served_first: Final = _chat(rig, limited, first, key=key) + served_second: Final = _chat(rig, limited, second, key=key) + assert (served_first.status_code, served_first.headers.get("x-litellm-model-group")) == (200, limited), ( + served_first.text + ) + assert (served_second.status_code, served_second.headers.get("x-litellm-model-group")) == (200, chat), ( + served_second.text + ) + assert (rig.upstream.targets(first), rig.upstream.targets(second)) == ( + ("/v1/chat/completions",), + ("/v1/chat/completions",), + ) + + +_COMPACT_PATHS: Final = ("/v1/responses/compact", "/responses/compact", "/openai/v1/responses/compact") + + +def _compact_body(model: str, marker: str) -> dict[str, JsonValue]: + return {"model": model, "input": [{"role": "user", "content": marker}]} + + +@pytest.mark.parametrize("path", _COMPACT_PATHS) +def test_responses_compact_rejects_an_embedding_only_model_before_the_upstream(rig: _Rig, path: str) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = rig.gateway.request("POST", path, _compact_body(model, marker)) + _assert_rejected(rig, response, marker, _openai_error("/v1/responses/compact", model, "/v1/embeddings")) + + +def test_responses_compact_keeps_reaching_the_upstream_for_a_chat_model(rig: _Rig) -> None: + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", "/v1/responses/compact", _compact_body(rig.names["chat_target"], marker) + ) + _assert_served(rig, response, marker, "/v1/responses/compact") + + +def test_count_tokens_keeps_counting_for_an_embedding_only_model(rig: _Rig) -> None: + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", + "/v1/messages/count_tokens", + {"model": rig.names["embedding_only"], "messages": [{"role": "user", "content": marker}]}, + ) + assert response.status_code == 200, response.text + assert JSON_OBJECT.validate_json(response.content)["input_tokens"] == _COUNTED_TOKENS, response.text + assert rig.upstream.targets(marker) == ("/v1/responses/input_tokens",) + + +def test_text_completions_reject_an_embedding_only_model_before_the_upstream(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = rig.gateway.request("POST", "/v1/completions", {"model": model, "prompt": marker}) + _assert_rejected(rig, response, marker, _openai_error("/v1/completions", model, "/v1/embeddings")) + + +@pytest.mark.parametrize("method", ("generateContent", "streamGenerateContent")) +def test_gemini_routes_reject_an_embedding_only_model_before_the_upstream(rig: _Rig, method: str) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + response: Final = rig.gateway.request( + "POST", f"/v1beta/models/{model}:{method}", {"contents": [{"role": "user", "parts": [{"text": marker}]}]} + ) + assert response.status_code == 400, response.text + assert f"{method} is not supported by model={model}. Supported endpoints: /v1/embeddings" in response.text + assert rig.upstream.calls(marker) == (), rig.upstream.targets(marker) + + +def test_repeated_rejections_log_one_failure_row_per_call(rig: _Rig) -> None: + model: Final = rig.names["embedding_only"] + marker: Final = _marker() + responses: Final = tuple( + rig.gateway.request("POST", "/v1/chat/completions", _chat_body(model, marker)) for _ in range(3) + ) + expected: Final = _openai_error("/v1/chat/completions", model, "/v1/embeddings") + for response in responses: + _assert_error(response, expected) + call_ids: Final = tuple(item.headers["x-litellm-call-id"] for item in responses) + assert len(set(call_ids)) == 3, call_ids + assert [_failure_row(call_id)["status"] for call_id in call_ids] == ["failure"] * 3 + assert rig.upstream.calls(marker) == () + + +def _closed_chat(rig: _Rig, model: str, marker: str) -> httpx.Response: + return rig.gateway.request("POST", "/v1/chat/completions", _chat_body(model, marker), headers=_CLOSE) + + +def test_live_declaration_update_turns_the_gate_on_and_off_on_every_worker(rig: _Rig) -> None: + chat_capable: Final[dict[str, JsonValue]] = { + "mode": "chat", + "supported_endpoints": ["/v1/chat/completions", "/v1/embeddings"], + } + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + name, identities = _deploy(scenario, rig.api_base, "flip", _single(chat_capable)) + _served_by_every_worker(rig.gateway, frozenset(identities)) + served: Final = _closed_chat(rig, name, marker) + _assert_served(rig, served, marker, "/v1/chat/completions") + patched: Final = rig.gateway.request( + "PATCH", f"/model/{identities[0]}/update", {"model_info": dict(_EMBED_ONLY)} + ) + assert patched.status_code == 200, patched.text + rejected: Final = held( + lambda: _closed_chat(rig, name, _marker()), + lambda response: response.status_code == 400, + holding=_RELOAD_SECONDS * 2, + seconds=_RELOAD_SECONDS * 12, + ) + _assert_error(rejected, _openai_error("/v1/chat/completions", name, "/v1/embeddings")) + restored: Final = rig.gateway.request("PATCH", f"/model/{identities[0]}/update", {"model_info": chat_capable}) + assert restored.status_code == 200, restored.text + held( + lambda: _closed_chat(rig, name, _marker()), + lambda response: response.status_code == 200, + holding=_RELOAD_SECONDS * 2, + seconds=_RELOAD_SECONDS * 12, + ) + + +_TESTS_ROOT: Final = Path(__file__).resolve().parents[2] +_CONFIG: Final = TypeAdapter(dict[str, JsonValue]) + + +def _owned_embedding(owned: Gateway, model: str, marker: str) -> httpx.Response: + return owned.request("POST", "/v1/embeddings", {"model": model, "input": marker, "encoding_format": "float"}) + + +@dataclass(frozen=True, slots=True) +class _Routed: + gateway: Gateway + upstream: _Upstream + names: Mapping[str, str] + + +def _routed_entry(name: str, api_base: str, litellm_model: str, info: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + return { + "model_name": name, + "litellm_params": {"model": litellm_model, "api_base": api_base, "api_key": _PROVIDER_KEY}, + "model_info": dict(info), + } + + +def _routed_config(directory: Path, api_base: str, names: Mapping[str, str]) -> Path: + shared: Final = _CONFIG.validate_python(yaml.safe_load((_TESTS_ROOT / "integration/proxy_config.yaml").read_text())) + config: Final = { + **shared, + "model_list": [ + _routed_entry(names["embed"], api_base, f"openai/{_EMBEDDING_MODEL}", _EMBED_ONLY), + _routed_entry(names["chat"], api_base, f"openai/{_CHAT_MODEL}", {"mode": "chat"}), + ], + "router_settings": { + **object_value(shared["router_settings"]), + "num_retries": 0, + "default_fallbacks": [names["chat"]], + "model_group_alias": {names["alias_embed"]: names["embed"], names["alias_chat"]: names["chat"]}, + }, + } + path: Final = directory / "gate-routed-config.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False)) + return path + + +@pytest.fixture(scope="module") +def routed(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Routed]: + directory: Final = tmp_path_factory.mktemp("gate-routed") + suffix: Final = uuid.uuid4().hex[:12] + names: Final = MappingProxyType( + {label: f"gate-routed-{label}-{suffix}" for label in ("embed", "chat", "alias_embed", "alias_chat")} + ) + with ( + wire_server(answering_model_discovery(_respond)) as wire, + gateway_from_environment() as rig_gateway, + owned_proxy_process( + rig_gateway, directory, {}, config=_routed_config(directory, wire.url + "/v1", names), workers=2 + ) as owned, + ): + eventually( + lambda: owned.gateway.request("POST", "/v1/chat/completions", _chat_body(names["chat"], "probe")), + lambda response: response.status_code == 200, + seconds=30, + ) + yield _Routed(owned.gateway, _Upstream(wire), names) + + +_ROUTED_TIMEOUT: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120) +_DEFAULT_FALLBACK_ROUTES: Final[Mapping[str, str]] = MappingProxyType( + {"chat_v1": "/v1/chat/completions", "messages": "/v1/responses", "responses_v1": "/v1/responses"} +) + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("route", tuple(_DEFAULT_FALLBACK_ROUTES)) +def test_router_default_fallbacks_answer_for_an_embedding_only_group(routed: _Routed, route: str) -> None: + marker: Final = _marker() + response: Final = _send(routed.gateway, _ROUTES[route], routed.names["embed"], marker) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert routed.upstream.targets(marker) == (_DEFAULT_FALLBACK_ROUTES[route],) + + +_OPT_OUTS: Final[Mapping[str, Mapping[str, JsonValue]]] = MappingProxyType( + { + "empty_fallbacks": {"fallbacks": []}, + "null_fallbacks": {"fallbacks": None}, + "disable_fallbacks": {"disable_fallbacks": True}, + } +) + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("opt_out", tuple(_OPT_OUTS)) +def test_request_opt_outs_keep_the_rejection_over_router_default_fallbacks(routed: _Routed, opt_out: str) -> None: + model: Final = routed.names["embed"] + marker: Final = _marker() + response: Final = routed.gateway.request( + "POST", "/v1/chat/completions", {**_chat_body(model, marker), **_OPT_OUTS[opt_out]} + ) + _assert_rejected_before( + routed.upstream, response, marker, _openai_error("/v1/chat/completions", model, "/v1/embeddings") + ) + + +@_ROUTED_TIMEOUT +def test_model_group_alias_to_an_embedding_only_group_falls_back_to_the_default(routed: _Routed) -> None: + marker: Final = _marker() + response: Final = routed.gateway.request( + "POST", "/v1/chat/completions", _chat_body(routed.names["alias_embed"], marker) + ) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert routed.upstream.targets(marker) == ("/v1/chat/completions",) + + +@_ROUTED_TIMEOUT +def test_model_group_alias_to_an_embedding_only_group_is_rejected_without_fallbacks(routed: _Routed) -> None: + alias: Final = routed.names["alias_embed"] + marker: Final = _marker() + response: Final = routed.gateway.request( + "POST", "/v1/chat/completions", {**_chat_body(alias, marker), "disable_fallbacks": True} + ) + _assert_rejected_before( + routed.upstream, response, marker, _openai_error("/v1/chat/completions", alias, "/v1/embeddings") + ) + + +@_ROUTED_TIMEOUT +def test_model_group_alias_to_a_chat_group_keeps_reaching_the_upstream(routed: _Routed) -> None: + marker: Final = _marker() + response: Final = routed.gateway.request( + "POST", "/v1/chat/completions", _chat_body(routed.names["alias_chat"], marker) + ) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert routed.upstream.targets(marker) == ("/v1/chat/completions",) + + +@_ROUTED_TIMEOUT +def test_model_group_alias_to_an_embedding_only_group_keeps_serving_embeddings(routed: _Routed) -> None: + marker: Final = _marker() + response: Final = _owned_embedding(routed.gateway, routed.names["alias_embed"], marker) + assert response.status_code == 200, response.text + assert list_value( + object_value(list_value(JSON_OBJECT.validate_json(response.content)["data"])[0])["embedding"] + ) == list(_VECTOR), response.text + assert routed.upstream.targets(marker) == ("/v1/embeddings",) + + +_REWRITER: Final = """from litellm.integrations.custom_logger import CustomLogger + + +class WebSocketModelRewriter(CustomLogger): + async def async_pre_call_hook(self, user_api_key_dict, cache, data, call_type): + if call_type == "_aresponses_websocket" and data.get("model") == "__SOURCE__": + data["model"] = "__TARGET__" + return data + + +rewriter = WebSocketModelRewriter() +""" + + +@dataclass(frozen=True, slots=True) +class _Rewritten: + gateway: Gateway + upstream: _Upstream + source: str + target: str + + +def _rewritten_config(directory: Path, api_base: str, source: str, target: str, module: str) -> Path: + shared: Final = _CONFIG.validate_python(yaml.safe_load((_TESTS_ROOT / "integration/proxy_config.yaml").read_text())) + config: Final = { + **shared, + "model_list": [ + _routed_entry(source, api_base, f"openai/{_CHAT_MODEL}", {"mode": "chat"}), + _routed_entry(target, api_base, f"openai/{_EMBEDDING_MODEL}", _EMBED_ONLY), + ], + "router_settings": {**object_value(shared["router_settings"]), "num_retries": 0}, + "litellm_settings": {**object_value(shared["litellm_settings"]), "callbacks": [f"{module}.rewriter"]}, + } + path: Final = directory / "gate-rewritten-config.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False)) + return path + + +@pytest.fixture(scope="module") +def rewritten(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rewritten]: + directory: Final = tmp_path_factory.mktemp("gate-rewritten") + suffix: Final = uuid.uuid4().hex[:12] + source: Final = f"gate-rewrite-chat-{suffix}" + target: Final = f"gate-rewrite-embed-{suffix}" + module: Final = f"gate_ws_rewriter_{suffix}" + (directory / f"{module}.py").write_text(_REWRITER.replace("__SOURCE__", source).replace("__TARGET__", target)) + with ( + wire_server(answering_model_discovery(_respond)) as wire, + gateway_from_environment() as rig_gateway, + owned_proxy_process( + rig_gateway, + directory, + {}, + config=_rewritten_config(directory, wire.url + "/v1", source, target, module), + workers=2, + ) as owned, + ): + eventually( + lambda: owned.gateway.request("POST", "/v1/chat/completions", _chat_body(source, "probe")), + lambda response: response.status_code == 200, + seconds=30, + ) + yield _Rewritten(owned.gateway, _Upstream(wire), source, target) + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("limit", tuple(_RESERVING_KEYS)) +def test_a_websocket_model_a_hook_rewrites_to_an_embedding_only_group_releases_the_key_reservation( + rewritten: _Rewritten, limit: str +) -> None: + markers: Final = tuple(_marker() for _ in range(3)) + served_marker: Final = _marker() + with rewritten.gateway.scenario() as scenario: + key: Final = scenario.key(**_RESERVING_KEYS[limit]) + closed: Final = tuple( + asyncio.run(_websocket_first_frame_and_close(rewritten.gateway, key, rewritten.source, marker)) + for marker in markers + ) + assert closed == (_websocket_rejection(rewritten.target),) * 3 + served: Final = rewritten.gateway.request( + "POST", "/v1/chat/completions", _chat_body(rewritten.source, served_marker), key=key + ) + assert served.status_code == 200, served.text + assert _SERVED_TEXT in served.text, served.text + assert [rewritten.upstream.calls(marker) for marker in markers] == [()] * 3 + assert rewritten.upstream.targets(served_marker) == ("/v1/chat/completions",) + + +@dataclass(frozen=True, slots=True) +class _Addressed: + gateway: Gateway + rig: Gateway + upstream: _Upstream + names: Mapping[str, str] + addresses: Mapping[str, str] + api_base: str + + +def _wildcard_backend_entry(name: str, api_base: str, backend: str) -> dict[str, JsonValue]: + entry: Final = _routed_entry(name, api_base, backend, _EMBED_ONLY) + return {**entry, "litellm_params": {**object_value(entry["litellm_params"]), "custom_llm_provider": "openai"}} + + +def _addressed_config( + directory: Path, api_base: str, names: Mapping[str, str], suffix: str, with_wildcards: bool +) -> Path: + shared: Final = _CONFIG.validate_python(yaml.safe_load((_TESTS_ROOT / "integration/proxy_config.yaml").read_text())) + wildcard: Final = f"gate-addr-wild-{suffix}" + wildcard_entries: Final = ( + ( + _routed_entry(f"{wildcard}/*", api_base, "openai/*", {}), + _wildcard_backend_entry(names["wildcard_backend"], api_base, f"{wildcard}/{_CHAT_MODEL}"), + ) + if with_wildcards + else () + ) + aliases: Final = {names["wildcard_alias"]: f"{wildcard}/{_CHAT_BACKEND_PREFIX}{suffix}"} if with_wildcards else {} + config: Final = { + **shared, + "model_list": [ + _routed_entry( + names["embed"], api_base, f"openai/gate-addr-backend-{suffix}", {**_EMBED_ONLY, "id": names["embed_id"]} + ), + _routed_entry(names["chat"], api_base, f"openai/{_CHAT_MODEL}", {"mode": "chat", "id": names["chat_id"]}), + _routed_entry(names["shared_embed"], api_base, f"openai/{_CHAT_MODEL}", _EMBED_ONLY), + *wildcard_entries, + _routed_entry(names["ghost_primary"], api_base, f"openai/{_EMBEDDING_MODEL}", _EMBED_ONLY), + ], + "router_settings": { + **object_value(shared["router_settings"]), + "num_retries": 0, + "model_group_alias": aliases, + "fallbacks": [{names["ghost_primary"]: [names["ghost"]]}, {names["ghost"]: [names["chat"]]}], + }, + "general_settings": { + **object_value(shared["general_settings"]), + "proxy_config_reload_interval_seconds": 3600, + }, + } + path: Final = directory / "gate-addressed-config.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False)) + return path + + +@contextlib.contextmanager +def _addressed_proxy(tmp_path_factory: pytest.TempPathFactory, with_wildcards: bool) -> Iterator[_Addressed]: + directory: Final = tmp_path_factory.mktemp("gate-addressed") + suffix: Final = uuid.uuid4().hex[:12] + names: Final = MappingProxyType( + { + label: f"gate-addr-{label}-{suffix}" + for label in ( + "embed", + "embed_id", + "chat", + "chat_id", + "shared_embed", + "wildcard_backend", + "wildcard_alias", + "ghost_primary", + "ghost", + ) + } + ) + with ( + wire_server(answering_model_discovery(_respond)) as wire, + gateway_from_environment() as rig_gateway, + owned_proxy_process( + rig_gateway, + directory, + {}, + config=_addressed_config(directory, wire.url + "/v1", names, suffix, with_wildcards), + workers=2, + ) as owned, + ): + eventually( + lambda: owned.gateway.request("POST", "/v1/chat/completions", _chat_body(names["chat"], "probe")), + lambda response: response.status_code == 200, + seconds=30, + ) + yield _Addressed( + owned.gateway, + rig_gateway, + _Upstream(wire), + names, + MappingProxyType( + { + "deployment_id": names["embed_id"], + "backend_name": f"openai/gate-addr-backend-{suffix}", + "wildcard_backend_name": f"gate-addr-wild-{suffix}/{_CHAT_MODEL}", + } + ), + wire.url + "/v1", + ) + + +@pytest.fixture(scope="module") +def addressed(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Addressed]: + with _addressed_proxy(tmp_path_factory, with_wildcards=True) as proxy: + yield proxy + + +@pytest.fixture(scope="module") +def addressed_without_wildcards(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Addressed]: + with _addressed_proxy(tmp_path_factory, with_wildcards=False) as proxy: + yield proxy + + +_ADDRESSED_ROUTES: Final = ("chat_v1", "messages", "responses_v1") + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("route", _ADDRESSED_ROUTES) +@pytest.mark.parametrize("address", ("deployment_id", "backend_name")) +def test_chat_sent_to_an_embedding_only_deployment_id_or_backend_name_is_rejected_before_the_upstream( + addressed: _Addressed, address: str, route: str +) -> None: + spec: Final = _ROUTES[route] + model: Final = addressed.addresses[address] + marker: Final = _marker() + response: Final = _send(addressed.gateway, spec, model, marker) + expected: Final = ( + _anthropic_error(model, "/v1/embeddings") + if spec.anthropic_shape + else _openai_error(spec.endpoint, model, "/v1/embeddings") + ) + _assert_rejected_before(addressed.upstream, response, marker, expected) + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("address", ("deployment_id", "backend_name")) +def test_embeddings_sent_to_an_embedding_only_deployment_id_or_backend_name_keep_working( + addressed: _Addressed, address: str +) -> None: + marker: Final = _marker() + response: Final = _owned_embedding(addressed.gateway, addressed.addresses[address], marker) + assert response.status_code == 200, response.text + assert list_value( + object_value(list_value(JSON_OBJECT.validate_json(response.content)["data"])[0])["embedding"] + ) == list(_VECTOR), response.text + assert addressed.upstream.targets(marker) == ("/v1/embeddings",) + + +@_ROUTED_TIMEOUT +@pytest.mark.parametrize("address", ("chat_deployment_id", "backend_name_a_chat_deployment_shares")) +def test_chat_sent_to_a_chat_capable_deployment_id_or_backend_name_keeps_reaching_the_upstream( + addressed: _Addressed, address: str +) -> None: + model: Final = addressed.names["chat_id"] if address == "chat_deployment_id" else f"openai/{_CHAT_MODEL}" + marker: Final = _marker() + response: Final = addressed.gateway.request("POST", "/v1/chat/completions", _chat_body(model, marker)) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert addressed.upstream.targets(marker) == ("/v1/chat/completions",) + + +@_ROUTED_TIMEOUT +def test_request_fallbacks_naming_a_chat_deployment_id_answer_for_an_embedding_only_primary( + addressed: _Addressed, +) -> None: + marker: Final = _marker() + response: Final = addressed.gateway.request( + "POST", + "/v1/chat/completions", + {**_chat_body(addressed.names["embed"], marker), "fallbacks": [addressed.names["chat_id"]]}, + ) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert addressed.upstream.targets(marker) == ("/v1/chat/completions",) + + +@_ROUTED_TIMEOUT +def test_chat_sent_to_an_embedding_only_backend_name_a_wildcard_also_matches_is_served_by_the_wildcard( + addressed: _Addressed, +) -> None: + marker: Final = _marker() + response: Final = addressed.gateway.request( + "POST", "/v1/chat/completions", _chat_body(addressed.addresses["wildcard_backend_name"], marker) + ) + assert response.status_code == 200, response.text + assert _SERVED_TEXT in response.text, response.text + assert [JSON_OBJECT.validate_json(call.body)["model"] for call in addressed.upstream.calls(marker)] == [_CHAT_MODEL] + + +@_ROUTED_TIMEOUT +def test_request_fallbacks_naming_an_alias_only_a_wildcard_serves_answer_for_an_embedding_only_primary( + addressed: _Addressed, +) -> None: + marker: Final = _marker() + response: Final = addressed.gateway.request( + "POST", + "/v1/chat/completions", + {**_chat_body(addressed.names["embed"], marker), "fallbacks": [addressed.names["wildcard_alias"]]}, + ) + _assert_fallback_answered(addressed.upstream, response, marker) + + +@_ROUTED_TIMEOUT +def test_yaml_fallbacks_through_a_group_with_no_deployments_answer_for_an_embedding_only_group( + addressed: _Addressed, +) -> None: + marker: Final = _marker() + response: Final = addressed.gateway.request( + "POST", "/v1/chat/completions", _chat_body(addressed.names["ghost_primary"], marker) + ) + _assert_fallback_answered(addressed.upstream, response, marker) + + +@_ROUTED_TIMEOUT +def test_an_embedding_only_model_the_read_through_loads_from_the_database_is_rejected_on_every_call( + addressed_without_wildcards: _Addressed, +) -> None: + addressed: Final = addressed_without_wildcards + name: Final = f"gate-addr-loaded-{uuid.uuid4().hex[:12]}" + markers: Final = tuple(_marker() for _ in range(6)) + with addressed.rig.scenario() as scenario: + identity: Final = _create( + addressed.rig, name, {"model": f"openai/{_EMBEDDING_MODEL}", "api_base": addressed.api_base}, _EMBED_ONLY + ) + scenario.cleanups.callback(scenario.delete_model, identity) + responses: Final = tuple( + addressed.gateway.request("POST", "/v1/chat/completions", _chat_body(name, marker), headers=_CLOSE) + for marker in markers + ) + expected: Final = _openai_error("/v1/chat/completions", name, "/v1/embeddings") + for response in responses: + _assert_error(response, expected) + assert [addressed.upstream.calls(marker) for marker in markers] == [()] * len(markers) + + +@pytest.mark.parametrize("metadata", ({"model_group": 7}, {"model_group": ["wrong-group"]})) +def test_rejection_with_invalid_metadata_is_logged_without_blocking_embeddings( + rig: _Rig, metadata: dict[str, JsonValue] +) -> None: + marker: Final = _marker() + model: Final = rig.names["embedding_only"] + rejected: Final = _chat(rig, model, marker, metadata=metadata) + _assert_rejected_chat(rig, rejected, marker, model) + embedded_marker: Final = _marker() + embedded: Final = rig.gateway.request( + "POST", "/v1/embeddings", {"model": model, "input": embedded_marker, "encoding_format": "float"} + ) + assert embedded.status_code == 200, embedded.text + assert rig.upstream.targets(embedded_marker) == ("/v1/embeddings",) + + +def test_existing_non_chat_metadata_on_a_chat_backend_remains_advisory(rig: _Rig) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + name, identities = _deploy( + scenario, + rig.api_base, + "legacy-audio", + _single({"mode": "audio_speech", "supported_endpoints": ["/v1/audio/speech"]}), + ) + _served_by_every_worker(rig.gateway, frozenset(identities)) + response: Final = _chat(rig, name, marker) + _assert_served(rig, response, marker, "/v1/chat/completions") + + +def test_a_mixed_group_skips_its_restricted_embedding_deployment(rig: _Rig) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + name: Final = f"gate-mixed-{uuid.uuid4().hex[:12]}" + embed: Final = _create( + scenario.gateway, + name, + {"model": f"openai/{_EMBEDDING_MODEL}", "api_base": rig.api_base, "order": 0}, + _EMBED_ONLY, + ) + scenario.cleanups.callback(scenario.delete_model, embed) + chat: Final = _create( + scenario.gateway, name, {"model": f"openai/{_CHAT_MODEL}", "api_base": rig.api_base, "order": 1}, {} + ) + scenario.cleanups.callback(scenario.delete_model, chat) + _served_by_every_worker(rig.gateway, frozenset({embed, chat})) + response: Final = _chat(rig, name, marker) + _assert_served(rig, response, marker, "/v1/chat/completions") + calls: Final = rig.upstream.calls(marker) + assert [JSON_OBJECT.validate_json(call.body)["model"] for call in calls] == [_CHAT_MODEL] + + +@pytest.mark.parametrize( + "fallbacks, expected_status", + [(("embedding_only",), 429), (("missing",), 429), (("missing", "chat_target"), 200)], +) +def test_local_rate_limit_skips_unusable_fallbacks(rig: _Rig, fallbacks: tuple[str, ...], expected_status: int) -> None: + limited: Final = rig.names["chat_rate_limited"] + targets: Final = tuple(rig.names[name] if name != "missing" else _marker() for name in fallbacks) + with rig.gateway.scenario() as scenario: + key: Final = scenario.key( + model_rpm_limit={limited: 1}, router_settings={"fallbacks": [{limited: list(targets)}]} + ) + first: Final = _chat(rig, limited, _marker(), key=key) + assert first.status_code == 200, first.text + marker: Final = _marker() + second: Final = _chat(rig, limited, marker, key=key) + assert second.status_code == expected_status, second.text + if expected_status == 200: + _assert_served(rig, second, marker, "/v1/chat/completions") + return + assert "Supported endpoints" not in second.text, second.text + assert rig.upstream.calls(marker) == () + + +def test_responses_websocket_uses_a_chat_fallback_without_calling_the_embedding_provider( + rig: _Rig, tmp_path: Path +) -> None: + marker: Final = _marker() + certificate: Final = write_self_signed_cert(tmp_path) + with ( + responses_ws_peer.responses_peer( + certificate, lambda identity: responses_ws_peer.events(identity, _SERVED_TEXT) + ) as peer, + owned_proxy_process(rig.gateway, tmp_path, {"SSL_CERT_FILE": str(certificate[0])}, workers=2) as owned, + owned.gateway.scenario() as scenario, + ): + fallback, fallback_ids = _deploy(scenario, peer.url, "ws-chat", _single({"mode": "chat"})) + primary, primary_ids = _deploy( + scenario, f"{rig.api_base}/{marker}", "ws-embed", _single(_EMBED_ONLY, f"openai/{_EMBEDDING_MODEL}") + ) + _served_by_every_worker(owned.gateway, frozenset((*primary_ids, *fallback_ids))) + key: Final = scenario.key(router_settings={"fallbacks": [{primary: [fallback]}]}) + (received,) = asyncio.run( + responses_ws_peer.session(str(owned.gateway.client.base_url), key, primary, (marker,)) + ) + assert received[-1]["type"] == "response.completed", received + assert responses_ws_peer.completed(received)["status"] == "completed" + assert any(event.get("delta") == _SERVED_TEXT for event in received), received + (forwarded,) = responses_ws_peer.drain(peer.frames) + assert marker in json.dumps(forwarded), forwarded + assert rig.upstream.calls(marker) == () + + +def test_client_routing_fields_cannot_disable_an_explicit_restriction(rig: _Rig) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + model, identities = _deploy(scenario, rig.api_base, "restricted-chat", _single(_EMBED_ONLY)) + _served_by_every_worker(rig.gateway, frozenset(identities)) + response: Final = _chat( + rig, + model, + marker, + original_generic_function="caller-value", + additional_drop_params=["original_generic_function"], + ) + _assert_rejected_chat(rig, response, marker, model) + + +def test_a_compaction_endpoint_declaration_keeps_the_model_available_for_compaction(rig: _Rig) -> None: + marker: Final = _marker() + with rig.gateway.scenario() as scenario: + model, identities = _deploy( + scenario, + rig.api_base, + "compact-declaration", + _single({"supported_endpoints": ["/responses/compact"], "enforce_supported_endpoints": True}), + ) + _served_by_every_worker(rig.gateway, frozenset(identities)) + response: Final = rig.gateway.request("POST", "/responses/compact", {"model": model, "input": marker}) + _assert_served(rig, response, marker, "/v1/responses/compact") + + +@pytest.mark.parametrize("alias_source", ["key_alias", "key_router", "team_router"]) +def test_local_rate_limit_fallbacks_resolve_saved_aliases(rig: _Rig, tmp_path: Path, alias_source: str) -> None: + suffix: Final = uuid.uuid4().hex[:12] + names: Final = {label: f"gate-fallback-alias-{label}-{suffix}" for label in ("primary", "target", "alias")} + shared: Final = _CONFIG.validate_python(yaml.safe_load((_TESTS_ROOT / "integration/proxy_config.yaml").read_text())) + config: Final = { + **shared, + "model_list": [ + _routed_entry(names["primary"], rig.api_base, f"openai/{_CHAT_MODEL}", {"mode": "chat"}), + _routed_entry(names["target"], rig.api_base, f"openai/{_CHAT_MODEL}", {"mode": "chat"}), + ], + "router_settings": { + **object_value(shared["router_settings"]), + "num_retries": 0, + "fallbacks": [{names["primary"]: [names["alias"]]}], + }, + } + path: Final = tmp_path / "fallback-alias.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False)) + aliases: Final = {names["alias"]: names["target"]} + with ( + owned_proxy_process(rig.gateway, tmp_path, {}, config=path, workers=2) as owned, + owned.gateway.scenario() as scenario, + ): + team: Final = ( + scenario.team(router_settings={"model_group_alias": aliases}) if alias_source == "team_router" else None + ) + key: Final = scenario.key( + team_id=team, + aliases=aliases if alias_source == "key_alias" else {}, + router_settings={"model_group_alias": aliases} if alias_source == "key_router" else None, + model_rpm_limit={names["primary"]: 1}, + ) + first: Final = owned.gateway.request( + "POST", "/v1/chat/completions", _chat_body(names["primary"], _marker()), key=key + ) + assert first.status_code == 200, first.text + marker: Final = _marker() + second: Final = owned.gateway.request( + "POST", "/v1/chat/completions", _chat_body(names["primary"], marker), key=key + ) + _assert_fallback_answered(rig.upstream, second, marker) diff --git a/tests/unit/proxy/common_utils/test_callback_utils.py b/tests/unit/proxy/common_utils/test_callback_utils.py index ecb2375d495..c8d909ac2eb 100644 --- a/tests/unit/proxy/common_utils/test_callback_utils.py +++ b/tests/unit/proxy/common_utils/test_callback_utils.py @@ -49,6 +49,19 @@ def test_get_remaining_tokens_and_requests_from_request_data(): } +@pytest.mark.parametrize("client_model_group", [7, ["embed-only"], {"name": "embed-only"}]) +def test_remaining_limit_headers_ignore_a_model_group_that_is_not_a_string(client_model_group: object): + casedata: Final = { + "metadata": { + "model_group": client_model_group, + f"litellm-key-remaining-requests-{client_model_group}": 100, + f"litellm-key-remaining-tokens-{client_model_group}": 200, + } + } + + assert get_remaining_tokens_and_requests_from_request_data(casedata) == {} + + @patch( "litellm.proxy.common_utils.callback_utils.CustomLogger.get_callback_env_vars", return_value=["API_KEY", "MISSING_VAR"], diff --git a/tests/unit/proxy/response_api_endpoints/test_endpoints.py b/tests/unit/proxy/response_api_endpoints/test_endpoints.py index 52e95901a16..55b83bf5b98 100644 --- a/tests/unit/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/unit/proxy/response_api_endpoints/test_endpoints.py @@ -2479,6 +2479,8 @@ class TestResponsesWSFirstFrameModelAuth: ) + + class TestReadWSModelFromFirstFrameErrors: @pytest.mark.asyncio async def test_transport_error_first_frame_closes_with_internal_error(self): diff --git a/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py b/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py index 262357fe1bd..dbba1cb9400 100644 --- a/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py +++ b/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py @@ -5056,6 +5056,28 @@ def test_get_logging_payload_router_rejected_request_takes_provider_from_model_g assert payload["custom_llm_provider"] == expected_provider +@pytest.mark.parametrize("client_model_group", [7, ["embed-only"], {"name": "embed-only"}]) +def test_get_logging_payload_drops_a_client_model_group_that_is_not_a_string(client_model_group: object): + payload: Final = get_logging_payload( + kwargs={ + "model": "embed-only", + "litellm_params": { + "metadata": {"user_api_key": "test-key", "model_group": client_model_group, "status": "failure"} + }, + "standard_logging_object": { + **_make_failed_request_standard_logging_payload(), + "model_group": client_model_group, + }, + }, + response_obj={}, + start_time=datetime.datetime.now(timezone.utc), + end_time=datetime.datetime.now(timezone.utc), + llm_router=None, + ) + + assert payload["model_group"] == "" + + def test_get_logging_payload_router_rejected_request_without_router_leaves_provider_empty(): assert _router_rejected_failure_payload("openai-group", None)["custom_llm_provider"] == "" diff --git a/tests/unit/proxy/test_common_request_processing.py b/tests/unit/proxy/test_common_request_processing.py index 679e1bfd85a..c447c6e43ca 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -452,7 +452,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" @@ -574,7 +574,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, @@ -713,6 +713,9 @@ class TestProxyBaseLLMRequestProcessing: assert returned_data["metadata"]["tags"] == ["existing-tag"] tag_budget_check.assert_not_awaited() + + + @pytest.mark.asyncio async def test_common_processing_pre_call_logic_skips_guardrail_tag_budget_check_for_zero_cost_model( self, monkeypatch @@ -1020,7 +1023,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 = { @@ -2042,7 +2045,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" @@ -7423,6 +7426,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: return processor.data, logging_obj mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{"gpt-4": ["gpt-3.5-turbo"]}] with patch.object( @@ -7464,6 +7468,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: ) mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = None with patch.object( @@ -7503,6 +7508,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: ) mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{"gpt-4": ["gpt-3.5-turbo", "claude-3-haiku"]}] with patch.object( @@ -7530,6 +7536,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: assert processor.data["model"] == "gpt-4" + @pytest.mark.asyncio async def test_fallback_uses_key_level_router_settings(self): from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError @@ -7546,6 +7553,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: return processor.data, MagicMock() mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{"gpt-4": ["gpt-3.5-turbo"]}] user_api_key_dict = MagicMock() @@ -7593,6 +7601,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: ) mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{"gpt-4": ["gpt-3.5-turbo"]}] with patch.object( @@ -7637,6 +7646,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: raise ValueError("unexpected auth failure on fallback") mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{"gpt-4": ["gpt-3.5-turbo"]}] with patch.object( @@ -7746,6 +7756,7 @@ class TestPreCallWithFallbacksOnLocalRateLimit: return processor.data, MagicMock() mock_router = MagicMock() + mock_router.endpoint_error_for_request.return_value = None mock_router.fallbacks = [{primary_model: [fallback_model]}] with patch.object( diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index 4e4e2f23950..83c63bb8de4 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -10,7 +10,10 @@ from unittest.mock import MagicMock from fastapi import HTTPException -from litellm.proxy.route_llm_request import ProxyModelNotFoundError, route_request +from litellm.proxy.route_llm_request import ( + ProxyModelNotFoundError, + route_request, +) @pytest.mark.parametrize( @@ -1524,6 +1527,8 @@ async def test_route_request_read_through_recovers_model_created_on_sibling_repl assert table.find_many_wheres[0] == {"model_name": model_name} + + @pytest.mark.asyncio async def test_route_request_unknown_model_raises_and_hits_db_once_within_ttl(monkeypatch): import litellm @@ -1765,6 +1770,8 @@ async def test_route_request_router_settings_override_skips_null_fields(): assert "model_group_retry_policy" not in call_kwargs + + def _router_with_defaults(default_litellm_params: dict[str, object]): import litellm diff --git a/tests/unit/router_utils/test_endpoint_eligibility.py b/tests/unit/router_utils/test_endpoint_eligibility.py new file mode 100644 index 00000000000..e129bf69b5c --- /dev/null +++ b/tests/unit/router_utils/test_endpoint_eligibility.py @@ -0,0 +1,87 @@ +from collections.abc import Mapping +from typing import Final + +import pytest +from pydantic import TypeAdapter + +from litellm.router_utils.endpoint_eligibility import UnsupportedEndpointError, eligible_deployments, endpoint_selection + +_DEPLOYMENT: Final = TypeAdapter(list[dict[str, object]]) + + +def _deployment(info: Mapping[str, object], identity: str = "embedding") -> dict[str, object]: + return _DEPLOYMENT.validate_python( + [ + { + "model_name": "shared", + "litellm_params": {"model": "openai/test-model"}, + "model_info": {"id": identity, **info}, + } + ] + )[0] + + +@pytest.mark.parametrize("call_type", ["acompletion", "anthropic_messages", "aresponses", "acreate_interaction"]) +def test_explicit_restriction_excludes_an_embedding_deployment_from_a_mixed_group(call_type: str) -> None: + embedding: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}) + chat: Final = _deployment({"supported_endpoints": ["/v1/chat/completions"]}, "chat") + with endpoint_selection(call_type): + result: Final = eligible_deployments("shared", (embedding, chat)) + assert result == (chat,) + + +@pytest.mark.parametrize( + "info", + [ + {"mode": "audio_speech", "supported_endpoints": ["/v1/audio/speech"]}, + {"enforce_supported_endpoints": False, "supported_endpoints": ["/v1/embeddings"]}, + {"enforce_supported_endpoints": True}, + {"enforce_supported_endpoints": True, "supported_endpoints": None}, + {"enforce_supported_endpoints": True, "supported_endpoints": "invalid"}, + {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/messages"]}, + ], +) +def test_advisory_or_incomplete_metadata_preserves_routing(info: Mapping[str, object]) -> None: + deployment: Final = _deployment(info) + with endpoint_selection("acompletion"): + result: Final = eligible_deployments("shared", (deployment,)) + assert result == (deployment,) + + +def test_unsupported_error_explains_the_requested_operation_and_available_endpoints() -> None: + deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}) + with endpoint_selection("acreate_interaction"): + result: Final = eligible_deployments("shared", (deployment,)) + assert isinstance(result, UnsupportedEndpointError) + assert result.status_code == 400 + assert "/v1beta/interactions" in result.message + assert "/v1/embeddings" in result.message + + +@pytest.mark.parametrize("call_type", ["aembedding", "aspeech", "_arealtime", "aget_responses"]) +def test_chat_restriction_does_not_block_other_operations(call_type: str) -> None: + deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}) + with endpoint_selection(call_type): + result: Final = eligible_deployments("shared", (deployment,)) + assert result == (deployment,) + + +def test_nested_non_chat_calls_do_not_inherit_the_outer_chat_restriction() -> None: + deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}) + with endpoint_selection("acompletion"): + with endpoint_selection("aembedding"): + embedding_result: Final = eligible_deployments("shared", (deployment,)) + chat_result: Final = eligible_deployments("shared", (deployment,)) + assert embedding_result == (deployment,) + assert isinstance(chat_result, UnsupportedEndpointError) + + +@pytest.mark.parametrize( + "endpoint", + ["/messages", "/interactions", "/responses/compact", "/openai/v1/responses/compact", "/openai/v1/chat/completions"], +) +def test_chat_endpoint_declarations_keep_their_meaning_across_prefixes(endpoint: str) -> None: + deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": [endpoint]}) + with endpoint_selection("acompletion"): + result: Final = eligible_deployments("shared", (deployment,)) + assert result == (deployment,) diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index bd3cf738af8..d348e6007c8 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -11285,6 +11285,57 @@ def test_is_deployment_blocked_static_helper_reflects_blocked_flag(): ) +def test_synchronous_text_completion_rejects_a_restricted_model_before_the_provider() -> None: + model: Final = "text-completion-openai/unit-text-model" + router: Final = Router( + model_list=[ + { + "model_name": model, + "litellm_params": { + "model": model, + "api_base": "https://sync-router.invalid/v1", + "api_key": "unit-provider-key", + }, + "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + } + ], + num_retries=0, + ) + with respx.mock(assert_all_called=False) as upstream: + provider: Final = upstream.post("https://sync-router.invalid/v1/completions").respond( + 200, + json={ + "id": "cmpl-unit", + "object": "text_completion", + "created": 1, + "model": "unit-text-model", + "choices": [{"text": "provider called", "index": 0, "finish_reason": "stop", "logprobs": None}], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }, + ) + with pytest.raises(litellm.BadRequestError, match="/v1/completions is not supported"): + router.text_completion(model=model, prompt="test") + assert provider.call_count == 0 + + +def test_endpoint_preflight_returns_unavailable_model_errors_as_values() -> None: + router: Final = Router( + model_list=[ + { + "model_name": "available", + "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "unit-provider-key"}, + } + ], + num_retries=0, + ) + missing: Final = "unregistered-fallback" + error: Final = router.endpoint_error_for_request(missing, "acompletion", {}) + assert isinstance(error, litellm.BadRequestError) + assert error.status_code == 400 + assert error.model == missing + assert router.endpoint_error_for_request("available", "acompletion", {}) is None + + def test_routing_groups_legacy_property_forwards_to_public_attribute(): router = _router_with_two_deployments([False, False]) diff --git a/tests/unit/test_router_model_cost_isolation.py b/tests/unit/test_router_model_cost_isolation.py index b353dc86d5f..7c092fd667b 100644 --- a/tests/unit/test_router_model_cost_isolation.py +++ b/tests/unit/test_router_model_cost_isolation.py @@ -2211,10 +2211,8 @@ def test_a_reload_rebuilds_exactly_what_a_fresh_boot_registered() -> None: f"the rebuild changed or dropped a field the boot registration wrote: " f"{ {k: (v, rebuilt.get(k)) for k, v in at_boot.items() if rebuilt.get(k) != v} }" ) - assert {field: rebuilt[field] for field in set(rebuilt) - set(at_boot)} == { - "db_model": False, - "member_auto_router": False, - } + replay_defaults: Final = {field: rebuilt[field] for field in set(rebuilt) - set(at_boot)} + assert replay_defaults.items() <= ModelInfo().model_dump(exclude_none=True).items() assert "member_auto_router" not in litellm.model_cost["gpt-4o"] assert router.model_list finally: diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index e5c1b9f2c26..5e0a6ccc4c7 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -49972,6 +49972,11 @@ export interface components { discoverable?: boolean | null; /** Enable Tag Filtering */ enable_tag_filtering?: boolean | null; + /** + * Enforce Supported Endpoints + * @default false + */ + enforce_supported_endpoints: boolean; /** Id */ id: string | null; /** Input Cost Per Character */