mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-05 07:37:50 +00:00
* fix(proxy): list slash-less wildcard groups like claude-* on /v1/models and /model/info * fix(proxy): expand team-owned wildcard rows on /model/info through their public name * fix(proxy): keep listing wildcard names wider than their deployment and skip the Azure placeholder A group like *gemini-3* over gemini/gemini-3* lists the catalog ids the name matches and routes back, as main did on /model/info, now on /v1/models too. Wildcard listings no longer list the Azure-LLM placeholder. Team-owned wildcard rows on /model/info go back to main's behavior. * fix(proxy): keep Azure listings on slashed wildcards and list bare deployments through custom_llm_provider get_provider_models returns main's list again, and only slash-less names skip the Azure-LLM catalog placeholder. A slash-less name over a bare wildcard deployment now lists the catalog of its custom_llm_provider. * test(proxy): pin custom_llm_provider guards and slashed wider-name listing * fix(proxy): list bare ids for wider names over a custom_llm_provider deployment --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
2424 lines
89 KiB
Python
2424 lines
89 KiB
Python
import datetime
|
|
import ipaddress
|
|
import json
|
|
import ssl
|
|
import threading
|
|
from collections.abc import Generator, Iterator, Mapping
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Final
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
from litellm import Router
|
|
from litellm.utils import AZURE_MODEL_PLACEHOLDER
|
|
|
|
|
|
def test_get_team_models_for_all_models_and_team_only_models():
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
|
|
team_models = ["all-proxy-models", "team-only-model", "team-only-model-2"]
|
|
proxy_model_list = ["model1", "model2", "model3"]
|
|
model_access_groups = {}
|
|
include_model_access_groups = False
|
|
|
|
result: Final = get_team_models(team_models, proxy_model_list, model_access_groups, include_model_access_groups)
|
|
combined_models = team_models + proxy_model_list
|
|
assert set(result) == set(combined_models)
|
|
|
|
|
|
def test_get_team_models_all_proxy_models_includes_access_groups():
|
|
"""
|
|
When a team has 'all-proxy-models' and include_model_access_groups=True,
|
|
the result should include model access group names (e.g. 'claude-model-group')
|
|
in addition to individual model names.
|
|
"""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
|
|
team_models = ["all-proxy-models"]
|
|
proxy_model_list = ["model1", "model2"]
|
|
model_access_groups = {
|
|
"group-a": ["model1"],
|
|
"group-b": ["model2"],
|
|
}
|
|
|
|
result = get_team_models(
|
|
team_models,
|
|
proxy_model_list,
|
|
model_access_groups,
|
|
include_model_access_groups=True,
|
|
)
|
|
assert "group-a" in result
|
|
assert "group-b" in result
|
|
assert "model1" in result
|
|
assert "model2" in result
|
|
assert len(result) == len(set(result)), "result should have no duplicates"
|
|
|
|
|
|
def test_get_team_models_all_proxy_models_without_include_flag():
|
|
"""
|
|
When include_model_access_groups=False, access group names should NOT
|
|
appear in the result even with 'all-proxy-models'.
|
|
"""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
|
|
team_models = ["all-proxy-models"]
|
|
proxy_model_list = ["model1", "model2"]
|
|
model_access_groups = {
|
|
"group-a": ["model1"],
|
|
"group-b": ["model2"],
|
|
}
|
|
|
|
result = get_team_models(
|
|
team_models,
|
|
proxy_model_list,
|
|
model_access_groups,
|
|
include_model_access_groups=False,
|
|
)
|
|
assert "group-a" not in result
|
|
assert "group-b" not in result
|
|
assert "model1" in result
|
|
assert "model2" in result
|
|
|
|
|
|
def test_get_key_models_all_proxy_models_includes_access_groups():
|
|
"""
|
|
When a key has 'all-proxy-models' and include_model_access_groups=True,
|
|
the result should include model access group names.
|
|
"""
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
|
|
user_api_key_dict = UserAPIKeyAuth(
|
|
models=["all-proxy-models"],
|
|
api_key="test-key",
|
|
)
|
|
proxy_model_list = ["model1", "model2"]
|
|
model_access_groups = {
|
|
"group-a": ["model1"],
|
|
}
|
|
|
|
result = get_key_models(
|
|
user_api_key_dict=user_api_key_dict,
|
|
proxy_model_list=proxy_model_list,
|
|
model_access_groups=model_access_groups,
|
|
include_model_access_groups=True,
|
|
)
|
|
assert "group-a" in result
|
|
assert "model1" in result
|
|
assert "model2" in result
|
|
assert len(result) == len(set(result)), "result should have no duplicates"
|
|
|
|
|
|
def test_get_key_models_passes_include_model_access_groups():
|
|
"""
|
|
When a key explicitly has an access group name in its models list and
|
|
include_model_access_groups=True, the group name should be retained
|
|
(not stripped by _get_models_from_access_groups).
|
|
"""
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
|
|
user_api_key_dict = UserAPIKeyAuth(
|
|
models=["group-a"],
|
|
api_key="test-key",
|
|
)
|
|
proxy_model_list = ["model1", "model2"]
|
|
model_access_groups = {
|
|
"group-a": ["model1", "model2"],
|
|
}
|
|
|
|
result = get_key_models(
|
|
user_api_key_dict=user_api_key_dict,
|
|
proxy_model_list=proxy_model_list,
|
|
model_access_groups=model_access_groups,
|
|
include_model_access_groups=True,
|
|
)
|
|
assert "group-a" in result
|
|
assert "model1" in result
|
|
assert "model2" in result
|
|
|
|
|
|
def test_get_key_models_keeps_literal_model_colliding_with_group_name():
|
|
"""A name that is BOTH a deployed model and an access group grants both at
|
|
runtime (_check_model_access_helper unions them), so the listing must keep
|
|
the literal model alongside the group members instead of dropping it."""
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
|
|
user_api_key_dict = UserAPIKeyAuth(models=["beta-models"], api_key="test-key")
|
|
|
|
result = get_key_models(
|
|
user_api_key_dict=user_api_key_dict,
|
|
proxy_model_list=["beta-models", "member-a", "unrelated"],
|
|
model_access_groups={"beta-models": ["member-a"]},
|
|
include_model_access_groups=False,
|
|
)
|
|
assert sorted(result) == ["beta-models", "member-a"]
|
|
|
|
|
|
def test_get_team_models_keeps_literal_model_colliding_with_group_name():
|
|
"""Team flavor of the collision case: literal deployment survives group expansion."""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
|
|
result = get_team_models(
|
|
team_models=["beta-models"],
|
|
proxy_model_list=["beta-models", "member-a", "unrelated"],
|
|
model_access_groups={"beta-models": ["member-a"]},
|
|
include_model_access_groups=False,
|
|
)
|
|
assert sorted(result) == ["beta-models", "member-a"]
|
|
|
|
|
|
def test_get_team_models_drops_group_name_that_is_not_a_deployed_model():
|
|
"""No collision: a pure access-group name is still replaced by its members."""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
|
|
result = get_team_models(
|
|
team_models=["beta-models"],
|
|
proxy_model_list=["member-a", "unrelated"],
|
|
model_access_groups={"beta-models": ["member-a"]},
|
|
include_model_access_groups=False,
|
|
)
|
|
assert result == ["member-a"]
|
|
|
|
|
|
def test_get_key_models_does_not_mutate_input():
|
|
"""
|
|
get_key_models must not mutate user_api_key_dict.models in-place.
|
|
_get_models_from_access_groups uses .pop()/.extend() which would corrupt
|
|
cached UserAPIKeyAuth objects if all_models were an alias instead of a copy.
|
|
"""
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
|
|
original_models = ["group-a", "extra-model"]
|
|
user_api_key_dict = UserAPIKeyAuth(
|
|
models=list(original_models), # give it a list
|
|
api_key="test-key",
|
|
)
|
|
model_access_groups = {
|
|
"group-a": ["model1", "model2"],
|
|
}
|
|
|
|
_ = get_key_models(
|
|
user_api_key_dict=user_api_key_dict,
|
|
proxy_model_list=["model1", "model2"],
|
|
model_access_groups=model_access_groups,
|
|
include_model_access_groups=False,
|
|
)
|
|
# The original models list on the auth object must be unchanged
|
|
assert user_api_key_dict.models == original_models
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"key_models,team_models,proxy_model_list,model_list,expected",
|
|
[
|
|
(
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
[],
|
|
[],
|
|
[{"model_name": "anthropic/*", "litellm_params": {"model": "anthropic/*"}}],
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
),
|
|
(
|
|
[],
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
[],
|
|
[{"model_name": "anthropic/*", "litellm_params": {"model": "anthropic/*"}}],
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
),
|
|
(
|
|
[],
|
|
[],
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
[{"model_name": "anthropic/*", "litellm_params": {"model": "anthropic/*"}}],
|
|
[
|
|
"anthropic/claude-3-haiku-20240307",
|
|
"anthropic/claude-3-5-haiku-20241022",
|
|
],
|
|
),
|
|
],
|
|
)
|
|
def test_get_complete_model_list_order(key_models, team_models, proxy_model_list, model_list, expected):
|
|
"""
|
|
Test that get_complete_model_list preserves order
|
|
"""
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
from litellm import Router
|
|
|
|
assert (
|
|
get_complete_model_list(
|
|
proxy_model_list=proxy_model_list,
|
|
key_models=key_models,
|
|
team_models=team_models,
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=Router(model_list=model_list),
|
|
)
|
|
== expected
|
|
)
|
|
|
|
|
|
def test_get_complete_model_list_byok_wildcard_expansion():
|
|
"""
|
|
Test that wildcard models (e.g., openai/*) are expanded when the router has
|
|
no deployment for them - BYOK case where team has openai/* but proxy has
|
|
no openai config.
|
|
"""
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
from litellm import Router
|
|
|
|
# Router with empty model_list - no openai/* deployment (BYOK scenario)
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=["openai/*"],
|
|
proxy_model_list=[],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=Router(model_list=[]),
|
|
)
|
|
# Should expand openai/* to actual OpenAI models
|
|
assert len(result) > 0
|
|
assert all(m.startswith("openai/") for m in result)
|
|
assert "openai/*" not in result
|
|
|
|
|
|
def test_get_complete_model_list_expands_team_scoped_wildcard_with_stored_credential(
|
|
monkeypatch,
|
|
):
|
|
"""
|
|
Team-scoped BYOK wildcard deployments are stored under an internal model_name,
|
|
with the public wildcard name in model_info.team_public_model_name.
|
|
"""
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="openai-credential",
|
|
credential_info={"provider": "openai"},
|
|
credential_values={
|
|
"api_key": "stored-openai-key",
|
|
"api_base": "https://example.openai.test/v1",
|
|
},
|
|
)
|
|
],
|
|
)
|
|
|
|
captured_params = {}
|
|
|
|
def fake_get_provider_models(provider, litellm_params=None):
|
|
captured_params["provider"] = provider
|
|
captured_params["api_key"] = litellm_params.api_key
|
|
captured_params["api_base"] = litellm_params.api_base
|
|
captured_params["credential_name"] = litellm_params.litellm_credential_name
|
|
return ["gpt-4o"]
|
|
|
|
monkeypatch.setattr(model_checks, "get_provider_models", fake_get_provider_models)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "model_name_team-1_generated",
|
|
"litellm_params": {
|
|
"model": "openai/*",
|
|
"custom_llm_provider": "openai",
|
|
"litellm_credential_name": "openai-credential",
|
|
},
|
|
"model_info": {
|
|
"team_id": "team-1",
|
|
"team_public_model_name": "openai/*",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=["openai/*"],
|
|
proxy_model_list=[],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
team_id="team-1",
|
|
)
|
|
|
|
assert "openai/gpt-4o" in result
|
|
assert captured_params == {
|
|
"provider": "openai",
|
|
"api_key": "stored-openai-key",
|
|
"api_base": "https://example.openai.test/v1",
|
|
"credential_name": None,
|
|
}
|
|
|
|
|
|
def test_wildcard_credential_hydration_preserves_deployment_params(
|
|
monkeypatch,
|
|
):
|
|
import litellm
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="openai-credential",
|
|
credential_info={"provider": "openai"},
|
|
credential_values={
|
|
"api_key": "stored-openai-key",
|
|
"api_version": "credential-version",
|
|
"model": "openai/wrong-model",
|
|
"unexpected_field": "unexpected-value",
|
|
},
|
|
)
|
|
],
|
|
)
|
|
|
|
captured_params = {}
|
|
|
|
def fake_get_provider_models(provider, litellm_params=None):
|
|
captured_params["provider"] = provider
|
|
captured_params["model"] = litellm_params.model
|
|
captured_params["api_key"] = litellm_params.api_key
|
|
captured_params["api_version"] = litellm_params.api_version
|
|
captured_params["credential_name"] = litellm_params.litellm_credential_name
|
|
captured_params["has_unexpected_field"] = hasattr(litellm_params, "unexpected_field")
|
|
return ["gpt-4o"]
|
|
|
|
monkeypatch.setattr(model_checks, "get_provider_models", fake_get_provider_models)
|
|
|
|
result = get_known_models_from_wildcard(
|
|
wildcard_model="openai/*",
|
|
litellm_params=LiteLLM_Params(
|
|
model="openai/*",
|
|
custom_llm_provider="openai",
|
|
api_version="deployment-version",
|
|
litellm_credential_name="openai-credential",
|
|
),
|
|
)
|
|
|
|
assert result == ["openai/gpt-4o"]
|
|
assert captured_params == {
|
|
"provider": "openai",
|
|
"model": "openai/*",
|
|
"api_key": "stored-openai-key",
|
|
"api_version": "deployment-version",
|
|
"credential_name": None,
|
|
"has_unexpected_field": False,
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_hosted_vllm_wildcard_lists_the_server_models_without_an_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
models_route: Final = respx.get("http://vllm-without-key.test:8000/v1/models").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"data": [
|
|
{"id": "meta-llama/Llama-3.1-8B-Instruct"},
|
|
{"id": "qwen2.5"},
|
|
{"id": "openai/gpt-oss-20b"},
|
|
]
|
|
},
|
|
)
|
|
)
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*",
|
|
LiteLLM_Params(model="hosted_vllm/*", api_base="http://vllm-without-key.test:8000/v1"),
|
|
)
|
|
|
|
assert listed == [
|
|
"hosted_vllm/meta-llama/Llama-3.1-8B-Instruct",
|
|
"hosted_vllm/qwen2.5",
|
|
"hosted_vllm/openai/gpt-oss-20b",
|
|
]
|
|
assert "authorization" not in models_route.calls.last.request.headers
|
|
|
|
|
|
@respx.mock
|
|
def test_hosted_vllm_wildcard_sends_the_configured_api_key_as_a_bearer_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
models_route: Final = respx.get(
|
|
"http://vllm-with-key.test:8000/v1/models", headers={"Authorization": "Bearer vllm-test-key"}
|
|
).mock(return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]}))
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*",
|
|
LiteLLM_Params(model="hosted_vllm/*", api_base="http://vllm-with-key.test:8000/v1", api_key="vllm-test-key"),
|
|
)
|
|
|
|
assert listed == ["hosted_vllm/qwen2.5"]
|
|
assert models_route.call_count == 1
|
|
|
|
|
|
@respx.mock
|
|
def test_hosted_vllm_wildcard_without_api_base_lists_the_server_chat_would_call(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.setenv("HOSTED_VLLM_API_BASE", "http://hosted-vllm-env.test:8000/v1")
|
|
monkeypatch.setenv("HOSTED_VLLM_API_KEY", "hosted-vllm-env-key")
|
|
monkeypatch.setenv("VLLM_API_BASE", "http://vllm-passthrough-env.test:8000")
|
|
passthrough_server_route: Final = respx.get("http://vllm-passthrough-env.test:8000/v1/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "model-on-the-passthrough-server"}]})
|
|
)
|
|
respx.get(
|
|
"http://hosted-vllm-env.test:8000/v1/models", headers={"Authorization": "Bearer hosted-vllm-env-key"}
|
|
).mock(return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]}))
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*", LiteLLM_Params(model="hosted_vllm/*", custom_llm_provider="hosted_vllm")
|
|
)
|
|
|
|
assert listed == ["hosted_vllm/qwen2.5"]
|
|
assert not passthrough_server_route.called
|
|
|
|
|
|
@respx.mock
|
|
def test_hosted_vllm_wildcard_listing_does_not_wait_the_completion_timeout_on_a_hung_server(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm.constants import NON_LLM_CONNECTION_TIMEOUT
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
models_route: Final = respx.get("http://vllm-timeout.test:8000/v1/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]})
|
|
)
|
|
|
|
get_known_models_from_wildcard(
|
|
"hosted_vllm/*", LiteLLM_Params(model="hosted_vllm/*", api_base="http://vllm-timeout.test:8000/v1")
|
|
)
|
|
|
|
read_timeout: Final = models_route.calls.last.request.extensions["timeout"]["read"]
|
|
assert read_timeout is not None
|
|
assert read_timeout <= NON_LLM_CONNECTION_TIMEOUT
|
|
|
|
|
|
@respx.mock
|
|
def test_hosted_vllm_wildcard_listing_sends_the_deployment_extra_headers(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
respx.get(
|
|
"http://vllm-behind-gateway.test:8000/v1/models",
|
|
headers={"X-Tenant": "team-a", "Authorization": "Bearer vllm-test-key"},
|
|
).mock(return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]}))
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*",
|
|
LiteLLM_Params(
|
|
model="hosted_vllm/*",
|
|
api_base="http://vllm-behind-gateway.test:8000/v1",
|
|
api_key="vllm-test-key",
|
|
extra_headers={"X-Tenant": "team-a"},
|
|
),
|
|
)
|
|
|
|
assert listed == ["hosted_vllm/qwen2.5"]
|
|
|
|
|
|
|
|
@respx.mock
|
|
def test_vllm_wildcard_never_lists_the_ids_of_a_configured_vllm_server(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.setenv("VLLM_API_BASE", "http://vllm-library-env.test:8000")
|
|
monkeypatch.setenv("VLLM_API_KEY", "vllm-library-env-key")
|
|
server_route: Final = respx.get(url__regex=r"http://vllm-library(-env)?\.test:8000/.*").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "model-only-a-server-serves"}]})
|
|
)
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"vllm/*",
|
|
LiteLLM_Params(model="vllm/*", api_base="http://vllm-library.test:8000", api_key="vllm-library-key"),
|
|
)
|
|
|
|
assert "vllm/model-only-a-server-serves" not in listed
|
|
assert not server_route.called
|
|
|
|
_SELF_SIGNED_CERT_NOT_VALID_BEFORE: Final = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
|
|
_SELF_SIGNED_CERT_NOT_VALID_AFTER: Final = datetime.datetime(9999, 12, 31, tzinfo=datetime.UTC)
|
|
|
|
|
|
def _self_signed_cert(directory: Path) -> tuple[Path, Path]:
|
|
key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
name: Final = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1")])
|
|
cert: Final = (
|
|
x509.CertificateBuilder()
|
|
.subject_name(name)
|
|
.issuer_name(name)
|
|
.public_key(key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(_SELF_SIGNED_CERT_NOT_VALID_BEFORE)
|
|
.not_valid_after(_SELF_SIGNED_CERT_NOT_VALID_AFTER)
|
|
.add_extension(
|
|
x509.SubjectAlternativeName([x509.IPAddress(ipaddress.IPv4Address("127.0.0.1"))]), critical=False
|
|
)
|
|
.sign(key, hashes.SHA256())
|
|
)
|
|
cert_path: Final = directory / "cert.pem"
|
|
key_path: Final = directory / "key.pem"
|
|
cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
|
key_path.write_bytes(
|
|
key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
|
|
)
|
|
return cert_path, key_path
|
|
|
|
|
|
@pytest.fixture
|
|
def self_signed_vllm_api_base(tmp_path: Path) -> Iterator[str]:
|
|
cert_path, key_path = _self_signed_cert(tmp_path)
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
timeout = 5
|
|
|
|
def do_GET(self) -> None:
|
|
body: Final = json.dumps({"data": [{"id": "qwen2.5"}]}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
pass
|
|
|
|
server: Final = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
context: Final = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
context.load_cert_chain(cert_path, key_path)
|
|
server.socket = context.wrap_socket(server.socket, server_side=True, do_handshake_on_connect=False)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
try:
|
|
yield f"https://127.0.0.1:{server.server_address[1]}/v1"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def test_hosted_vllm_wildcard_lists_a_self_signed_server_only_when_the_deployment_turns_off_ssl_verify(
|
|
monkeypatch: pytest.MonkeyPatch, self_signed_vllm_api_base: str
|
|
) -> None:
|
|
import litellm
|
|
from litellm.caching.llm_caching_handler import LLMClientCache
|
|
from litellm.llms.custom_httpx import http_handler
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.setattr(litellm, "ssl_verify", True)
|
|
monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache())
|
|
monkeypatch.setattr(http_handler, "_ssl_context_cache", {})
|
|
monkeypatch.delenv("SSL_VERIFY", raising=False)
|
|
monkeypatch.delenv("SSL_CERT_FILE", raising=False)
|
|
|
|
verified: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*", LiteLLM_Params(model="hosted_vllm/*", api_base=self_signed_vllm_api_base)
|
|
)
|
|
unverified: Final = get_known_models_from_wildcard(
|
|
"hosted_vllm/*",
|
|
LiteLLM_Params(model="hosted_vllm/*", api_base=self_signed_vllm_api_base, ssl_verify=False),
|
|
)
|
|
|
|
assert verified == []
|
|
assert unverified == ["hosted_vllm/qwen2.5"]
|
|
|
|
|
|
@respx.mock
|
|
def test_custom_prefix_over_a_litellm_proxy_wildcard_lists_upstream_models_under_the_prefix_alone(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
respx.get("http://upstream-proxy.test:4000/v1/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "gpt-4o"}, {"id": "claude-sonnet"}]})
|
|
)
|
|
|
|
listed: Final = get_known_models_from_wildcard(
|
|
"team-proxy/*",
|
|
LiteLLM_Params(model="litellm_proxy/*", api_base="http://upstream-proxy.test:4000", api_key="sk-upstream"),
|
|
)
|
|
|
|
assert listed == ["team-proxy/gpt-4o", "team-proxy/claude-sonnet"]
|
|
|
|
|
|
def test_wildcard_custom_prefix_does_not_stack_provider_prefix(monkeypatch):
|
|
"""Regression test for #30358.
|
|
|
|
A wildcard with a custom prefix (e.g. ``ollama_server1/*`` to distinguish multiple Ollama
|
|
instances) must not stack the provider's own prefix onto the expanded model ids. The expanded
|
|
ids should be ``ollama_server1/gemma3:1b`` rather than ``ollama_server1/ollama/gemma3:1b``.
|
|
"""
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(
|
|
model_checks,
|
|
"get_provider_models",
|
|
lambda provider, litellm_params=None: ["ollama/gemma3:1b", "ollama/llama3:8b"],
|
|
)
|
|
|
|
result = get_known_models_from_wildcard(
|
|
wildcard_model="ollama_server1/*",
|
|
litellm_params=LiteLLM_Params(model="ollama_chat/*", custom_llm_provider="ollama_chat"),
|
|
)
|
|
|
|
assert result == ["ollama_server1/gemma3:1b", "ollama_server1/llama3:8b"]
|
|
|
|
|
|
def test_wildcard_custom_prefix_keeps_org_segment_for_non_provider_first_segment(
|
|
monkeypatch,
|
|
):
|
|
"""Only a known provider prefix should be stripped before re-prefixing.
|
|
|
|
If ``get_provider_models`` returns ids whose first segment is an org rather than a litellm
|
|
provider (e.g. ``meta-llama/Llama-3-8B``), stripping the first slash segment would drop the
|
|
org and produce an uncallable id. The org segment must be preserved.
|
|
"""
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(
|
|
model_checks,
|
|
"get_provider_models",
|
|
lambda provider, litellm_params=None: ["meta-llama/Llama-3-8B"],
|
|
)
|
|
|
|
result = get_known_models_from_wildcard(
|
|
wildcard_model="my_hf/*",
|
|
litellm_params=LiteLLM_Params(model="huggingface/*", custom_llm_provider="huggingface"),
|
|
)
|
|
|
|
assert result == ["my_hf/meta-llama/Llama-3-8B"]
|
|
|
|
|
|
def test_wildcard_credential_hydration_preserves_missing_credential_name(
|
|
monkeypatch,
|
|
):
|
|
import litellm
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "credential_list", [])
|
|
|
|
captured_params = {}
|
|
|
|
def fake_get_provider_models(provider, litellm_params=None):
|
|
captured_params["provider"] = provider
|
|
captured_params["api_key"] = litellm_params.api_key
|
|
captured_params["credential_name"] = litellm_params.litellm_credential_name
|
|
return ["gpt-4o"]
|
|
|
|
monkeypatch.setattr(model_checks, "get_provider_models", fake_get_provider_models)
|
|
|
|
result = get_known_models_from_wildcard(
|
|
wildcard_model="openai/*",
|
|
litellm_params=LiteLLM_Params(
|
|
model="openai/*",
|
|
custom_llm_provider="openai",
|
|
api_key=None,
|
|
litellm_credential_name="missing-credential",
|
|
),
|
|
)
|
|
|
|
assert result == ["openai/gpt-4o"]
|
|
assert captured_params == {
|
|
"provider": "openai",
|
|
"api_key": None,
|
|
"credential_name": "missing-credential",
|
|
}
|
|
|
|
|
|
def test_hydrate_credential_name_none_leaves_params_untouched(monkeypatch):
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import _hydrate_litellm_credential_name
|
|
from litellm.types.router import LiteLLM_Params
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="shared-credential",
|
|
credential_info={},
|
|
credential_values={"api_key": "sk-shared"},
|
|
)
|
|
],
|
|
)
|
|
params = LiteLLM_Params(model="openai/gpt-4o", litellm_credential_name=None)
|
|
|
|
result = _hydrate_litellm_credential_name(params)
|
|
|
|
assert result is not None
|
|
assert result.api_key is None
|
|
assert result.litellm_credential_name is None
|
|
|
|
|
|
def test_hydrate_replaced_credential_uses_new_credential_values(monkeypatch):
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import _hydrate_litellm_credential_name
|
|
from litellm.types.router import LiteLLM_Params
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="shared-credential",
|
|
credential_info={},
|
|
credential_values={"api_key": "sk-shared"},
|
|
),
|
|
CredentialItem(
|
|
credential_name="other-credential",
|
|
credential_info={},
|
|
credential_values={"api_key": "sk-other"},
|
|
),
|
|
],
|
|
)
|
|
params = LiteLLM_Params(model="openai/gpt-4o", litellm_credential_name="other-credential")
|
|
|
|
result = _hydrate_litellm_credential_name(params)
|
|
|
|
assert result is not None
|
|
assert result.api_key == "sk-other"
|
|
assert result.litellm_credential_name is None
|
|
|
|
|
|
def test_hydrate_inline_api_key_wins_over_stored_credential(monkeypatch):
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import _hydrate_litellm_credential_name
|
|
from litellm.types.router import LiteLLM_Params
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="shared-credential",
|
|
credential_info={},
|
|
credential_values={"api_key": "sk-shared"},
|
|
)
|
|
],
|
|
)
|
|
params = LiteLLM_Params(
|
|
model="openai/gpt-4o",
|
|
api_key="sk-inline",
|
|
litellm_credential_name="shared-credential",
|
|
)
|
|
|
|
result = _hydrate_litellm_credential_name(params)
|
|
|
|
assert result is not None
|
|
assert result.api_key == "sk-inline"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_available_models_for_user_expands_query_team_wildcard(
|
|
monkeypatch,
|
|
):
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.proxy.auth import model_checks
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.proxy.utils import get_available_models_for_user
|
|
from litellm.types.utils import CredentialItem
|
|
|
|
monkeypatch.setattr(
|
|
litellm,
|
|
"credential_list",
|
|
[
|
|
CredentialItem(
|
|
credential_name="openai-credential",
|
|
credential_info={"provider": "openai"},
|
|
credential_values={"api_key": "stored-openai-key"},
|
|
)
|
|
],
|
|
)
|
|
|
|
def fake_get_provider_models(provider, litellm_params=None):
|
|
assert litellm_params.api_key == "stored-openai-key"
|
|
assert litellm_params.litellm_credential_name is None
|
|
return ["gpt-4o-mini"]
|
|
|
|
monkeypatch.setattr(model_checks, "get_provider_models", fake_get_provider_models)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "model_name_team-1_generated",
|
|
"litellm_params": {
|
|
"model": "openai/*",
|
|
"custom_llm_provider": "openai",
|
|
"litellm_credential_name": "openai-credential",
|
|
},
|
|
"model_info": {
|
|
"team_id": "team-1",
|
|
"team_public_model_name": "openai/*",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
result = await get_available_models_for_user(
|
|
user_api_key_dict=UserAPIKeyAuth(
|
|
api_key="sk-test",
|
|
models=[],
|
|
team_id="team-1",
|
|
team_models=["openai/*"],
|
|
),
|
|
llm_router=router,
|
|
general_settings={},
|
|
user_model=None,
|
|
team_id="team-1",
|
|
)
|
|
|
|
assert "openai/gpt-4o-mini" in result
|
|
|
|
|
|
def test_get_key_models_all_team_models_recursive_team():
|
|
"""GH#30619: when key and team both have all-team-models,
|
|
the sentinel should expand to proxy_model_list."""
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
from litellm.proxy._types import SpecialModelNames
|
|
|
|
user_api_key_dict = type(
|
|
"obj",
|
|
(object,),
|
|
{
|
|
"models": [SpecialModelNames.all_team_models.value],
|
|
"team_id": "team-1",
|
|
"team_models": [SpecialModelNames.all_team_models.value],
|
|
},
|
|
)()
|
|
proxy_model_list = ["model-a", "model-b"]
|
|
result = get_key_models(user_api_key_dict, proxy_model_list, {})
|
|
assert SpecialModelNames.all_team_models.value not in result
|
|
assert set(result) == {"model-a", "model-b"}
|
|
|
|
|
|
def test_get_key_models_all_team_models_keeps_mixed_team_entries():
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
from litellm.proxy._types import SpecialModelNames
|
|
|
|
user_api_key_dict = type(
|
|
"obj",
|
|
(object,),
|
|
{
|
|
"models": [SpecialModelNames.all_team_models.value],
|
|
"team_id": "team-1",
|
|
"team_models": [
|
|
SpecialModelNames.all_team_models.value,
|
|
"restricted-model",
|
|
],
|
|
},
|
|
)()
|
|
result = get_key_models(user_api_key_dict, ["model-a", "model-b"], {})
|
|
assert SpecialModelNames.all_team_models.value not in result
|
|
assert set(result) == {"model-a", "model-b", "restricted-model"}
|
|
|
|
|
|
def test_get_team_models_all_team_models_expands():
|
|
"""GH#30619: all-team-models in team_models should expand."""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
from litellm.proxy._types import SpecialModelNames
|
|
|
|
result = get_team_models(
|
|
[SpecialModelNames.all_team_models.value],
|
|
["model-a", "model-b"],
|
|
{},
|
|
)
|
|
assert SpecialModelNames.all_team_models.value not in result
|
|
assert set(result) == {"model-a", "model-b"}
|
|
|
|
|
|
def test_get_team_models_all_team_models_expands_with_access_groups():
|
|
"""GH#30619: all-team-models with include_model_access_groups
|
|
should include access group keys."""
|
|
from litellm.proxy.auth.model_checks import get_team_models
|
|
from litellm.proxy._types import SpecialModelNames
|
|
|
|
result = get_team_models(
|
|
[SpecialModelNames.all_team_models.value],
|
|
["model-a", "model-b"],
|
|
{"group-1": ["g1-model"], "group-2": ["g2-model"]},
|
|
include_model_access_groups=True,
|
|
)
|
|
assert SpecialModelNames.all_team_models.value not in result
|
|
assert "model-a" in result
|
|
assert "model-b" in result
|
|
assert "group-1" in result
|
|
assert "group-2" in result
|
|
|
|
|
|
def test_get_key_models_teamless_all_team_models_returns_unrestricted():
|
|
"""Teamless key with all-team-models must resolve the same as leaving the
|
|
models field empty ([] = unrestricted). The sentinel must not leak into
|
|
the returned list. Fails if someone adds a team_id guard to the sentinel
|
|
expansion in get_key_models."""
|
|
from litellm.proxy._types import SpecialModelNames
|
|
from litellm.proxy.auth.model_checks import get_key_models
|
|
|
|
user_api_key_dict = type(
|
|
"obj",
|
|
(object,),
|
|
{
|
|
"models": [SpecialModelNames.all_team_models.value],
|
|
"team_id": None,
|
|
"team_models": [],
|
|
},
|
|
)()
|
|
proxy_model_list = ["gpt-4o", "claude-sonnet-4-20250514"]
|
|
result = get_key_models(user_api_key_dict, proxy_model_list, {})
|
|
assert SpecialModelNames.all_team_models.value not in result
|
|
assert result == [], "should return [] (unrestricted), same as an unscoped key"
|
|
|
|
|
|
def test_expand_wildcard_deployments_non_wildcard_passthrough():
|
|
"""Non-wildcard deployments must be returned unchanged."""
|
|
from litellm.proxy.auth.model_checks import (
|
|
expand_wildcard_deployments_for_model_info,
|
|
)
|
|
|
|
deployment = {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o"}}
|
|
result = expand_wildcard_deployments_for_model_info([deployment])
|
|
assert result == [deployment]
|
|
|
|
|
|
def test_expand_wildcard_deployments_openai_wildcard():
|
|
"""openai/* should expand into ≥1 known openai model entries."""
|
|
|
|
from litellm.proxy.auth.model_checks import (
|
|
expand_wildcard_deployments_for_model_info,
|
|
)
|
|
|
|
fake_models = ["openai/gpt-4o", "openai/gpt-4o-mini"]
|
|
deployment = {
|
|
"model_name": "openai/*",
|
|
"litellm_params": {"model": "openai/*"},
|
|
}
|
|
with patch(
|
|
"litellm.proxy.auth.model_checks.get_known_models_from_wildcard",
|
|
return_value=fake_models,
|
|
):
|
|
result = expand_wildcard_deployments_for_model_info([deployment])
|
|
|
|
assert len(result) == 2
|
|
assert all(r["model_name"] in fake_models for r in result)
|
|
assert all(r["litellm_params"]["model"] in fake_models for r in result)
|
|
|
|
|
|
def test_expand_wildcard_concrete_model_name_with_wildcard_litellm_params():
|
|
"""Concrete model_name must not be overwritten when only litellm_params.model is wildcard."""
|
|
from litellm.proxy.auth.model_checks import (
|
|
expand_wildcard_deployments_for_model_info,
|
|
)
|
|
|
|
deployment = {
|
|
"model_name": "my-custom-alias",
|
|
"litellm_params": {"model": "openai/*"},
|
|
}
|
|
result = expand_wildcard_deployments_for_model_info([deployment])
|
|
# model_name is not a wildcard, so the deployment passes through unchanged
|
|
assert result == [deployment]
|
|
|
|
|
|
def test_expand_wildcard_invalid_litellm_params_passthrough():
|
|
"""Deployments with invalid litellm_params must pass through unchanged (no 500)."""
|
|
from litellm.proxy.auth.model_checks import (
|
|
expand_wildcard_deployments_for_model_info,
|
|
)
|
|
|
|
deployment = {
|
|
"model_name": "openai/*",
|
|
"litellm_params": {
|
|
"model": "openai/*",
|
|
"max_retries": "not-an-int-field-that-breaks",
|
|
},
|
|
}
|
|
# Even if LiteLLM_Params construction fails the deployment should survive
|
|
result = expand_wildcard_deployments_for_model_info([deployment])
|
|
assert result == [deployment]
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("allowed_model_names", "expected_names", "expected_listings"),
|
|
[
|
|
pytest.param(None, ["qwen-qwen2.5"], 1, id="unscoped"),
|
|
pytest.param(["qwen-qwen2.5"], ["qwen-qwen2.5"], 1, id="granted-a-listed-name"),
|
|
pytest.param(["hosted_vllm/qwen2.5"], [], 0, id="granted-only-the-provider-id"),
|
|
],
|
|
)
|
|
@respx.mock
|
|
def test_expand_wildcard_alias_without_a_slash_lists_and_filters_by_its_own_pattern(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
request: pytest.FixtureRequest,
|
|
allowed_model_names: list[str] | None,
|
|
expected_names: list[str],
|
|
expected_listings: int,
|
|
) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
api_base: Final = f"http://vllm-alias-{request.node.callspec.id}.test:8000/v1"
|
|
listing: Final = respx.get(f"{api_base}/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]})
|
|
)
|
|
deployment: Final = {"model_name": "qwen-*", "litellm_params": {"model": "hosted_vllm/*", "api_base": api_base}}
|
|
|
|
rows: Final = expand_wildcard_deployments_for_model_info([deployment], allowed_model_names=allowed_model_names)
|
|
|
|
assert ([row["model_name"] for row in rows], listing.call_count) == (expected_names, expected_listings)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("allowed_model_names", "keeps_the_alias"),
|
|
[
|
|
pytest.param(["gpt-4o-mini"], True, id="granted-a-name-the-alias-matches"),
|
|
pytest.param(["claude-sonnet"], False, id="granted-only-other-names"),
|
|
],
|
|
)
|
|
def test_expand_wildcard_alias_over_one_model_is_kept_only_for_keys_with_a_name_it_matches(
|
|
allowed_model_names: list[str], keeps_the_alias: bool
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info
|
|
|
|
deployment: Final = {"model_name": "gpt-4o*", "litellm_params": {"model": "openai/gpt-4o"}}
|
|
|
|
rows: Final = expand_wildcard_deployments_for_model_info([deployment], allowed_model_names=allowed_model_names)
|
|
|
|
assert rows == ([deployment] if keeps_the_alias else [])
|
|
|
|
def test_get_complete_model_list_excludes_wildcard_routes_by_default():
|
|
"""Regression (LIT-4108): a wildcard with a matching router deployment leaked into /v1/models."""
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "bedrock/*",
|
|
"litellm_params": {"model": "bedrock/*"},
|
|
},
|
|
{
|
|
"model_name": "gpt-4",
|
|
"litellm_params": {"model": "openai/gpt-4"},
|
|
},
|
|
]
|
|
)
|
|
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["bedrock/*", "gpt-4"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
return_wildcard_routes=False,
|
|
llm_router=router,
|
|
)
|
|
|
|
assert "bedrock/*" not in result
|
|
assert "gpt-4" in result
|
|
assert any(m.startswith("bedrock/") for m in result)
|
|
|
|
|
|
def test_get_complete_model_list_excludes_wildcard_routes_without_router():
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["bedrock/*", "gpt-4"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
return_wildcard_routes=False,
|
|
llm_router=None,
|
|
)
|
|
|
|
assert "bedrock/*" not in result
|
|
assert "gpt-4" in result
|
|
assert any(m.startswith("bedrock/") for m in result)
|
|
|
|
|
|
def test_get_complete_model_list_includes_wildcard_routes_when_requested():
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "bedrock/*",
|
|
"litellm_params": {"model": "bedrock/*"},
|
|
},
|
|
]
|
|
)
|
|
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["bedrock/*"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
return_wildcard_routes=True,
|
|
llm_router=router,
|
|
)
|
|
|
|
assert result.count("bedrock/*") == 1
|
|
assert any(m.startswith("bedrock/") and m != "bedrock/*" for m in result)
|
|
|
|
|
|
def test_add_known_models_refreshes_models_by_provider_for_wildcard_expansion():
|
|
"""models_by_provider was a frozen import-time snapshot of set unions, so cost map
|
|
reloads (which call add_known_models) never reached wildcard expansion until a
|
|
process restart (LIT-4947)."""
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
fake_model = "vertex_ai/gemini-lit4947-regression"
|
|
captured_reference = litellm.models_by_provider
|
|
assert fake_model not in litellm.models_by_provider["vertex_ai"]
|
|
try:
|
|
litellm.add_known_models(
|
|
model_cost_map={fake_model: {"litellm_provider": "vertex_ai-language-models", "mode": "chat"}}
|
|
)
|
|
assert fake_model in litellm.models_by_provider["vertex_ai"]
|
|
assert litellm.models_by_provider is captured_reference
|
|
assert fake_model in captured_reference["vertex_ai"]
|
|
assert fake_model in get_known_models_from_wildcard("vertex_ai/*")
|
|
finally:
|
|
litellm.vertex_language_models.discard(fake_model)
|
|
litellm.add_known_models(model_cost_map={})
|
|
assert fake_model not in litellm.models_by_provider["vertex_ai"]
|
|
|
|
|
|
def test_wildcard_lists_a_model_once_when_registered_both_bare_and_provider_prefixed():
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
bare: Final = "gpt-dedupe-regression"
|
|
entry: Final = {"litellm_provider": "openai", "mode": "chat"}
|
|
try:
|
|
litellm.register_model({bare: entry, f"openai/{bare}": entry}, persist_across_reloads=False)
|
|
|
|
listed: Final = get_known_models_from_wildcard("openai/*")
|
|
|
|
assert listed.count(f"openai/{bare}") == 1
|
|
finally:
|
|
litellm.model_cost.pop(bare, None)
|
|
litellm.model_cost.pop(f"openai/{bare}", None)
|
|
litellm.open_ai_chat_completion_models.discard(bare)
|
|
litellm.open_ai_chat_completion_models.discard(f"openai/{bare}")
|
|
litellm.add_known_models(model_cost_map={})
|
|
|
|
|
|
def _is_vertex_ai_label(label: object) -> bool:
|
|
return label == "vertex_ai" or (isinstance(label, str) and label.startswith("vertex_ai-"))
|
|
|
|
|
|
def test_vertex_wildcard_lists_every_vertex_cost_map_row_under_its_callable_id():
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
vertex_rows: Final = {
|
|
f"vertex_ai/{name.removeprefix('vertex_ai/')}"
|
|
for name, info in litellm.model_cost.items()
|
|
if _is_vertex_ai_label(info.get("litellm_provider"))
|
|
}
|
|
assert vertex_rows
|
|
|
|
listed: Final = set(get_known_models_from_wildcard("vertex_ai/*"))
|
|
|
|
assert sorted(vertex_rows - listed) == []
|
|
|
|
|
|
VERTEX_COST_MAP_LABELS: Final = (
|
|
"vertex_ai",
|
|
"vertex_ai-language-models",
|
|
"vertex_ai-vision-models",
|
|
"vertex_ai-chat-models",
|
|
"vertex_ai-code-chat-models",
|
|
"vertex_ai-text-models",
|
|
"vertex_ai-code-text-models",
|
|
"vertex_ai-embedding-models",
|
|
"vertex_ai-anthropic_models",
|
|
"vertex_ai-llama_models",
|
|
"vertex_ai-deepseek_models",
|
|
"vertex_ai-mistral_models",
|
|
"vertex_ai-ai21_models",
|
|
"vertex_ai-image-models",
|
|
"vertex_ai-video-models",
|
|
"vertex_ai-openai_models",
|
|
"vertex_ai-minimax_models",
|
|
"vertex_ai-moonshot_models",
|
|
"vertex_ai-zai_models",
|
|
"vertex_ai-qwen_models",
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def _registered_rows(label_per_key: Mapping[str, str]) -> Generator[None, None, None]:
|
|
import litellm
|
|
|
|
try:
|
|
litellm.register_model(
|
|
{key: {"litellm_provider": label, "mode": "chat"} for key, label in label_per_key.items()},
|
|
persist_across_reloads=False,
|
|
)
|
|
yield
|
|
finally:
|
|
provider_model_sets: Final = tuple(value for value in vars(litellm).values() if isinstance(value, set))
|
|
for key in label_per_key:
|
|
for provider_models in provider_model_sets:
|
|
provider_models.discard(key)
|
|
provider_models.discard(key.removeprefix("vertex_ai/"))
|
|
litellm.add_known_models(model_cost_map={})
|
|
|
|
|
|
@pytest.mark.parametrize("label", VERTEX_COST_MAP_LABELS)
|
|
def test_vertex_wildcard_lists_a_row_registered_under_any_vertex_label(label: str):
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
key: Final = f"vertex_ai/fake-{label}-row"
|
|
with _registered_rows({key: label}):
|
|
listed: Final = get_known_models_from_wildcard("vertex_ai/*")
|
|
|
|
assert key in listed
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("wildcard_model", "listed_prefix"),
|
|
[("vertex_ai/*", "vertex_ai"), ("my_vertex/*", "my_vertex")],
|
|
)
|
|
def test_wildcard_keeps_a_publisher_segment_that_names_another_provider(wildcard_model: str, listed_prefix: str):
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
with _registered_rows(
|
|
{
|
|
"vertex_ai/openai/fake-publisher-maas": "vertex_ai-openai_models",
|
|
"vertex_ai/meta/fake-publisher-maas": "vertex_ai-llama_models",
|
|
}
|
|
):
|
|
listed: Final = get_known_models_from_wildcard(
|
|
wildcard_model=wildcard_model,
|
|
litellm_params=LiteLLM_Params(model="vertex_ai/*"),
|
|
)
|
|
|
|
assert f"{listed_prefix}/openai/fake-publisher-maas" in listed
|
|
assert f"{listed_prefix}/meta/fake-publisher-maas" in listed
|
|
assert f"{listed_prefix}/fake-publisher-maas" not in listed
|
|
|
|
|
|
@pytest.mark.parametrize("provider", ["replicate", "wandb"])
|
|
def test_wildcard_lists_each_model_once_under_its_full_owner_id(provider: str):
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
full_ids: Final = {
|
|
f"{provider}/{model.removeprefix(f'{provider}/')}" for model in litellm.models_by_provider[provider]
|
|
}
|
|
assert any(model.count("/") > 1 for model in full_ids)
|
|
|
|
assert sorted(get_known_models_from_wildcard(f"{provider}/*")) == sorted(full_ids)
|
|
|
|
|
|
def test_get_complete_model_list_drops_no_default_models_sentinel():
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
result = get_complete_model_list(
|
|
key_models=["no-default-models", "model-a"],
|
|
team_models=[],
|
|
proxy_model_list=["model-a", "model-b"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
)
|
|
assert result == ["model-a"]
|
|
|
|
|
|
def test_get_complete_model_list_sentinel_only_grants_nothing():
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
result = get_complete_model_list(
|
|
key_models=["no-default-models"],
|
|
team_models=["no-default-models"],
|
|
proxy_model_list=["model-a", "model-b"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
)
|
|
assert result == []
|
|
|
|
|
|
def test_get_provider_models_admits_providers_without_a_static_catalog():
|
|
"""Providers without a static model list are no longer rejected up front.
|
|
|
|
With endpoint discovery off, get_valid_models falls back to the (empty)
|
|
static list, so the result is [] rather than None. Before the fix this
|
|
returned None and the wildcard was never expanded.
|
|
"""
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_provider_models
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
assert "litellm_proxy" not in litellm.models_by_provider
|
|
assert "hosted_vllm" not in litellm.models_by_provider
|
|
|
|
result = get_provider_models(
|
|
"litellm_proxy",
|
|
litellm_params=LiteLLM_Params(
|
|
model="litellm_proxy/*",
|
|
api_base="http://upstream:4000",
|
|
api_key="sk-upstream",
|
|
),
|
|
)
|
|
|
|
assert result == []
|
|
|
|
|
|
def test_get_complete_model_list_discovers_litellm_proxy_wildcard_models(monkeypatch):
|
|
"""A litellm_proxy/* deployment lists the upstream proxy's models when endpoint discovery is on."""
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
captured = {}
|
|
|
|
def fake_get_models(self, api_key=None, api_base=None):
|
|
captured["api_key"] = api_key
|
|
captured["api_base"] = api_base
|
|
return ["gpt-4o", "claude-sonnet"]
|
|
|
|
monkeypatch.setattr(OpenAIGPTConfig, "get_models", fake_get_models)
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "litellm_proxy/*",
|
|
"litellm_params": {
|
|
"model": "litellm_proxy/*",
|
|
"api_base": "http://upstream:4000",
|
|
"api_key": "sk-upstream",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
result = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["litellm_proxy/*"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
|
|
assert captured == {"api_key": "sk-upstream", "api_base": "http://upstream:4000"}
|
|
assert "litellm_proxy/gpt-4o" in result
|
|
assert "litellm_proxy/claude-sonnet" in result
|
|
|
|
|
|
def test_get_provider_models_returns_none_for_an_unknown_provider():
|
|
from litellm.proxy.auth.model_checks import get_provider_models
|
|
|
|
assert get_provider_models("not-a-real-provider") is None
|
|
|
|
|
|
def test_transcribe_is_a_known_provider_for_wildcard_expansion():
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import (
|
|
get_known_models_from_wildcard,
|
|
get_provider_models,
|
|
)
|
|
|
|
assert "transcribe" in litellm.models_by_provider
|
|
assert "transcribe/StartTranscriptionJob" in litellm.models_by_provider["transcribe"]
|
|
assert get_provider_models("transcribe") == ["transcribe/StartTranscriptionJob"]
|
|
assert get_known_models_from_wildcard("transcribe/*") == ["transcribe/StartTranscriptionJob"]
|
|
|
|
|
|
@respx.mock
|
|
def test_partial_bedrock_wildcard_filters_the_discovered_ids(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
|
|
region: Final = "ap-south-1"
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/foundation-models", params={"byInferenceType": "ON_DEMAND"}
|
|
).mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"modelSummaries": [
|
|
{"modelId": "anthropic.claude-haiku-4-5-20251001-v1:0"},
|
|
{"modelId": "amazon.nova-micro-v1:0"},
|
|
]
|
|
},
|
|
)
|
|
)
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/inference-profiles", params={"typeEquals": "SYSTEM_DEFINED"}
|
|
).mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"inferenceProfileSummaries": [
|
|
{"inferenceProfileId": "us.anthropic.claude-haiku-4-5-20251001-v1:0", "status": "ACTIVE"}
|
|
]
|
|
},
|
|
)
|
|
)
|
|
deployment: Final = LiteLLM_Params(
|
|
model="bedrock/anthropic.*",
|
|
aws_access_key_id="AKIAPARTIALWILDCARD",
|
|
aws_secret_access_key="partial-wildcard-secret",
|
|
aws_region_name=region,
|
|
)
|
|
|
|
assert get_known_models_from_wildcard("bedrock/anthropic.*", deployment) == [
|
|
"bedrock/anthropic.claude-haiku-4-5-20251001-v1:0"
|
|
]
|
|
|
|
|
|
@respx.mock
|
|
def test_partial_bedrock_wildcard_lists_nothing_when_no_discovered_id_carries_its_prefix(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
|
|
region: Final = "ap-southeast-2"
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/foundation-models", params={"byInferenceType": "ON_DEMAND"}
|
|
).mock(return_value=httpx.Response(200, json={"modelSummaries": [{"modelId": "amazon.nova-micro-v1:0"}]}))
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/inference-profiles", params={"typeEquals": "SYSTEM_DEFINED"}
|
|
).mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"inferenceProfileSummaries": [
|
|
{"inferenceProfileId": "us.anthropic.claude-haiku-4-5-20251001-v1:0", "status": "ACTIVE"}
|
|
]
|
|
},
|
|
)
|
|
)
|
|
deployment: Final = LiteLLM_Params(
|
|
model="bedrock/anthropic.*",
|
|
aws_access_key_id="AKIAPROFILEONLYACCOUNT",
|
|
aws_secret_access_key="profile-only-secret",
|
|
aws_region_name=region,
|
|
)
|
|
|
|
assert get_known_models_from_wildcard("bedrock/anthropic.*", deployment) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("wildcard", "deployment_model", "listed_ids"),
|
|
[
|
|
(
|
|
"openrouter/*",
|
|
"openrouter/*",
|
|
["openrouter/openai/gpt-5.4-mini", "openrouter/anthropic/claude-haiku-5-5"],
|
|
),
|
|
("or-team/*", "openrouter/*", ["or-team/openai/gpt-5.4-mini", "or-team/anthropic/claude-haiku-5-5"]),
|
|
("openrouter/openai/*", "openrouter/openai/*", ["openrouter/openai/gpt-5.4-mini"]),
|
|
("*", "openrouter/*", ["openai/gpt-5.4-mini", "anthropic/claude-haiku-5-5"]),
|
|
],
|
|
)
|
|
@respx.mock
|
|
def test_openrouter_wildcard_lists_the_live_catalog_with_each_vendor_namespace(
|
|
monkeypatch: pytest.MonkeyPatch, wildcard: str, deployment_model: str, listed_ids: list[str]
|
|
) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("OPENROUTER_API_BASE", raising=False)
|
|
respx.get("https://openrouter.ai/api/v1/models/user").mock(
|
|
return_value=httpx.Response(
|
|
200, json={"data": [{"id": "openai/gpt-5.4-mini"}, {"id": "anthropic/claude-haiku-5-5"}]}
|
|
)
|
|
)
|
|
|
|
assert (
|
|
get_known_models_from_wildcard(wildcard, LiteLLM_Params(model=deployment_model, api_key=f"sk-or-{wildcard}"))
|
|
== listed_ids
|
|
)
|
|
|
|
|
|
@respx.mock
|
|
def test_key_granted_openrouter_wildcard_without_an_openrouter_deployment_lists_no_environment_catalog(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-environment-only")
|
|
catalog: Final = respx.get("https://openrouter.ai/api/v1/models/user").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "anthropic/claude-haiku-5-5"}]})
|
|
)
|
|
router: Final = Router(
|
|
model_list=[{"model_name": "or-team/*", "litellm_params": {"model": "openrouter/*", "api_key": "sk-or-team"}}]
|
|
)
|
|
|
|
listed: Final = get_complete_model_list(
|
|
key_models=["openrouter/*"],
|
|
team_models=[],
|
|
proxy_model_list=router.model_names,
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
|
|
assert (listed, catalog.call_count) == ([], 0)
|
|
|
|
|
|
@respx.mock
|
|
def test_key_granted_hosted_vllm_wildcard_without_a_hosted_vllm_deployment_lists_no_environment_server(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.setenv("HOSTED_VLLM_API_BASE", "http://hosted-vllm-environment-only.test:8000/v1")
|
|
environment_server: Final = respx.get("http://hosted-vllm-environment-only.test:8000/v1/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "qwen2.5"}]})
|
|
)
|
|
router: Final = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "vllm-team/*",
|
|
"litellm_params": {"model": "hosted_vllm/*", "api_base": "http://vllm-team.test:8000/v1"},
|
|
}
|
|
]
|
|
)
|
|
|
|
listed: Final = get_complete_model_list(
|
|
key_models=["hosted_vllm/*"],
|
|
team_models=[],
|
|
proxy_model_list=router.model_names,
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
|
|
assert (listed, environment_server.call_count) == ([], 0)
|
|
|
|
|
|
@contextmanager
|
|
def _openai_catalog_behind_a_catch_all(
|
|
monkeypatch: pytest.MonkeyPatch, concrete_name: str, concrete_model_info: Mapping[str, str]
|
|
) -> Generator[Router, None, None]:
|
|
import litellm
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
|
|
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
|
|
with respx.mock(assert_all_called=False) as respx_mock:
|
|
respx_mock.get("https://api.openai.com/v1/models").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": "gpt-4o"}, {"id": "gpt-4o-mini"}]})
|
|
)
|
|
yield Router(
|
|
model_list=[
|
|
{"model_name": "*", "litellm_params": {"model": "openai/*", "api_key": "sk-openai-catch-all"}},
|
|
{
|
|
"model_name": concrete_name,
|
|
"litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-or-gpt-4o"},
|
|
"model_info": dict(concrete_model_info),
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("concrete_name", "concrete_model_info", "team_id"),
|
|
[
|
|
("gpt-4o", {}, None),
|
|
("model_name_team-x_gpt-4o", {"team_id": "team-x", "team_public_model_name": "gpt-4o"}, "team-x"),
|
|
],
|
|
)
|
|
def test_catch_all_wildcard_lists_a_name_a_concrete_deployment_serves_once(
|
|
monkeypatch: pytest.MonkeyPatch, concrete_name: str, concrete_model_info: Mapping[str, str], team_id: str | None
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
with _openai_catalog_behind_a_catch_all(monkeypatch, concrete_name, concrete_model_info) as router:
|
|
listed: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=router.get_model_names(team_id=team_id),
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
team_id=team_id,
|
|
)
|
|
|
|
assert sorted(listed) == ["gpt-4o", "gpt-4o-mini"]
|
|
|
|
|
|
def test_catch_all_wildcard_model_info_keeps_only_the_concrete_row_for_a_deployed_name(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info
|
|
|
|
with _openai_catalog_behind_a_catch_all(monkeypatch, "gpt-4o", {}) as router:
|
|
rows: Final = expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or ()], llm_router=router
|
|
)
|
|
|
|
assert sorted((row["model_name"], row["litellm_params"]["model"]) for row in rows) == [
|
|
("gpt-4o", "openrouter/openai/gpt-4o"),
|
|
("gpt-4o-mini", "openai/gpt-4o-mini"),
|
|
]
|
|
|
|
|
|
@contextmanager
|
|
def _openrouter_catalog_beside_a_named_deployment(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
wildcard_name: str,
|
|
named: str,
|
|
catalog: tuple[str, ...] = ("openai/gpt-5.4-mini", "anthropic/claude-haiku-5-5"),
|
|
wildcard_key: str | None = None,
|
|
) -> Generator[Router, None, None]:
|
|
import litellm
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("OPENROUTER_API_BASE", raising=False)
|
|
with respx.mock(assert_all_called=False) as respx_mock:
|
|
respx_mock.get("https://openrouter.ai/api/v1/models/user").mock(
|
|
return_value=httpx.Response(200, json={"data": [{"id": catalog_id} for catalog_id in catalog]})
|
|
)
|
|
yield Router(
|
|
model_list=[
|
|
{
|
|
"model_name": wildcard_name,
|
|
"litellm_params": {
|
|
"model": "openrouter/*",
|
|
"api_key": wildcard_key or f"sk-or-wildcard-{wildcard_name}",
|
|
},
|
|
},
|
|
{
|
|
"model_name": named,
|
|
"litellm_params": {"model": "openrouter/openai/gpt-5.4-mini", "api_key": "sk-or-named"},
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("wildcard_name", "named"),
|
|
[("openrouter/*", "openrouter/openai/gpt-5.4-mini"), ("or-team/*", "or-team/openai/gpt-5.4-mini")],
|
|
)
|
|
def test_provider_wildcard_lists_a_name_a_named_deployment_serves_once(
|
|
monkeypatch: pytest.MonkeyPatch, wildcard_name: str, named: str
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
|
|
with _openrouter_catalog_beside_a_named_deployment(monkeypatch, wildcard_name, named) as router:
|
|
listed: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=list(router.model_names),
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
rows: Final = expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or ()], llm_router=router
|
|
)
|
|
|
|
vendor_prefix: Final = wildcard_name.removesuffix("*")
|
|
assert sorted(listed) == [f"{vendor_prefix}anthropic/claude-haiku-5-5", named]
|
|
assert sorted((row["model_name"], row["litellm_params"]["api_key"]) for row in rows) == [
|
|
(f"{vendor_prefix}anthropic/claude-haiku-5-5", f"sk-or-wildcard-{wildcard_name}"),
|
|
(named, "sk-or-named"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("wildcard_name", "named"),
|
|
[("openrouter/*", "openrouter/openai/gpt-5.4-mini"), ("or-team/*", "or-team/openai/gpt-5.4-mini")],
|
|
)
|
|
def test_provider_wildcard_whose_whole_catalog_is_deployed_by_name_shows_only_the_named_row(
|
|
monkeypatch: pytest.MonkeyPatch, wildcard_name: str, named: str
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info
|
|
|
|
with _openrouter_catalog_beside_a_named_deployment(
|
|
monkeypatch,
|
|
wildcard_name,
|
|
named,
|
|
catalog=("openai/gpt-5.4-mini",),
|
|
wildcard_key=f"sk-or-one-model-{wildcard_name}",
|
|
) as router:
|
|
rows: Final = expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or ()], llm_router=router
|
|
)
|
|
|
|
assert [(row["model_name"], row["litellm_params"]["api_key"]) for row in rows] == [(named, "sk-or-named")]
|
|
|
|
|
|
@respx.mock
|
|
def test_custom_prefix_that_starts_a_discovered_id_still_prefixes_every_listed_model(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
monkeypatch.setattr(litellm, "check_provider_endpoint", True)
|
|
monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
|
|
region: Final = "ca-central-1"
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/foundation-models", params={"byInferenceType": "ON_DEMAND"}
|
|
).mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"modelSummaries": [
|
|
{"modelId": "anthropic.claude-haiku-4-5-20251001-v1:0"},
|
|
{"modelId": "amazon.nova-micro-v1:0"},
|
|
]
|
|
},
|
|
)
|
|
)
|
|
respx.get(
|
|
f"https://bedrock.{region}.amazonaws.com/inference-profiles", params={"typeEquals": "SYSTEM_DEFINED"}
|
|
).mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"inferenceProfileSummaries": [
|
|
{"inferenceProfileId": "us.anthropic.claude-haiku-4-5-20251001-v1:0", "status": "ACTIVE"}
|
|
]
|
|
},
|
|
)
|
|
)
|
|
deployment: Final = LiteLLM_Params(
|
|
model="bedrock/*",
|
|
aws_access_key_id="AKIACUSTOMPREFIXACCOUNT",
|
|
aws_secret_access_key="custom-prefix-secret",
|
|
aws_region_name=region,
|
|
)
|
|
|
|
assert get_known_models_from_wildcard("anthropic/*", deployment) == [
|
|
"anthropic/amazon.nova-micro-v1:0",
|
|
"anthropic/anthropic.claude-haiku-4-5-20251001-v1:0",
|
|
"anthropic/us.anthropic.claude-haiku-4-5-20251001-v1:0",
|
|
]
|
|
|
|
|
|
@contextmanager
|
|
def _registered_chat_models(provider_by_model: Mapping[str, str]) -> Generator[None, None, None]:
|
|
import litellm
|
|
|
|
litellm.register_model(
|
|
{model: {"litellm_provider": provider, "mode": "chat"} for model, provider in provider_by_model.items()},
|
|
persist_across_reloads=False,
|
|
)
|
|
try:
|
|
yield
|
|
finally:
|
|
provider_model_sets: Final = tuple(known for known in vars(litellm).values() if isinstance(known, set))
|
|
for model in provider_by_model:
|
|
litellm.model_cost.pop(model, None)
|
|
for provider_models in provider_model_sets:
|
|
provider_models.discard(model)
|
|
litellm.add_known_models(model_cost_map={})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("wildcard", "deployment_model", "catalog", "expected"),
|
|
[
|
|
pytest.param(
|
|
"cloudflare/@cf/partial-wildcard-org/*",
|
|
"cloudflare/@cf/partial-wildcard-org/*",
|
|
("cloudflare/@cf/partial-wildcard-org/model-a", "cloudflare/@hf/partial-wildcard-org/model-b"),
|
|
["cloudflare/@cf/partial-wildcard-org/model-a"],
|
|
id="prefixed-catalog-with-deployment",
|
|
),
|
|
pytest.param(
|
|
"cloudflare/@cf/partial-wildcard-org/*",
|
|
None,
|
|
("cloudflare/@cf/partial-wildcard-org/model-a", "cloudflare/@hf/partial-wildcard-org/model-b"),
|
|
["cloudflare/@cf/partial-wildcard-org/model-a"],
|
|
id="prefixed-catalog-without-deployment",
|
|
),
|
|
pytest.param(
|
|
"gemini/gemini-partial-wildcard-9*",
|
|
"gemini/gemini-partial-wildcard-9*",
|
|
(
|
|
"gemini-partial-wildcard-9-flash",
|
|
"gemini/gemini-partial-wildcard-9-pro",
|
|
"gemini/gemini-other-partial-wildcard-9",
|
|
),
|
|
["gemini/gemini-partial-wildcard-9-flash", "gemini/gemini-partial-wildcard-9-pro"],
|
|
id="mixed-bare-and-prefixed-catalog",
|
|
),
|
|
],
|
|
)
|
|
def test_partial_wildcard_lists_catalog_ids_stored_with_the_provider_prefix(
|
|
wildcard: str, deployment_model: str | None, catalog: tuple[str, ...], expected: list[str]
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
provider: Final = wildcard.split("/", 1)[0]
|
|
deployment: Final = None if deployment_model is None else LiteLLM_Params(model=deployment_model)
|
|
with _registered_chat_models(dict.fromkeys(catalog, provider)):
|
|
listed: Final = get_known_models_from_wildcard(wildcard, deployment)
|
|
|
|
assert sorted(listed) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("provider", "routing_segment"), [("bedrock", "converse"), ("bedrock", "us-east-1"), ("openai", "responses")]
|
|
)
|
|
def test_routing_segment_lists_the_same_catalog_ids_as_the_bare_provider_wildcard(
|
|
provider: str, routing_segment: str
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
bare_listing: Final = get_known_models_from_wildcard(f"{provider}/*", LiteLLM_Params(model=f"{provider}/*"))
|
|
routed_listing: Final = get_known_models_from_wildcard(
|
|
f"{provider}/*", LiteLLM_Params(model=f"{provider}/{routing_segment}/*")
|
|
)
|
|
|
|
assert bare_listing
|
|
assert sorted(routed_listing) == sorted(bare_listing)
|
|
|
|
|
|
def test_provider_wildcard_lists_cost_map_models_but_not_pricing_tiers() -> None:
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
model_id: Final = "together_ai/partial-wildcard-org/model-t"
|
|
pricing_tier: Final = "together-ai-partial-wildcard-tier"
|
|
with _registered_chat_models({model_id: "together_ai", pricing_tier: "together_ai"}):
|
|
listed: Final = get_known_models_from_wildcard("together_ai/*")
|
|
|
|
assert model_id in listed
|
|
assert [listed_id for listed_id in listed if pricing_tier in listed_id] == []
|
|
|
|
|
|
def test_provider_wildcard_stops_listing_a_cost_map_model_once_it_is_retired() -> None:
|
|
import litellm
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
|
|
model_id: Final = "together_ai/partial-wildcard-org/model-retired"
|
|
with _registered_chat_models({model_id: "together_ai"}):
|
|
listed_while_served: Final = get_known_models_from_wildcard("together_ai/*")
|
|
litellm.add_known_models({model_id: {"litellm_provider": "together_ai", "deprecation_date": "2000-01-01"}})
|
|
listed_once_retired: Final = get_known_models_from_wildcard("together_ai/*")
|
|
|
|
assert model_id in listed_while_served
|
|
assert model_id not in listed_once_retired
|
|
|
|
|
|
_ROUTE_CF_ID: Final = "cloudflare/@cf/partial-route-org/model-a-instruct"
|
|
_ROUTE_CF_SUFFIXED_ID: Final = "cloudflare/@cf/partial-route-org/model-b-instruct-v2"
|
|
_ROUTE_HF_ID: Final = "cloudflare/@hf/partial-route-org/model-c"
|
|
_ROUTE_GEMINI_ID: Final = "gemini-partial-route-9-flash"
|
|
_ROUTE_OPENAI_ID: Final = "gpt-partial-route-9-mini"
|
|
_ROUTE_BEDROCK_ID: Final = "anthropic.partial-route-9-v1:0"
|
|
_ROUTE_TOGETHER_ID: Final = "together_ai/partial-route-org/model-t"
|
|
_ROUTE_CF_IDS: Final = {_ROUTE_CF_ID, _ROUTE_CF_SUFFIXED_ID}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model_name", "deployment_model", "key_model", "reachable"),
|
|
[
|
|
pytest.param("cloudflare/@cf/*", "cloudflare/@cf/*", None, _ROUTE_CF_IDS, id="filter"),
|
|
pytest.param(
|
|
"cloudflare/@cf/*-instruct", "cloudflare/@cf/*-instruct", None, {_ROUTE_CF_ID}, id="filter-with-suffix"
|
|
),
|
|
pytest.param(
|
|
"cloudflare/@cf/*",
|
|
"cloudflare/*",
|
|
None,
|
|
{*_ROUTE_CF_IDS, _ROUTE_HF_ID},
|
|
id="alias-whose-prefix-matches-catalog-ids",
|
|
),
|
|
pytest.param("cloudflare/team-*", "cloudflare/*", None, {*_ROUTE_CF_IDS, _ROUTE_HF_ID}, id="alias"),
|
|
pytest.param("cloudflare/cf-*", "cloudflare/@cf/*", None, _ROUTE_CF_IDS, id="alias-over-partial-deployment"),
|
|
pytest.param("my-cf/*", "cloudflare/@cf/*", None, _ROUTE_CF_IDS, id="bare-alias-over-partial-deployment"),
|
|
pytest.param(
|
|
"cloudflare/team-*", "cloudflare/@cf/*/*", None, set(), id="alias-with-fewer-wildcards-than-its-deployment"
|
|
),
|
|
pytest.param("partial-route-v/*/*", "cloudflare/@cf/*/*", None, _ROUTE_CF_IDS, id="multi-wildcard-alias"),
|
|
pytest.param(
|
|
"partial-route-v/*-*", "cloudflare/@cf/*/*", None, set(), id="multi-wildcard-alias-routing-splits-elsewhere"
|
|
),
|
|
pytest.param(
|
|
"partial-route-team/*", "openai/gpt-partial-route-9*", None, set(), id="alias-over-mid-word-deployment"
|
|
),
|
|
pytest.param(
|
|
"team/gpt-partial-route-9*",
|
|
"openai/gpt-partial-route-9*",
|
|
None,
|
|
{f"openai/{_ROUTE_OPENAI_ID}"},
|
|
id="alias-repeating-the-deployment-partial-word",
|
|
),
|
|
pytest.param(
|
|
"team/gpt-partial-route-9-prod*",
|
|
"openai/gpt-partial-route-9*",
|
|
None,
|
|
{f"openai/{_ROUTE_OPENAI_ID}"},
|
|
id="alias-extending-the-deployment-partial-word",
|
|
),
|
|
pytest.param(
|
|
"partial-route-g/*",
|
|
"gemini/gemini-partial-route-*",
|
|
None,
|
|
{_ROUTE_GEMINI_ID},
|
|
id="alias-over-deployment-cut-at-a-word-boundary",
|
|
),
|
|
pytest.param(
|
|
"cloudflare/*", "cloudflare/*", "cloudflare/@cf/*", _ROUTE_CF_IDS, id="key-filter-over-provider-wildcard"
|
|
),
|
|
pytest.param("gemini/gemini-*", "gemini/*", None, {_ROUTE_GEMINI_ID}, id="alias-over-bare-id"),
|
|
pytest.param(
|
|
"gemini/*", "gemini/*", "gemini/gemini-partial-route-9*", {_ROUTE_GEMINI_ID}, id="key-filter-over-bare-id"
|
|
),
|
|
pytest.param(
|
|
"openai/*", "openai/responses/*", None, {f"openai/responses/{_ROUTE_OPENAI_ID}"}, id="responses-route"
|
|
),
|
|
pytest.param(
|
|
"openai/*",
|
|
"openai/responses/*",
|
|
"openai/gpt-partial-route-9*",
|
|
{f"openai/responses/{_ROUTE_OPENAI_ID}"},
|
|
id="key-filter-over-responses-route",
|
|
),
|
|
pytest.param(
|
|
"bedrock/*", "bedrock/converse/*", None, {f"bedrock/converse/{_ROUTE_BEDROCK_ID}"}, id="converse-route"
|
|
),
|
|
pytest.param(
|
|
"openai/partial-route-org/*",
|
|
"openai/partial-route-org/*",
|
|
None,
|
|
set(),
|
|
id="same-name-namespace-the-catalog-lacks",
|
|
),
|
|
pytest.param(
|
|
"together/*", "together_ai/partial-route-missing-org/*", None, set(), id="alias-over-org-the-catalog-lacks"
|
|
),
|
|
pytest.param(
|
|
"openai/*",
|
|
"openai/chat_completions/*",
|
|
None,
|
|
{f"openai/chat_completions/{_ROUTE_OPENAI_ID}"},
|
|
id="chat-completions-bridge-route",
|
|
),
|
|
pytest.param(
|
|
"bedrock/*", "bedrock/llama/*", None, {f"bedrock/llama/{_ROUTE_BEDROCK_ID}"}, id="invoke-format-route"
|
|
),
|
|
pytest.param(
|
|
"together/*",
|
|
"together_ai/partial-route-org/*",
|
|
None,
|
|
{_ROUTE_TOGETHER_ID},
|
|
id="alias-over-org-a-cost-map-catalog-carries",
|
|
),
|
|
pytest.param(
|
|
"bedrock/*", "bedrock/us-east-1/*", None, {f"bedrock/us-east-1/{_ROUTE_BEDROCK_ID}"}, id="region-route"
|
|
),
|
|
pytest.param(
|
|
"*", "openai/*", "openai/gpt-partial-route-9*", set(), id="provider-prefixed-key-filter-over-catch-all"
|
|
),
|
|
pytest.param(
|
|
"gpt-partial-route-*",
|
|
"openai/gpt-partial-route-*",
|
|
None,
|
|
{f"openai/{_ROUTE_OPENAI_ID}"},
|
|
id="slashless-name-repeating-the-deployment-partial-word",
|
|
),
|
|
pytest.param("team-*", "cloudflare/@cf/*", None, _ROUTE_CF_IDS, id="slashless-alias-over-partial-deployment"),
|
|
pytest.param(
|
|
"team-*", "openai/*", None, {f"openai/{_ROUTE_OPENAI_ID}"}, id="slashless-alias-over-provider-catch-all"
|
|
),
|
|
pytest.param(
|
|
"team-*", "cloudflare/@cf/*/*", None, set(), id="slashless-alias-with-fewer-wildcards-than-its-deployment"
|
|
),
|
|
pytest.param(
|
|
"*partial-route-9*",
|
|
"openai/gpt-partial-route-9*",
|
|
None,
|
|
{f"openai/{_ROUTE_OPENAI_ID}"},
|
|
id="slashless-name-with-more-wildcards-than-its-deployment",
|
|
),
|
|
pytest.param(
|
|
"gemini/*partial-route-9*",
|
|
"gemini/gemini-*",
|
|
None,
|
|
{_ROUTE_GEMINI_ID},
|
|
id="slashed-name-with-more-wildcards-than-its-deployment",
|
|
),
|
|
pytest.param("*", "cloudflare/@cf/*", None, _ROUTE_CF_IDS, id="catch-all-over-partial-deployment"),
|
|
pytest.param("*", "gemini/gemini-partial-route-9*", None, set(), id="catch-all-over-mid-word-deployment"),
|
|
pytest.param("*", "openai/responses/gpt-partial-route-9*", None, set(), id="catch-all-over-mid-word-route"),
|
|
pytest.param(
|
|
"*", "gemini/gemini-partial-route-9-flash*", None, set(), id="catch-all-whose-only-match-is-its-fixed-text"
|
|
),
|
|
],
|
|
)
|
|
def test_every_listed_partial_wildcard_id_routes_to_the_catalog_model_it_names(
|
|
model_name: str, deployment_model: str, key_model: str | None, reachable: set[str]
|
|
) -> None:
|
|
from pydantic import TypeAdapter
|
|
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
routed_params: Final = TypeAdapter(tuple[LiteLLM_Params, ...])
|
|
router: Final = Router(model_list=[{"model_name": model_name, "litellm_params": {"model": deployment_model}}])
|
|
|
|
def routed_model(listed_id: str) -> str:
|
|
routed: Final = routed_params.validate_python(
|
|
tuple(deployment["litellm_params"] for deployment in router.wildcard_route_deployments(listed_id, None))
|
|
)
|
|
return routed[0].model if routed else listed_id
|
|
|
|
def routed_catalog_ids(listed: list[str]) -> list[str]:
|
|
routed_models: Final = [routed_model(listed_id) for listed_id in listed]
|
|
return sorted(
|
|
routed.removeprefix("gemini/") if routed.startswith("gemini/gemini-") else routed
|
|
for routed in routed_models
|
|
if "partial-route" in routed
|
|
)
|
|
|
|
with _registered_chat_models(
|
|
{
|
|
_ROUTE_CF_ID: "cloudflare",
|
|
_ROUTE_CF_SUFFIXED_ID: "cloudflare",
|
|
_ROUTE_HF_ID: "cloudflare",
|
|
_ROUTE_GEMINI_ID: "gemini",
|
|
_ROUTE_OPENAI_ID: "openai",
|
|
_ROUTE_BEDROCK_ID: "bedrock",
|
|
_ROUTE_TOGETHER_ID: "together_ai",
|
|
}
|
|
):
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=[key_model or model_name],
|
|
team_models=[],
|
|
proxy_model_list=[model_name],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
model_info_names: Final = [
|
|
str(row["model_name"])
|
|
for row in expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or []], router
|
|
)
|
|
if row["model_name"] != model_name
|
|
]
|
|
listed_v1_models: Final = routed_catalog_ids(v1_models)
|
|
listed_model_info_names: Final = routed_catalog_ids(model_info_names)
|
|
|
|
unrouted_ids: Final = [
|
|
listed_id
|
|
for listed_id in [*v1_models, *model_info_names]
|
|
if not router.wildcard_route_deployments(listed_id, None)
|
|
]
|
|
assert unrouted_ids == []
|
|
assert all(v1_models) and all(model_info_names)
|
|
if model_name == "*":
|
|
assert [listed_id for listed_id in v1_models if not routed_model(listed_id).endswith(f"/{listed_id}")] == []
|
|
assert listed_v1_models == sorted(reachable)
|
|
if key_model is None:
|
|
assert listed_model_info_names == sorted(reachable)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model_name", "expected_v1_models"),
|
|
[
|
|
pytest.param("azure/*", [f"azure/{AZURE_MODEL_PLACEHOLDER}"], id="provider-wildcard-lists-as-before"),
|
|
pytest.param("*", [AZURE_MODEL_PLACEHOLDER], id="catch-all-lists-as-before"),
|
|
pytest.param("my-*", [], id="slashless-alias-skips-the-placeholder"),
|
|
],
|
|
)
|
|
def test_only_slashless_azure_wildcards_skip_the_catalog_placeholder(
|
|
model_name: str, expected_v1_models: list[str]
|
|
) -> None:
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
|
|
router: Final = Router(model_list=[{"model_name": model_name, "litellm_params": {"model": "azure/*"}}])
|
|
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=[model_name],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
model_info_names: Final = [
|
|
str(row["model_name"])
|
|
for row in expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or []], router
|
|
)
|
|
]
|
|
|
|
assert v1_models == expected_v1_models
|
|
assert model_info_names == (expected_v1_models or [model_name])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model_name", "deployment_model"),
|
|
[
|
|
pytest.param("gpt-*", "gpt-*", id="same-wildcards-as-its-deployment"),
|
|
pytest.param("*gpt-5*", "gpt-5*", id="more-wildcards-than-its-deployment"),
|
|
],
|
|
)
|
|
def test_slashless_wildcard_over_a_bare_deployment_lists_its_custom_llm_provider_catalog(
|
|
model_name: str, deployment_model: str
|
|
) -> None:
|
|
from fnmatch import fnmatchcase
|
|
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
from litellm.utils import get_valid_models
|
|
|
|
router: Final = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": model_name,
|
|
"litellm_params": {"model": deployment_model, "custom_llm_provider": "openai", "api_key": "sk-test"},
|
|
}
|
|
]
|
|
)
|
|
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=[model_name],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
model_info_rows: Final = expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or []], router
|
|
)
|
|
openai_ids: Final = frozenset(model.removeprefix("openai/") for model in get_valid_models(custom_llm_provider="openai"))
|
|
routed_models: Final = {
|
|
listed_id: [deployment["litellm_params"]["model"] for deployment in router.wildcard_route_deployments(listed_id, None)]
|
|
for listed_id in v1_models
|
|
}
|
|
|
|
assert v1_models
|
|
assert [
|
|
listed_id for listed_id in v1_models if listed_id not in openai_ids or not fnmatchcase(listed_id, model_name)
|
|
] == []
|
|
assert {listed_id: models for listed_id, models in routed_models.items() if models != [listed_id]} == {}
|
|
assert sorted(str(row["model_name"]) for row in model_info_rows) == sorted(v1_models)
|
|
assert [row for row in model_info_rows if row["litellm_params"]["model"] != row["model_name"]] == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("model_name", "deployment_model"),
|
|
[
|
|
pytest.param("*", "*", id="catch-all-over-a-bare-catch-all"),
|
|
pytest.param("foo/gpt-*", "gpt-*", id="slashed-name-over-a-bare-deployment"),
|
|
pytest.param("gpt-*", "openai/gpt-*", id="slashless-name-over-a-provider-prefixed-deployment"),
|
|
],
|
|
)
|
|
def test_custom_llm_provider_changes_no_listing_outside_slashless_names_over_bare_deployments(
|
|
model_name: str, deployment_model: str
|
|
) -> None:
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
with_provider: Final = get_known_models_from_wildcard(
|
|
model_name, LiteLLM_Params(model=deployment_model, custom_llm_provider="openai")
|
|
)
|
|
without_provider: Final = get_known_models_from_wildcard(model_name, LiteLLM_Params(model=deployment_model))
|
|
|
|
assert with_provider == without_provider
|
|
|
|
|
|
def test_inference_provider_route_lists_every_id_the_plain_provider_wildcard_lists() -> None:
|
|
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
plain: Final = get_known_models_from_wildcard("huggingface/*", LiteLLM_Params(model="huggingface/*"))
|
|
routed: Final = get_known_models_from_wildcard(
|
|
"huggingface/*", LiteLLM_Params(model="huggingface/featherless-ai/*")
|
|
)
|
|
|
|
assert plain
|
|
assert [model for model in plain if "/" not in model.removeprefix("huggingface/")] == []
|
|
assert sorted(routed) == sorted(plain)
|
|
|
|
|
|
def test_catch_all_lists_no_id_a_more_specific_sibling_wildcard_routes_elsewhere() -> None:
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
|
|
unshadowed_id: Final = "cloudflare/@cf/unshadowed-route-org/model-d"
|
|
router: Final = Router(
|
|
model_list=[
|
|
{"model_name": "*", "litellm_params": {"model": "cloudflare/@cf/*"}},
|
|
{"model_name": "partial-route-org/*", "litellm_params": {"model": "openai/*"}},
|
|
]
|
|
)
|
|
|
|
def routed_models(listed: list[str]) -> dict[str, object]:
|
|
return {
|
|
listed_id: router.wildcard_route_deployments(listed_id, None)[0]["litellm_params"]["model"]
|
|
for listed_id in listed
|
|
if "route-org/model-" in listed_id
|
|
}
|
|
|
|
with _registered_chat_models(
|
|
{_ROUTE_CF_ID: "cloudflare", _ROUTE_CF_SUFFIXED_ID: "cloudflare", unshadowed_id: "cloudflare"}
|
|
):
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["*", "partial-route-org/*"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
model_info_names: Final = [
|
|
str(row["model_name"])
|
|
for row in expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or []], router
|
|
)
|
|
]
|
|
|
|
assert routed_models(v1_models) == {"unshadowed-route-org/model-d": unshadowed_id}
|
|
assert routed_models(model_info_names) == {"unshadowed-route-org/model-d": unshadowed_id}
|
|
|
|
|
|
def test_wildcard_lists_an_id_a_more_specific_sibling_serves_with_the_same_model() -> None:
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import expand_wildcard_deployments_for_model_info, get_complete_model_list
|
|
|
|
router: Final = Router(
|
|
model_list=[
|
|
{"model_name": "openai/gpt-partial-route-9*", "litellm_params": {"model": "openai/gpt-partial-route-9*"}},
|
|
{
|
|
"model_name": "openai/gpt-partial-route-9-mini*",
|
|
"litellm_params": {"model": "openai/gpt-partial-route-9-mini*"},
|
|
},
|
|
]
|
|
)
|
|
with _registered_chat_models({_ROUTE_OPENAI_ID: "openai"}):
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=["openai/gpt-partial-route-9*"],
|
|
team_models=[],
|
|
proxy_model_list=["openai/gpt-partial-route-9*", "openai/gpt-partial-route-9-mini*"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
model_info_names: Final = [
|
|
str(row["model_name"])
|
|
for row in expand_wildcard_deployments_for_model_info(
|
|
[dict(deployment) for deployment in router.get_model_list() or []], router
|
|
)
|
|
]
|
|
|
|
assert v1_models == [f"openai/{_ROUTE_OPENAI_ID}"]
|
|
assert "openai/gpt-partial-route-9*" not in model_info_names
|
|
|
|
|
|
def test_prefixed_alias_over_a_bare_catch_all_lists_ids_that_route_through_it(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from litellm import Router
|
|
from litellm.proxy.auth.model_checks import get_complete_model_list
|
|
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-partial-route-test")
|
|
router: Final = Router(model_list=[{"model_name": "team-partial-route/*", "litellm_params": {"model": "*"}}])
|
|
with _registered_chat_models({_ROUTE_OPENAI_ID: "openai"}):
|
|
v1_models: Final = get_complete_model_list(
|
|
key_models=[],
|
|
team_models=[],
|
|
proxy_model_list=["team-partial-route/*"],
|
|
user_model=None,
|
|
infer_model_from_keys=False,
|
|
llm_router=router,
|
|
)
|
|
|
|
unrouted_ids: Final = [
|
|
listed_id for listed_id in v1_models if not router.wildcard_route_deployments(listed_id, None)
|
|
]
|
|
assert f"team-partial-route/{_ROUTE_OPENAI_ID}" in v1_models
|
|
assert unrouted_ids == []
|