litellm/tests/unit/proxy/auth/test_model_checks.py
simonp9 1f15088fc9
feat(proxy): mixed Codex model picker and LiteLLM routing on native codex routes (#49408)
* feat(proxy): mixed Codex model picker and LiteLLM routing on native codex routes

The native Codex catalog now lists the key's other chat and responses models after
ChatGPT's, within Codex's catalog size limit, and drops the ETag so the merged list
refreshes on Codex's timer. A request for a model configured by exact name on another
provider goes through /v1/responses without the ChatGPT bearer or account header,
while ChatGPT names and wildcard-only matches keep the subscription passthrough.
Reasoning items another provider produced are dropped before forwarding to ChatGPT.

* fix(proxy): keep Codex's picker from turning reasoning off for LiteLLM models

Codex fills a missing default_reasoning_level with effort "none" when a model is picked in /model, while
`codex -m <model>` sends no effort at all. A thinking-only provider model then answers 400. On the native
/backend-api/codex/responses route, drop effort "none" for a LiteLLM model whose catalog entry names no
default, so the provider applies its own default exactly as `codex -m` does

* fix(proxy): harden Codex mixed-model routing on the native ChatGPT routes

Keep the ChatGPT credentials out of hook raw headers on the inner /v1/responses
request, route to LiteLLM and merge LiteLLM catalog rows only for keys that may
call /v1/responses, drop unpersisted reasoning ids on the passthrough when store
is not true, keep any slug the caller's ChatGPT account lists on the subscription
via a shared account catalog cache, and keep X-Models-Etag for keys whose catalog
added no LiteLLM rows. The guide covers the new behavior and the /v1 wildcard
precedence over a catch-all deployment.

* fix(proxy): keep Codex's native route on the subscription when the account list is unknown

A name OpenAI lists for the caller's ChatGPT account now goes to LiteLLM only when the
account's list is known and leaves it out. A failed catalog read is remembered for 60 s and
the Codex version comes only from a User-Agent whose product matches the originator header.

The merged catalog serves a stable ETag digest of OpenAI's ETag and the key's LiteLLM rows,
and /responses replaces X-Models-Etag with the same value, memoized per key for 300 s, so
Codex stops refetching or dropping its catalog cache.

LiteLLM-routed native replies wrap their reasoning with the model id, so an OpenAI or Azure
blob is dropped before chatgpt.com and unwrapped for LiteLLM, and the chat-completions bridge
decodes wrapped thinking blocks with their signature.

The access-group recipe uses chatgpt/ model names so /v1 routing does not change, and the
native route checks the chatgpt/ form of a name the key holds only through that group. A
none reasoning effort is dropped only under drop_params. The guide covers app tools, every
routing case, and the ETag, and the catalog logs how many models the byte limit left out.

* fix(fireworks_ai): move Codex additional_tools items into tools on the Responses API

* fix(proxy): keep Codex native turns working when the key's model listing fails

A listing callback that raises no longer turns chatgpt.com's catalog or
responses answer into a 500. The catalog keeps upstream's body and ETag, and
responses keep upstream's X-Models-Etag.

Under drop_params, a `none` reasoning effort is now dropped only when a
candidate deployment does not explicitly support it, so an alias to a model
like openai/gpt-5.4 keeps it.

The account slug list and its unavailable marker are keyed on the account id
plus the caller's ChatGPT Authorization header, so a failed fetch with one
bearer cannot turn off LiteLLM routing for another caller on the same account.

The auth step checks the Codex route before importing the Codex module.

* fix(proxy): keep Codex model switches working across LiteLLM providers

On the Codex native route, a LiteLLM model now gets only the encrypted reasoning every deployment it can route to is able to decrypt, so ChatGPT or another deployment's blobs no longer make xAI answer 400

Codex additional_tools items are hoisted into tools for every provider except the OpenAI API, and namespace tools become flat tools named mcp__docs__search, the shape xAI accepts

A JWT caller's models ETag is cached on a hash of its LiteLLM credential, so ChatGPT turns stop rebuilding the catalog rows

* fix(proxy): keep Codex namespace tool calls intact across LiteLLM providers

Fireworks and Bedrock Mantle accept Codex namespace tools natively, so the
native route now hoists them without flattening, like OpenAI and Azure and
the chat completions bridge. Other native Responses providers still get
flat names, now built with the bridge's namespace_tool_name_map, and each
function_call they return gets its tool name and namespace back in the
non-streaming body and in streamed output_item.added, output_item.done,
and completed/incomplete/failed events. Every other event passes through
unchanged

* fix(proxy): bridge Codex turns for providers whose Responses API rejects Codex tools

Codex sends its freeform exec tool as a custom tool on every turn, and xAI's
Responses API answers 422 for custom, namespace and additional_tools shapes.
On the Codex native route, a model whose deployments include a native
Responses provider other than OpenAI, Azure, Fireworks or Bedrock Mantle now
asks for the chat completions bridge with use_chat_completions_api, which
already turns custom and namespace tools into functions, hoists
additional_tools, and puts the namespace back on each returned call

OpenAI and Azure still get the request unchanged, and Fireworks and Bedrock
Mantle still get it on their own Responses API with additional_tools hoisted.
The route's namespace flattening and the response-side namespace restore are
removed, since no path sends flattened names anymore

* fix(proxy): drop reasoning ids a Codex turn's LiteLLM target cannot resolve

A reasoning item another deployment produced, replayed with store false and no blob the
target can decrypt, now reaches the target as its readable summary without id or status,
and is dropped when nothing readable is left. OpenAI answered 400 on the status field and
404 on the id for a glm-5p3 item replayed whole to an openai deployment. Same-deployment
replays and stored conversations keep the item whole

* refactor(codex): reach encrypted reasoning stripping through the router registry

The Codex route read Router.optional_callbacks and imported EncryptedContentAffinityCheck from
litellm.router_utils, which the router surface fence forbids outside the router. RouterRegistry
gains decryptable_reasoning(model_group, request_input, targets), which keeps the same rule: with
encrypted content affinity on for the group, reasoning one target can decrypt stays, otherwise
every target must decrypt it

* fix(proxy): keep Codex's priority tier off LiteLLM rows in the native model catalog

* fix(proxy): drop Codex's none effort for an alias whose native entry names no default

* fix(proxy): start a native Codex alias at its pinned reasoning effort and build the catalog off the event loop

* test(proxy): map the pinned reasoning efforts field in the router registry stub

* fix(proxy): honor admin-only /v1/responses and keep Codex's default tier off native aliases

Codex native routes now send a key to LiteLLM only when admin_only_routes allows it,
a native alias with configured service tiers starts with no tier picked, and the
account slug cache declares its Redis key family.

* fix(proxy): check the key's tool allowlist on native Codex LiteLLM turns and additional_tools items

* fix(proxy): check vector store grants on file_search tools inside additional_tools items

A Responses request can define file_search inside an additional_tools input item, which the native Codex route hoists for LiteLLM-served models and OpenAI runs as is on /v1/responses. The vector store check now reads those tools in both strict and legacy modes. The input item scan uses a Mapping guard instead of a pydantic error per item, since auth now runs it on every list input.

* fix(proxy): apply the proxy-wide allowed_routes to LiteLLM models on native Codex routes

* fix(proxy): cap native Codex account catalog reads per LiteLLM caller before rate limits apply

* fix(guardrails): check additional_tools items next to top-level function tools in tool_policy

ToolPolicyGuardrail read tools from the request body only when no top-level
function tool existed, so a blocked tool defined in an additional_tools item
reached Fireworks once the Responses transform moved the item into tools after
the pre-call hooks. Check every list on each request, including namespace
members inside items.

* fix(router): serve decryptable_reasoning from the Rust core's optional checks

The merge with main removed the Python router core, so the registry's host has no
optional_callbacks and no routing context, and native Codex requests with encrypted
reasoning raised RustCoreUnsupported. The filter now lives with the optional checks,
which own both. The ported pinned_reasoning_efforts field gets a test on the Rust core

Also give the static litellm.Router a declared constructor. Since the Python core was
deleted a bare __init__ assignment reads as RustCore's, so every Router(model_list=...)
is a type error and make lint fails on main. TYPE_CHECKING only, no runtime change

* fix(proxy): check the budget of a saved alias target on native Codex LiteLLM turns

Native /backend-api/codex/responses binds {endpoint}, so auth skipped the key's or team's model_group_alias and read budgets on the requested name, while the later /v1/responses handoff served the target unchecked. Auth now resolves the alias on a turn that routes to LiteLLM, as it does on /v1, and leaves ChatGPT-only names on the subscription. Also makes the tool_policy tests build a real ToolPolicyRegistry and type their fixtures.

* fix(proxy): keep a name the ChatGPT account lists on the subscription despite a saved alias

* fix(codex): keep LiteLLM keys off ChatGPT, enforce tool policy on namespaces and pin reasoning on aliases

Auth no longer asks ChatGPT for the account's model list when the bearer is
missing or is the LiteLLM key, so a key sent as the bearer never reaches it.

tool_policy now reads the members of a top-level namespace, which is where
the native route's hoist puts them.

A global alias now reads encrypted content affinity from its target group.

A non-list top-level tools value and a non-string function name now reach the
provider, which refuses them, instead of raising a 500.

The catalog ETag no longer depends on the order a process iterates priced
service tiers in, so workers and pods agree on it.

* fix(proxy): list a team's models in the order the team set them

get_team_models built the names in a set, so each process listed a team key's models in its own hash order and the native Codex catalog sent a different row order and ETag from each worker

* fix(proxy): apply JWT route policy before native Codex routes serve LiteLLM models

A standard JWT caller now gets a LiteLLM model on /backend-api/codex/models
and /backend-api/codex/responses only when the JWT auth flow would admit a
direct /v1/responses call for that model, so team_allowed_routes,
admin_allowed_routes and RBAC route limits hold on the handoff too.

The Fireworks Responses transform also leaves additional_tools input items
in place when the tools param is not a list, instead of raising while hoisting

* fix(fireworks): hoist additional_tools when tools is a tuple

litellm.responses passes a tuple tools argument through to the Fireworks
transform unchanged, so the non-list guard skipped hoisting and left the
additional_tools item in the input. Treat tuples like lists and cover both
shapes in the drop_params merge test

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
2026-11-06 11:14:47 -08:00

2486 lines
92 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_lists_models_in_the_order_the_team_set_them() -> None:
from litellm.proxy.auth.model_checks import get_team_models
team_models: Final = ["zeta", "alpha", "mu", "beta", "omega", "kappa", "delta", "all-proxy-models"]
result: Final = get_team_models(team_models, ["proxy-b", "proxy-a"], {"grp": ["g1"]}, True)
assert result == [*team_models, "proxy-b", "proxy-a", "grp", "g1"]
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 == []
@pytest.fixture
def fresh_available_models_cache(monkeypatch: pytest.MonkeyPatch) -> None:
import litellm.utils as litellm_utils
monkeypatch.setattr(litellm_utils, "_model_cache", litellm_utils.AvailableModelsCache())
def test_get_provider_models_admits_providers_without_a_static_catalog():
from litellm.proxy.auth.model_checks import get_provider_models
from litellm.types.router import LiteLLM_Params
result = get_provider_models(
"hosted_vllm",
litellm_params=LiteLLM_Params(
model="hosted_vllm/*",
api_base="http://upstream:4000",
api_key="sk-upstream",
),
)
assert result == []
@pytest.mark.parametrize("wildcard", ["litellm_proxy/*", "team-proxy/*"])
@pytest.mark.usefixtures("fresh_available_models_cache")
@respx.mock
def test_get_complete_model_list_discovers_litellm_proxy_wildcard_models_by_default(
monkeypatch: pytest.MonkeyPatch, wildcard: str
) -> None:
import litellm
from litellm.proxy.auth.model_checks import get_complete_model_list
monkeypatch.setattr(litellm, "check_provider_endpoint", False)
catalog: Final = respx.get(
"http://upstream-discovery.test:4000/gateway/v1/models",
headers={"Authorization": "Bearer test-upstream-key"},
).mock(return_value=httpx.Response(200, json={"data": [{"id": "chat-model"}, {"id": "vendor/chat-model"}]}))
router: Final = Router(
model_list=[
{
"model_name": wildcard,
"litellm_params": {
"model": "litellm_proxy/*",
"api_base": "http://upstream-discovery.test:4000/gateway/v1",
"api_key": "test-upstream-key",
},
}
]
)
result: Final = get_complete_model_list(
key_models=[],
team_models=[],
proxy_model_list=list(router.registry().names(kind="group_set")),
user_model=None,
infer_model_from_keys=False,
llm_router=router,
)
assert result == [wildcard.replace("*", "chat-model"), wildcard.replace("*", "vendor/chat-model")]
assert catalog.call_count == 1
@pytest.mark.usefixtures("fresh_available_models_cache")
@respx.mock
def test_litellm_proxy_discovery_cache_is_scoped_to_the_upstream_key_and_gateway(
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", False)
first: Final = respx.get(
"http://first-proxy.test:4000/v1/models", headers={"Authorization": "Bearer test-first-key"}
).mock(return_value=httpx.Response(200, json={"data": [{"id": "first-key-model"}]}))
second: Final = respx.get(
"http://first-proxy.test:4000/v1/models", headers={"Authorization": "Bearer test-second-key"}
).mock(return_value=httpx.Response(200, json={"data": [{"id": "second-key-model"}]}))
other_gateway: Final = respx.get(
"http://other-proxy.test:4000/v1/models", headers={"Authorization": "Bearer test-first-key"}
).mock(return_value=httpx.Response(200, json={"data": [{"id": "other-gateway-model"}]}))
deployment: Final = LiteLLM_Params(
model="litellm_proxy/*", api_base="http://first-proxy.test:4000/v1", api_key="test-first-key"
)
assert get_known_models_from_wildcard("litellm_proxy/*", deployment) == ["litellm_proxy/first-key-model"]
assert get_known_models_from_wildcard(
"litellm_proxy/*", deployment.model_copy(update={"api_key": "test-second-key"})
) == ["litellm_proxy/second-key-model"]
assert get_known_models_from_wildcard(
"litellm_proxy/*", deployment.model_copy(update={"api_base": "http://other-proxy.test:4000/v1"})
) == ["litellm_proxy/other-gateway-model"]
assert get_known_models_from_wildcard("litellm_proxy/*", deployment) == ["litellm_proxy/first-key-model"]
assert (first.call_count, second.call_count, other_gateway.call_count) == (1, 1, 1)
@pytest.mark.usefixtures("fresh_available_models_cache")
@respx.mock
def test_undeployed_litellm_proxy_wildcard_does_not_list_the_environment_gateway(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import litellm
from litellm.proxy.auth.model_checks import get_complete_model_list
monkeypatch.setattr(litellm, "check_provider_endpoint", False)
monkeypatch.setenv("LITELLM_PROXY_API_BASE", "http://environment-proxy.test:4000")
monkeypatch.setenv("LITELLM_PROXY_API_KEY", "test-environment-key")
catalog: Final = respx.get("http://environment-proxy.test:4000/v1/models").mock(
return_value=httpx.Response(200, json={"data": [{"id": "environment-only-model"}]})
)
listed: Final = get_complete_model_list(
key_models=["litellm_proxy/*"],
team_models=[],
proxy_model_list=[],
user_model=None,
infer_model_from_keys=False,
llm_router=Router(model_list=[]),
)
assert (listed, catalog.call_count) == ([], 0)
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 == []