scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-31 06:56:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 245

Merge 6de7d49616 into a392a39627

Browse source
This commit is contained in:
paveln83 2026-10-31 06:37:46 +00:00 • committed by GitHub
commit 119c4036f9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 887 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
litellm/proxy/auth/model_checks.py
View file

@ -8,6 +8,7 @@ import itertools
import re
from collections.abc import Callable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, replace
from types import MappingProxyType
from typing import Final, TypeVar
@ -17,10 +18,12 @@ from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.litellm_core_utils.pattern_match_deployments import PatternMatchRouter
from litellm.llms.azure.ptu_capacity import is_str_keyed_mapping
from litellm.proxy._types import SpecialModelNames, UserAPIKeyAuth
from litellm.proxy.common_utils.model_listing_utils import CallerAliases, alias_map, dispatched_model
from litellm.proxy.common_utils.token_count_settings import cost_map_entry
from litellm.repositories.object_permission_repository import ObjectPermissionRepository
from litellm.responses.main import strip_responses_routing_prefix
from litellm.router import Router
from litellm.router_api import fallback_chain
from litellm.router_api import RouterRegistry, fallback_chain
from litellm.types.router import CredentialLiteLLMParams, LiteLLM_Params
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager, get_valid_models, potential_model_names_in_lookup_order
@ -626,6 +629,105 @@ def _known_models_without_a_deployment(wildcard_model: str) -> Sequence[str]:
return get_known_models_from_wildcard(wildcard_model=wildcard_model, litellm_params=None)
def _litellm_model(deployment: Mapping[str, object]) -> object:
litellm_params: Final = deployment.get("litellm_params")
return litellm_params.get("model") if is_str_keyed_mapping(litellm_params) else None
def _router_deployment(registry: RouterRegistry, model_id: str) -> Mapping[str, object] | None:
deployment: Final[object] = registry.deployment_dict(model_id)
return deployment if is_str_keyed_mapping(deployment) else None
GLOBAL_ALIASES_ONLY: Final = CallerAliases(team=None, key=None)
@dataclass(frozen=True, slots=True)
class RoutedElsewhere:
pass
ROUTED_ELSEWHERE: Final = RoutedElsewhere()
def wildcard_row_priced_model(
row: Mapping[str, object], llm_router: Router | None, caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY
) -> str | RoutedElsewhere | None:
model_id: Final = _model_info_field(row, "id")
listed_name: Final = row.get("model_name")
if llm_router is None or not isinstance(model_id, str) or not isinstance(listed_name, str):
return None
registry: Final = llm_router.registry()
source: Final = _router_deployment(registry, model_id)
if source is None:
return None
source_pattern: Final = source.get("model_name")
source_model: Final = _litellm_model(source)
if not isinstance(source_pattern, str) or not isinstance(source_model, str) or _litellm_model(row) == source_model:
return None
return wildcard_deployment_priced_model(
llm_router,
WildcardDeployment(model_id, source_pattern, source_model, _deployment_team_id(row)),
listed_name,
caller_aliases,
)
@dataclass(frozen=True, slots=True)
class WildcardDeployment:
model_id: str
model_name_pattern: str
litellm_model: str
team_id: str | None
def wildcard_deployment_priced_model(
llm_router: Router, deployment: WildcardDeployment, listed_name: str, caller_aliases: CallerAliases
) -> str | RoutedElsewhere:
served_name: Final = _served_name(llm_router, listed_name, caller_aliases)
candidates: Final = llm_router.registry().names(
served_name, kind="route_candidates", team_id=deployment.team_id or caller_aliases.team_id
)
if deployment.model_id not in candidates:
return ROUTED_ELSEWHERE
return _routed_deployment_model(
deployment.model_name_pattern, routed_model_name(llm_router, served_name), deployment.litellm_model
)
def wildcard_row_filled_model(
llm_router: Router, name_pattern: str, litellm_model: str, listed_name: str, caller_aliases: CallerAliases
) -> str:
served_name: Final = _served_name(llm_router, listed_name, caller_aliases)
return _routed_deployment_model(name_pattern, routed_model_name(llm_router, served_name), litellm_model)
def _served_name(llm_router: Router, listed_name: str, caller_aliases: CallerAliases) -> str:
return dispatched_model(listed_name, _without_deleted_team_targets(caller_aliases, llm_router.registry()))
def _without_deleted_team_targets(caller_aliases: CallerAliases, registry: RouterRegistry) -> CallerAliases:
deleted_prefix: Final = f"model_name_{caller_aliases.team_id}_"
live_names: Final = registry.index("model_name")
return replace(
caller_aliases,
team={
alias: target
for alias, target in alias_map(caller_aliases.team).items()
if not target.startswith(deleted_prefix) or target in live_names
},
)
def routed_model_name(llm_router: Router, listed_name: str) -> str:
return llm_router.registry().resolve(listed_name, kind="alias") or listed_name
def model_info_cost_map_entry(model: str) -> Mapping[str, object]:
provider, _, _ = model.partition("/")
return cost_map_entry(model) or cost_map_entry(model.rsplit("/", 1)[-1], custom_llm_provider=provider)
def _get_wildcard_models(
unique_models: list[str],
return_wildcard_routes: bool | None = False,

7
litellm/proxy/common_utils/model_listing_utils.py
View file

@ -160,8 +160,8 @@ class ClaudeCodeRoutingNames:
@dataclass(frozen=True, slots=True)
class CallerAliases:
"""The caller's `team` and `key` alias maps, the names `/v1/models` lists for it, and `router`,
its key's or team's `router_settings.model_group_alias`.
"""The caller's `team` and `key` alias maps, the names `/v1/models` lists for it, `router`,
its key's or team's `router_settings.model_group_alias`, and `team_id`, the key's team.
`rewrite` are the maps `/chat/completions` rewrites its model through after auth's `router`
rewrite, in the order it applies them: the team's, the key's in `add_litellm_data_to_request`,
then the global `model_alias_map` and the key's again in `common_processing_pre_call_logic`,
@ -170,6 +170,7 @@ class CallerAliases:
team: object
key: object
router: object = None
team_id: str | None = None
@property
def own(self) -> tuple[object, ...]:
@ -214,7 +215,7 @@ def caller_alias_maps(
"""Team aliases count only when listing the team the key authenticated as."""
if listed_team_id is not None and listed_team_id != key_team_id:
return CallerAliases(None, key_aliases, router_aliases)
return CallerAliases(team_aliases, key_aliases, router_aliases)
return CallerAliases(team_aliases, key_aliases, router_aliases, key_team_id)
def alias_map(aliases: object) -> Mapping[str, str]:

86
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -15,7 +15,7 @@ import datetime
import itertools
import json
from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping, Sequence
from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from dataclasses import dataclass
from fnmatch import fnmatchcase
from functools import partial
@ -61,6 +61,11 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.auth.litellm_license import AUTO_ROUTER_LICENSE_REMEDY
from litellm.proxy.auth.model_checks import (
GLOBAL_ALIASES_ONLY,
model_info_cost_map_entry,
wildcard_row_filled_model,
)
from litellm.proxy.auth.team_grants import team_model_aliases
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.config_sync_pubsub import (
@ -77,6 +82,7 @@ from litellm.proxy.common_utils.encrypt_decrypt_utils import (
decrypt_value_helper,
encrypt_value_helper,
)
from litellm.proxy.common_utils.model_listing_utils import CallerAliases
from litellm.proxy.common_utils.openai_error_payload import proxy_exception_for
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.db.routing_prisma_wrapper import WriterPinnedClient
@ -1287,14 +1293,31 @@ def _ptu_priced_deployment(model_params: Deployment) -> Deployment:
)
def _cost_map_entry(db_model: Deployment, incoming_model_info: Mapping[str, object]) -> Mapping[str, object]:
@dataclass(frozen=True, slots=True)
class _ListedRow:
name: str | None = None
llm_router: Router | None = None
caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY
_NO_LISTED_ROW: Final = _ListedRow()
def _live_cost_map_model(db_model: Deployment, listed_row: _ListedRow) -> str | None:
stored_model: Final = _decrypted_model(db_model.litellm_params.model)
if stored_model is None or "*" not in stored_model or listed_row.name is None or listed_row.llm_router is None:
return stored_model
return wildcard_row_filled_model(
listed_row.llm_router, db_model.model_name, stored_model, listed_row.name, listed_row.caller_aliases
)
def _cost_map_entry(
db_model: Deployment, incoming_model_info: Mapping[str, object], listed_row: _ListedRow = _NO_LISTED_ROW
) -> Mapping[str, object]:
base_model: Final = incoming_model_info.get("base_model")
lookup: Final = base_model if isinstance(base_model, str) else _decrypted_model(db_model.litellm_params.model)
if lookup is None:
return MappingProxyType({})
with suppress(Exception):
return MappingProxyType(dict(litellm.get_model_info(model=lookup)))
return MappingProxyType({})
lookup: Final = base_model if isinstance(base_model, str) else _live_cost_map_model(db_model, listed_row)
return MappingProxyType({}) if lookup is None else model_info_cost_map_entry(lookup)
LoadedCatalog: TypeAlias = Callable[[], Mapping[str, Mapping[str, object]]]
@ -1334,6 +1357,7 @@ def legacy_update_model_info_changes(
incoming_model_info: ModelInfo,
discovered_model_info: Mapping[str, object] = MappingProxyType({}),
loaded_catalog: LoadedCatalog = GetModelCostMap.loaded_model_cost_map,
listed_row: _ListedRow = _NO_LISTED_ROW,
) -> Mapping[str, object]:
"""The ``model_info`` fields ``POST /model/update`` persists from its request.
@ -1351,7 +1375,7 @@ def legacy_update_model_info_changes(
echoed: Final = frozenset(
echoed_cost_map_fields(
requested,
_cost_map_entry(db_model, {**stored, **sent}),
_cost_map_entry(db_model, {**stored, **sent}, listed_row),
_loaded_catalog_entry(requested, loaded_catalog),
discovered_model_info,
)
@ -1370,9 +1394,10 @@ def _legacy_update_model_info_writes(
incoming_model_info: ModelInfo,
discovered_model_info: Mapping[str, object],
member_marker: bool | None,
listed_row: _ListedRow,
) -> Mapping[str, object]:
changes: Final = {
**legacy_update_model_info_changes(current, incoming_model_info, discovered_model_info),
**legacy_update_model_info_changes(current, incoming_model_info, discovered_model_info, listed_row=listed_row),
**({"member_auto_router": member_marker} if member_marker is not None else {}),
}
return MappingProxyType({k: v for k, v in changes.items() if getattr(current.model_info, k, None) != v})
@ -1384,13 +1409,14 @@ async def _locked_legacy_update_model_info_json(
incoming_model_info: ModelInfo,
discovered_model_info: Mapping[str, object],
member_marker: bool | None,
listed_row: _ListedRow,
) -> str | None:
row: Final = await table.find_unique(where={"model_id": model_id})
if row is None:
return None
payload: Final = row.model_dump()
writes: Final = _legacy_update_model_info_writes(
Deployment(**payload), incoming_model_info, discovered_model_info, member_marker
Deployment(**payload), incoming_model_info, discovered_model_info, member_marker, listed_row
)
if not writes:
return None
@ -1420,6 +1446,9 @@ def update_db_model(
db_model: Deployment,
updated_patch: updateDeployment,
loaded_catalog: LoadedCatalog = GetModelCostMap.loaded_model_cost_map,
llm_router: Router | None = None,
caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY,
listed_name: str | None = None,
) -> PrismaCompatibleUpdateDBModel:
if updated_patch.model_info is not None:
_raise_if_ptu_cost_attribution_disabled(updated_patch.model_info.model_dump(exclude_none=True))
@ -1451,7 +1480,11 @@ def update_db_model(
incoming_model_info: Final = updated_patch.model_info.model_dump(exclude_none=True)
echoed_fields: Final = echoed_cost_map_fields(
incoming_model_info,
_cost_map_entry(db_model, incoming_model_info),
_cost_map_entry(
db_model,
incoming_model_info,
_ListedRow(listed_name or updated_patch.model_name, llm_router, caller_aliases),
),
_loaded_catalog_entry(incoming_model_info, loaded_catalog),
)
stored_model_info_fields: Final[Mapping[str, object]] = without_server_derived_pricing(incoming_model_info)
@ -1631,6 +1664,7 @@ async def patch_model(
llm_router,
premium_user,
prisma_client,
request_caller_aliases,
store_model_in_db,
user_api_key_cache,
)
@ -1800,6 +1834,8 @@ async def patch_model(
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
write_row=write_row,
llm_router=llm_router,
caller_aliases=await request_caller_aliases(user_api_key_dict),
)
if updated_model is None:
@ -2155,6 +2191,8 @@ async def _update_team_model_in_db(
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
write_row: Callable[[PrismaCompatibleUpdateDBModel], Awaitable[_RowT]],
llm_router: Router | None = None,
caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY,
) -> _RowT:
"""
Handle team model updates with proper alias management.
@ -2199,7 +2237,11 @@ async def _update_team_model_in_db(
# No team_id in patch, proceed with standard update
if patch_team_id is None:
return await write_row(update_db_model(db_model=db_model, updated_patch=patch_data))
return await write_row(
update_db_model(
db_model=db_model, updated_patch=patch_data, llm_router=llm_router, caller_aliases=caller_aliases
)
)
# Determine public model name
public_model_name: Final = _get_public_model_name(
@ -2220,7 +2262,15 @@ async def _update_team_model_in_db(
),
}
)
row: Final = await write_row(update_db_model(db_model=db_model, updated_patch=team_patch))
row: Final = await write_row(
update_db_model(
db_model=db_model,
updated_patch=team_patch,
llm_router=llm_router,
caller_aliases=caller_aliases,
listed_name=patch_data.model_name,
)
)
if is_new_team_assignment:
await _setup_new_team_model_assignment(
@ -3265,6 +3315,7 @@ async def update_model(
llm_router,
premium_user,
prisma_client,
request_caller_aliases,
store_model_in_db,
user_api_key_cache,
)
@ -3401,8 +3452,11 @@ async def update_model(
if llm_router is not None
else MappingProxyType({})
)
listed_row: Final = _ListedRow(
model_params.model_name, llm_router, await request_caller_aliases(user_api_key_dict)
)
requested_model_info: Final = _legacy_update_model_info_writes(
deployment, _model_info, discovered_model_info, member_marker
deployment, _model_info, discovered_model_info, member_marker, listed_row
)
_raise_if_access_groups_malformed(requested_model_info)
async with _auto_router_capability_slot(
@ -3414,7 +3468,7 @@ async def update_model(
) as table:
model_info_json: Final = (
await _locked_legacy_update_model_info_json(
table, _model_id, _model_info, discovered_model_info, member_marker
table, _model_id, _model_info, discovered_model_info, member_marker, listed_row
)
if requested_model_info
else None

103
litellm/proxy/proxy_server.py
View file

@ -439,13 +439,17 @@ from litellm.proxy.auth.master_key_boot_check import (
with_stored_secrets_counted,
)
from litellm.proxy.auth.model_checks import (
GLOBAL_ALIASES_ONLY,
RoutedElsewhere,
expand_wildcard_deployments_for_model_info,
get_all_fallbacks,
get_complete_model_list,
get_key_models,
get_mcp_server_ids,
get_team_models,
model_info_cost_map_entry,
run_model_listing,
wildcard_row_priced_model,
)
from litellm.proxy.auth.password_policy import validate_password_not_breached, validate_password_policy
from litellm.proxy.auth.user_api_key_auth import ( # noqa: F401, RUF100 # legacy module exports
@ -517,6 +521,7 @@ from litellm.proxy.common_utils.http_parsing_utils import ( # noqa: F401, RUF10
from litellm.proxy.common_utils.load_config_utils import get_config_from_bucket
from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations
from litellm.proxy.common_utils.model_listing_utils import (
CallerAliases,
ClaudeCodeRoutingNames,
DeployedModelGroups,
TeamModelNameTranslator,
@ -12412,16 +12417,7 @@ async def model_list(
http_request is not None and http_request.headers.get("anthropic-version") is not None
)
client_headers: Final[Mapping[str, str]] = http_request.headers if http_request is not None else _EMPTY_HEADERS
caller_router_settings: Final = await proxy_config.get_hierarchical_router_settings(
user_api_key_dict, prisma_client, proxy_logging_obj
)
caller_aliases: Final = caller_alias_maps(
user_api_key_dict.aliases,
user_api_key_dict.team_model_aliases,
user_api_key_dict.team_id,
team_id,
caller_router_settings.get("model_group_alias") if isinstance(caller_router_settings, Mapping) else None,
)
caller_aliases: Final = await request_caller_aliases(user_api_key_dict, team_id)
routing_names: Final = ClaudeCodeRoutingNames(
llm_router,
team_id or user_api_key_dict.team_id,
@ -12703,16 +12699,7 @@ async def model_info(
)
hidden_names: Final = blocked_names | unhealthy_names
internal_to_public: Final = TeamModelNameTranslator.build_internal_to_public_map(llm_router, settings)
caller_router_settings: Final = await proxy_config.get_hierarchical_router_settings(
user_api_key_dict, prisma_client, proxy_logging_obj
)
caller_aliases: Final = caller_alias_maps(
user_api_key_dict.aliases,
user_api_key_dict.team_model_aliases,
user_api_key_dict.team_id,
team_id,
caller_router_settings.get("model_group_alias") if isinstance(caller_router_settings, Mapping) else None,
)
caller_aliases: Final = await request_caller_aliases(user_api_key_dict, team_id)
callback_hidden_names: Final = await _names_hidden_by_listing_callbacks(
user_api_key_dict,
tuple(
@ -16129,7 +16116,10 @@ async def get_all_team_and_direct_access_models(
def _enrich_model_info_with_litellm_data(
model: dict[str, Any], debug: bool = False, llm_router: Router | None = None
model: dict[str, Any],
debug: bool = False,
llm_router: Router | None = None,
caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY,
) -> dict[str, Any]:
"""
Enrich a model dictionary with litellm model info (pricing, context window, etc.)
@ -16139,6 +16129,7 @@ def _enrich_model_info_with_litellm_data(
model: Model dictionary to enrich
debug: Whether to include debug information like openai_client
llm_router: Optional router instance for debug info
caller_aliases: The caller's alias maps, applied before deciding which model a wildcard row prices as
Returns:
Enriched model dictionary with sensitive info removed
@ -16156,30 +16147,23 @@ def _enrich_model_info_with_litellm_data(
# read litellm model_prices_and_context_window.json to get the following:
# input_cost_per_token, output_cost_per_token, max_tokens
litellm_model_info = get_litellm_model_info(model=model)
# 2nd pass on the model, try seeing if we can find model in litellm model_cost map
if litellm_model_info == {}:
# use litellm_param model_name to get model_info
litellm_params = model.get("litellm_params", {})
litellm_model = litellm_params.get("model", None)
try:
litellm_model_info = litellm.get_model_info(model=litellm_model)
except Exception:
litellm_model_info = {}
# 3rd pass on the model, try seeing if we can find model but without the "/" in model cost map
if litellm_model_info == {}:
# use litellm_param model_name to get model_info
litellm_params = model.get("litellm_params", {})
litellm_model = litellm_params.get("model", None)
if litellm_model:
split_model: Final = litellm_model.split("/")
if len(split_model) > 0:
litellm_model = split_model[-1]
try:
litellm_model_info = litellm.get_model_info(model=litellm_model, custom_llm_provider=split_model[0])
except Exception:
litellm_model_info = {}
priced_as: Final = wildcard_row_priced_model(model, llm_router, caller_aliases)
priced_model: Final = (
{**model, "litellm_params": {**model["litellm_params"], "model": priced_as}}
if isinstance(priced_as, str)
else model
)
priced_litellm_model: Final = priced_model.get("litellm_params", {}).get("model")
litellm_model_info: Final = (
MappingProxyType({})
if isinstance(priced_as, RoutedElsewhere)
else get_litellm_model_info(model=priced_model)
or (
model_info_cost_map_entry(priced_litellm_model)
if isinstance(priced_litellm_model, str)
else MappingProxyType({})
)
)
discovered_model_info: Final = (
llm_router.registry().discovered_model_info(model_info.get("id"))
if llm_router is not None
@ -17773,8 +17757,23 @@ def _translate_model_name_for_response(model: dict) -> dict[str, object]:
return {**model, "model_name": team_public}
def _get_proxy_model_info(model: dict) -> dict:
return _translate_model_name_for_response(_enrich_model_info_with_litellm_data(model=model, llm_router=llm_router))
def _get_proxy_model_info(model: dict, caller_aliases: CallerAliases = GLOBAL_ALIASES_ONLY) -> dict:
return _translate_model_name_for_response(
_enrich_model_info_with_litellm_data(model=model, llm_router=llm_router, caller_aliases=caller_aliases)
)
async def request_caller_aliases(user_api_key_dict: UserAPIKeyAuth, listed_team_id: str | None = None) -> CallerAliases:
router_settings: Final = await proxy_config.get_hierarchical_router_settings(
user_api_key_dict, prisma_client, proxy_logging_obj
)
return caller_alias_maps(
user_api_key_dict.aliases,
user_api_key_dict.team_model_aliases,
user_api_key_dict.team_id,
listed_team_id,
router_settings.get("model_group_alias") if isinstance(router_settings, Mapping) else None,
)
def _model_info_json_response(data: Sequence[Mapping[str, object]] | Mapping[str, object]) -> Response:
@ -17911,7 +17910,10 @@ async def model_info_v1(
status_code=400,
detail={"error": f"Model id = {litellm_model_id} not found on litellm proxy"},
)
_deployment_info_dict = _get_proxy_model_info(model=deployment_info.model_dump(exclude_none=True))
_deployment_info_dict = _get_proxy_model_info(
model=deployment_info.model_dump(exclude_none=True),
caller_aliases=await request_caller_aliases(user_api_key_dict),
)
single_model_list: list[dict] = [_deployment_info_dict]
if prisma_client is not None:
single_model_list = await _populate_team_access_on_models(
@ -17980,8 +17982,11 @@ async def model_info_v1(
if include_team_models:
all_models = _filter_models_to_user_accessible(all_models)
caller_aliases: Final = await request_caller_aliases(user_api_key_dict)
all_models = [
_translate_model_name_for_response(_enrich_model_info_with_litellm_data(model=model, llm_router=llm_router))
_translate_model_name_for_response(
_enrich_model_info_with_litellm_data(model=model, llm_router=llm_router, caller_aliases=caller_aliases)
)
for model in all_models
]

140
tests/unit/proxy/management_endpoints/test_model_management_endpoints.py
View file

@ -4860,6 +4860,146 @@ class TestModelInfoCostMapEchoFilter:
"cost-map metadata must not be persisted from an unchanged /model/info echo"
)
def test_echo_of_an_expanded_wildcard_row_is_not_persisted_on_the_wildcard(self):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
router: Final = litellm.Router(
model_list=[
{
"model_name": "my-openai/*",
"litellm_params": {"model": "openai/*", "api_key": "x"},
"model_info": {"id": "dep-echo-wildcard"},
}
]
)
entry: Final = litellm.get_model_info("openai/gpt-5.6")
echo: Final = {**entry, "id": "dep-echo-wildcard", "db_model": True, "access_groups": ["prod"]}
db_model: Final = Deployment(
model_name="my-openai/*",
litellm_params=LiteLLM_Params(model="openai/*"),
model_info=ModelInfo(id="dep-echo-wildcard"),
)
result: Final = update_db_model(
db_model=db_model,
updated_patch=updateDeployment(model_name="my-openai/gpt-5.6", model_info=ModelInfo(**echo)),
llm_router=router,
)
info: Final = json.loads(result["model_info"])
assert info["access_groups"] == ["prod"]
assert set(info).isdisjoint(entry), "one listed model's metadata must not be saved onto the whole wildcard"
def test_echo_saved_on_a_worker_that_has_not_loaded_the_wildcard_is_not_persisted(self):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
router_without_the_wildcard: Final = litellm.Router(
model_list=[
{
"model_name": "openai-mini",
"litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "x"},
"model_info": {"id": "dep-other"},
}
]
)
entry: Final = litellm.get_model_info("openai/gpt-5.6")
db_model: Final = Deployment(
model_name="my-openai/*",
litellm_params=LiteLLM_Params(model="openai/*"),
model_info=ModelInfo(id="dep-echo-wildcard"),
)
result: Final = update_db_model(
db_model=db_model,
updated_patch=updateDeployment(
model_name="my-openai/gpt-5.6",
model_info=ModelInfo(**{**entry, "id": "dep-echo-wildcard", "access_groups": ["prod"]}),
),
llm_router=router_without_the_wildcard,
)
info: Final = json.loads(result["model_info"])
assert info["access_groups"] == ["prod"]
assert set(info).isdisjoint(entry), "one listed model's metadata must not be saved onto the whole wildcard"
def test_legacy_update_echo_of_an_expanded_wildcard_row_is_not_persisted_on_the_wildcard(self):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import (
_ListedRow,
legacy_update_model_info_changes,
)
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
router: Final = litellm.Router(
model_list=[
{
"model_name": "my-openai/*",
"litellm_params": {"model": "openai/*", "api_key": "x"},
"model_info": {"id": "dep-echo-wildcard"},
}
]
)
entry: Final = litellm.get_model_info("openai/gpt-5.6")
db_model: Final = Deployment(
model_name="my-openai/*",
litellm_params=LiteLLM_Params(model="openai/*"),
model_info=ModelInfo(id="dep-echo-wildcard"),
)
changes: Final = legacy_update_model_info_changes(
db_model,
ModelInfo(**{**entry, "id": "dep-echo-wildcard", "access_groups": ["prod"]}),
loaded_catalog=lambda: {},
listed_row=_ListedRow("my-openai/gpt-5.6", router),
)
assert dict(changes) == {"access_groups": ["prod"]}
@pytest.mark.asyncio
async def test_echo_of_a_team_wildcard_row_is_not_persisted_on_the_wildcard(self):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import _update_team_model_in_db
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
router: Final = litellm.Router(
model_list=[
{
"model_name": "team-openai/*",
"litellm_params": {"model": "openai/*", "api_key": "x"},
"model_info": {"id": "dep-team-wildcard", "team_id": "team-a"},
}
]
)
entry: Final = litellm.get_model_info("openai/gpt-5.6")
echo: Final = {**entry, "id": "dep-team-wildcard", "team_id": "team-a", "access_groups": ["prod"]}
db_model: Final = Deployment(
model_name="team-openai/*",
litellm_params=LiteLLM_Params(model="openai/*"),
model_info=ModelInfo(id="dep-team-wildcard", team_id="team-a"),
)
with (
patch("litellm.proxy.proxy_server.premium_user", True),
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"),
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"),
):
result: Final = await _update_team_model_in_db(
db_model=db_model,
patch_data=updateDeployment(model_name="team-openai/gpt-5.6", model_info=ModelInfo(**echo)),
user_api_key_dict=UserAPIKeyAuth(user_id="test_user", user_role=LitellmUserRoles.PROXY_ADMIN),
prisma_client=MockPrismaClient(team_exists=True),
write_row=_passthrough_row,
llm_router=router,
)
info: Final = json.loads(result["model_info"])
assert info["access_groups"] == ["prod"]
assert set(info).isdisjoint(entry), "one listed model's metadata must not be saved onto a team's wildcard"
def test_an_edited_value_survives_the_echo_filter(self):
import litellm
from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model

517
tests/unit/proxy/proxy_server/test_routes_model_info.py
View file

@ -12,7 +12,7 @@ from __future__ import annotations
import asyncio
import copy
import json
from collections.abc import Callable
from collections.abc import Callable, Mapping
from contextlib import AbstractContextManager
from typing import Final
from unittest.mock import AsyncMock, MagicMock
@ -25,7 +25,7 @@ import litellm
from litellm.caching.llm_caching_handler import LLMClientCache
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.proxy import proxy_server
from litellm.proxy._types import LiteLLM_ProxyModelTable
from litellm.proxy._types import LiteLLM_ProxyModelTable, UserAPIKeyAuth
from litellm.utils import _invalidate_model_cost_lowercase_map
from .conftest import normalize, route_entry_points_to_stubs # type: ignore[import-not-found]
@ -314,7 +314,7 @@ def configured_router(monkeypatch):
monkeypatch.setattr(proxy_server, "llm_router", router)
monkeypatch.setattr(proxy_server, "llm_model_list", [{"model_name": "gpt-4"}])
monkeypatch.setattr(proxy_server, "user_model", None)
monkeypatch.setattr(proxy_server, "_get_proxy_model_info", lambda model: model)
monkeypatch.setattr(proxy_server, "_get_proxy_model_info", lambda model, caller_aliases: model)
yield router
@ -539,6 +539,11 @@ def test_model_info_id_lookup_reports_the_same_cost_as_the_list(
_invalidate_model_cost_lowercase_map()
def _discoverable_models(monkeypatch: pytest.MonkeyPatch, models: Mapping[str, list[str]]) -> None:
for provider, provider_models in models.items():
monkeypatch.setitem(litellm.models_by_provider, provider, provider_models)
async def test_v1_model_info_star_wildcard_filter_keeps_provider_expansion(monkeypatch):
from litellm.proxy._types import SpecialModelNames, UserAPIKeyAuth
from litellm.proxy.auth import model_checks
@ -576,6 +581,512 @@ async def test_v1_model_info_star_wildcard_filter_keeps_provider_expansion(monke
assert router.registry().names("gpt-4o", kind="wildcard") == frozenset({"openai/gpt-4o"})
def test_model_info_prices_expanded_wildcard_rows_as_the_model_their_calls_route_to(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""A catch-all over Fireworks lists ``acme-chat``, which also names a DeepSeek entry, and a
prefixed OpenAI wildcard lists ``my-openai/gpt-acme``, which names no entry at all. Each row must
show the provider and price its calls are billed at."""
catalog: Final = {
"fireworks_ai/accounts/fireworks/models/acme-chat": {
"input_cost_per_token": 1e-07,
"output_cost_per_token": 2e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"acme-chat": {
"input_cost_per_token": 9e-07,
"output_cost_per_token": 9e-06,
"litellm_provider": "deepseek",
"mode": "chat",
},
"gpt-acme": {
"input_cost_per_token": 3e-07,
"output_cost_per_token": 4e-07,
"litellm_provider": "openai",
"mode": "chat",
},
}
monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost, **catalog})
_invalidate_model_cost_lowercase_map()
router: Final = litellm.Router(
model_list=[
{
"model_name": "*",
"litellm_params": {"model": "fireworks_ai/accounts/fireworks/models/*", "api_key": "x"},
"model_info": {"id": "fireworks-wildcard"},
},
{
"model_name": "my-openai/*",
"litellm_params": {"model": "openai/*", "api_key": "x"},
"model_info": {"id": "openai-wildcard"},
},
]
)
monkeypatch.setattr(proxy_server, "llm_router", router)
_discoverable_models(
monkeypatch,
{
"fireworks_ai": ["fireworks_ai/accounts/fireworks/models/acme-chat"],
"openai": ["gpt-acme"],
},
)
monkeypatch.setattr(proxy_server, "llm_model_list", router.get_model_list())
monkeypatch.setattr(proxy_server, "user_model", None)
with auth_as():
response: Final = client.get("/model/info")
assert response.status_code == 200, response.text
listed: Final = {
row["model_name"]: (
row["litellm_params"]["model"],
row["model_info"].get("litellm_provider"),
row["model_info"].get("input_cost_per_token"),
row["model_info"].get("output_cost_per_token"),
)
for row in response.json()["data"]
}
assert listed == {
"acme-chat": ("fireworks_ai/accounts/fireworks/models/acme-chat", "fireworks_ai", 1e-07, 2e-07),
"my-openai/gpt-acme": ("openai/gpt-acme", "openai", 3e-07, 4e-07),
}
calls: Final = {
model_name: router.completion(
model=model_name, messages=[{"role": "user", "content": "hi"}], mock_response="hello there"
)
for model_name in listed
}
assert {model_name: call._hidden_params["response_cost"] for model_name, call in calls.items()} == pytest.approx(
{
model_name: listed[model_name][2] * call.usage.prompt_tokens
+ listed[model_name][3] * call.usage.completion_tokens
for model_name, call in calls.items()
}
)
_invalidate_model_cost_lowercase_map()
def test_model_info_never_prices_a_wildcard_row_as_a_route_its_calls_skip(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""``acme-pro`` is an alias of the ``acme-lite`` deployment and ``acme-max`` is a routing group over it, so
calls to those names never reach the catch-all, which still lists them. Such a row shows no price or
provider, even where another provider's model goes by the same name, and every price shown is the one
calls are billed at."""
catalog: Final = {
"fireworks_ai/accounts/fireworks/models/acme-lite": {
"input_cost_per_token": 5e-07,
"output_cost_per_token": 6e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"fireworks_ai/accounts/fireworks/models/acme-pro": {
"input_cost_per_token": 1.5e-07,
"output_cost_per_token": 2.5e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"fireworks_ai/accounts/fireworks/models/acme-max": {
"input_cost_per_token": 3.5e-07,
"output_cost_per_token": 4.5e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"deepseek/acme-lite": {
"input_cost_per_token": 7e-07,
"output_cost_per_token": 8e-07,
"litellm_provider": "deepseek",
"mode": "chat",
},
"acme-pro": {
"input_cost_per_token": 9e-07,
"output_cost_per_token": 1e-06,
"litellm_provider": "deepseek",
"mode": "chat",
},
"acme-max": {
"input_cost_per_token": 1.1e-06,
"output_cost_per_token": 1.2e-06,
"litellm_provider": "deepseek",
"mode": "chat",
},
}
monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost, **catalog})
_invalidate_model_cost_lowercase_map()
router: Final = litellm.Router(
model_list=[
{
"model_name": "*",
"litellm_params": {"model": "fireworks_ai/accounts/fireworks/models/*", "api_key": "x"},
"model_info": {"id": "fireworks-wildcard"},
},
{
"model_name": "acme-lite",
"litellm_params": {"model": "deepseek/acme-lite", "api_key": "x"},
"model_info": {"id": "deepseek-acme-lite"},
},
],
model_group_alias={"acme-pro": "acme-lite"},
routing_groups=[{"group_name": "acme-max", "models": ["acme-lite"], "routing_strategy": "simple-shuffle"}],
)
monkeypatch.setattr(proxy_server, "llm_router", router)
_discoverable_models(
monkeypatch,
{
"fireworks_ai": [
"fireworks_ai/accounts/fireworks/models/acme-lite",
"fireworks_ai/accounts/fireworks/models/acme-pro",
"fireworks_ai/accounts/fireworks/models/acme-max",
],
},
)
monkeypatch.setattr(proxy_server, "llm_model_list", router.get_model_list())
monkeypatch.setattr(proxy_server, "user_model", None)
with auth_as():
response: Final = client.get("/model/info")
assert response.status_code == 200, response.text
listed: Final = {
(row["model_info"]["id"], row["model_name"]): (
row["model_info"].get("input_cost_per_token"),
row["model_info"].get("output_cost_per_token"),
)
for row in response.json()["data"]
if "*" not in row["model_name"]
}
assert sorted(listed) == [
("deepseek-acme-lite", "acme-lite"),
("deepseek-acme-lite", "acme-pro"),
("fireworks-wildcard", "acme-max"),
("fireworks-wildcard", "acme-pro"),
]
shown_models: Final = {
(row["model_info"]["id"], row["model_name"]): (
row["model_info"].get("litellm_provider"),
row["model_info"].get("key"),
)
for row in response.json()["data"]
if "*" not in row["model_name"] and row["model_info"]["id"] != "deepseek-acme-lite"
}
assert shown_models == dict.fromkeys(
[
("fireworks-wildcard", "acme-max"),
("fireworks-wildcard", "acme-pro"),
],
(None, None),
)
assert listed[("deepseek-acme-lite", "acme-lite")] == (7e-07, 8e-07)
assert {key: prices for key, prices in listed.items() if key[0] != "deepseek-acme-lite"} == {
("fireworks-wildcard", "acme-max"): (None, None),
("fireworks-wildcard", "acme-pro"): (None, None),
}
calls: Final = {
model_name: router.completion(
model=model_name, messages=[{"role": "user", "content": "hi"}], mock_response="hello there"
)
for _, model_name in listed
}
priced_rows: Final = {key: prices for key, prices in listed.items() if prices[0] is not None}
assert {
key: calls[key[1]]._hidden_params["response_cost"] for key in priced_rows
} == pytest.approx(
{
key: input_cost * calls[key[1]].usage.prompt_tokens + output_cost * calls[key[1]].usage.completion_tokens
for key, (input_cost, output_cost) in priced_rows.items()
}
)
_invalidate_model_cost_lowercase_map()
def _catch_all_and_shadowed_wildcard_router(monkeypatch: pytest.MonkeyPatch) -> litellm.Router:
catalog: Final = {
"gpt-5.6-acme": {
"input_cost_per_token": 3e-07,
"output_cost_per_token": 4e-07,
"litellm_provider": "openai",
"mode": "chat",
},
"fireworks_ai/accounts/fireworks/models/acme-pro": {
"input_cost_per_token": 1.5e-07,
"output_cost_per_token": 2.5e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"fireworks_ai/accounts/fireworks/models/acme-max": {
"input_cost_per_token": 9e-07,
"output_cost_per_token": 9.5e-07,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"fireworks_ai/accounts/fireworks/models/acme-mini": {
"input_cost_per_token": 5e-08,
"output_cost_per_token": 6e-08,
"litellm_provider": "fireworks_ai",
"mode": "chat",
},
"deepseek/acme-lite": {
"input_cost_per_token": 7e-07,
"output_cost_per_token": 8e-07,
"litellm_provider": "deepseek",
"mode": "chat",
},
}
monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost, **catalog})
_invalidate_model_cost_lowercase_map()
router: Final = litellm.Router(
model_list=[
{
"model_name": "*",
"litellm_params": {"model": "openai/*", "api_key": "x"},
"model_info": {"id": "openai-catch-all"},
},
{
"model_name": "my-fw/*",
"litellm_params": {"model": "fireworks_ai/accounts/fireworks/models/*", "api_key": "x"},
"model_info": {"id": "fireworks-wildcard"},
},
{
"model_name": "acme-lite",
"litellm_params": {"model": "deepseek/acme-lite", "api_key": "x"},
"model_info": {"id": "deepseek-acme-lite"},
},
],
model_group_alias={"my-fw/acme-pro": "acme-lite", "my-fw/acme-max": "my-fw/acme-mini"},
)
monkeypatch.setattr(proxy_server, "llm_router", router)
_discoverable_models(
monkeypatch,
{
"openai": ["gpt-5.6-acme"],
"fireworks_ai": [
"fireworks_ai/accounts/fireworks/models/acme-pro",
"fireworks_ai/accounts/fireworks/models/acme-max",
"fireworks_ai/accounts/fireworks/models/acme-mini",
],
},
)
monkeypatch.setattr(proxy_server, "llm_model_list", router.get_model_list())
monkeypatch.setattr(proxy_server, "user_model", None)
return router
def _expanded_wildcard_rows(client: TestClient) -> dict[str, dict[str, object]]:
response: Final = client.get("/model/info")
assert response.status_code == 200, response.text
return {
row["model_name"]: row
for row in response.json()["data"]
if row["model_info"]["id"] != "deepseek-acme-lite" and "*" not in row["model_name"]
}
def test_model_info_prices_a_wildcard_row_aliased_to_a_sibling_name_as_the_alias_target(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""An alias from ``my-fw/acme-max`` to ``my-fw/acme-mini`` still reaches the same wildcard, but calls run
``acme-mini``, so the ``my-fw/acme-max`` row must show the price those calls are billed at."""
router: Final = _catch_all_and_shadowed_wildcard_router(monkeypatch)
with auth_as():
rows: Final = _expanded_wildcard_rows(client)
shown: Final = rows["my-fw/acme-max"]["model_info"]
call: Final = router.completion(
model="my-fw/acme-max", messages=[{"role": "user", "content": "hi"}], mock_response="hello there"
)
assert call._hidden_params["response_cost"] == pytest.approx(
shown["input_cost_per_token"] * call.usage.prompt_tokens
+ shown["output_cost_per_token"] * call.usage.completion_tokens
)
_invalidate_model_cost_lowercase_map()
def test_model_info_prices_a_wildcard_row_the_proxy_renames_as_the_model_its_calls_reach(
client: TestClient,
auth_as: Callable[..., AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""The proxy renames a requested model through the global ``model_alias_map``, the key's ``aliases`` and the
key's ``router_settings.model_group_alias`` before the router sees it. A row renamed to another deployment
shows no price or provider, and a row renamed to a sibling name of its own wildcard shows that sibling's
price."""
_catch_all_and_shadowed_wildcard_router(monkeypatch)
monkeypatch.setattr(litellm, "model_alias_map", {"my-fw/acme-max": "acme-lite"})
with auth_as(
aliases={"gpt-5.6-acme": "acme-lite"},
router_settings={"model_group_alias": {"my-fw/acme-pro": "my-fw/acme-mini"}},
):
rows: Final = _expanded_wildcard_rows(client)
shown: Final = {
name: (
row["model_info"].get("input_cost_per_token"),
row["model_info"].get("output_cost_per_token"),
row["model_info"].get("litellm_provider"),
)
for name, row in rows.items()
}
assert shown == {
"my-fw/acme-max": (None, None, None),
"my-fw/acme-mini": (5e-08, 6e-08, "fireworks_ai"),
"my-fw/acme-pro": (5e-08, 6e-08, "fireworks_ai"),
"gpt-5.6-acme": (None, None, None),
}
_invalidate_model_cost_lowercase_map()
def test_model_info_prices_a_wildcard_row_whose_team_alias_targets_a_deleted_team_deployment(
client: TestClient,
auth_as: Callable[..., AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""Calls skip a team alias whose target team deployment was deleted, so the row that alias names keeps the
price its calls through the wildcard are billed at, while the team's live aliases still apply."""
from litellm.proxy.litellm_pre_call_utils import _update_model_if_team_alias_exists
_catch_all_and_shadowed_wildcard_router(monkeypatch)
caller: Final = {
"team_id": "team-a",
"team_model_aliases": {"my-fw/acme-mini": "model_name_team-a_deleted", "gpt-5.6-acme": "my-fw/acme-mini"},
}
with auth_as(**caller):
rows: Final = _expanded_wildcard_rows(client)
stale_request: Final = {"model": "my-fw/acme-mini"}
live_request: Final = {"model": "gpt-5.6-acme"}
_update_model_if_team_alias_exists(data=stale_request, user_api_key_dict=UserAPIKeyAuth(**caller))
_update_model_if_team_alias_exists(data=live_request, user_api_key_dict=UserAPIKeyAuth(**caller))
shown: Final = rows["my-fw/acme-mini"]["model_info"]
live_alias_row: Final = rows["gpt-5.6-acme"]["model_info"]
assert (stale_request["model"], live_request["model"]) == ("my-fw/acme-mini", "my-fw/acme-mini")
assert (
shown.get("input_cost_per_token"),
shown.get("output_cost_per_token"),
shown.get("litellm_provider"),
) == (5e-08, 6e-08, "fireworks_ai")
assert (live_alias_row.get("input_cost_per_token"), live_alias_row.get("litellm_provider")) == (None, None)
_invalidate_model_cost_lowercase_map()
def _fireworks_wildcard_rows(client: TestClient) -> dict[str, dict[str, object]]:
response: Final = client.get("/model/info")
assert response.status_code == 200, response.text
return {
row["model_name"]: row["model_info"]
for row in response.json()["data"]
if row["model_info"]["id"] == "fireworks-wildcard" and "*" not in row["model_name"]
}
_FIREWORKS_WILDCARD: Final = {
"model_name": "my-fw/*",
"litellm_params": {"model": "fireworks_ai/accounts/fireworks/models/*", "api_key": "x"},
"model_info": {"id": "fireworks-wildcard"},
}
def test_model_info_shows_no_wildcard_price_for_a_name_the_callers_team_publishes(
client: TestClient,
auth_as: Callable[..., AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""Team ``team-a`` publishes ``my-fw/acme-mini`` on its own DeepSeek deployment, so a team-a key's calls to that
name skip the global wildcard and the wildcard's row shows that key no price, while a key outside the team
still sees the Fireworks price its calls bill."""
_catch_all_and_shadowed_wildcard_router(monkeypatch)
router: Final = litellm.Router(
model_list=[
_FIREWORKS_WILDCARD,
{
"model_name": "model_name_team-a_acme-mini",
"litellm_params": {"model": "deepseek/acme-lite", "api_key": "x"},
"model_info": {"id": "team-deepseek", "team_id": "team-a", "team_public_model_name": "my-fw/acme-mini"},
},
]
)
monkeypatch.setattr(proxy_server, "llm_router", router)
_discoverable_models(monkeypatch, {"fireworks_ai": ["fireworks_ai/accounts/fireworks/models/acme-mini"]})
monkeypatch.setattr(proxy_server, "llm_model_list", router.get_model_list())
with auth_as(team_id="team-a"):
team_rows: Final = _fireworks_wildcard_rows(client)
with auth_as():
other_rows: Final = _fireworks_wildcard_rows(client)
assert team_rows["my-fw/acme-mini"].get("input_cost_per_token") is None
assert other_rows["my-fw/acme-mini"].get("input_cost_per_token") == 5e-08
_invalidate_model_cost_lowercase_map()
def test_saving_a_replayed_model_info_wildcard_row_stores_none_of_its_shown_metadata(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[object]],
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
) -> None:
"""An API client that saves a ``/model/info`` row back unchanged, apart from one edit, must not pin the
cost-map metadata the row showed onto the wildcard, whether the row is a catch-all name or a name an alias
sends to a sibling name. A name an alias sends to another deployment shows
no metadata to pin."""
from litellm.proxy.management_endpoints.model_management_endpoints import update_db_model
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo, updateDeployment
router: Final = _catch_all_and_shadowed_wildcard_router(monkeypatch)
with auth_as():
rows: Final = _expanded_wildcard_rows(client)
assert sorted(rows) == ["gpt-5.6-acme", "my-fw/acme-max", "my-fw/acme-mini", "my-fw/acme-pro"]
sources: Final = {
deployment["model_info"]["id"]: deployment for deployment in router.get_model_list() or []
}
stored: Final = {
name: json.loads(
update_db_model(
db_model=Deployment(
model_name=sources[row["model_info"]["id"]]["model_name"],
litellm_params=LiteLLM_Params(model=sources[row["model_info"]["id"]]["litellm_params"]["model"]),
model_info=ModelInfo(id=row["model_info"]["id"]),
),
updated_patch=updateDeployment(
model_name=name, model_info=ModelInfo(**{**row["model_info"], "access_groups": ["prod"]})
),
llm_router=router,
)["model_info"]
)
for name, row in rows.items()
}
shown_metadata: Final = {
name: set(row["model_info"]) - set(sources[row["model_info"]["id"]]["model_info"]) for name, row in rows.items()
}
assert sorted(name for name, row in rows.items() if row["model_info"].get("key")) == [
"gpt-5.6-acme",
"my-fw/acme-max",
"my-fw/acme-mini",
]
assert {name: info["access_groups"] for name, info in stored.items()} == dict.fromkeys(rows, ["prod"])
assert {name: shown_metadata[name] & set(info) for name, info in stored.items()} == dict.fromkeys(rows, set())
_invalidate_model_cost_lowercase_map()
# ---------------------------------------------------------------------------
# GET /model/info — team BYOK scoping (issue #30983)
# ---------------------------------------------------------------------------

4
tests/unit/proxy/proxy_server/test_team_model_name_translation.py
View file

@ -808,7 +808,7 @@ async def test_model_info_v1_litellm_model_id_include_team_models_filters_inacce
monkeypatch.setattr(ps, "llm_model_list", [team_row])
monkeypatch.setattr(ps, "llm_router", router)
monkeypatch.setattr(ps, "prisma_client", MagicMock())
monkeypatch.setattr(ps, "_get_proxy_model_info", lambda model: team_row)
monkeypatch.setattr(ps, "_get_proxy_model_info", lambda model, caller_aliases: team_row)
monkeypatch.setattr(ps, "_populate_team_access_on_models", _fake_populate)
caller = UserAPIKeyAuth(
@ -843,7 +843,7 @@ async def test_model_info_v1_litellm_model_id_team_id_applies_team_filter(monkey
monkeypatch.setattr(ps, "llm_model_list", [team_row])
monkeypatch.setattr(ps, "llm_router", router)
monkeypatch.setattr(ps, "prisma_client", MagicMock())
monkeypatch.setattr(ps, "_get_proxy_model_info", lambda model: team_row)
monkeypatch.setattr(ps, "_get_proxy_model_info", lambda model, caller_aliases: team_row)
monkeypatch.setattr(ps, "_populate_team_access_on_models", _fake_populate)
monkeypatch.setattr(ps, "_filter_models_by_team_id", team_filter)