mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
6a0d03914c
* test: drop the cwd-relative sys.path.insert calls from the test suite
TQ003 stands at 1,077 across 1,058 files, and 1,015 of them are the same shape:
sys.path.insert(0, os.path.abspath("../..")) and its deeper siblings. The
argument resolves against the working directory rather than the file, so from
the repo root, where every job runs pytest, it inserts the directory two levels
above the checkout. It has never pointed at litellm. The package is installed
into the environment anyway, which is what actually makes the import work, and
what the rule's message has said all along.
Removing them leaves 1,634 imports of sys and os with no remaining reference,
and those go too, except where another test module imports the name back out of
the file. The rest of TQ003 is 62 call sites that resolve against __file__ or a
variable, which are a different question and are left alone.
Collection is identical either way: 45,871 tests and the same 51 pre-existing
collection errors before and after, and ruff reports no new undefined name.
* test: drop the duplicate imports the sys.path sweep exposed to F811
* test(pre-call-utils): restore the os import the new bedrock tests need
1299 lines
45 KiB
Python
1299 lines
45 KiB
Python
|
|
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 == []
|