diff --git a/litellm/exceptions.py b/litellm/exceptions.py index 61e3a8fb5dc..28211b3dab2 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -323,6 +323,17 @@ class UnsupportedEndpointError(BadRequestError): pass +class EndpointNotSupportedByModelError(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" + + class UnprocessableEntityError(openai.UnprocessableEntityError): def __init__( self, diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index fc29a6e3a08..4884e75af4e 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -129,6 +129,7 @@ from litellm.proxy.guardrails.auto_router_compression import arm_pre_call as _ar from litellm.proxy.hooks.parallel_request_limiter_v3 import get_or_create_request_stash 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 @@ -871,6 +872,17 @@ _DECODED_JSON: Final = TypeAdapter(object) _ROUTER_SETTINGS_ADAPTER: Final[TypeAdapter[Mapping[str, object] | None]] = TypeAdapter(Mapping[str, object] | None) +def _fallback_probe_data( + pristine: object, fallback_model: str, key_aliases: object, group_alias: object +) -> Mapping[str, object]: + return { + **_REQUEST_FIELDS.validate_python(pristine), + "model": resolve_served_model( + fallback_model, _DECODED_JSON.validate_python(litellm.model_alias_map), key_aliases, group_alias + ), + } + + def _request_model(data: Mapping[str, object]) -> str | list[str] | None: try: return _REQUEST_MODEL.validate_python(data.get("model"), strict=True) @@ -2676,6 +2688,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: @@ -2691,6 +2714,13 @@ class ProxyBaseLLMRequestProcessing: self.data = independent_snapshot(pristine) self.data["model"] = fallback_model try: + if endpoint_unsupported_by_model( + route_type, + _fallback_probe_data(pristine, fallback_model, key_aliases, group_alias), + llm_router, + user_api_key_dict.team_id, + ): + continue return await self.common_processing_pre_call_logic( request=request, general_settings=general_settings, diff --git a/litellm/proxy/common_utils/callback_utils.py b/litellm/proxy/common_utils/callback_utils.py index f889748c504..d1787f3ff8b 100644 --- a/litellm/proxy/common_utils/callback_utils.py +++ b/litellm/proxy/common_utils/callback_utils.py @@ -436,10 +436,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 9e642a02d49..772bb3e53d2 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -34,6 +34,7 @@ from litellm._logging import verbose_proxy_logger from litellm.anthropic_interface.exceptions.exception_mapping_utils import AnthropicExceptionMapping from litellm.caching.dual_cache import DualCache from litellm.constants import EMPTY_MAPPING, MANAGED_BACKGROUND_RESPONSE_METADATA_KEY, TERMINAL_RESPONSE_STATUSES +from litellm.exceptions import EndpointNotSupportedByModelError from litellm.integrations.custom_guardrail import ModifyResponseException from litellm.litellm_core_utils.core_helpers import as_str_mapping, get_metadata_variable_name_from_kwargs from litellm.litellm_core_utils.internal_call_metadata import is_background_response, requests_background @@ -2086,6 +2087,14 @@ def _routing_hints_from_first_ws_frame(first_message: str) -> Mapping[str, objec def _responses_ws_failure_frame(failure: Exception) -> str: + if isinstance(failure, EndpointNotSupportedByModelError): + return json.dumps( + { + "type": "error", + "status": 400, + "error": {"type": "invalid_request_error", "message": failure.message.split("\n\n", 1)[0]}, + } + ) proxy_exception: Final = handle_exception_on_proxy(failure) status: Final = int(proxy_exception.code) if str(proxy_exception.code).isdigit() else 500 error_type: Final = ( @@ -2444,7 +2453,12 @@ async def _responses_websocket_session( with contextlib.suppress(Exception): await websocket.send_text(_responses_ws_failure_frame(e)) with contextlib.suppress(Exception): - await websocket.close(code=1011, reason="Internal server error") + await websocket.close( + code=1008 if isinstance(e, EndpointNotSupportedByModelError) else 1011, + reason="Pre-call error" + if isinstance(e, EndpointNotSupportedByModelError) + else "Internal server error", + ) return crash_row return bill_socket_failure(outcome) if isinstance(outcome, Exception) else None diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index db1951ec9e2..fea4f342b91 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -431,6 +431,28 @@ 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, +) -> bool: + model: Final = data.get("model") + if not isinstance(model, str) or llm_router is None: + return False + metadata: Final = data.get("metadata") + request_kwargs: Final = { + **data, + "metadata": { + **(metadata if isinstance(metadata, Mapping) else {}), + "user_api_key_team_id": team_id, + }, + } + return bool(llm_router.get_routable_deployments(model, request_kwargs)) and not llm_router.get_routable_deployments( + model, request_kwargs, call_type=route_type + ) + + 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 833ee71ae9c..df9ed611a47 100644 --- a/litellm/proxy/spend_tracking/spend_tracking_utils.py +++ b/litellm/proxy/spend_tracking/spend_tracking_utils.py @@ -464,7 +464,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: @@ -682,7 +686,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 ff31f48c49e..31d679f616e 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -287,6 +287,11 @@ from litellm.router_utils.embedding_batch_size_check import ( embedding_deployment_selection, log_rejection_before_provider_call, ) +from litellm.router_utils.endpoint_eligibility import ( + eligible_deployments, + endpoint_selection, + warn_on_unenforced_endpoint_flag, +) from litellm.router_utils.fallback_event_handlers import ( MID_STREAM_FALLBACK_CONTROLS_KEY, AttemptedFallbackTargets, @@ -1449,7 +1454,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 @@ -6054,12 +6059,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(): @@ -6915,6 +6921,7 @@ class Router: """ model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group retry_kwargs: Final = mid_stream_retry_kwargs(initial_kwargs) + handler: Final = retry_kwargs.get("original_generic_function") healthy_deployments, all_deployments = await self._async_get_healthy_deployments( model=model_group, parent_otel_span=get_parent_otel_span_from_kwargs(retry_kwargs), @@ -6945,7 +6952,10 @@ class Router: ) carry_superseded_attempt_usage(retry_kwargs) try: - response = await attempt_function(**retry_kwargs) + with endpoint_selection( + getattr(handler, "__name__", None), getattr(handler, "__module__", None), model=model_group + ): + response = await attempt_function(**retry_kwargs) except Exception as retry_error: # noqa: BLE001 # the fallback chain judges the failed retry wrapped = _stream_fallback_error_for_raised(retry_error, model_group, False, e.llm_provider) if wrapped is None: @@ -9315,9 +9325,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) @@ -10638,6 +10657,7 @@ class Router: declared_params: Final = _LITELLM_PARAMS_ADAPTER.validate_python(_litellm_params) warn_on_provider_credential_mismatch(model_name=_model_name, litellm_params=declared_params) warn_on_ignored_model_capability_flags(model_name=_model_name, litellm_params=declared_params) + warn_on_unenforced_endpoint_flag(model_name=_model_name, model_info=_model_info) deployment = without_model_capability_flags( Deployment( **deployment_info, @@ -11441,6 +11461,9 @@ class Router: declared_params: Final = deployment.litellm_params.model_dump(exclude_none=True) warn_on_provider_credential_mismatch(model_name=deployment.model_name, litellm_params=declared_params) warn_on_ignored_model_capability_flags(model_name=deployment.model_name, litellm_params=declared_params) + warn_on_unenforced_endpoint_flag( + model_name=deployment.model_name, model_info=deployment.model_info.model_dump() + ) served: Final = with_admin_api_key_secret_reference(without_model_capability_flags(deployment)) # add to model list @@ -14797,9 +14820,15 @@ class Router: """ return [m for m in self.model_list if m["litellm_params"]["model"] == model] - def get_routable_deployments(self, model: str) -> tuple[DeploymentTypedDict, ...]: + def get_routable_deployments( + self, + model: str, + request_kwargs: Mapping[str, object] = MappingProxyType({}), + call_type: str | None = None, + ) -> tuple[DeploymentTypedDict, ...]: try: - _, candidates = self._common_checks_available_deployment(model=model, request_kwargs={}) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # its signature leaves input and the result untyped + with endpoint_selection(call_type, model=model): + _, candidates = self._common_checks_available_deployment(model=model, request_kwargs=request_kwargs) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # its signature leaves input and the result untyped except (litellm.BadRequestError, ValueError): return () if isinstance(candidates, dict): @@ -15075,7 +15104,10 @@ class Router: model=model, llm_provider="", ) - return selectable + eligible: Final = eligible_deployments(model, selectable) + if isinstance(eligible, tuple): + return list(eligible) + raise 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..4a87e582c97 --- /dev/null +++ b/litellm/router_utils/endpoint_eligibility.py @@ -0,0 +1,120 @@ +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, TypeAdapter, ValidationError, field_validator + +from litellm._logging import verbose_router_logger +from litellm.exceptions import EndpointNotSupportedByModelError + +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: + rooted: Final = "/" + endpoint.strip().strip("/").casefold() + path: Final = rooted.removeprefix("/openai").removeprefix("/v1beta").removeprefix("/v1") + return "/" + path.rsplit(":", 1)[-1] if ":" in path else path + + +CHAT_ENDPOINTS: Final = frozenset(_endpoint_path(endpoint) for endpoint in CHAT_ENDPOINT_BY_CALL.values()) +_Deployment: Final = TypeVar("_Deployment", bound=Mapping[str, object]) +_ENFORCE_FLAG: Final = TypeAdapter(bool | None) +_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 | None = None + supported_endpoints: tuple[str, ...] | None = None + + @field_validator("supported_endpoints", mode="before") + @classmethod + def _one_endpoint_as_a_list(cls, value: object) -> object: + return (value,) if isinstance(value, str) else value + + +@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 _enforced_endpoints(model_info: object) -> tuple[str, ...] | None: + try: + declaration: Final = _EndpointDeclaration.model_validate(model_info) + except ValidationError: + return None + return declaration.supported_endpoints if declaration.enforce_supported_endpoints else None + + +def warn_on_unenforced_endpoint_flag(model_name: str, model_info: Mapping[str, object]) -> str | None: + try: + flagged: Final = _ENFORCE_FLAG.validate_python(model_info.get("enforce_supported_endpoints")) + except ValidationError: + return None + if not flagged or _enforced_endpoints(model_info) is not None: + return None + warning: Final = ( + f"Deployment '{model_name}' sets enforce_supported_endpoints without a supported_endpoints list of " + 'strings, so litellm enforces nothing for it. Set supported_endpoints, e.g. ["/v1/embeddings"].' + ) + verbose_router_logger.warning(warning) + return warning + + +def non_chat_endpoints(deployment: Mapping[str, object]) -> frozenset[str] | None: + declared: Final = _enforced_endpoints(deployment.get("model_info")) + if declared is None: + return None + endpoints: Final = frozenset(declared) + 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, ...] | EndpointNotSupportedByModelError: + 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 EndpointNotSupportedByModelError( + 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 db8be9e91a3..56dcad4051a 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -270,6 +270,7 @@ def _whole_float_as_int(value: object) -> object: class ModelInfo(MirroredPricingParams): + enforce_supported_endpoints: bool | None = None 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..367135eae26 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, refuse chat with a gateway 400, and route chat through a live fallback; 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..c45adfeeb84 --- /dev/null +++ b/tests/e2e/llm_translation/test_embedding_only_supported_endpoints_e2e.py @@ -0,0 +1,119 @@ +"""Live provider checks for embeddings, the chat refusal, 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 BadRequestError, 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}" + ) + + @pytest.mark.covers("llm.embeddings.openai.supported_endpoints_gate.nonstream.works") + @meta(_subject(Route.CHAT_COMPLETIONS)) + def test_chat_completions_without_a_fallback_are_refused_by_the_gateway( + self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients + ) -> None: + model: Final = _register(proxy, resources, "e2e-embed-only", EMBEDDINGS_ONLY) + with pytest.raises(BadRequestError) as refused: + _ = _openai_at(sdk, "/v1", resources.key()).chat.completions.create( + model=model, messages=[{"role": "user", "content": PROMPT}], extra_body=NO_PROXY_CACHE + ) + assert ( + f"/v1/chat/completions is not supported by model={model}. Supported endpoints: /v1/embeddings" + in refused.value.message + ), f"chat on {model} should be refused by the gateway before any provider call, got {refused.value!r}" diff --git a/tests/e2e/models.py b/tests/e2e/models.py index afddc7e25d6..504adec5401 100644 --- a/tests/e2e/models.py +++ b/tests/e2e/models.py @@ -1416,7 +1416,7 @@ class LiteLLMParamsBody(BaseModel): order: int | None = None -ModelMode = Literal["chat", "batch", "realtime", "image_generation"] +ModelMode = Literal["chat", "embedding", "batch", "realtime", "image_generation"] class TokenCountModelBody(BaseModel, frozen=True): @@ -1425,6 +1425,7 @@ class TokenCountModelBody(BaseModel, frozen=True): 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..465f5801692 --- /dev/null +++ b/tests/integration/routing/test_non_chat_model_endpoint_gate.py @@ -0,0 +1,1515 @@ +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 OwnedState, 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", + "status": 400, + "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, + state=OwnedState.SHARED_WITH_WORKER, + ) 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, + state=OwnedState.SHARED_WITH_WORKER, + ) 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, + state=OwnedState.SHARED_WITH_WORKER, + ) 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), (("embedding_only", "chat_target"), 200), (("missing",), 400)], +) +def test_local_rate_limit_skips_restricted_fallbacks_and_still_tries_unloaded_ones( + 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 _websocket_fallback_config(directory: Path, primary: tuple[str, str], fallback: tuple[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(primary[0], primary[1], f"openai/{_EMBEDDING_MODEL}", _EMBED_ONLY), + _routed_entry(fallback[0], fallback[1], f"openai/{_CHAT_MODEL}", {"mode": "chat"}), + ], + "router_settings": { + **object_value(shared["router_settings"]), + "num_retries": 0, + "fallbacks": [{primary[0]: [fallback[0]]}], + }, + } + path: Final = directory / "gate-websocket-fallback.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False)) + return path + + +def test_responses_websocket_uses_a_configured_chat_fallback_without_calling_the_embedding_provider( + rig: _Rig, tmp_path: Path +) -> None: + marker: Final = _marker() + suffix: Final = uuid.uuid4().hex[:12] + primary: Final = f"gate-ws-embed-{suffix}" + fallback: Final = f"gate-ws-chat-{suffix}" + 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])}, + config=_websocket_fallback_config(tmp_path, (primary, f"{rig.api_base}/{marker}"), (fallback, peer.url)), + workers=2, + state=OwnedState.SHARED_WITH_WORKER, + ) as owned, + owned.gateway.scenario() as scenario, + ): + key: Final = scenario.key() + (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, state=OwnedState.SHARED_WITH_WORKER + ) 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/management_endpoints/test_model_management_endpoints.py b/tests/unit/proxy/management_endpoints/test_model_management_endpoints.py index b892dc97e5c..835e7605955 100644 --- a/tests/unit/proxy/management_endpoints/test_model_management_endpoints.py +++ b/tests/unit/proxy/management_endpoints/test_model_management_endpoints.py @@ -3617,6 +3617,33 @@ class TestDeleteTeamModels: assert any(e[0] == "delete_many" for e in prisma.events) +@pytest.mark.parametrize( + ("model_info_patch", "expected_gate"), + [ + pytest.param({"tpm": 5}, True, id="unrelated-field"), + pytest.param({"supported_endpoints": ["/v1/embeddings", "/v1/rerank"]}, True, id="endpoint-list-only"), + pytest.param({"enforce_supported_endpoints": False}, False, id="explicit-false"), + ], +) +def test_model_info_patch_changes_the_endpoint_gate_only_when_it_sends_the_flag( + model_info_patch: dict[str, object], expected_gate: bool +): + from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model + + result: Final = update_db_model( + db_model=Deployment( + model_name="embedding-only", + litellm_params=LiteLLM_Params(model="openai/text-embedding-3-small"), + model_info=ModelInfo( + id="embedding-only", enforce_supported_endpoints=True, supported_endpoints=["/v1/embeddings"] + ), + ), + updated_patch=updateDeployment.model_validate({"model_info": model_info_patch}), + ) + + assert json.loads(result["model_info"]).get("enforce_supported_endpoints") is expected_gate + + def _build_db_model_for_blocked_test(): from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo diff --git a/tests/unit/proxy/response_api_endpoints/test_endpoints.py b/tests/unit/proxy/response_api_endpoints/test_endpoints.py index e3d514b2ea9..b11dcc58ef6 100644 --- a/tests/unit/proxy/response_api_endpoints/test_endpoints.py +++ b/tests/unit/proxy/response_api_endpoints/test_endpoints.py @@ -2055,6 +2055,22 @@ def test_responses_ws_failure_frame_uses_the_http_status_for_the_same_failure( assert frame["error"]["message"] == handle_exception_on_proxy(failure).message +def test_responses_ws_failure_frame_shows_only_the_refusal_for_a_model_that_refuses_the_endpoint(): + from litellm.exceptions import EndpointNotSupportedByModelError + from litellm.proxy.response_api_endpoints.endpoints import _responses_ws_failure_frame + + refusal: Final = "/v1/responses is not supported by model=embed-only. Supported endpoints: /v1/embeddings" + failure: Final = EndpointNotSupportedByModelError( + message=f"{refusal}\n\nLiteLLM: model group 'embed-only' failed with the error above", model="embed-only" + ) + + assert json.loads(_responses_ws_failure_frame(failure)) == { + "type": "error", + "status": 400, + "error": {"type": "invalid_request_error", "message": refusal}, + } + + class TestWSSessionCostTracking: @pytest.mark.asyncio async def test_router_budget_limiter_skips_arealtime_call_type(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 23f9ac0e2d8..740334252d2 100644 --- a/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py +++ b/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py @@ -5384,6 +5384,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 fdcbcbfcd05..e58a1de9eb5 100644 --- a/tests/unit/proxy/test_common_request_processing.py +++ b/tests/unit/proxy/test_common_request_processing.py @@ -456,7 +456,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" @@ -740,7 +740,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, @@ -1186,7 +1186,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 = { @@ -2208,7 +2208,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" @@ -7706,6 +7706,56 @@ class TestPreCallWithFallbacksOnLocalRateLimit: assert processor.data["model"] == fallback_model assert call_count == 2 + @pytest.mark.asyncio + async def test_local_rate_limit_fallback_skips_an_aliased_embedding_only_model(self): + from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing + from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError + + router: Final = Router( + model_list=[ + { + "model_name": "embed-only", + "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "sk-synthetic"}, + "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + }, + {"model_name": "chat-model", "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "sk-synthetic"}}, + ], + fallbacks=[{"rate-limited": ["embed-alias", "chat-model"]}], + ) + processor: Final = ProxyBaseLLMRequestProcessing( + data={ + "model": "rate-limited", + "router_settings_override": {"model_group_alias": {"embed-alias": "embed-only"}}, + } + ) + attempted: Final[list[object]] = [] + + async def pre_call_logic(**_: object): + attempted.append(processor.data.get("model")) + if processor.data.get("model") == "rate-limited": + raise ProxyRateLimitError(detail="key rpm limit exceeded", headers={"retry-after": "30"}) + return processor.data, MagicMock() + + with patch.object(processor, "common_processing_pre_call_logic", side_effect=pre_call_logic): + await processor._pre_call_with_fallbacks( + request=MagicMock(), + general_settings={}, + proxy_logging_obj=MagicMock(), + user_api_key_dict=ProxyUserAPIKeyAuth(), + version=None, + proxy_config=MagicMock(), + user_model=None, + user_temperature=None, + user_request_timeout=None, + user_max_tokens=None, + user_api_base=None, + model="rate-limited", + route_type="acompletion", + llm_router=router, + ) + + assert attempted == ["rate-limited", "chat-model"] + @pytest.mark.asyncio async def test_raises_when_no_fallbacks_configured(self): from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index 9e31d3774f9..8f364e51e2b 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -28,6 +28,7 @@ from litellm.proxy.route_llm_request import ( JSON_OBJECT_ADAPTER, ProxyModelNotFoundError, UserConfigRouters, + endpoint_unsupported_by_model, route_request, user_config_routers, ) @@ -3027,3 +3028,27 @@ async def test_user_config_routers_discard_the_least_recently_used_router_past_c assert first_handler() is not None assert second_handler() is None routers.clear() + + +def test_fallback_probe_skips_only_a_loaded_model_that_refuses_the_route() -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "chat-fallback", + "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "unit-provider-key"}, + }, + { + "model_name": "embed-only", + "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "unit-provider-key"}, + "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + }, + ], + num_retries=0, + ) + + def skipped(model: str) -> bool: + return endpoint_unsupported_by_model("acompletion", {"model": model}, router, None) + + assert skipped("embed-only") is True + assert skipped("chat-fallback") is False + assert skipped("added-on-another-worker") is False 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..5e5d77254d0 --- /dev/null +++ b/tests/unit/router_utils/test_endpoint_eligibility.py @@ -0,0 +1,139 @@ +from collections.abc import Mapping +from typing import Final + +import pytest +from pydantic import TypeAdapter + +from litellm.exceptions import EndpointNotSupportedByModelError +from litellm.router_utils.endpoint_eligibility import ( + eligible_deployments, + endpoint_selection, + warn_on_unenforced_endpoint_flag, +) + +_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": 7}, + {"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_a_single_endpoint_string_is_enforced_like_a_one_item_list() -> None: + deployment: Final = _deployment({"enforce_supported_endpoints": True, "supported_endpoints": "/v1/embeddings"}) + with endpoint_selection("acompletion"): + result: Final = eligible_deployments("shared", (deployment,)) + assert isinstance(result, EndpointNotSupportedByModelError) + assert result.message.endswith("Supported endpoints: /v1/embeddings") + + +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, EndpointNotSupportedByModelError) + 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, EndpointNotSupportedByModelError) + + +@pytest.mark.parametrize( + "endpoint", + [ + "/messages", + "/interactions", + "/responses/compact", + "/openai/v1/responses/compact", + "/openai/v1/chat/completions", + "chat/completions", + "v1/responses", + " /v1/chat/completions ", + "/v1/chat/completions/", + "/V1/Chat/Completions", + "/v1beta/models/gemini-3-flash:generateContent", + ], +) +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,) + + +@pytest.mark.parametrize( + "info", + [ + {"enforce_supported_endpoints": True}, + {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings", None]}, + {"enforce_supported_endpoints": "true", "supported_endpoints": 7}, + ], +) +def test_a_flag_without_a_usable_endpoint_list_is_reported(info: Mapping[str, object]) -> None: + warning: Final = warn_on_unenforced_endpoint_flag("embed-only", info) + assert warning is not None + assert "'embed-only'" in warning + + +@pytest.mark.parametrize( + "info", + [ + {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + {"enforce_supported_endpoints": True, "supported_endpoints": "/v1/embeddings"}, + {"enforce_supported_endpoints": False}, + {"supported_endpoints": ["/v1/embeddings", None]}, + ], +) +def test_an_enforceable_or_unflagged_deployment_is_not_reported(info: Mapping[str, object]) -> None: + assert warn_on_unenforced_endpoint_flag("embed-only", info) is None diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index ee3a70c7c1c..62510dac26d 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -12270,6 +12270,67 @@ 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_routable_deployments_leave_out_models_that_refuse_the_call_type() -> None: + router: Final = Router( + model_list=[ + { + "model_name": "available", + "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "unit-provider-key"}, + }, + { + "model_name": "embed-only", + "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "unit-provider-key"}, + "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + }, + ], + num_retries=0, + ) + + def routable(model: str, call_type: str | None) -> tuple[str, ...]: + return tuple( + deployment["model_name"] for deployment in router.get_routable_deployments(model, {}, call_type=call_type) + ) + + assert routable("unregistered-fallback", "acompletion") == () + assert routable("available", "acompletion") == ("available",) + assert routable("embed-only", "acompletion") == () + assert routable("embed-only", "aembedding") == ("embed-only",) + assert routable("embed-only", None) == ("embed-only",) + + def test_routing_groups_legacy_property_forwards_to_public_attribute(): router = _router_with_two_deployments([False, False]) @@ -18803,6 +18864,35 @@ async def test_anthropic_messages_stream_dropped_before_content_is_retried_withi assert [(attempted, budget) for _, attempted, budget in provider.calls] == [(0, 2), (1, 2)] +@pytest.mark.asyncio +async def test_anthropic_messages_same_group_retry_skips_an_embedding_only_deployment(): + """The retry of a /v1/messages stream dropped before content must not land on an enforced embedding-only + deployment of the same group, even when that deployment is the group's preferred order.""" + router: Final = Router( + model_list=[ + {"model_name": "glm", "litellm_params": {"model": "anthropic/glm-chat", "api_key": "sk-test", "order": 2}}, + { + "model_name": "glm", + "litellm_params": {"model": "anthropic/glm-embed", "api_key": "sk-test", "order": 1}, + "model_info": {"enforce_supported_endpoints": True, "supported_endpoints": ["/v1/embeddings"]}, + }, + ], + num_retries=1, + ) + provider: Final = _AnthropicMessagesScriptedProvider( + _anthropic_messages_dropped_before_content, _anthropic_messages_retried_stream + ) + + async def anthropic_messages(**kwargs: object) -> object: + return await provider(**kwargs) + + stream: Final = await _anthropic_messages_stream_through_router(router, anthropic_messages) + body: Final = [chunk async for chunk in stream] + + assert body == [_anthropic_messages_message_start_chunk(), _anthropic_messages_content_chunk("pong")] + assert [model for model, _, _ in provider.calls] == ["anthropic/glm-chat", "anthropic/glm-chat"] + + class _AnthropicMessagesRetryStashRecorder: def __init__(self) -> None: self.observed: tuple[Usage | None, ...] = () @@ -31431,3 +31521,26 @@ def test_router_keeps_supported_fallback_shapes( router: Final = Router(model_list=list(_FALLBACK_VALIDATION_MODEL_LIST), **router_kwargs) assert router.fallbacks == expected_fallbacks + + +def test_router_warns_when_a_deployment_enforces_endpoints_without_a_list(caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING, logger="LiteLLM Router"): + router: Final = Router( + model_list=[ + { + "model_name": "embed-from-config", + "litellm_params": {"model": "openai/text-embedding-3-small", "api_key": "sk-fake"}, + "model_info": {"enforce_supported_endpoints": True}, + } + ] + ) + router.add_deployment( + Deployment( + model_name="embed-from-api", + litellm_params=LiteLLM_Params(model="openai/text-embedding-3-small", api_key="sk-fake"), + model_info=ModelInfo(enforce_supported_endpoints=True, supported_endpoints=["/v1/embeddings", 7]), + ) + ) + + assert "Deployment 'embed-from-config' sets enforce_supported_endpoints without" in caplog.text + assert "Deployment 'embed-from-api' sets enforce_supported_endpoints without" in caplog.text diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index f114c1c73d2..6faffd7a050 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -50433,6 +50433,8 @@ export interface components { discoverable?: boolean | null; /** Enable Tag Filtering */ enable_tag_filtering?: boolean | null; + /** Enforce Supported Endpoints */ + enforce_supported_endpoints?: boolean | null; /** Explicit Cache Read Input Token Cost */ explicit_cache_read_input_token_cost?: number | null; /** Id */