diff --git a/litellm/litellm_core_utils/routing_common_utils.py b/litellm/litellm_core_utils/routing_common_utils.py index 154c3942edc..363854f70de 100644 --- a/litellm/litellm_core_utils/routing_common_utils.py +++ b/litellm/litellm_core_utils/routing_common_utils.py @@ -38,7 +38,6 @@ _V = TypeVar("_V") MID_STREAM_CONTINUATION_KWARG: Final = "_mid_stream_continuation" ROUTER_ONLY_CALL_KWARGS: Final = frozenset({"silent_model", "include_fallback_errors", MID_STREAM_CONTINUATION_KWARG}) -_OBJECT_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object]) _DEPLOYMENT_ROWS_ADAPTER: Final[TypeAdapter[tuple[Mapping[str, object], ...]]] = TypeAdapter( tuple[Mapping[str, object], ...] ) @@ -215,30 +214,6 @@ def resolve_model_group_alias(model_group_alias: object, model: str) -> str | No return target -def resolve_served_model( - model: str, - model_alias_map: object, - key_aliases: object, - model_group_alias: object, -) -> str: - """Model the request preparation serves for `model`. - - Apply the global alias map, then key aliases, then `router_settings.model_group_alias`. - """ - global_aliases: Final = ( - _OBJECT_MAPPING_ADAPTER.validate_python(model_alias_map) if isinstance(model_alias_map, Mapping) else None - ) - global_target: Final = global_aliases.get(model) if global_aliases is not None else None - after_global: Final = global_target if isinstance(global_target, str) else model - key_alias_mapping: Final = ( - _OBJECT_MAPPING_ADAPTER.validate_python(key_aliases) if isinstance(key_aliases, Mapping) else None - ) - key_target: Final = key_alias_mapping.get(after_global) if key_alias_mapping is not None else None - after_key: Final = key_target if isinstance(key_target, str) else after_global - router_target: Final = resolve_model_group_alias(model_group_alias, after_key) - return router_target if router_target is not None else after_key - - def truncate_fallback_error_detail(detail: str) -> str: """ Bound a fallback failure detail before it is logged or appended to an exception message. diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py index f0c9632837d..fa331b37950 100644 --- a/litellm/proxy/auth/auth_checks.py +++ b/litellm/proxy/auth/auth_checks.py @@ -557,7 +557,7 @@ def _get_router_zero_cost_cache(llm_router: RegistryReader) -> dict[str, bool] | return cache if isinstance(cache, dict) else None -def is_model_cost_zero(model: str | list[str] | None, llm_router: RegistryReader | None) -> bool: +def is_model_cost_zero(model: str | Sequence[str] | None, llm_router: RegistryReader | None) -> bool: """ Check if a model has zero cost (no configured pricing). @@ -1122,7 +1122,7 @@ def route_skips_budget_checks(route: str) -> bool: def request_skips_budget_checks( route: str, - model: str | list[str] | None, + model: str | Sequence[str] | None, llm_router: Router | None, *, request_body: Mapping[str, object] | None = None, @@ -1146,7 +1146,7 @@ async def common_checks( proxy_logging_obj: ProxyLogging, valid_token: UserAPIKeyAuth | None, request: Request, - skip_budget_checks: bool = False, + skip_budget_checks: bool | None = None, project_object: LiteLLM_ProjectTableCachedObj | None = None, ) -> bool: """ @@ -1181,11 +1181,15 @@ async def common_checks( team_id=valid_token.team_id if valid_token is not None else None, ) - skip_all_budget_checks: Final = skip_budget_checks or request_skips_budget_checks( - route=route, - model=_model, - llm_router=llm_router, - request_body=_typed_request_body(request_body), + skip_all_budget_checks: Final = ( + skip_budget_checks + if skip_budget_checks is not None + else request_skips_budget_checks( + route=route, + model=_model, + llm_router=llm_router, + request_body=_typed_request_body(request_body), + ) ) membership_user_id: Final = ( diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 5bf4db6a725..d7a8d4a0700 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -41,7 +41,7 @@ from litellm.integrations.otel.model.config import is_otel_v2_enabled from litellm.integrations.otel.runtime import phase_event, phase_span, seed_request_identity from litellm.litellm_core_utils.dd_tracing import tracer from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value -from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias, resolve_served_model +from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias from litellm.models.end_user import LiteLLM_EndUserTable from litellm.proxy._types import * from litellm.proxy.agent_endpoints.auth.agent_caller import agent_caller_from_headers @@ -95,6 +95,7 @@ from litellm.proxy.auth.auth_checks import ( # noqa: F401 # legacy module expo route_skips_budget_checks, stamp_matched_model_access_groups, team_member_max_budget_alert_check, + team_model_aliases_for_auth_check, virtual_key_max_budget_alert_check, virtual_key_max_budget_check, virtual_key_soft_budget_check, @@ -173,7 +174,11 @@ from litellm.proxy.common_utils.user_api_key_cache import ( from litellm.proxy.db.db_lookup_gate import bounded_db_lookup from litellm.proxy.db.db_span import db_span from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler -from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup +from litellm.proxy.litellm_pre_call_utils import ( + LiteLLMProxyRequestSetup, + pre_call_served_model, + request_served_model, +) from litellm.proxy.spend_tracking.carried_budget_state import ( carry_committed_budget_spend, carry_team_and_user_budget_state, @@ -332,6 +337,65 @@ def _is_codex_native_litellm_turn(route: str, model: object, llm_router: litellm return routes_to_litellm(model, llm_router, team_id) +_SERVED_MODEL_BUDGET_WAIVER_ROUTES: Final = frozenset( + { + "/chat/completions", + "/v1/chat/completions", + "/completions", + "/v1/completions", + "/responses", + "/v1/responses", + "/messages", + "/v1/messages", + } +) + + +def _inference_model_overridden() -> bool: + from litellm.proxy import proxy_server + + cli_model: Final[object] = proxy_server.user_model # pyright: ignore[reportUnknownMemberType] # legacy untyped CLI global + completion_model: Final = proxy_server.general_settings_view().get("completion_model") + return (isinstance(completion_model, str) and completion_model != "") or ( + isinstance(cli_model, str) and cli_model != "" + ) + + +async def budget_served_model( + model: str | Sequence[str] | None, + request_data: Mapping[str, object], + request: Request | None, + llm_router: litellm.Router | None, + valid_token: UserAPIKeyAuth, +) -> str | Sequence[str] | None: + """The model the zero-cost budget waiver judges. For a client's own HTTP request to a chat, completions, + responses or messages route that names a string body `model`, it is the model request preparation serves it + with through the caller's team, key, global and router aliases. Everywhere else it stays the requested model: + requests built inside the proxy, a path or query `model`, a configured `completion_model` or CLI `--model`, + and requests whose body auth has not yet passed through the router aliases.""" + if request is None or llm_router is None or not isinstance(model, str): + return model + if get_request_route_template(request) not in _SERVED_MODEL_BUDGET_WAIVER_ROUTES: + return model + if request.scope.get(MODEL_GROUP_ALIAS_RESOLVED_SCOPE_KEY) is not True or request_data.get("model") != model: + return model + if "model" in safe_get_request_query_params(request=request) or _inference_model_overridden(): + return model + if not ( + key_model_aliases_for_auth_check(valid_token) + or team_model_aliases_for_auth_check(valid_token) + or model in litellm.model_alias_map + ): + return model + from litellm.proxy.proxy_server import prisma_client, proxy_config, proxy_logging_obj + + settings: Final = await proxy_config.get_hierarchical_router_settings( + user_api_key_dict=valid_token, prisma_client=prisma_client, proxy_logging_obj=proxy_logging_obj + ) + group_alias: Final[object] = settings.get("model_group_alias") if isinstance(settings, Mapping) else None + return request_served_model(model, valid_token, group_alias, llm_router) + + async def _resolve_router_settings_model_group_alias( request_data: dict[str, object], # mutable-ok: the request body is rewritten in place for every downstream reader valid_token: UserAPIKeyAuth, @@ -971,9 +1035,6 @@ async def _try_team_member_model_budget_fallback( model_group_aliases: Final[object] = ( router_settings.get("model_group_alias") if router_settings is not None else None ) - key_aliases: Final[object] = cast( # cast-ok: auth aliases are dynamically typed at runtime - object, valid_token.aliases - ) resolved_fallback_candidates: Final[tuple[tuple[str, str], ...]] = tuple( ( candidate, @@ -985,12 +1046,7 @@ async def _try_team_member_model_budget_fallback( ( fallback_candidate, fallback_model, - resolve_served_model( - fallback_model, - litellm.model_alias_map, - key_aliases, - model_group_aliases, - ), + pre_call_served_model(fallback_model, valid_token, model_group_aliases), ) for fallback_candidate, fallback_model in resolved_fallback_candidates ) @@ -3137,7 +3193,10 @@ async def validate_resolved_virtual_key( # noqa: C901 # Preserve ordering of e team_id=valid_token.team_id, ) skip_budget_checks: Final = request_skips_budget_checks( - route=route, model=model, llm_router=llm_router, request_body=request_data + route=route, + model=await budget_served_model(model, request_data, request, llm_router, valid_token), + llm_router=llm_router, + request_body=request_data, ) # Check 3. Check if user is in their team budget @@ -3968,12 +4027,13 @@ async def run_centralized_common_checks( dict(end_user_model_max_budget) if end_user_model_max_budget else None ) - skip_budget_checks: Final = _should_skip_budget_checks( + skip_budget_checks: Final = await _should_skip_budget_checks( request_data=request_data, route=route, request=request, llm_router=llm_router, - team_id=user_api_key_auth_obj.team_id, + valid_token=user_api_key_auth_obj, + sub_call=sub_call, ) typed_team_object: Final[LiteLLM_TeamTable | None] = ( team_object if isinstance(team_object, LiteLLM_TeamTable) else None @@ -4216,21 +4276,27 @@ async def _reserve_budget_after_common_checks( request.state.budget_reservation = reservation # rebind-ok: read by the release middleware -def _should_skip_budget_checks( +async def _should_skip_budget_checks( request_data: dict[str, object], route: str, request: Request | None, llm_router: litellm.Router | None, - team_id: str | None = None, + valid_token: UserAPIKeyAuth, + sub_call: bool = False, ) -> bool: model: Final = _get_model_from_request_context( request_data=request_data, route=route, request=request, llm_router=llm_router, - team_id=team_id, + team_id=valid_token.team_id, + ) + return request_skips_budget_checks( + route=route, + model=(model if sub_call else await budget_served_model(model, request_data, request, llm_router, valid_token)), + llm_router=llm_router, + request_body=request_data, ) - return request_skips_budget_checks(route=route, model=model, llm_router=llm_router, request_body=request_data) def _resolve_request_principal(request: Request, valid_token: UserAPIKeyAuth) -> Principal: @@ -4908,7 +4974,10 @@ async def _run_post_custom_auth_checks( # every budget check for these; this path did not, so the same request could # be refused under custom auth and served under the other two. skip_budget_checks: Final = ( - is_model_cost_zero(model=current_model, llm_router=llm_router) + is_model_cost_zero( + model=await budget_served_model(current_model, request_data, request, llm_router, valid_token), + llm_router=llm_router, + ) if current_model is not None and llm_router is not None else False ) diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index bfa559f86c5..46a9f9910d7 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -89,7 +89,7 @@ from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import ( from litellm.litellm_core_utils.llm_response_utils.get_headers import ( get_response_headers, ) -from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias, resolve_served_model +from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.litellm_core_utils.served_output_texts import ( record_served_output_texts, @@ -310,6 +310,8 @@ from litellm.proxy.litellm_pre_call_utils import ( LiteLLMProxyRequestSetup, add_litellm_data_to_request, get_metadata_variable_name, + global_then_key_alias_target, + pre_call_served_model, refresh_proxy_server_request_body_snapshot, reject_url_valued_destination, sanitize_for_log, @@ -1017,13 +1019,11 @@ _ROUTER_SETTINGS_ADAPTER: Final[TypeAdapter[Mapping[str, object] | None]] = Type def _fallback_probe_data( - pristine: object, fallback_model: str, key_aliases: object, group_alias: object + pristine: object, fallback_model: str, user_api_key_dict: UserAPIKeyAuth, group_alias: object ) -> Mapping[str, object]: return { **_REQUEST_FIELDS.validate_python(pristine), - "model": resolve_served_model( - fallback_model, _DECODED_JSON.validate_python(litellm.model_alias_map), key_aliases, group_alias - ), + "model": pre_call_served_model(fallback_model, user_api_key_dict, group_alias), } @@ -2246,13 +2246,9 @@ async def _team_member_within_fallback_model_budget( if proxy_config is not None and llm_router is not None else None ) - raw_aliases: Final[object] = cast( # cast-ok: auth aliases are dynamically typed at runtime - object, user_api_key_dict.aliases - ) - served_model: Final = resolve_served_model( + served_model: Final = pre_call_served_model( model, - litellm.model_alias_map, - raw_aliases, + user_api_key_dict, router_settings.get("model_group_alias") if router_settings is not None else None, ) if is_model_cost_zero(model=served_model, llm_router=llm_router): @@ -2714,17 +2710,9 @@ class ProxyBaseLLMRequestProcessing: ### MODEL ALIAS MAPPING ### # check if model name in model alias map # get the actual model name - if isinstance(self.data["model"], str) and self.data["model"] in litellm.model_alias_map: - self.data["model"] = litellm.model_alias_map[self.data["model"]] - - # Check key-specific aliases - if ( - isinstance(self.data["model"], str) - and user_api_key_dict.aliases - and isinstance(user_api_key_dict.aliases, dict) - and self.data["model"] in user_api_key_dict.aliases - ): - self.data["model"] = user_api_key_dict.aliases[self.data["model"]] + prepared_model: Final = self.data["model"] + if isinstance(prepared_model, str): + self.data["model"] = global_then_key_alias_target(prepared_model, user_api_key_dict) if llm_router is not None and proxy_config is not None: self.data.update( @@ -2945,9 +2933,6 @@ class ProxyBaseLLMRequestProcessing: if isinstance(raw_router_settings, Mapping) else None ) - key_aliases: Final[object] = cast( # cast-ok: auth aliases are dynamically typed at runtime - object, user_api_key_dict.aliases - ) try: for fallback_model in fallback_models: if fallback_model == original_model: @@ -2965,7 +2950,7 @@ class ProxyBaseLLMRequestProcessing: try: if endpoint_unsupported_by_model( route_type, - _fallback_probe_data(pristine, fallback_model, key_aliases, group_alias), + _fallback_probe_data(pristine, fallback_model, user_api_key_dict, group_alias), llm_router, user_api_key_dict.team_id, ): diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index d5eac099fb6..447b41f9cfd 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -14,7 +14,6 @@ from typing import ( Any, Final, Literal, - Protocol, TypeAlias, cast, # noqa: TID251 # narrows untyped deployment and credential payloads ) @@ -57,7 +56,6 @@ from litellm.litellm_core_utils.core_helpers import ( attribute_or_none, is_codex_user_agent, normalize_drop_params, - require_attributes, ) from litellm.litellm_core_utils.credential_accessor import CredentialAccessor from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( @@ -325,6 +323,7 @@ def sanitize_for_log(value: object) -> str: return _sanitize_for_log(value) +from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias from litellm.router import Router from litellm.secret_managers.main import get_secret_bool from litellm.types.litellm_params import AGENTIC_LOOP_KWARG_NAMES, STREAM_USAGE_INJECTED_FIELD @@ -3604,12 +3603,8 @@ def _warn_stale_team_alias_once(warning_key: str, message: str, *args: str) -> N verbose_proxy_logger.warning(message, *args) -class _SupportsStartswith(Protocol): - def startswith(self, prefix: str, /) -> bool: ... - - def _update_model_if_team_alias_exists( - data: dict, + data: dict[str, object], # mutable-ok: request preparation rewrites the body model in place user_api_key_dict: UserAPIKeyAuth, ) -> None: """ @@ -3634,63 +3629,91 @@ def _update_model_if_team_alias_exists( Keeping the requested model name lets it resolve against the deployments that still exist (e.g. a gateway-level model group shared with the team). """ - _model: Final = data.get("model") - team_model_aliases: Final = as_str_mapping(user_api_key_dict.team_model_aliases) - if not _model or not team_model_aliases or _model not in team_model_aliases: - return - from litellm.proxy.proxy_server import llm_router - # Skip alias rewrite if this model resolves to team-specific deployments - # (team models use team_public_model_name, not model_aliases) - aliased_target: Final = team_model_aliases[_model] + rewrite: Final = _team_alias_rewrite(data.get("model"), user_api_key_dict, llm_router) + if rewrite.warning is not None: + _warn_stale_team_alias_once(rewrite.warning.key, rewrite.warning.message, *rewrite.warning.args) + if rewrite.target is not None: + data["model"] = rewrite.target - # Optional bypass for stale aliases from pre-PR deployments: - # only enabled via feature flag to preserve backwards compatibility. - # Cached at module level to avoid hot-path secret lookups on every request. - global _ENABLE_TEAM_STALE_ALIAS_BYPASS - if _ENABLE_TEAM_STALE_ALIAS_BYPASS is None: - _ENABLE_TEAM_STALE_ALIAS_BYPASS = get_secret_bool("LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS", False) - enable_stale_alias_bypass: Final = _ENABLE_TEAM_STALE_ALIAS_BYPASS - # Check if the alias points to a team-scoped UUID name - # (format: "model_name_{team_id}_{uuid}") - is_stale_team_alias: Final = require_attributes(aliased_target, _SupportsStartswith, "startswith").startswith( - f"model_name_{user_api_key_dict.team_id}_" - ) - if is_stale_team_alias and llm_router: - if aliased_target not in llm_router.registry().index("model_name"): - _warn_stale_team_alias_once( - f"deleted:{user_api_key_dict.team_id}:{_model}:{aliased_target}", + +@dataclass(frozen=True, slots=True) +class _StaleTeamAliasWarning: + key: str + message: str + args: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class _TeamAliasRewrite: + target: str | None + warning: _StaleTeamAliasWarning | None = None + + +_NO_TEAM_ALIAS_REWRITE: Final = _TeamAliasRewrite(target=None) + + +def team_alias_target(model: object, user_api_key_dict: UserAPIKeyAuth, llm_router: Router | None) -> str | None: + return _team_alias_rewrite(model, user_api_key_dict, llm_router).target + + +def _team_alias_rewrite( + model: object, user_api_key_dict: UserAPIKeyAuth, llm_router: Router | None +) -> _TeamAliasRewrite: + team_model_aliases: Final = as_str_mapping(user_api_key_dict.team_model_aliases) + if not isinstance(model, str) or not model or not team_model_aliases or model not in team_model_aliases: + return _NO_TEAM_ALIAS_REWRITE + aliased_target: Final = team_model_aliases[model] + if not isinstance(aliased_target, str): + return _NO_TEAM_ALIAS_REWRITE + is_stale_team_alias: Final = aliased_target.startswith(f"model_name_{user_api_key_dict.team_id}_") + if not is_stale_team_alias or not llm_router: + return _TeamAliasRewrite(target=aliased_target) + if aliased_target not in llm_router.registry().index("model_name"): + return _TeamAliasRewrite( + target=None, + warning=_StaleTeamAliasWarning( + f"deleted:{user_api_key_dict.team_id}:{model}:{aliased_target}", "Team model alias for model='%s', team_id='%s' targets '%s', which has no live " "deployment. Routing with the requested model name instead; remove the stale " "entry from the team's model_aliases to silence this warning.", - _sanitize_for_log(_model), - _sanitize_for_log(user_api_key_dict.team_id), - _sanitize_for_log(aliased_target), - ) - return - # This is a stale alias from pre-PR deployments. - # Check if current team deployments exist for the public name. - key: Final = (user_api_key_dict.team_id, _model) - if key in llm_router.registry().index("team_model"): - if enable_stale_alias_bypass: - # Team deployments exist; skip stale alias - return - _warn_stale_team_alias_once( - f"{user_api_key_dict.team_id}:{_model}:{aliased_target}", - "Stale team model alias detected for model='%s', team_id='%s'. " - "New sibling deployments may be unreachable. " - "Set LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS=true to enable " - "team-scoped sibling routing.", - _sanitize_for_log(_model), - _sanitize_for_log(user_api_key_dict.team_id), - ) + ( + _sanitize_for_log(model), + _sanitize_for_log(user_api_key_dict.team_id), + _sanitize_for_log(aliased_target), + ), + ), + ) + if (user_api_key_dict.team_id, model) not in llm_router.registry().index("team_model"): + return _TeamAliasRewrite(target=aliased_target) + if _stale_team_alias_bypass_enabled(): + return _NO_TEAM_ALIAS_REWRITE + return _TeamAliasRewrite( + target=aliased_target, + warning=_StaleTeamAliasWarning( + f"{user_api_key_dict.team_id}:{model}:{aliased_target}", + "Stale team model alias detected for model='%s', team_id='%s'. " + "New sibling deployments may be unreachable. " + "Set LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS=true to enable " + "team-scoped sibling routing.", + (_sanitize_for_log(model), _sanitize_for_log(user_api_key_dict.team_id)), + ), + ) - data["model"] = aliased_target + +def _stale_team_alias_bypass_enabled() -> bool: + global _ENABLE_TEAM_STALE_ALIAS_BYPASS + cached: Final = _ENABLE_TEAM_STALE_ALIAS_BYPASS + if cached is not None: + return cached + enabled: Final = get_secret_bool("LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS", False) is True + _ENABLE_TEAM_STALE_ALIAS_BYPASS = enabled + return enabled def _update_model_if_key_alias_exists( - data: dict, + data: dict[str, object], # mutable-ok: request preparation rewrites the body model in place user_api_key_dict: UserAPIKeyAuth, ) -> None: """ @@ -3705,14 +3728,45 @@ def _update_model_if_key_alias_exists( } - requested_model = "xai/grok-4-fast-non-reasoning" """ - _model: Final = data.get("model") - if ( - _model - and user_api_key_dict.aliases - and isinstance(user_api_key_dict.aliases, dict) - and _model in user_api_key_dict.aliases - ): - data["model"] = user_api_key_dict.aliases[_model] + requested: Final = data.get("model") + if isinstance(requested, str): + data["model"] = key_alias_target(requested, user_api_key_dict) + + +def key_alias_target(model: str, user_api_key_dict: UserAPIKeyAuth) -> str: + from litellm.proxy.auth.auth_checks import ( # noqa: PLC0415 # auth checks import request preparation + key_model_aliases_for_auth_check, + ) + + key_aliases: Final = key_model_aliases_for_auth_check(user_api_key_dict) + return key_aliases.get(model, model) if key_aliases else model + + +def caller_alias_target(model: str, user_api_key_dict: UserAPIKeyAuth, llm_router: Router | None) -> str: + """The model `add_litellm_data_to_request` leaves after the team's and then the key's aliases.""" + team_target: Final = team_alias_target(model, user_api_key_dict, llm_router) + return key_alias_target(team_target if team_target is not None else model, user_api_key_dict) + + +def global_then_key_alias_target(model: str, user_api_key_dict: UserAPIKeyAuth) -> str: + """The model `common_processing_pre_call_logic` leaves after the global `model_alias_map` and then the + key's aliases, ahead of the key's or team's `router_settings.model_group_alias`.""" + return key_alias_target(litellm.model_alias_map.get(model, model), user_api_key_dict) + + +def pre_call_served_model(model: str, user_api_key_dict: UserAPIKeyAuth, model_group_alias: object) -> str: + """The model group `common_processing_pre_call_logic` sends `model` to.""" + prepared: Final = global_then_key_alias_target(model, user_api_key_dict) + return resolve_model_group_alias(model_group_alias, prepared) or prepared + + +def request_served_model( + model: str, user_api_key_dict: UserAPIKeyAuth, model_group_alias: object, llm_router: Router | None +) -> str: + """The model group a request whose body names `model` is sent to, through every alias rewrite.""" + return pre_call_served_model( + caller_alias_target(model, user_api_key_dict, llm_router), user_api_key_dict, model_group_alias + ) def _apply_credential_overrides_from_model_config( diff --git a/tests/unit/proxy/auth/test_custom_auth_end_user_budget.py b/tests/unit/proxy/auth/test_custom_auth_end_user_budget.py index 63ded55c630..e75a77325c6 100644 --- a/tests/unit/proxy/auth/test_custom_auth_end_user_budget.py +++ b/tests/unit/proxy/auth/test_custom_auth_end_user_budget.py @@ -1,8 +1,9 @@ from typing import Final -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from starlette.requests import Request +from starlette.routing import Route import litellm from litellm.proxy._types import ( @@ -451,3 +452,64 @@ def test_apply_budget_limits_copies_per_model_limits_onto_end_user_params(): "end_user_model_tpm_limit": {"gpt-4o": 400000}, "end_user_model_rpm_limit": {"gpt-4o": 200}, } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "requested_model, key_aliases, run_common_checks, expect_enforced", + [ + ("free-key", {"free-key": "free-model"}, True, False), + ("free-model", {"free-model": "paid-model"}, True, True), + ("free-key", {"free-key": "free-model"}, False, True), + ], +) +async def test_custom_auth_judges_the_zero_cost_waiver_by_the_key_alias_target( + requested_model: str, + key_aliases: dict[str, str], + run_common_checks: bool, + expect_enforced: bool, + free_and_paid_router: litellm.Router, +) -> None: + valid_token: Final = UserAPIKeyAuth( + token="test_token", + aliases=key_aliases, + end_user_id="end-user", + end_user_model_max_budget={requested_model: {"budget_limit": 1.0, "time_period": "1d"}}, + ) + limiter: Final = MagicMock() + limiter.is_end_user_within_model_budget = AsyncMock( + side_effect=litellm.BudgetExceededError(current_cost=5.0, max_budget=1.0) + ) + http_request: Final = Request( + { + "type": "http", + "method": "POST", + "path": "/v1/chat/completions", + "headers": [], + "query_string": b"", + "route": Route("/v1/chat/completions", endpoint=lambda request: None), + } + ) + + async def run() -> UserAPIKeyAuth: + return await _run_post_custom_auth_checks( + valid_token=valid_token, + request=http_request, + request_data={"model": requested_model}, + route="/v1/chat/completions", + parent_otel_span=None, + ) + + with ( + patch("litellm.proxy.proxy_server.llm_router", free_and_paid_router), + patch("litellm.proxy.proxy_server.model_max_budget_limiter", limiter), + patch("litellm.proxy.proxy_server.general_settings", {"custom_auth_run_common_checks": run_common_checks}), + ): + if expect_enforced: + with pytest.raises(litellm.BudgetExceededError): + await run() + assert limiter.is_end_user_within_model_budget.await_args.kwargs["model"] == requested_model + else: + result: Final = await run() + assert result.token == "test_token" + limiter.is_end_user_within_model_budget.assert_not_awaited() diff --git a/tests/unit/proxy/auth/test_model_checks.py b/tests/unit/proxy/auth/test_model_checks.py index cd5410ffd18..1c576b1ee24 100644 --- a/tests/unit/proxy/auth/test_model_checks.py +++ b/tests/unit/proxy/auth/test_model_checks.py @@ -2484,3 +2484,58 @@ def test_prefixed_alias_over_a_bare_catch_all_lists_ids_that_route_through_it(mo ] assert f"team-partial-route/{_ROUTE_OPENAI_ID}" in v1_models assert unrouted_ids == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "requested_model, token_aliases", + [ + ("free-key", {"aliases": {"free-key": "free-model"}}), + ("free-team", {"team_model_aliases": {"free-team": "free-model"}}), + ("free-team", {"team_model_aliases": {"free-team": "free-model"}, "aliases": {"free-model": "paid-model"}}), + ("free-model", {"aliases": {"paid-model": "free-model"}}), + ("free-team", {"team_model_aliases": {"free-team": ""}}), + ( + "free-team", + {"team_id": "t1", "team_model_aliases": {"free-team": "model_name_t1_deleted"}}, + ), + ], +) +async def test_caller_alias_target_matches_the_model_add_litellm_data_to_request_dispatches( + requested_model: str, token_aliases: Mapping[str, object] +) -> None: + from unittest.mock import MagicMock + + from fastapi import Request + + from litellm.proxy._types import UserAPIKeyAuth + from litellm.proxy.litellm_pre_call_utils import caller_alias_target + from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request + + router: Final = Router( + model_list=[ + {"model_name": "free-model", "litellm_params": {"model": "openai/free-backend", "api_key": "sk-fixture"}}, + {"model_name": "paid-model", "litellm_params": {"model": "openai/paid-backend", "api_key": "sk-fixture"}}, + ] + ) + valid_token: Final = UserAPIKeyAuth(api_key="hashed-key", **token_aliases) + request: Final = MagicMock(spec=Request) + request.url = MagicMock() + request.url.path = "/v1/chat/completions" + request.method = "POST" + request.query_params = {} + request.headers = {"Content-Type": "application/json"} + request.client = MagicMock() + request.client.host = "127.0.0.1" + + with patch("litellm.proxy.proxy_server.llm_router", router): + dispatched: Final = await add_litellm_data_to_request( + data={"model": requested_model}, + request=request, + user_api_key_dict=valid_token, + proxy_config=MagicMock(), + general_settings={}, + version="test-version", + ) + served: Final = caller_alias_target(requested_model, valid_token, router) + assert served == dispatched["model"] diff --git a/tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py b/tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py index 8a5d4f7969b..0d2433d9707 100644 --- a/tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py +++ b/tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py @@ -21,6 +21,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch import fastapi import orjson import pytest +from starlette.routing import Route from litellm.redis.stores.connection import connect from litellm.redis.stores.spend_counters import SpendCounterStore @@ -13915,6 +13916,89 @@ async def test_exhausted_key_budget_spares_only_zero_spend_mcp_discovery( assert result.spend == 5.0 +@pytest.mark.asyncio +@pytest.mark.parametrize( + "requested_model, key_aliases, team_model_aliases, expect_blocked", + [ + ("free-key", {"free-key": "free-model"}, None, False), + ("free-team", None, {"free-team": "free-model"}, False), + ("free-model", {"free-model": "paid-model"}, None, True), + ("free-model", None, {"free-model": "paid-model"}, True), + ("free-model", None, None, False), + ("paid-model", None, None, True), + ], +) +async def test_exhausted_key_budget_is_waived_only_when_the_served_model_is_free( + requested_model: str, + key_aliases: dict[str, str] | None, + team_model_aliases: dict[str, str] | None, + expect_blocked: bool, + free_and_paid_router: litellm.Router, +) -> None: + from litellm.proxy.utils import hash_token + + api_key: Final = "sk-exhausted-aliased-free-model" + hashed_token: Final = hash_token(api_key) + user_api_key_cache: Final = DualCache() + await cache_key_object( + hashed_token=hashed_token, + user_api_key_obj=UserAPIKeyAuth( + token=hashed_token, + max_budget=2.0, + spend=5.0, + aliases=key_aliases or {}, + team_model_aliases=team_model_aliases, + ), + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=None, + ) + + mock_request: Final = MagicMock() + mock_request.url.path = "/v1/chat/completions" + mock_request.method = "POST" + mock_request.headers = {"authorization": f"Bearer {api_key}"} + mock_request.query_params = {} + mock_request.state = SimpleNamespace() + mock_request.scope = {"route": Route("/v1/chat/completions", endpoint=lambda request: None)} + + proxy_logging_obj: Final = MagicMock() + proxy_logging_obj.budget_alerts = AsyncMock() + proxy_logging_obj.post_call_failure_hook = AsyncMock(return_value=None) + + async def authenticate() -> UserAPIKeyAuth: + return await user_api_key_auth_builder( + request=mock_request, + api_key=f"Bearer {api_key}", + azure_api_key_header="", + anthropic_api_key_header=None, + google_ai_studio_api_key_header=None, + azure_apim_header=None, + request_data={"model": requested_model}, + ) + + with ( + patch("litellm.proxy.proxy_server.general_settings", {}), + patch("litellm.proxy.proxy_server.master_key", "sk-master"), + patch("litellm.proxy.proxy_server.prisma_client", MagicMock()), + patch("litellm.proxy.proxy_server.llm_router", free_and_paid_router), + patch("litellm.proxy.proxy_server.user_api_key_cache", user_api_key_cache), + patch("litellm.proxy.proxy_server.proxy_logging_obj", proxy_logging_obj), + patch( + "litellm.proxy.auth.user_api_key_auth.virtual_key_max_budget_alert_check", + new_callable=AsyncMock, + ), + ): + if expect_blocked: + with pytest.raises(ProxyException) as exc_info: + await authenticate() + assert "Budget has been exceeded! Key=key Current cost: 5.0, Max budget: 2.0" in exc_info.value.message + else: + result: Final = await authenticate() + assert result.token == hashed_token + assert result.spend == 5.0 + + +@pytest.mark.asyncio @pytest.mark.parametrize( "route, request_data, expected", [ @@ -13924,9 +14008,14 @@ async def test_exhausted_key_budget_spares_only_zero_spend_mcp_discovery( ("/mcp-rest/tools/call", {}, False), ], ) -def test_should_skip_budget_checks_spares_only_zero_spend_mcp_discovery( +async def test_should_skip_budget_checks_spares_only_zero_spend_mcp_discovery( route: str, request_data: dict[str, object], expected: bool ) -> None: from litellm.proxy.auth.user_api_key_auth import _should_skip_budget_checks - assert _should_skip_budget_checks(request_data=request_data, route=route, request=None, llm_router=None) is expected + assert ( + await _should_skip_budget_checks( + request_data=request_data, route=route, request=None, llm_router=None, valid_token=UserAPIKeyAuth() + ) + is expected + ) diff --git a/tests/unit/proxy/conftest.py b/tests/unit/proxy/conftest.py index 8d8781be0a8..7ff922179ce 100644 --- a/tests/unit/proxy/conftest.py +++ b/tests/unit/proxy/conftest.py @@ -546,3 +546,27 @@ def pbkdf2_on_event_loop(monkeypatch: pytest.MonkeyPatch) -> list[bool]: monkeypatch.setattr(hashlib, "pbkdf2_hmac", recording_pbkdf2) return on_event_loop + + +def _per_token_priced(model_name: str, backend: str, input_cost: float, output_cost: float) -> dict[str, object]: + return { + "model_name": model_name, + "litellm_params": { + "model": backend, + "api_key": "sk-fixture", + "input_cost_per_token": input_cost, + "output_cost_per_token": output_cost, + }, + } + + +@pytest.fixture +def free_and_paid_router() -> "litellm.Router": + import litellm + + return litellm.Router( + model_list=[ + _per_token_priced("free-model", "openai/free-backend", 0.0, 0.0), + _per_token_priced("paid-model", "openai/paid-backend", 1e-6, 2e-6), + ] + ) diff --git a/tests/unit/proxy/test_litellm_pre_call_utils.py b/tests/unit/proxy/test_litellm_pre_call_utils.py index e8ee31b27f5..f4a836b4b4b 100644 --- a/tests/unit/proxy/test_litellm_pre_call_utils.py +++ b/tests/unit/proxy/test_litellm_pre_call_utils.py @@ -11607,3 +11607,61 @@ def test_caller_tags_reach_litellm_metadata_whether_metadata_is_a_dict_or_a_json LiteLLMProxyRequestSetup.merge_caller_body_tags(request_data=data) assert set(data["litellm_metadata"]["tags"]) == {"key-tag", "client-tag"} assert data["metadata"] == caller_metadata + + +@pytest.mark.parametrize( + ("model", "global_aliases", "key_aliases", "group_alias", "expected"), + [ + pytest.param("a", {"a": "b"}, {"b": "c"}, {"c": "d"}, "d", id="global-then-key-then-router"), + pytest.param("a", {}, {}, {"a": {"model": "x"}}, "x", id="router-item-entry"), + pytest.param("a", {"a": "b"}, {}, None, "b", id="global-only"), + pytest.param("a", {}, {}, None, "a", id="no-alias"), + ], +) +def test_pre_call_served_model_applies_aliases_in_request_preparation_order( + monkeypatch: pytest.MonkeyPatch, + model: str, + global_aliases: dict[str, str], + key_aliases: dict[str, str], + group_alias: object, + expected: str, +) -> None: + from litellm.proxy.litellm_pre_call_utils import pre_call_served_model + + monkeypatch.setattr(litellm, "model_alias_map", global_aliases) + + assert pre_call_served_model(model, UserAPIKeyAuth(aliases=key_aliases), group_alias) == expected + + +@pytest.mark.parametrize( + ("token_fields", "expected"), + [ + pytest.param({"aliases": {"requested": 7}}, "requested", id="non-string-key-alias"), + pytest.param({"team_model_aliases": {"requested": 7}}, "requested", id="non-string-team-alias"), + pytest.param({"aliases": {"mid": 7}, "team_model_aliases": {"requested": "mid"}}, "mid", id="after-team"), + ], +) +def test_request_served_model_ignores_a_malformed_alias_like_request_preparation_does( + token_fields: dict[str, object], expected: str +) -> None: + from litellm.proxy.litellm_pre_call_utils import _update_model_if_key_alias_exists, _update_model_if_team_alias_exists + from litellm.proxy.litellm_pre_call_utils import request_served_model + + valid_token: Final = UserAPIKeyAuth(**token_fields) + prepared: Final[dict[str, object]] = {"model": "requested"} # mutable-ok: the rewrite functions edit the body + with patch("litellm.proxy.proxy_server.llm_router", None): + _update_model_if_team_alias_exists(data=prepared, user_api_key_dict=valid_token) + _update_model_if_key_alias_exists(data=prepared, user_api_key_dict=valid_token) + + assert request_served_model("requested", valid_token, None, None) == prepared["model"] == expected + + +def test_stale_team_alias_bypass_flag_is_read_once_per_process(monkeypatch: pytest.MonkeyPatch) -> None: + import litellm.proxy.litellm_pre_call_utils as pre_call_utils + + monkeypatch.setattr(pre_call_utils, "_ENABLE_TEAM_STALE_ALIAS_BYPASS", None) + monkeypatch.setenv("LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS", "true") + first: Final = pre_call_utils._stale_team_alias_bypass_enabled() + monkeypatch.setenv("LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS", "false") + + assert (first, pre_call_utils._stale_team_alias_bypass_enabled()) == (True, True) diff --git a/tests/unit/proxy/test_zero_cost_model_budget_bypass.py b/tests/unit/proxy/test_zero_cost_model_budget_bypass.py index ce1dfce1646..35e75a85520 100644 --- a/tests/unit/proxy/test_zero_cost_model_budget_bypass.py +++ b/tests/unit/proxy/test_zero_cost_model_budget_bypass.py @@ -6,12 +6,15 @@ to models with zero cost (e.g., on-premises models). """ import asyncio -from typing import Optional +from typing import Final, Optional from unittest.mock import MagicMock, patch import pytest +from starlette.requests import Request as StarletteRequest +from starlette.routing import Route import litellm +from litellm.constants import MODEL_GROUP_ALIAS_RESOLVED_SCOPE_KEY from litellm.caching.caching import DualCache from litellm.proxy._types import ( LiteLLM_BudgetTable, @@ -653,8 +656,8 @@ class TestOverBudgetRequestThroughModelGroupAlias: router = self._router() request_data = {"model": model} - skip_budget_checks = _should_skip_budget_checks( - request_data=request_data, route=self.ROUTE, request=None, llm_router=router + skip_budget_checks = await _should_skip_budget_checks( + request_data=request_data, route=self.ROUTE, request=None, llm_router=router, valid_token=UserAPIKeyAuth() ) return await common_checks( request_body=request_data, @@ -692,3 +695,164 @@ class TestOverBudgetRequestThroughModelGroupAlias: async def test_over_budget_request_for_paid_model_is_blocked(self, mock_proxy_logging): with pytest.raises(litellm.BudgetExceededError): await self._request("paid-model", mock_proxy_logging) + + +_CHAT: Final = "/v1/chat/completions" + + +def _routed_request( + template: str | None = _CHAT, query: bytes = b"", router_aliases_applied: bool = True +) -> StarletteRequest: + scope: Final[dict[str, object]] = { # mutable-ok: ASGI scope literal + "type": "http", + "method": "POST", + "path": _CHAT, + "headers": [], + "query_string": query, + MODEL_GROUP_ALIAS_RESOLVED_SCOPE_KEY: router_aliases_applied, + } + if template is not None: + scope["route"] = Route(template, endpoint=lambda request: None) + return StarletteRequest(scope) + + +_KEY_FREE: Final = {"aliases": {"free-key": "free-model"}} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("body_model", "token_fields", "request_kwargs", "settings", "expected_skip"), + [ + pytest.param("free-key", _KEY_FREE, {}, {}, True, id="key-alias-to-free"), + pytest.param("free-team", {"team_model_aliases": {"free-team": "free-model"}}, {}, {}, True, id="team-to-free"), + pytest.param("free-model", {"aliases": {"free-model": "paid-model"}}, {}, {}, False, id="key-alias-to-paid"), + pytest.param( + "free-model", {"team_model_aliases": {"free-model": "paid-model"}}, {}, {}, False, id="team-alias-to-paid" + ), + pytest.param( + "free-team", + {"team_model_aliases": {"free-team": "free-model"}, "aliases": {"free-model": "paid-model"}}, + {}, + {}, + False, + id="team-then-key", + ), + pytest.param( + "free-team", + { + "team_model_aliases": {"free-team": "free-model"}, + "router_settings": {"model_group_alias": {"free-model": "paid-model"}}, + }, + {}, + {}, + False, + id="router-alias-after-team", + ), + pytest.param(["free-key"], _KEY_FREE, {}, {}, False, id="list-body"), + pytest.param("free-key", _KEY_FREE, {"query": b"model=free-key"}, {}, False, id="query-model"), + pytest.param( + "free-key", _KEY_FREE, {"template": "/engines/{model:path}/chat/completions"}, {}, False, id="path" + ), + pytest.param("free-key", _KEY_FREE, {"template": None}, {}, False, id="request-built-inside-the-proxy"), + pytest.param("free-key", _KEY_FREE, {"router_aliases_applied": False}, {}, False, id="before-router-alias"), + pytest.param("free-key", _KEY_FREE, {}, {"completion_model": "free-model"}, False, id="completion-model-set"), + ], +) +async def test_budget_waiver_judges_the_model_request_preparation_serves( + body_model: str | list[str], + token_fields: dict[str, object], + request_kwargs: dict[str, object], + settings: dict[str, object], + expected_skip: bool, + free_and_paid_router: Router, +) -> None: + from litellm.proxy.auth.user_api_key_auth import _should_skip_budget_checks + + with patch("litellm.proxy.proxy_server.general_settings", settings): + skip: Final = await _should_skip_budget_checks( + request_data={"model": body_model}, + route=_CHAT, + request=_routed_request(**request_kwargs), + llm_router=free_and_paid_router, + valid_token=UserAPIKeyAuth(token="t", **token_fields), + ) + + assert skip is expected_skip + + +@pytest.mark.asyncio +async def test_sub_call_is_judged_by_its_own_model_not_the_callers_aliases(free_and_paid_router: Router) -> None: + from litellm.proxy.auth.user_api_key_auth import _should_skip_budget_checks + + assert ( + await _should_skip_budget_checks( + request_data={"model": "paid-model"}, + route=_CHAT, + request=_routed_request(), + llm_router=free_and_paid_router, + valid_token=UserAPIKeyAuth(token="t", aliases={"paid-model": "free-model"}), + sub_call=True, + ) + is False + ) + + +@pytest.mark.asyncio +async def test_common_checks_honours_the_callers_decision_not_to_skip_budgets( + mock_proxy_logging: ProxyLogging, free_and_paid_router: Router +) -> None: + with pytest.raises(litellm.BudgetExceededError): + await common_checks( + request_body={"model": "free-model"}, + team_object=None, + user_object=LiteLLM_UserTable(user_id="test-user", spend=100.0, max_budget=50.0), + end_user_object=None, + global_proxy_spend=None, + general_settings={}, + route=_CHAT, + llm_router=free_and_paid_router, + proxy_logging_obj=mock_proxy_logging, + valid_token=UserAPIKeyAuth(token="t", aliases={"free-model": "paid-model"}), + request=MagicMock(), + skip_budget_checks=False, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("bypass_flag", "expected_skip"), [("false", True), ("true", False)]) +async def test_waiver_skips_a_stale_team_alias_exactly_when_dispatch_does( + monkeypatch: pytest.MonkeyPatch, bypass_flag: str, expected_skip: bool +) -> None: + import litellm.proxy.litellm_pre_call_utils as pre_call_utils + from litellm.proxy.auth.user_api_key_auth import _should_skip_budget_checks + + monkeypatch.setenv("LITELLM_ENABLE_TEAM_STALE_ALIAS_BYPASS", bypass_flag) + monkeypatch.setattr(pre_call_utils, "_ENABLE_TEAM_STALE_ALIAS_BYPASS", None) + team_deployment: Final = {"api_key": "sk-fixture", "output_cost_per_token": 0.0} + router: Final = Router( + model_list=[ + { + "model_name": "model_name_team-1_legacy", + "litellm_params": {"model": "openai/free-backend", "input_cost_per_token": 0.0, **team_deployment}, + "model_info": {"team_id": "team-1", "team_public_model_name": "legacy"}, + }, + { + "model_name": "model_name_team-1_current", + "litellm_params": {"model": "openai/paid-backend", "input_cost_per_token": 1e-6, **team_deployment}, + "model_info": {"team_id": "team-1", "team_public_model_name": "team-model"}, + }, + ] + ) + + assert ( + await _should_skip_budget_checks( + request_data={"model": "team-model"}, + route=_CHAT, + request=_routed_request(), + llm_router=router, + valid_token=UserAPIKeyAuth( + token="t", team_id="team-1", team_model_aliases={"team-model": "model_name_team-1_legacy"} + ), + ) + is expected_skip + ) diff --git a/tests/unit/router_utils/test_router_utils_common_utils.py b/tests/unit/router_utils/test_router_utils_common_utils.py index 42b21692a8f..ea384d837fd 100644 --- a/tests/unit/router_utils/test_router_utils_common_utils.py +++ b/tests/unit/router_utils/test_router_utils_common_utils.py @@ -19,7 +19,6 @@ from litellm.litellm_core_utils.routing_common_utils import ( format_fallback_outcome_message, provider_for_generic_call, resolve_model_group_alias, - resolve_served_model, truncate_fallback_error_detail, PROVIDER_SCOPED_CREDENTIAL_PARAMS, warn_on_provider_credential_mismatch, @@ -579,25 +578,6 @@ class TestResolveModelGroupAlias: assert router.get_model_from_alias("group-b") is None -@pytest.mark.parametrize( - "model, model_alias_map, key_aliases, model_group_alias, expected", - [ - pytest.param("a", {"a": "b"}, {"b": "c"}, {"c": "d"}, "d", id="global-key-router-order"), - pytest.param("a", {}, ["a", "b"], {}, "a", id="non-mapping-key-aliases-ignored"), - pytest.param("a", {}, {}, {"a": {"model": "x"}}, "x", id="router-item-entry"), - pytest.param("a", {}, {}, {}, "a", id="no-alias"), - ], -) -def test_resolves_served_model_aliases_in_order( - model: str, - model_alias_map: object, - key_aliases: object, - model_group_alias: object, - expected: str, -) -> None: - assert resolve_served_model(model, model_alias_map, key_aliases, model_group_alias) == expected - - class TestTruncateFallbackErrorDetail: def test_short_detail_is_returned_unchanged(self): assert truncate_fallback_error_detail("boom") == "boom"