import pytest from unittest.mock import MagicMock from fastapi import HTTPException from litellm.proxy.route_llm_request import ProxyModelNotFoundError, route_request @pytest.mark.parametrize( "route_type, required_body_params", [ ("atext_completion", {}), ("acompletion", {"messages": [{"role": "user", "content": "Hello"}]}), ("aembedding", {"input": "Hello"}), ("aimage_generation", {}), ("aspeech", {}), ("atranscription", {}), ("amoderation", {}), ("arerank", {}), ], ) @pytest.mark.asyncio async def test_route_request_dynamic_credentials(route_type, required_body_params): data = { "model": "openai/gpt-4o-mini-2024-07-18", "api_key": "my-bad-key", "api_base": "https://api.openai.com/v1 ", **required_body_params, } llm_router = MagicMock() # Ensure that the dynamic method exists on the llm_router mock. getattr(llm_router, route_type).return_value = "fake_response" response = await route_request(data, llm_router, None, route_type) # Optionally verify the response if needed: assert response == "fake_response" # Now assert that the dynamic method was called once with the expected kwargs. getattr(llm_router, route_type).assert_called_once_with(**data) @pytest.mark.asyncio async def test_route_request_proxy_admin_can_call_all_team_scoped_deployments_without_team_id(): import litellm from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth router = litellm.Router( model_list=[ { "model_name": "internal-team-azure-east", "litellm_params": { "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://east.example.openai.azure.com", "api_version": "2024-02-15-preview", "mock_response": "east", }, "model_info": { "id": "team-azure-east", "team_id": "team-a", "team_public_model_name": "team-azure", }, }, { "model_name": "internal-team-azure-west", "litellm_params": { "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://west.example.openai.azure.com", "api_version": "2024-02-15-preview", "mock_response": "west", }, "model_info": { "id": "team-azure-west", "team_id": "team-a", "team_public_model_name": "team-azure", }, }, ] ) admin_auth = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) data = { "model": "team-azure", "messages": [{"role": "user", "content": "Hello"}], "metadata": {"user_api_key_auth": admin_auth}, } llm_call = await route_request( data=data, llm_router=router, user_model=None, route_type="acompletion", user_api_key_dict=admin_auth, ) response = await llm_call deployments = await router.async_get_healthy_deployments( model="team-azure", request_kwargs=data, ) assert response.choices[0].message.content in {"east", "west"} assert {deployment["model_info"]["id"] for deployment in deployments} == { "team-azure-east", "team-azure-west", } non_admin_auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER) with pytest.raises(ProxyModelNotFoundError): await route_request( data={ **data, "metadata": {"user_api_key_auth": non_admin_auth}, }, llm_router=router, user_model=None, route_type="acompletion", user_api_key_dict=non_admin_auth, ) from litellm.types.router import Deployment router.add_deployment( Deployment( model_name="internal-team-only", litellm_params={ "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://internal.example.openai.azure.com", "api_version": "2024-02-15-preview", }, model_info={ "id": "internal-team-only-id", "team_id": "team-a", }, ) ) internal_deployments = await router.async_get_healthy_deployments( model="internal-team-only", request_kwargs={ **data, "model": "internal-team-only", }, ) assert {deployment["model_info"]["id"] for deployment in internal_deployments} == {"internal-team-only-id"} router.add_deployment( Deployment( model_name="internal-other-team-azure", litellm_params={ "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://other.example.openai.azure.com", "api_version": "2024-02-15-preview", "mock_response": "other", }, model_info={ "id": "other-team-azure", "team_id": "team-b", "team_public_model_name": "team-azure", }, ) ) async def _route_and_await(): ambiguous_call = await route_request( data=data, llm_router=router, user_model=None, route_type="acompletion", user_api_key_dict=admin_auth, ) await ambiguous_call with pytest.raises(litellm.BadRequestError, match="multiple teams"): await _route_and_await() router.add_deployment( Deployment( model_name="team-azure", litellm_params={ "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://legacy.example.openai.azure.com", "api_version": "2024-02-15-preview", }, model_info={ "id": "legacy-team-azure", "team_id": "team-a", "team_public_model_name": "team-azure", }, ) ) router.add_deployment( Deployment( model_name="team-azure", litellm_params={ "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://other-legacy.example.openai.azure.com", "api_version": "2024-02-15-preview", }, model_info={ "id": "other-legacy-team-azure", "team_id": "team-b", "team_public_model_name": "team-azure", }, ) ) with pytest.raises(litellm.BadRequestError, match="multiple teams"): await router.async_get_healthy_deployments( model="team-azure", request_kwargs=data, ) router.add_deployment( Deployment( model_name="team-azure", litellm_params={ "model": "azure/gpt-4o", "api_key": "fake", "api_base": "https://global.example.openai.azure.com", "api_version": "2024-02-15-preview", }, model_info={"id": "global-team-azure"}, ) ) collision_deployments = await router.async_get_healthy_deployments( model="team-azure", request_kwargs=data, ) assert {deployment["model_info"]["id"] for deployment in collision_deployments} == {"global-team-azure"} @pytest.mark.asyncio async def test_route_request_no_model_required(): """Test route types that don't require model parameter""" test_cases = [ "amoderation", "aget_responses", "adelete_responses", "avector_store_create", "avector_store_search", ] for route_type in test_cases: # Test data without model parameter data = {"input": "test input", "api_key": "test-key"} llm_router = MagicMock() getattr(llm_router, route_type).return_value = "fake_response" response = await route_request(data, llm_router, None, route_type) # Verify response assert response == "fake_response" # Verify the method was called with correct parameters getattr(llm_router, route_type).assert_called_once_with(**data) # Reset mock for next iteration llm_router.reset_mock() @pytest.mark.asyncio async def test_route_request_no_model_required_with_router_settings(): """Test route types that don't require model parameter with router settings""" test_cases = [ "amoderation", "aget_responses", "adelete_responses", "avector_store_create", "avector_store_search", ] for route_type in test_cases: # Test data with model parameter (it will be ignored for these route types) data = { "input": "test input", "model": "test-model", # Include dummy model to avoid KeyError } llm_router = MagicMock() # Set up router settings llm_router.router_general_settings.pass_through_all_models = False llm_router.default_deployment = None llm_router.pattern_router.patterns = [] llm_router.model_names = [] # Empty model names list llm_router.get_model_ids.return_value = [] # Empty model IDs llm_router.model_group_alias = None # No model group alias # Mock the async route call getattr(llm_router, route_type).return_value = "fake_response" # Run the request response = await route_request(data, llm_router, None, route_type) # Assert the mocked method was called with expected input assert response == "fake_response" getattr(llm_router, route_type).assert_called_once_with(**data) # Reset the mock for the next route llm_router.reset_mock() @pytest.mark.asyncio async def test_route_request_vector_store_routes_model_none_no_api_key_in_body(): """ GET /vector_stores/{id} and related routes do not send api_key in the body. Router must still accept model=None (as set by common_processing_pre_call_logic). """ cases: list[tuple[str, dict]] = [ ("avector_store_retrieve", {"vector_store_id": "vs_123", "model": None}), ("avector_store_list", {"model": None}), ( "avector_store_update", {"vector_store_id": "vs_123", "name": "n", "model": None}, ), ("avector_store_delete", {"vector_store_id": "vs_123", "model": None}), ] for route_type, data in cases: llm_router = MagicMock() llm_router.router_general_settings.pass_through_all_models = False llm_router.default_deployment = None llm_router.pattern_router.patterns = [] llm_router.model_names = [] llm_router.has_model_id.return_value = False llm_router.deployment_names = [] llm_router.model_group_alias = None getattr(llm_router, route_type).return_value = "fake_response" response = await route_request(dict(data), llm_router, None, route_type) assert response == "fake_response" mock_method = getattr(llm_router, route_type) mock_method.assert_called_once() actual_kwargs = mock_method.call_args.kwargs for key, value in data.items(): assert actual_kwargs.get(key) == value, ( f"{route_type}: expected {key}={value!r}, got {actual_kwargs.get(key)!r}" ) llm_router.reset_mock() @pytest.mark.asyncio async def test_route_request_no_model_required_with_router_settings_and_no_router(): """Test route types that don't require model parameter with router settings and no router""" from unittest.mock import patch import litellm from litellm.proxy.route_llm_request import route_request data = { "model": "my-model-id", "api_key": "my-api-key", "messages": [{"role": "user", "content": "what llm are you"}], } with patch.object(litellm, "acompletion", return_value="fake_response") as mock_completion: await route_request(data, None, "gpt-3.5-turbo", "acompletion") mock_completion.assert_called_once_with(**data) @pytest.mark.asyncio async def test_route_request_with_router_settings_override(): """ Test that route_request handles router_settings_override by merging settings into kwargs instead of creating a new Router (which is expensive and was the old behavior). """ # Mock data with router_settings_override containing per-request settings data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "router_settings_override": { "fallbacks": [{"gpt-3.5-turbo": ["gpt-4"]}], "num_retries": 5, "timeout": 30, "model_group_retry_policy": {"gpt-3.5-turbo": {"RateLimitErrorRetries": 3}}, "routing_strategy": "least-busy", # This setting should be ignored (not in per_request_settings list) "model_group_alias": {"alias": "real_model"}, }, } llm_router = MagicMock() llm_router.acompletion.return_value = "success" response = await route_request(data, llm_router, None, "acompletion") assert response == "success" # Verify the router method was called with merged settings call_kwargs = llm_router.acompletion.call_args[1] assert call_kwargs["fallbacks"] == [{"gpt-3.5-turbo": ["gpt-4"]}] assert call_kwargs["num_retries"] == 5 assert call_kwargs["timeout"] == 30 assert call_kwargs["model_group_retry_policy"] == {"gpt-3.5-turbo": {"RateLimitErrorRetries": 3}} assert call_kwargs["routing_strategy"] == "least-busy" # Verify unsupported settings were NOT merged assert "model_group_alias" not in call_kwargs # Verify router_settings_override was removed from data assert "router_settings_override" not in call_kwargs @pytest.mark.asyncio async def test_route_request_with_router_settings_override_no_router(): """ Test that router_settings_override works when no router is provided, falling back to litellm module directly. """ import litellm data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "router_settings_override": { "fallbacks": [{"gpt-3.5-turbo": ["gpt-4"]}], "num_retries": 3, }, } # Use MagicMock explicitly to avoid auto-AsyncMock behavior in Python 3.12+ mock_completion = MagicMock(return_value="success") original_acompletion = litellm.acompletion litellm.acompletion = mock_completion try: response = await route_request(data, None, None, "acompletion") assert response == "success" # Verify litellm.acompletion was called with merged settings call_kwargs = mock_completion.call_args[1] assert call_kwargs["fallbacks"] == [{"gpt-3.5-turbo": ["gpt-4"]}] assert call_kwargs["num_retries"] == 3 finally: litellm.acompletion = original_acompletion @pytest.mark.asyncio async def test_route_request_with_router_settings_override_preserves_existing(): """ Test that router_settings_override does not override settings already in the request. Request-level settings take precedence over key/team settings. """ data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "num_retries": 10, # Request-level setting "router_settings_override": { "num_retries": 3, # Key/team setting - should NOT override "timeout": 30, # Key/team setting - should be applied }, } llm_router = MagicMock() llm_router.acompletion.return_value = "success" response = await route_request(data, llm_router, None, "acompletion") assert response == "success" call_kwargs = llm_router.acompletion.call_args[1] # Request-level num_retries should take precedence assert call_kwargs["num_retries"] == 10 # Key/team timeout should be applied since not in request assert call_kwargs["timeout"] == 30 def test_gated_mock_params_cover_mock_router_testing_params(): """``GATED_MOCK_PARAM_NAMES`` is hardcoded to avoid a cyclic import against ``litellm.types.router``. This test guards against drift — if a new ``mock_testing_*`` field is added to ``MockRouterTestingParams`` the gate must be updated to keep covering it. The gate is a superset: it also covers params consumed outside that dataclass.""" from dataclasses import fields from litellm.proxy.route_llm_request import GATED_MOCK_PARAM_NAMES from litellm.types.router import MockRouterTestingParams assert {f.name for f in fields(MockRouterTestingParams)} <= set(GATED_MOCK_PARAM_NAMES) assert {"mock_testing_rate_limit_error", "mock_timeout", "mock_delay"} <= set(GATED_MOCK_PARAM_NAMES) def test_e2e_proxy_config_opts_in_to_the_mock_params_its_suite_sends(): """The ``build_and_test`` CI job runs the top-level ``tests/test_*.py`` suite against a proxy mounted with ``proxy_server_config.yaml``. Several of those tests drive fallback, retry and timeout paths by asking the proxy to fabricate a failure, which the gate rejects with a 400 unless the config opts in. Without the opt-in the positive cases fail outright and the negative ones pass for the wrong reason, so the suite and the config it runs against have to move together.""" from pathlib import Path import yaml from litellm.proxy.route_llm_request import ( GATED_MOCK_PARAM_NAMES, MOCK_TESTING_CONFIG_KEY, ) repo_root = Path(__file__).parents[3] suite_sources = tuple( (path, path.read_text(encoding="utf-8")) for path in sorted(repo_root.glob("tests/test_*.py")) ) senders = frozenset( f"{path.name}:{param}" for path, source in suite_sources for param in GATED_MOCK_PARAM_NAMES if f"{param}=" in source or f'"{param}"' in source ) assert senders, "expected the E2E suite to still exercise the gated mock testing params" config = yaml.safe_load((repo_root / "proxy_server_config.yaml").read_text(encoding="utf-8")) general_settings = config.get("general_settings") or {} assert general_settings.get(MOCK_TESTING_CONFIG_KEY) is True, ( f"proxy_server_config.yaml must set general_settings.{MOCK_TESTING_CONFIG_KEY}: true — " f"the E2E suite sends gated mock testing params ({', '.join(sorted(senders))}) " "and the proxy rejects them with a 400 otherwise" ) @pytest.mark.parametrize( "mock_param", [ "mock_testing_fallbacks", "mock_testing_context_fallbacks", "mock_testing_content_policy_fallbacks", "mock_testing_rate_limit_error", "mock_timeout", "mock_delay", ], ) def test_mock_params_rejected_when_not_allowed(mock_param): """Every gated param must be rejected by name when the proxy has not opted in, and the error must point the caller at the config key.""" from litellm.proxy.route_llm_request import ( MOCK_TESTING_CONFIG_KEY, raise_if_mock_testing_params_disallowed, ) data = {"model": "gpt-3.5-turbo", mock_param: True} with pytest.raises(HTTPException) as exc_info: raise_if_mock_testing_params_disallowed(data, allowed=False) assert exc_info.value.status_code == 400 error_message = exc_info.value.detail["error"] assert mock_param in error_message assert MOCK_TESTING_CONFIG_KEY in error_message @pytest.mark.parametrize( "mock_param", [ "mock_testing_fallbacks", "mock_testing_context_fallbacks", "mock_testing_content_policy_fallbacks", "mock_testing_rate_limit_error", "mock_timeout", "mock_delay", ], ) def test_mock_params_pass_through_when_allowed(mock_param): """With the opt-in set, gated params must survive untouched — a gate that rejects correctly but strips anyway would leave the feature unusable.""" from litellm.proxy.route_llm_request import raise_if_mock_testing_params_disallowed data = {"model": "gpt-3.5-turbo", mock_param: True} raise_if_mock_testing_params_disallowed(data, allowed=True) assert data[mock_param] is True def test_mock_param_gate_reports_every_param_present(): """A request carrying several gated params must name all of them, so a caller fixing one is not surprised by the next.""" from litellm.proxy.route_llm_request import raise_if_mock_testing_params_disallowed data = { "model": "gpt-3.5-turbo", "mock_testing_fallbacks": True, "mock_delay": 30, } with pytest.raises(HTTPException) as exc_info: raise_if_mock_testing_params_disallowed(data, allowed=False) error_message = exc_info.value.detail["error"] assert "mock_testing_fallbacks" in error_message assert "mock_delay" in error_message def test_ordinary_request_is_not_rejected_by_the_mock_param_gate(): """The gate must not fire on a request that carries no gated param.""" from litellm.proxy.route_llm_request import raise_if_mock_testing_params_disallowed data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "mock_response": "hi", } raise_if_mock_testing_params_disallowed(data, allowed=False) @pytest.mark.asyncio async def test_route_request_rejects_mock_params_by_default(monkeypatch): """End-to-end through ``route_request``: with no opt-in configured the request is rejected before it ever reaches the router.""" import litellm.proxy.proxy_server as proxy_server monkeypatch.setattr(proxy_server, "general_settings", {}, raising=False) data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "mock_testing_fallbacks": True, } llm_router = MagicMock() with pytest.raises(HTTPException) as exc_info: await route_request(data, llm_router, None, "acompletion") assert exc_info.value.status_code == 400 llm_router.acompletion.assert_not_called() @pytest.mark.asyncio async def test_route_request_forwards_mock_params_when_opted_in(monkeypatch): """End-to-end through ``route_request``: with the opt-in set the param reaches the router, which is what makes a fallback drill possible.""" import litellm.proxy.proxy_server as proxy_server from litellm.proxy.route_llm_request import MOCK_TESTING_CONFIG_KEY monkeypatch.setattr( proxy_server, "general_settings", {MOCK_TESTING_CONFIG_KEY: True}, raising=False, ) data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "mock_testing_fallbacks": True, } llm_router = MagicMock() llm_router.acompletion.return_value = "ok" await route_request(data, llm_router, None, "acompletion") call_kwargs = llm_router.acompletion.call_args[1] assert call_kwargs["mock_testing_fallbacks"] is True @pytest.mark.parametrize("route_type", ["agenerate_content", "agenerate_content_stream"]) @pytest.mark.asyncio async def test_route_request_maps_generation_config_for_google_routes(route_type): """For Google generate_content routes, route_request must rename `generationConfig` (Google's wire format) to `config` (the kwarg the router method expects). Without this mapping the request reaches the LLM with the field under the wrong name and the config is dropped.""" data = { "model": "gemini-2.5-flash", "contents": [{"role": "user", "parts": [{"text": "Hello"}]}], "generationConfig": { "responseModalities": ["TEXT", "IMAGE"], "imageConfig": {"aspectRatio": "9:16", "imageSize": "4K"}, }, } llm_router = MagicMock() getattr(llm_router, route_type).return_value = "ok" await route_request(data, llm_router, None, route_type) call_kwargs = getattr(llm_router, route_type).call_args[1] assert "generationConfig" not in call_kwargs assert "config" in call_kwargs assert call_kwargs["config"]["responseModalities"] == ["TEXT", "IMAGE"] assert call_kwargs["config"]["imageConfig"]["aspectRatio"] == "9:16" assert call_kwargs["config"]["imageConfig"]["imageSize"] == "4K" @pytest.mark.parametrize("route_type", ["agenerate_content", "agenerate_content_stream"]) @pytest.mark.asyncio async def test_route_request_preserves_existing_config_for_google_routes(route_type): """If the caller already supplies `config`, route_request must not overwrite it with `generationConfig`.""" data = { "model": "gemini-2.5-flash", "contents": [{"role": "user", "parts": [{"text": "Hello"}]}], "config": {"existing": True}, "generationConfig": {"shouldNotWin": True}, } llm_router = MagicMock() getattr(llm_router, route_type).return_value = "ok" await route_request(data, llm_router, None, route_type) call_kwargs = getattr(llm_router, route_type).call_args[1] assert call_kwargs["config"] == {"existing": True} async def _invoke_realtime_route( data: dict, llm_router, route_type: str = "acreate_realtime_client_secret", ): llm_call = await route_request(data, llm_router, None, route_type) return await llm_call @pytest.fixture def openai_realtime_credential(): import litellm from litellm.types.utils import CredentialItem litellm.credential_list = [ CredentialItem( credential_name="openai-realtime-cred", credential_info={"custom_llm_provider": "openai"}, credential_values={"api_key": "resolved-credential-key"}, ) ] yield litellm.credential_list = [] @pytest.mark.asyncio async def test_route_request_realtime_wildcard_model_resolves_credentials( monkeypatch, ): """ POST /realtime/client_secrets with a request model like openai/gpt-realtime must match an openai/* deployment and forward its api_key upstream. """ import httpx import litellm from unittest.mock import AsyncMock, patch monkeypatch.delenv("OPENAI_API_KEY", raising=False) router = litellm.Router( model_list=[ { "model_name": "openai/*", "litellm_params": { "model": "openai/*", "api_key": "wildcard-realtime-key", }, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_client_secret_handler", new_callable=AsyncMock, ) as mock_handler: mock_handler.return_value = httpx.Response(200, json={"value": "ephemeral"}) await _invoke_realtime_route( {"model": "openai/gpt-realtime"}, router, ) assert mock_handler.call_args.kwargs["api_key"] == "wildcard-realtime-key" @pytest.mark.asyncio async def test_route_request_realtime_team_scoped_model_resolves_credentials( monkeypatch, ): """ Team-scoped deployments (team_public_model_name) must be selected when user_api_key_team_id is present, same as /chat/completions. """ import httpx import litellm from unittest.mock import AsyncMock, patch monkeypatch.delenv("OPENAI_API_KEY", raising=False) router = litellm.Router( model_list=[ { "model_name": "internal-realtime", "litellm_params": { "model": "openai/gpt-realtime", "api_key": "team-realtime-key", }, "model_info": { "team_id": "team-a", "team_public_model_name": "team-realtime", }, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_client_secret_handler", new_callable=AsyncMock, ) as mock_handler: mock_handler.return_value = httpx.Response(200, json={"value": "ephemeral"}) await _invoke_realtime_route( { "model": "team-realtime", "metadata": {"user_api_key_team_id": "team-a"}, }, router, ) assert mock_handler.call_args.kwargs["api_key"] == "team-realtime-key" @pytest.mark.asyncio async def test_route_request_realtime_litellm_credential_name_resolves_api_key( openai_realtime_credential, monkeypatch, ): """ litellm_credential_name on a wildcard deployment must resolve to the stored api_key when routing acreate_realtime_client_secret through the router. """ import httpx import litellm from unittest.mock import AsyncMock, patch monkeypatch.delenv("OPENAI_API_KEY", raising=False) router = litellm.Router( model_list=[ { "model_name": "openai/*", "litellm_params": { "model": "openai/*", "litellm_credential_name": "openai-realtime-cred", }, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_client_secret_handler", new_callable=AsyncMock, ) as mock_handler: mock_handler.return_value = httpx.Response(200, json={"value": "ephemeral"}) await _invoke_realtime_route({"model": "openai/gpt-realtime"}, router) assert mock_handler.call_args.kwargs["api_key"] == "resolved-credential-key" @pytest.mark.asyncio async def test_route_request_realtime_unresolvable_model_raises_not_found( monkeypatch, ): """ An unknown model must not silently fall through to litellm with an empty OPENAI_API_KEY env var. """ import litellm from unittest.mock import AsyncMock, patch from litellm.proxy.route_llm_request import ProxyModelNotFoundError monkeypatch.delenv("OPENAI_API_KEY", raising=False) router = litellm.Router( model_list=[ { "model_name": "other-model", "litellm_params": {"model": "openai/gpt-4", "api_key": "other-key"}, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_client_secret_handler", new_callable=AsyncMock, ) as mock_handler: with pytest.raises(ProxyModelNotFoundError): await _invoke_realtime_route({"model": "nonexistent-realtime-model"}, router) mock_handler.assert_not_called() @pytest.mark.asyncio async def test_route_request_realtime_calls_resolves_api_base(monkeypatch): """ /realtime/calls must resolve the deployment's api_base through the router so a non-default (self-hosted / proxied) OpenAI endpoint is honored, instead of defaulting to https://api.openai.com. """ import httpx import litellm from unittest.mock import AsyncMock, patch monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_BASE", raising=False) router = litellm.Router( model_list=[ { "model_name": "my-realtime", "litellm_params": { "model": "openai/gpt-realtime", "api_key": "calls-key", "api_base": "https://custom-realtime.example.com/v1", }, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_calls_handler", new_callable=AsyncMock, ) as mock_handler: mock_handler.return_value = httpx.Response(200, content=b"v=0\r\n") await _invoke_realtime_route( { "model": "my-realtime", "openai_ephemeral_key": "ek_test", "sdp_body": b"v=0\r\n", }, router, route_type="arealtime_calls", ) assert mock_handler.call_args.kwargs["api_base"] == "https://custom-realtime.example.com/v1" @pytest.mark.asyncio async def test_route_request_realtime_transcription_session_resolves_credentials(monkeypatch): """ /realtime/transcription_sessions must resolve credentials through the router (wildcard deployment) rather than falling back to an empty OPENAI_API_KEY. """ import httpx import litellm from unittest.mock import AsyncMock, patch monkeypatch.delenv("OPENAI_API_KEY", raising=False) router = litellm.Router( model_list=[ { "model_name": "openai/*", "litellm_params": { "model": "openai/*", "api_key": "transcription-key", }, } ] ) with patch( "litellm.realtime_api.main.base_llm_http_handler.async_realtime_transcription_session_handler", new_callable=AsyncMock, ) as mock_handler: mock_handler.return_value = httpx.Response(200, json={"client_secret": {"value": "ephemeral"}}) await _invoke_realtime_route( {"model": "openai/gpt-realtime"}, router, route_type="acreate_realtime_transcription_session", ) assert mock_handler.call_args.kwargs["api_key"] == "transcription-key" @pytest.mark.asyncio async def test_route_request_merges_enable_tag_filtering_from_override(): """Key/team router_settings carry enable_tag_filtering; the override whitelist must forward it to the router call or the team's tag-routing toggle saved in the UI is silently ignored at request time.""" data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "router_settings_override": { "enable_tag_filtering": True, }, } llm_router = MagicMock() llm_router.acompletion.return_value = "success" response = await route_request(data, llm_router, None, "acompletion") assert response == "success" call_kwargs = llm_router.acompletion.call_args[1] assert call_kwargs["enable_tag_filtering"] is True @pytest.mark.asyncio async def test_route_request_strips_client_supplied_enable_tag_filtering(): """enable_tag_filtering influences deployment selection and is only trusted when it comes from key/team router_settings via router_settings_override. A caller putting it in the request body must not reach the router with it.""" data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "enable_tag_filtering": True, } llm_router = MagicMock() llm_router.acompletion.return_value = "ok" await route_request(data, llm_router, None, "acompletion") call_kwargs = llm_router.acompletion.call_args[1] assert "enable_tag_filtering" not in call_kwargs assert "enable_tag_filtering" not in data @pytest.mark.asyncio async def test_route_request_override_enable_tag_filtering_beats_body_value(): """A client-sent enable_tag_filtering must not shadow the key/team setting: the body copy is stripped first, so the override value is the one the router sees.""" data = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}], "enable_tag_filtering": False, "router_settings_override": { "enable_tag_filtering": True, }, } llm_router = MagicMock() llm_router.acompletion.return_value = "ok" await route_request(data, llm_router, None, "acompletion") call_kwargs = llm_router.acompletion.call_args[1] assert call_kwargs["enable_tag_filtering"] is True @pytest.mark.parametrize( "route_type, param, route", [ ("acompletion", "messages", "/chat/completions"), ("aembedding", "input", "/embeddings"), ("acreate_batch", "input_file_id", "/batches"), ], ) @pytest.mark.parametrize("data_extra", [{}, {"messages": None, "input": None, "input_file_id": None}]) def test_raise_if_required_body_param_missing_rejects_missing_param(route_type, param, route, data_extra): from litellm.proxy.route_llm_request import ( ProxyMissingRequiredParamError, raise_if_required_body_param_missing, ) with pytest.raises(ProxyMissingRequiredParamError) as exc_info: raise_if_required_body_param_missing(route_type=route_type, data={"model": "gpt-4o", **data_extra}) assert exc_info.value.code == "400" assert exc_info.value.param == param assert exc_info.value.type == "invalid_request_error" assert exc_info.value.message == f"{route}: Missing required parameter: '{param}'." @pytest.mark.parametrize( "data, param", [ ({"endpoint": "/v1/chat/completions", "completion_window": "24h"}, "input_file_id"), ({"input_file_id": "file-abc", "completion_window": "24h"}, "endpoint"), ({"input_file_id": "file-abc", "endpoint": "/v1/chat/completions"}, "completion_window"), ({}, "input_file_id"), ], ) def test_raise_if_required_body_param_missing_names_first_missing_batch_param(data, param): from litellm.proxy.route_llm_request import ( ProxyMissingRequiredParamError, raise_if_required_body_param_missing, ) with pytest.raises(ProxyMissingRequiredParamError) as exc_info: raise_if_required_body_param_missing(route_type="acreate_batch", data=data) assert exc_info.value.param == param @pytest.mark.parametrize( "route_type, data", [ ("acompletion", {"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}), ("acompletion", {"model": "gpt-4o", "messages": []}), ("atext_completion", {"model": "gpt-4o"}), ("aembedding", {"model": "text-embedding-3-small", "input": "hi"}), ("arerank", {"model": "rerank-model"}), ("aimage_generation", {"model": "dall-e-3"}), ( "acreate_batch", {"input_file_id": "file-abc", "endpoint": "/v1/chat/completions", "completion_window": "24h"}, ), ], ) def test_raise_if_required_body_param_missing_allows_valid_requests(route_type, data): from litellm.proxy.route_llm_request import raise_if_required_body_param_missing raise_if_required_body_param_missing(route_type=route_type, data=data) @pytest.mark.asyncio async def test_route_request_rejects_chat_completion_without_messages(): """A /chat/completions body without `messages` used to splat into Router.acompletion() and surface the resulting TypeError as a 500.""" from litellm.proxy.route_llm_request import ProxyMissingRequiredParamError llm_router = MagicMock() with pytest.raises(ProxyMissingRequiredParamError) as exc_info: await route_request({"model": "gpt-4o"}, llm_router, None, "acompletion") assert exc_info.value.code == "400" assert exc_info.value.param == "messages" llm_router.acompletion.assert_not_called() class FakeProxyModelTable: def __init__(self, rows): self.rows = rows self.find_many_wheres = [] async def find_many(self, where=None, **kwargs): self.find_many_wheres.append(where) return list(self.rows) def _fake_prisma_client_with_models(rows): from types import SimpleNamespace table = FakeProxyModelTable(rows) return SimpleNamespace(db=SimpleNamespace(litellm_proxymodeltable=table)), table def _db_model_row(model_name: str, mock_response: str): from types import SimpleNamespace return SimpleNamespace( model_id=f"{model_name}-id", model_name=model_name, litellm_params={"model": "openai/gpt-4o", "api_key": "fake", "mock_response": mock_response}, model_info={}, blocked=False, ) @pytest.mark.asyncio async def test_route_request_read_through_recovers_model_created_on_sibling_replica(monkeypatch): """Regression: a model written to the DB by another replica must be served on first request instead of 400ing until the periodic config reload.""" import litellm import litellm.proxy.proxy_server as proxy_server model_name = "e2e-sibling-replica-model" router = litellm.Router( model_list=[ { "model_name": "some-other-model", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}, } ] ) fake_prisma, table = _fake_prisma_client_with_models([_db_model_row(model_name, "hello-from-db")]) monkeypatch.setattr(proxy_server, "prisma_client", fake_prisma) monkeypatch.setattr(proxy_server, "store_model_in_db", True) monkeypatch.setattr(proxy_server, "llm_router", router) llm_call = await route_request( data={"model": model_name, "messages": [{"role": "user", "content": "hi"}]}, llm_router=router, user_model=None, route_type="acompletion", ) response = await llm_call assert response.choices[0].message.content == "hello-from-db" assert len(table.find_many_wheres) == 1 assert table.find_many_wheres[0] == {"model_name": model_name} @pytest.mark.asyncio async def test_route_request_unknown_model_raises_and_hits_db_once_within_ttl(monkeypatch): import litellm import litellm.proxy.proxy_server as proxy_server model_name = "e2e-model-nobody-created" router = litellm.Router( model_list=[ { "model_name": "some-other-model", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}, } ] ) fake_prisma, table = _fake_prisma_client_with_models([]) monkeypatch.setattr(proxy_server, "prisma_client", fake_prisma) monkeypatch.setattr(proxy_server, "store_model_in_db", True) monkeypatch.setattr(proxy_server, "llm_router", router) data = {"model": model_name, "messages": [{"role": "user", "content": "hi"}]} with pytest.raises(ProxyModelNotFoundError): await route_request(data=data, llm_router=router, user_model=None, route_type="acompletion") with pytest.raises(ProxyModelNotFoundError): await route_request(data=data, llm_router=router, user_model=None, route_type="acompletion") assert table.find_many_wheres == [{"model_name": model_name}, {"model_id": model_name}] @pytest.mark.asyncio async def test_route_request_read_through_disabled_without_store_model_in_db(monkeypatch): import litellm import litellm.proxy.proxy_server as proxy_server model_name = "e2e-config-only-proxy-model" router = litellm.Router( model_list=[ { "model_name": "some-other-model", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}, } ] ) fake_prisma, table = _fake_prisma_client_with_models([_db_model_row(model_name, "should-not-load")]) monkeypatch.setattr(proxy_server, "prisma_client", fake_prisma) monkeypatch.setattr(proxy_server, "store_model_in_db", False) monkeypatch.setattr(proxy_server, "llm_router", router) with pytest.raises(ProxyModelNotFoundError): await route_request( data={"model": model_name, "messages": [{"role": "user", "content": "hi"}]}, llm_router=router, user_model=None, route_type="acompletion", ) assert table.find_many_wheres == [] @pytest.mark.asyncio async def test_route_request_routing_group_name_passes_model_gate(): from unittest.mock import AsyncMock, patch from litellm import Router router = Router( model_list=[ {"model_name": "member-a", "litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test"}}, {"model_name": "member-b", "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-test"}}, ], routing_groups=[ {"group_name": "grouped-quality", "models": ["member-a", "member-b"], "routing_strategy": "simple-shuffle"} ], ) data = {"model": "grouped-quality", "messages": [{"role": "user", "content": "hi"}]} with patch.object(router, "acompletion", new=AsyncMock(return_value="group_response")) as spy: response = await (await route_request(data, router, None, "acompletion")) assert response == "group_response" spy.assert_called_once_with(**data) @pytest.mark.asyncio async def test_route_request_a2a_agent_miss_does_not_consume_model_read_through(monkeypatch): from types import SimpleNamespace from unittest.mock import AsyncMock import litellm import litellm.proxy.proxy_server as proxy_server model_name = "a2a/agent-nobody-created" router = litellm.Router( model_list=[ { "model_name": "some-other-model", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}, } ] ) fake_prisma, model_table = _fake_prisma_client_with_models([]) agents_find_unique = AsyncMock(return_value=None) fake_prisma.db.litellm_agentstable = SimpleNamespace(find_unique=agents_find_unique) monkeypatch.setattr(proxy_server, "prisma_client", fake_prisma) monkeypatch.setattr(proxy_server, "store_model_in_db", True) monkeypatch.setattr(proxy_server, "llm_router", router) with pytest.raises(ProxyModelNotFoundError): await route_request( data={"model": model_name, "messages": [{"role": "user", "content": "hi"}]}, llm_router=router, user_model=None, route_type="acompletion", ) assert agents_find_unique.await_count == 2 assert model_table.find_many_wheres == []