scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 455

fix(proxy): judge the zero-cost budget waiver by the model a key or team alias serves (#51058)

Browse source 
* fix(proxy): judge the zero-cost budget waiver by the model a key or team alias serves

The zero-cost budget waiver judged the name the client sent, but key `aliases` and team
`model_aliases` are applied later during request preparation. An over-budget key got a 422
for an alias of a $0 model, and an alias named like a $0 group that points at a paid model
was waived and served past the budget.

For a string body `model` on chat, completions, responses and messages, the key-path,
centralized and custom-auth waivers now judge the model request preparation serves. That
model comes from the same team, key, global and router alias functions that request
preparation uses, so the two cannot drift. Other routes and the JWT path keep judging the
requested model. common_checks no longer overrides a caller's explicit decision not to skip
budgets, and proxy-made sub-calls are judged by their own model.

Resolves LIT-10283

* fix(proxy): waive only once auth has applied the router aliases, and keep empty team alias targets

The served-model waiver now applies only after auth's router-alias rewrite has run on the
request body. Custom auth without common checks runs its waiver before that rewrite, so it
keeps judging the requested model instead of a key or team alias of a name the router
rewrites elsewhere. The waiver also keeps an empty-string team alias target, matching
request preparation.

Resolves LIT-10283

* fix(proxy): waive by the served model only for a client's own body-model request

The waiver resolved aliases for any request on a chat, completions, responses or messages
path, including the HTTP request the Responses WebSocket builds for first-frame admission.
That session dispatches the model as an endpoint model, which skips the team alias, so a
team alias of a paid model to a free one waived the paid model's budget.

The waiver now resolves aliases only when FastAPI matched the request to one of the body
model routes, which requests built inside the proxy and path model routes never are. The
waiver tests are folded into one table, share a priced router fixture, and cover the
stale team alias flag cache.

Resolves LIT-10283

---------

Co-authored-by: Joshua Valluru <326636767+joshua-berri@users.noreply.github.com>
This commit is contained in:
joshua-berri 2026-11-10 18:28:49 -08:00 • committed by GitHub
parent 35e501974e
commit 8092edc7a1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 685 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
litellm/litellm_core_utils/routing_common_utils.py
View file

@ -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.

20
litellm/proxy/auth/auth_checks.py
View file

@ -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 = (

107
litellm/proxy/auth/user_api_key_auth.py
View file

@ -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
)

37
litellm/proxy/common_request_processing.py
View file

@ -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,
):

178
litellm/proxy/litellm_pre_call_utils.py
View file

@ -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(

64
tests/unit/proxy/auth/test_custom_auth_end_user_budget.py
View file

@ -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()

55
tests/unit/proxy/auth/test_model_checks.py
View file

@ -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"]

93
tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py
View file

@ -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
)

24
tests/unit/proxy/conftest.py
View file

@ -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),
]
)

58
tests/unit/proxy/test_litellm_pre_call_utils.py
View file

@ -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)

170
tests/unit/proxy/test_zero_cost_model_budget_bypass.py
View file

@ -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
)

20
tests/unit/router_utils/test_router_utils_common_utils.py
View file

@ -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"