scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-24 00:52:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

feat(router): time-windowed team reservation of deployments via model_info.access_windows (#42398)

Browse source 
* feat(router): time-windowed team reservation of deployments via model_info.access_windows

Deployments can declare model_info.access_windows, a list of local wall-clock windows (IANA timezone, cross-midnight allowed) that reserve the deployment for the listed team_ids. While a window is active the router drops the deployment for every other request, including no-team and admin requests, on every candidate path (model name, model id, specific_deployment, early-resolve, wildcard, litellm_params.model lookup, fallbacks). If every candidate is reserved the request fails with a 400 naming the window end instead of falling back. Outside a window routing is unchanged and reserved deployments stay visible in /model/info and /v1/models. Malformed windows (bad time, unknown timezone, empty team_ids, start equal to end, offset-aware times) fail proxy startup with a clear error since the proxy router runs with ignore_invalid_deployments=True

Resolves LIT-8308

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(router): cover _filter_reserved_deployments directly for coverage gate

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(router): keep reservation filtering immutable

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(router): drop strategy markers before reservation filtering

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-22 13:22:15 -05:00 • committed by GitHub
parent 8a9305fa27
commit 21c442759e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 769 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
litellm/proxy/proxy_server.py
View file

@ -132,6 +132,7 @@ from litellm.proxy.common_utils.callback_utils import (
strip_callback_config,
)
from litellm.proxy.common_utils.realtime_utils import _realtime_request_body
from litellm.router_utils.access_windows import access_windows_config_error
from litellm.router_utils.add_retry_fallback_headers import (
get_fallback_errors_from_headers,
get_hidden_params_dict,
@ -4852,6 +4853,22 @@ def validate_deployment_complexity_router_placement(model: Mapping[str, object])
raise ValueError(f"model {model.get('model_name', '')!r}: {violation}")
def validate_deployment_access_windows(model: Mapping[str, object]) -> None:
"""
Reject a malformed `model_info.access_windows` instead of silently dropping the deployment.
Checked here rather than on `ModelInfo` because the proxy builds its router with
`ignore_invalid_deployments=True`, so a rejection further down turns a bad
deployment into a silently missing model instead of a refusal to start.
"""
model_info: Final = model.get("model_info")
if not isinstance(model_info, Mapping):
return
error: Final = access_windows_config_error(model_info, model_name=str(model.get("model_name", "")))
if error is not None:
raise ValueError(error)
def validate_auto_router_capability_limits(model_list: Sequence[Mapping[str, object]], *, limit: int | None) -> None:
"""
Refuse to start when config.yaml defines more auto-routers claiming a licensed capability than allowed.
@ -6552,6 +6569,7 @@ class ProxyConfig:
model["litellm_params"][k] = get_secret(v)
validate_deployment_max_agentic_loops(model)
validate_deployment_complexity_router_placement(model)
validate_deployment_access_windows(model)
pin_complexity_router_model_id(model)
complexity_router_config = model["litellm_params"].get("complexity_router_config")
if isinstance(complexity_router_config, dict):

97
litellm/router.py
View file

@ -31,6 +31,7 @@ from collections.abc import (
MutableMapping,
Sequence,
)
from datetime import datetime, timezone
from functools import lru_cache, partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypeAlias, TypeVar, Union, cast
@ -138,6 +139,7 @@ from litellm.router_strategy.tag_based_routing import (
get_deployments_for_tag,
is_valid_deployment_tag,
)
from litellm.router_utils.access_windows import access_windows_config_error, filter_reserved_deployments
from litellm.router_utils.add_retry_fallback_headers import (
_HiddenParamsHost,
add_fallback_headers_to_response,
@ -8774,6 +8776,9 @@ class Router:
)
if ptu_error is not None and is_ptu_cost_attribution_enabled():
raise ValueError(ptu_error)
access_windows_error: Final = access_windows_config_error(_model_info, model_name=_model_name)
if access_windows_error is not None:
raise ValueError(access_windows_error)
zeroed_pricing: Final = zeroed_ptu_pricing(_model_info, _litellm_params) if config_sourced else None
litellm_params: Final[LiteLLM_Params] = LiteLLM_Params(
**( # pyright: ignore[reportArgumentType] # untyped merged dict; already true for every field here
@ -12535,12 +12540,30 @@ class Router:
request_team_id: Final = get_request_team_id(request_kwargs)
# check if aliases set on litellm model alias map
if specific_deployment is True:
return model, self._get_deployment_by_litellm_model(model=model)
return model, self._drop_strategy_markers(
model,
self._filter_reserved_deployments(
model=model,
healthy_deployments=self._get_deployment_by_litellm_model(model=model),
request_team_id=request_team_id,
),
)
elif model not in self.model_names and self.has_model_id(model):
deployment: Final = self.get_deployment(model_id=model)
if deployment is not None:
deployment_model: Final = deployment.litellm_params.model
return deployment_model, deployment.model_dump(exclude_none=True)
return deployment_model, cast( # cast-ok: contract requires a plain dict for a single deployment
dict,
self._filter_reserved_deployments(
model=deployment_model,
healthy_deployments=(
cast( # cast-ok: model_dump of a router deployment
DeploymentTypedDict, deployment.model_dump(exclude_none=True)
),
),
request_team_id=request_team_id,
)[0],
)
raise ValueError(
f"LiteLLM Router: Trying to call specific deployment, but Model ID :{model} does not exist in Model ID map"
)
@ -12558,8 +12581,26 @@ class Router:
)
if early is not None:
if not isinstance(early[1], list):
return early
return early[0], self._drop_strategy_markers(early[0], early[1])
return early[0], cast( # cast-ok: contract requires a plain dict for a single deployment
dict,
self._filter_reserved_deployments(
model=early[0],
healthy_deployments=(
cast( # cast-ok: early resolve returns a router deployment
DeploymentTypedDict, early[1]
),
),
request_team_id=request_team_id,
)[0],
)
return early[0], self._drop_strategy_markers(
early[0],
self._filter_reserved_deployments(
model=early[0],
healthy_deployments=early[1],
request_team_id=request_team_id,
),
)
## get healthy deployments
### get all deployments
@ -12569,10 +12610,14 @@ class Router:
else self._get_all_deployments(model_name=model, team_id=request_team_id)
)
_pre_model_access_group_filter_len: Final = len(healthy_deployments)
healthy_deployments = self._filter_deployments_by_model_access_groups(
healthy_deployments = self._filter_reserved_deployments(
model=model,
healthy_deployments=healthy_deployments,
request_kwargs=request_kwargs,
healthy_deployments=self._filter_deployments_by_model_access_groups(
model=model,
healthy_deployments=healthy_deployments,
request_kwargs=request_kwargs,
request_team_id=request_team_id,
),
request_team_id=request_team_id,
)
_access_group_filter_emptied_candidates = (
@ -12585,10 +12630,14 @@ class Router:
# _get_deployment_by_litellm_model does not re-apply that filter.
if _pre_model_access_group_filter_len == 0:
_litellm_model_deployments: Final = self._get_deployment_by_litellm_model(model=model)
healthy_deployments = self._filter_deployments_by_model_access_groups(
healthy_deployments = self._filter_reserved_deployments(
model=model,
healthy_deployments=_litellm_model_deployments,
request_kwargs=request_kwargs,
healthy_deployments=self._filter_deployments_by_model_access_groups(
model=model,
healthy_deployments=_litellm_model_deployments,
request_kwargs=request_kwargs,
request_team_id=request_team_id,
),
request_team_id=request_team_id,
)
# If the litellm-model lookup produced candidates that access-group
@ -12616,10 +12665,14 @@ class Router:
# Re-assign model to the fallback and try to get deployments again
model = fallback_model
healthy_deployments = self._get_all_deployments(model_name=model, team_id=request_team_id)
healthy_deployments = self._filter_deployments_by_model_access_groups(
healthy_deployments = self._filter_reserved_deployments(
model=model,
healthy_deployments=healthy_deployments,
request_kwargs=request_kwargs,
healthy_deployments=self._filter_deployments_by_model_access_groups(
model=model,
healthy_deployments=healthy_deployments,
request_kwargs=request_kwargs,
request_team_id=request_team_id,
),
request_team_id=request_team_id,
)
@ -12653,6 +12706,24 @@ class Router:
)
return selectable
def _filter_reserved_deployments(
self,
model: str,
healthy_deployments: Sequence[DeploymentTypedDict],
request_team_id: str | None,
) -> tuple[DeploymentTypedDict, ...]:
result: Final = filter_reserved_deployments(
self._drop_strategy_markers(model, healthy_deployments), request_team_id, now=datetime.now(timezone.utc)
)
if result.blocking_window is not None and len(result.deployments) == 0:
raise litellm.BadRequestError(
message=f"Deployment {model} is reserved for another team until "
f"{result.blocking_window.end:%H:%M} {result.blocking_window.timezone}",
model=model,
llm_provider="",
)
return result.deployments
def _filter_deployments_by_model_access_groups(
self,
model: str,

75
litellm/router_utils/access_windows.py Normal file
View file

@ -0,0 +1,75 @@
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Final, Generic, TypeVar
from zoneinfo import ZoneInfo
from pydantic import TypeAdapter, ValidationError
from litellm.types.router import ModelAccessWindow
_WINDOWS_ADAPTER: Final = TypeAdapter(tuple[ModelAccessWindow, ...])
_DeploymentT = TypeVar("_DeploymentT", bound=Mapping[str, object])
def parse_access_windows(model_info: Mapping[str, object]) -> tuple[ModelAccessWindow, ...]:
raw: Final = model_info.get("access_windows")
if raw is None:
return ()
return _WINDOWS_ADAPTER.validate_python(raw)
def access_windows_config_error(model_info: Mapping[str, object], *, model_name: str) -> str | None:
if model_info.get("access_windows") is None:
return None
try:
parse_access_windows(model_info)
except ValidationError as exc:
first: Final = exc.errors()[0]
loc: Final = ".".join(str(part) for part in first["loc"])
return f"model '{model_name}': invalid model_info.access_windows: {loc}: {first['msg']}"
return None
def is_window_active(window: ModelAccessWindow, now: datetime) -> bool:
aware: Final = now if now.tzinfo is not None else now.replace(tzinfo=timezone.utc)
local: Final = aware.astimezone(ZoneInfo(window.timezone)).time()
if window.start < window.end:
return window.start <= local < window.end
return local >= window.start or local < window.end
@dataclass(frozen=True, slots=True)
class ReservationFilterResult(Generic[_DeploymentT]):
deployments: tuple[_DeploymentT, ...]
blocking_window: ModelAccessWindow | None
def _reservation_blocking_window(
deployment: Mapping[str, object], request_team_id: str | None, now: datetime
) -> ModelAccessWindow | None:
model_info: Final = deployment.get("model_info")
if not isinstance(model_info, Mapping):
return None
active: Final = tuple(window for window in parse_access_windows(model_info) if is_window_active(window, now))
if not active:
return None
if request_team_id is not None and any(request_team_id in window.team_ids for window in active):
return None
return active[0]
def filter_reserved_deployments(
healthy_deployments: Sequence[_DeploymentT],
request_team_id: str | None,
now: datetime,
) -> ReservationFilterResult[_DeploymentT]:
checks: Final = tuple(
(deployment, _reservation_blocking_window(deployment, request_team_id, now))
for deployment in healthy_deployments
)
return ReservationFilterResult(
deployments=tuple(deployment for deployment, blocking in checks if blocking is None),
blocking_window=next((blocking for _, blocking in checks if blocking is not None), None),
)

41
litellm/types/router.py
View file

@ -7,6 +7,7 @@ import enum
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import httpx
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
@ -163,6 +164,44 @@ def _as_utc(value: datetime.datetime | None) -> datetime.datetime | None:
return value.astimezone(datetime.timezone.utc)
class ModelAccessWindow(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
start: datetime.time
end: datetime.time
timezone: str
team_ids: tuple[str, ...] = Field(min_length=1)
@field_validator("start", "end")
@classmethod
def _naive_wall_clock(cls, value: datetime.time) -> datetime.time:
if value.tzinfo is not None:
raise ValueError("start and end must be local wall-clock times without a UTC offset")
return value
@field_validator("timezone")
@classmethod
def _known_iana_timezone(cls, value: str) -> str:
try:
ZoneInfo(value)
except (ZoneInfoNotFoundError, ValueError) as exc:
raise ValueError(f"unknown IANA timezone '{value}'") from exc
return value
@field_validator("team_ids")
@classmethod
def _non_empty_team_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]:
if any(not team_id for team_id in value):
raise ValueError("team_ids entries must be non-empty")
return value
@model_validator(mode="after")
def _start_differs_from_end(self) -> "ModelAccessWindow":
if self.start == self.end:
raise ValueError("start and end must differ")
return self
class ModelInfo(MirroredPricingParams):
id: str | None # Allow id to be optional on input, but it will always be present as a str in the model instance
db_model: bool = False # used for proxy - to separate models which are stored in the db vs. config.
@ -188,6 +227,8 @@ class ModelInfo(MirroredPricingParams):
# admin-toggled pause flag; mirrors LiteLLM_ProxyModelTable.blocked
blocked: bool | None = None
access_windows: tuple[ModelAccessWindow, ...] | None = None
# Bounds live on the model rather than litellm.constants: names there reach
# litellm/__init__ through several modules' star re-exports, and a Final rebound that
# way trips the basedpyright gate.

57
tests/test_litellm/proxy/proxy_server/test_proxy_config.py
View file

@ -32,6 +32,7 @@ from litellm.proxy.proxy_server import (
_scrub_guardrail_inner,
resolve_complexity_router_plugins,
resolve_routing_plugins,
validate_deployment_access_windows,
validate_deployment_complexity_router_placement,
validate_deployment_max_agentic_loops,
validate_auto_router_capability_limits,
@ -4714,3 +4715,59 @@ def test_websearch_interception_settings_can_be_named_in_supported_db_objects(mo
monkeypatch.setattr(proxy_server, "general_settings", {"supported_db_objects": ["models"]})
assert proxy_server.should_load_db_object(object_type="websearch_interception_settings") is False
def test_validate_deployment_access_windows_rejects_malformed_time():
model = {
"model_name": "gpt-4o-shared",
"litellm_params": {"model": "gpt-4o"},
"model_info": {
"access_windows": [
{"start": "25:00", "end": "06:00", "timezone": "America/New_York", "team_ids": ["t"]}
]
},
}
with pytest.raises(ValueError, match="access_windows") as exc_info:
validate_deployment_access_windows(model)
assert "gpt-4o-shared" in str(exc_info.value)
def test_validate_deployment_access_windows_rejects_unknown_timezone():
model = {
"model_name": "gpt-4o-shared",
"litellm_params": {"model": "gpt-4o"},
"model_info": {
"access_windows": [
{"start": "22:00", "end": "06:00", "timezone": "Mars/Olympus", "team_ids": ["t"]}
]
},
}
with pytest.raises(ValueError, match="Mars/Olympus"):
validate_deployment_access_windows(model)
def test_validate_deployment_access_windows_accepts_valid_and_absent():
assert (
validate_deployment_access_windows(
{
"model_name": "gpt-4o-shared",
"litellm_params": {"model": "gpt-4o"},
"model_info": {
"access_windows": [
{"start": "22:00", "end": "06:00", "timezone": "America/New_York", "team_ids": ["t"]}
]
},
}
)
is None
)
assert validate_deployment_access_windows({"model_name": "m", "litellm_params": {"model": "m"}}) is None
assert (
validate_deployment_access_windows(
{"model_name": "m", "litellm_params": {"model": "m"}, "model_info": {"id": "x"}}
)
is None
)

190
tests/test_litellm/router_utils/test_access_windows.py Normal file
View file

@ -0,0 +1,190 @@
from datetime import datetime, time, timezone
from typing import Final
from litellm.router_utils.access_windows import (
access_windows_config_error,
filter_reserved_deployments,
is_window_active,
)
from litellm.types.router import ModelAccessWindow
_NIGHT_NY: Final = ModelAccessWindow(
start=time(22, 0),
end=time(6, 0),
timezone="America/New_York",
team_ids=("team-nightly",),
)
def _window(start: str, end: str, tz: str = "UTC", team_ids=("team-a",)) -> ModelAccessWindow:
return ModelAccessWindow(
start=time.fromisoformat(start),
end=time.fromisoformat(end),
timezone=tz,
team_ids=tuple(team_ids),
)
def _deployment(windows: object = None) -> dict:
if windows is None:
return {"model_info": {}}
return {"model_info": {"access_windows": windows}}
def test_same_day_window_active_and_inactive():
window: Final = _window("09:00", "17:00")
assert is_window_active(window, datetime(2026, 3, 9, 12, 0, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 9, 20, 0, tzinfo=timezone.utc)) is False
def test_cross_midnight_window():
window: Final = _window("22:00", "06:00")
assert is_window_active(window, datetime(2026, 3, 9, 23, 0, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 10, 5, 59, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 9, 12, 0, tzinfo=timezone.utc)) is False
def test_start_boundary_inclusive_and_end_boundary_exclusive():
window: Final = _window("22:00", "06:00")
assert is_window_active(window, datetime(2026, 3, 9, 22, 0, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 10, 6, 0, tzinfo=timezone.utc)) is False
def test_dst_spring_forward_gap_uses_real_local_time():
window: Final = ModelAccessWindow(
start=time(1, 30),
end=time(3, 30),
timezone="America/New_York",
team_ids=("team-a",),
)
assert is_window_active(window, datetime(2026, 3, 8, 6, 30, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 8, 7, 0, tzinfo=timezone.utc)) is True
assert is_window_active(window, datetime(2026, 3, 8, 7, 30, tzinfo=timezone.utc)) is False
def test_naive_now_is_treated_as_utc():
window: Final = _window("09:00", "17:00")
assert is_window_active(window, datetime(2026, 3, 9, 12, 0)) is True
def test_team_in_second_window_is_kept():
deployments: Final = (
_deployment([
{"start": "01:00", "end": "02:00", "timezone": "UTC", "team_ids": ["team-other"]},
{"start": "20:00", "end": "23:59", "timezone": "UTC", "team_ids": ["team-a"]},
]),
)
result: Final = filter_reserved_deployments(
deployments, "team-a", now=datetime(2026, 3, 9, 21, 0, tzinfo=timezone.utc)
)
assert result.deployments == deployments
assert result.blocking_window is None
def test_unlisted_team_is_dropped_with_blocking_window():
deployments: Final = (_deployment([_NIGHT_NY.model_dump()]),)
result: Final = filter_reserved_deployments(
deployments, "team-b", now=datetime(2026, 3, 10, 4, 0, tzinfo=timezone.utc)
)
assert result.deployments == ()
assert result.blocking_window == _NIGHT_NY
def test_missing_team_id_is_dropped():
result: Final = filter_reserved_deployments(
(_deployment([_NIGHT_NY.model_dump()]),),
None,
now=datetime(2026, 3, 10, 4, 0, tzinfo=timezone.utc),
)
assert result.deployments == ()
assert result.blocking_window == _NIGHT_NY
def test_listed_team_is_kept():
deployments: Final = (_deployment([_NIGHT_NY.model_dump()]),)
result: Final = filter_reserved_deployments(
deployments, "team-nightly", now=datetime(2026, 3, 10, 4, 0, tzinfo=timezone.utc)
)
assert result.deployments == deployments
assert result.blocking_window is None
def test_deployment_without_windows_kept_for_anyone():
deployments: Final = (_deployment(),)
result: Final = filter_reserved_deployments(
deployments, None, now=datetime(2026, 3, 10, 4, 0, tzinfo=timezone.utc)
)
assert result.deployments == deployments
assert result.blocking_window is None
def test_inactive_window_keeps_deployment_for_unlisted_team():
deployments: Final = (_deployment([_NIGHT_NY.model_dump()]),)
result: Final = filter_reserved_deployments(
deployments, "team-b", now=datetime(2026, 3, 10, 16, 0, tzinfo=timezone.utc)
)
assert result.deployments == deployments
assert result.blocking_window is None
def test_unreserved_deployment_survives_for_other_team():
reserved: Final = _deployment([_NIGHT_NY.model_dump()])
open_deployment: Final = _deployment()
result: Final = filter_reserved_deployments(
(reserved, open_deployment),
"team-b",
now=datetime(2026, 3, 10, 4, 0, tzinfo=timezone.utc),
)
assert result.deployments == (open_deployment,)
assert result.blocking_window == _NIGHT_NY
def test_config_error_unknown_timezone():
error: Final = access_windows_config_error(
{"access_windows": [{"start": "22:00", "end": "06:00", "timezone": "Mars/Olympus", "team_ids": ["t"]}]},
model_name="nightly-model",
)
assert error is not None
assert "nightly-model" in error
assert "access_windows" in error
assert "Mars/Olympus" in error
def test_config_error_bad_time():
error: Final = access_windows_config_error(
{"access_windows": [{"start": "25:00", "end": "06:00", "timezone": "UTC", "team_ids": ["t"]}]},
model_name="m",
)
assert error is not None
assert "access_windows" in error
def test_config_error_empty_team_ids():
error: Final = access_windows_config_error(
{"access_windows": [{"start": "22:00", "end": "06:00", "timezone": "UTC", "team_ids": []}]},
model_name="m",
)
assert error is not None
def test_config_error_start_equals_end():
error: Final = access_windows_config_error(
{"access_windows": [{"start": "22:00", "end": "22:00", "timezone": "UTC", "team_ids": ["t"]}]},
model_name="m",
)
assert error is not None
def test_config_error_none_when_absent():
assert access_windows_config_error({}, model_name="m") is None
assert access_windows_config_error({"access_windows": None}, model_name="m") is None
def test_config_error_offset_aware_time():
error: Final = access_windows_config_error(
{"access_windows": [{"start": "22:00+05:00", "end": "06:00", "timezone": "UTC", "team_ids": ["t"]}]},
model_name="m",
)
assert error is not None
assert "UTC offset" in error

241
tests/test_litellm/test_router.py
View file

@ -9,7 +9,7 @@ import sys
import threading
import warnings
from collections.abc import Awaitable, Callable, Mapping
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from typing import Final, Literal
from unittest.mock import AsyncMock, MagicMock, patch
@ -17620,3 +17620,242 @@ class TestMemberAutoRouterInference:
monkeypatch.setitem(sys.modules, "fastapi", None)
monkeypatch.delitem(sys.modules, "litellm.proxy.auth.auto_router_checks", raising=False)
assert (await self._route(router, {"metadata": {"user_api_key_team_id": "router-team"}})).model == "restricted-model"
def _access_window_offsets(start_hours: float, end_hours: float, team_ids: list) -> dict:
now_utc = datetime.now(timezone.utc)
return {
"start": (now_utc + timedelta(hours=start_hours)).strftime("%H:%M"),
"end": (now_utc + timedelta(hours=end_hours)).strftime("%H:%M"),
"timezone": "UTC",
"team_ids": team_ids,
}
def _reserved_model_list(windows_for_reserved=None, windows_for_open=None) -> list:
reserved: dict = {
"model_name": "gpt-4o-ptu",
"litellm_params": {"model": "gpt-4o", "mock_response": "reserved"},
"model_info": {"id": "reserved-deployment"},
}
if windows_for_reserved is not None:
reserved["model_info"]["access_windows"] = windows_for_reserved
unreserved: dict = {
"model_name": "gpt-4o-ptu",
"litellm_params": {"model": "gpt-4o", "mock_response": "open"},
"model_info": {"id": "open-deployment"},
}
if windows_for_open is not None:
unreserved["model_info"]["access_windows"] = windows_for_open
return [reserved, unreserved]
def test_access_windows_hide_reserved_deployment_from_other_teams():
router = Router(
model_list=_reserved_model_list(
windows_for_reserved=[_access_window_offsets(-1, 1, ["team-a"])],
),
)
_, deployments = router._common_checks_available_deployment(
model="gpt-4o-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["open-deployment"]
def test_access_windows_raise_when_only_reserved_deployments_remain():
router = Router(model_list=_reserved_model_list(
windows_for_reserved=[_access_window_offsets(-1, 1, ["team-a"])],
windows_for_open=[_access_window_offsets(-1, 1, ["team-a"])],
)[:1])
for request_kwargs in ({"metadata": {"user_api_key_team_id": "team-b"}}, {}):
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._common_checks_available_deployment(model="gpt-4o-ptu", request_kwargs=request_kwargs)
_, deployments = router._common_checks_available_deployment(
model="gpt-4o-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-a"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["reserved-deployment"]
def test_reserved_deployments_drop_strategy_markers_before_filtering():
router = Router(model_list=_reserved_model_list()[:1])
marker = {"model_name": "gpt-4o-ptu", "litellm_params": {"model": "auto_router/semantic"}}
reserved = {
"model_name": "gpt-4o-ptu",
"litellm_params": {"model": "gpt-4o"},
"model_info": {"access_windows": [_access_window_offsets(-1, 1, ["team-a"])]},
}
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._filter_reserved_deployments(
model="gpt-4o-ptu",
healthy_deployments=[marker, reserved],
request_team_id="team-b",
)
def test_access_windows_invalid_timezone_fails_router_construction():
with pytest.raises(ValueError, match=r"gpt-4o-ptu.*access_windows"):
Router(
model_list=[
{
"model_name": "gpt-4o-ptu",
"litellm_params": {"model": "gpt-4o", "mock_response": "x"},
"model_info": {
"access_windows": [
{
"start": "22:00",
"end": "06:00",
"timezone": "Mars/Olympus",
"team_ids": ["team-a"],
}
]
},
}
],
)
def test_access_windows_inactive_window_leaves_deployments_available():
router = Router(
model_list=_reserved_model_list(
windows_for_reserved=[_access_window_offsets(2, 3, ["team-a"])],
),
)
_, deployments = router._common_checks_available_deployment(
model="gpt-4o-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
assert {d["model_info"]["id"] for d in deployments} == {"reserved-deployment", "open-deployment"}
def test_access_windows_apply_when_calling_by_model_id():
router = Router(model_list=_reserved_model_list(
windows_for_reserved=[_access_window_offsets(-1, 1, ["team-a"])],
))
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._common_checks_available_deployment(
model="reserved-deployment",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
_, deployment = router._common_checks_available_deployment(
model="reserved-deployment",
request_kwargs={"metadata": {"user_api_key_team_id": "team-a"}},
)
assert deployment["model_info"]["id"] == "reserved-deployment"
def test_access_windows_apply_when_calling_by_litellm_model_name():
router = Router(
model_list=[
{
"model_name": "gpt-4o-ptu",
"litellm_params": {
"model": "openai/gpt-5.6-bypass-probe",
"mock_response": "reserved",
},
"model_info": {
"id": "reserved-litellm-model",
"access_windows": [_access_window_offsets(-1, 1, ["team-a"])],
},
}
],
)
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._common_checks_available_deployment(
model="openai/gpt-5.6-bypass-probe",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
_, deployments = router._common_checks_available_deployment(
model="openai/gpt-5.6-bypass-probe",
request_kwargs={"metadata": {"user_api_key_team_id": "team-a"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["reserved-litellm-model"]
def test_access_windows_apply_to_specific_deployment_calls():
router = Router(
model_list=[
{
"model_name": "gpt-4o-ptu",
"litellm_params": {
"model": "openai/gpt-5.6-specific-probe",
"mock_response": "reserved",
},
"model_info": {
"id": "reserved-specific",
"access_windows": [_access_window_offsets(-1, 1, ["team-a"])],
},
}
],
)
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._common_checks_available_deployment(
model="openai/gpt-5.6-specific-probe",
specific_deployment=True,
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
_, deployments = router._common_checks_available_deployment(
model="openai/gpt-5.6-specific-probe",
specific_deployment=True,
request_kwargs={"metadata": {"user_api_key_team_id": "team-a"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["reserved-specific"]
def test_access_windows_apply_to_wildcard_early_resolve():
router = Router(
model_list=[
{
"model_name": "openai/*",
"litellm_params": {"model": "openai/*", "mock_response": "reserved"},
"model_info": {
"id": "reserved-wildcard",
"access_windows": [_access_window_offsets(-1, 1, ["team-a"])],
},
}
],
)
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._common_checks_available_deployment(
model="openai/gpt-probe-wildcard",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
_, deployments = router._common_checks_available_deployment(
model="openai/gpt-probe-wildcard",
request_kwargs={"metadata": {"user_api_key_team_id": "team-a"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["reserved-wildcard"]
def test_access_windows_filter_reserved_deployments_method():
router = Router(model_list=_reserved_model_list())
reserved: dict = {
"model_info": {
"id": "reserved-deployment",
"access_windows": [_access_window_offsets(-1, 1, ["team-a"])],
}
}
open_deployment: dict = {"model_info": {"id": "open-deployment"}}
assert [
d["model_info"]["id"]
for d in router._filter_reserved_deployments(
model="gpt-4o-ptu",
healthy_deployments=[reserved, open_deployment],
request_team_id="team-b",
)
] == ["open-deployment"]
with pytest.raises(litellm.BadRequestError, match="reserved for another team"):
router._filter_reserved_deployments(
model="gpt-4o-ptu",
healthy_deployments=[reserved],
request_team_id="team-b",
)
assert [
d["model_info"]["id"]
for d in router._filter_reserved_deployments(
model="gpt-4o-ptu",
healthy_deployments=[reserved, open_deployment],
request_team_id="team-a",
)
] == ["reserved-deployment", "open-deployment"]

45
tests/test_litellm/types/test_router.py
View file

@ -146,3 +146,48 @@ def test_aws_session_tags_round_trip_as_sts_shaped_pairs():
def test_aws_session_tags_reject_shapes_sts_would_refuse(aws_session_tags):
with pytest.raises(ValidationError, match="aws_session_tags"):
LiteLLM_Params(model="bedrock/anthropic.claude-opus-5", aws_session_tags=aws_session_tags)
def test_model_info_parses_access_windows_time_strings():
import datetime
info = ModelInfo(
id="x",
access_windows=[
{
"start": "22:00",
"end": "06:00",
"timezone": "America/New_York",
"team_ids": ["team-nightly"],
}
],
)
window = info.access_windows[0]
assert window.start == datetime.time(22, 0)
assert window.end == datetime.time(6, 0)
assert window.timezone == "America/New_York"
assert window.team_ids == ("team-nightly",)
@pytest.mark.parametrize(
"access_windows",
[
[{"start": "25:00", "end": "06:00", "timezone": "UTC", "team_ids": ["t"]}],
[{"start": "22:00", "end": "06:00", "timezone": "Mars/Olympus", "team_ids": ["t"]}],
[{"start": "22:00", "end": "06:00", "timezone": "UTC", "team_ids": []}],
],
ids=["invalid-time", "unknown-timezone", "empty-team-ids"],
)
def test_model_info_rejects_invalid_access_windows(access_windows):
with pytest.raises(ValidationError):
ModelInfo(id="x", access_windows=access_windows)
def test_model_info_rejects_offset_aware_access_window_times():
with pytest.raises(ValidationError):
ModelInfo(
id="x",
access_windows=[
{"start": "22:00+05:00", "end": "06:00", "timezone": "UTC", "team_ids": ["t"]}
],
)

19
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -33722,6 +33722,23 @@ export interface components {
[key: string]: string | string[];
};
};
/** ModelAccessWindow */
ModelAccessWindow: {
/**
* End
* Format: time
*/
end: string;
/**
* Start
* Format: time
*/
start: string;
/** Team Ids */
team_ids: string[];
/** Timezone */
timezone: string;
};
/** ModelDeprecationInfo */
ModelDeprecationInfo: {
/**
@ -41879,6 +41896,8 @@ export interface components {
};
/** ModelInfo */
litellm__types__router__ModelInfo: {
/** Access Windows */
access_windows?: components["schemas"]["ModelAccessWindow"][] | null;
/** Allow Fail Open */
allow_fail_open?: boolean | null;
/** Base Model */