mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
Merge cc7050d244 into 9071ca503e
This commit is contained in:
commit
46cc179bd0
8 changed files with 1182 additions and 2 deletions
|
|
@ -253,7 +253,10 @@ class InMemoryCache(BaseCache):
|
|||
) -> list[float] | None:
|
||||
results: Final = []
|
||||
for increment in increment_list:
|
||||
result = await self.async_increment(increment["key"], increment["increment_value"], **kwargs)
|
||||
# forward each op's own ttl; allow_ttl_override leaves an already-live ttl untouched
|
||||
result = await self.async_increment(
|
||||
increment["key"], increment["increment_value"], ttl=increment.get("ttl"), **kwargs
|
||||
)
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
|
|
|
|||
235
litellm/proxy/hooks/tag_rate_limits_shared.py
Normal file
235
litellm/proxy/hooks/tag_rate_limits_shared.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
"""Primitives shared by both tag-scoped rate-limit hooks (model_based_tag_rate_limits_hook.py,
|
||||
global_tag_rate_limits_hook.py): identity/scope extraction, policy fingerprinting, bucket-key
|
||||
hashing, and cache-partitioning. Every name here is a genuine public export (no leading
|
||||
underscore); each hook imports what it needs aliased back to its own historical private name."""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
from collections.abc import Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Literal, TypeAlias
|
||||
|
||||
from litellm.exceptions import RateLimitType
|
||||
from litellm.types.caching import RedisPipelineIncrementOperation
|
||||
from litellm.types.router import TagRateLimitEntry, TagRateLimitScope
|
||||
|
||||
LimitUnit: TypeAlias = Literal["tokens", "requests", "dollars", "concurrency"]
|
||||
LIMIT_UNITS: Final[tuple[LimitUnit, ...]] = ("tokens", "requests", "dollars", "concurrency")
|
||||
|
||||
# requests/concurrency admit via atomic check-and-increment; tokens/dollars are only known
|
||||
# after the response, so they stay a read-then-account-on-success check with an unavoidable race.
|
||||
ATOMIC_UNITS: Final[frozenset[LimitUnit]] = frozenset({"requests", "concurrency"})
|
||||
|
||||
UNIT_TO_GROUP_FIELD: Final[Mapping[LimitUnit, str]] = MappingProxyType(
|
||||
{
|
||||
"tokens": "token_limits",
|
||||
"requests": "request_limits",
|
||||
"dollars": "dollar_limits",
|
||||
"concurrency": "concurrency_limits",
|
||||
}
|
||||
)
|
||||
UNIT_TO_RATE_LIMIT_TYPE: Final[Mapping[LimitUnit, RateLimitType]] = MappingProxyType(
|
||||
{
|
||||
"tokens": RateLimitType.TOKENS,
|
||||
"requests": RateLimitType.REQUESTS,
|
||||
"dollars": RateLimitType.BUDGET,
|
||||
"concurrency": RateLimitType.CONCURRENT_REQUESTS,
|
||||
}
|
||||
)
|
||||
|
||||
# shared read-only fallback for an absent/None mapping, so call sites don't build a fresh {}
|
||||
EMPTY_MAPPING: Final[Mapping[str, object]] = MappingProxyType({})
|
||||
|
||||
# holds a strong reference to every fire-and-forget background task (concurrency release,
|
||||
# token/dollar accounting) so the event loop's weak-ref-only tracking can't garbage-collect
|
||||
# one mid-execution; each task's own completion callback discards its entry
|
||||
BACKGROUND_TASKS: Final[set["asyncio.Task[None]"]] = set() # mutable-ok: see comment above
|
||||
|
||||
# floor for a concurrency reservation's self-heal ttl, regardless of period_seconds, so an
|
||||
# expiring-while-in-flight reservation can't silently admit past the limit
|
||||
CONCURRENCY_MIN_SAFETY_TTL_SECONDS: Final = 3600
|
||||
|
||||
# single-key atomic check-and-increment (one key per call: every tag_rl key carries its own
|
||||
# {..} hash tag, so a multi-key call could span shards and cross-slot error). refresh_ttl
|
||||
# (ARGV[4]) distinguishes callers: "requests" is a fixed window whose ttl is set once and never
|
||||
# extended; "concurrency" isn't windowed, so its crash-safety ttl must refresh on every admission.
|
||||
TAG_RL_CHECK_AND_INCR_SCRIPT: Final = """
|
||||
local key = KEYS[1]
|
||||
local limit = tonumber(ARGV[1])
|
||||
local increment = tonumber(ARGV[2])
|
||||
local ttl = tonumber(ARGV[3])
|
||||
local refresh_ttl = tonumber(ARGV[4])
|
||||
local current = tonumber(redis.call('GET', key) or 0)
|
||||
if current + increment > limit then
|
||||
return { 0, current }
|
||||
end
|
||||
local new_value = redis.call('INCRBY', key, increment)
|
||||
if ttl > 0 then
|
||||
if refresh_ttl == 1 then
|
||||
redis.call('EXPIRE', key, ttl)
|
||||
else
|
||||
local current_ttl = redis.call('TTL', key)
|
||||
if current_ttl == -1 then
|
||||
redis.call('EXPIRE', key, ttl)
|
||||
end
|
||||
end
|
||||
end
|
||||
return { 1, new_value }
|
||||
"""
|
||||
|
||||
# atomic decrement floored at 0 (refund or concurrency release); floors via DEL rather than
|
||||
# `SET key 0`, since a release on an already-expired key would otherwise recreate it with no ttl
|
||||
TAG_RL_DECR_FLOOR_ZERO_SCRIPT: Final = """
|
||||
local key = KEYS[1]
|
||||
local delta = tonumber(ARGV[1])
|
||||
local new_value = redis.call('INCRBY', key, delta)
|
||||
if new_value < 0 then
|
||||
redis.call('DEL', key)
|
||||
new_value = 0
|
||||
end
|
||||
return new_value
|
||||
"""
|
||||
|
||||
# a (tag_id, values) pair mirroring TagRateLimitScope, used to fold enabled_for/disabled_for
|
||||
# into a hashable form without depending on TagRateLimitScope's own hashability
|
||||
ScopeSignature: TypeAlias = tuple[str, tuple[str, ...]] | None
|
||||
|
||||
|
||||
def scope_signature(scope: TagRateLimitScope | None) -> ScopeSignature:
|
||||
"""Normalizes a TagRateLimitScope into a hashable tuple for policy fingerprinting/dedup."""
|
||||
return None if scope is None else (scope.tag_id, scope.values)
|
||||
|
||||
|
||||
def extract_identity(tags: Sequence[str], tag_id: str) -> str | None:
|
||||
"""First tag matching `f"{tag_id}:"`, value after the colon. `!`-prefixed tag-routing
|
||||
negation markers are skipped so they're never misread as an identity value."""
|
||||
prefix: Final = f"{tag_id}:"
|
||||
for tag in tags:
|
||||
if tag.startswith("!"):
|
||||
continue
|
||||
if tag.startswith(prefix):
|
||||
return tag[len(prefix) :]
|
||||
return None
|
||||
|
||||
|
||||
def entry_applies(entry: TagRateLimitEntry, tags: Sequence[str], key_alias: str | None, model: str | None) -> bool:
|
||||
"""Applies entry's own scoping fields in order (deny before allow): disabled_for excludes a
|
||||
matching value; enabled_for requires a matching value (absence fails, unlike disabled_for);
|
||||
apply_to_models and apply_to_key_alias are allowlists an absent model/alias never satisfies.
|
||||
An entry with none of these set always applies."""
|
||||
if entry.disabled_for is not None:
|
||||
disabled_gate_value: Final = extract_identity(tags, entry.disabled_for.tag_id)
|
||||
if disabled_gate_value is not None and disabled_gate_value in entry.disabled_for.values:
|
||||
return False
|
||||
if entry.enabled_for is not None:
|
||||
enabled_gate_value: Final = extract_identity(tags, entry.enabled_for.tag_id)
|
||||
if enabled_gate_value is None or enabled_gate_value not in entry.enabled_for.values:
|
||||
return False
|
||||
if entry.apply_to_models is not None and model not in entry.apply_to_models:
|
||||
return False
|
||||
if entry.apply_to_key_alias is None:
|
||||
return True
|
||||
return key_alias in entry.apply_to_key_alias
|
||||
|
||||
|
||||
def resolve_authoritative_metadata_variable_name(
|
||||
metadata_source: Mapping[str, object],
|
||||
) -> Literal["metadata", "litellm_metadata"]:
|
||||
"""Picks metadata vs litellm_metadata by the unforgeable `user_api_key_auth` marker
|
||||
`add_litellm_data_to_request` stamps into whichever bucket is actually authoritative --
|
||||
key presence or truthiness alone can be forged by a caller onto the wrong bucket."""
|
||||
litellm_metadata: Final = metadata_source.get("litellm_metadata")
|
||||
if isinstance(litellm_metadata, Mapping) and "user_api_key_auth" in litellm_metadata:
|
||||
return "litellm_metadata"
|
||||
return "metadata"
|
||||
|
||||
|
||||
def _active_metadata_bucket(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> Mapping[str, object]:
|
||||
"""request_kwargs carries metadata at its own top level at admission time, but only nested
|
||||
under litellm_params by async_log_success_event/async_log_failure_event time; checks both."""
|
||||
top_level: Final = request_kwargs.get(metadata_variable_name)
|
||||
if isinstance(top_level, Mapping):
|
||||
return top_level
|
||||
litellm_params: Final = request_kwargs.get("litellm_params")
|
||||
if isinstance(litellm_params, Mapping):
|
||||
nested: Final = litellm_params.get(metadata_variable_name)
|
||||
if isinstance(nested, Mapping):
|
||||
return nested
|
||||
return EMPTY_MAPPING
|
||||
|
||||
|
||||
def extract_key_hash(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> str | None:
|
||||
"""Reads the calling virtual key's hash from the authoritative metadata bucket (nested under
|
||||
litellm_params by log time, same as _active_metadata_bucket's other callers); `metadata["user_api_key"]`
|
||||
is already the hashed token despite the plain name."""
|
||||
active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
|
||||
key_hash: Final = active.get("user_api_key")
|
||||
return key_hash if isinstance(key_hash, str) else None
|
||||
|
||||
|
||||
def extract_key_alias(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> str | None:
|
||||
"""Reads the calling virtual key's own key_alias from the authoritative metadata bucket."""
|
||||
active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
|
||||
key_alias: Final = active.get("user_api_key_alias")
|
||||
return key_alias if isinstance(key_alias, str) else None
|
||||
|
||||
|
||||
def order_tags_for_identity_resolution(
|
||||
tags: Sequence[str], request_kwargs: Mapping[str, object], metadata_variable_name: str
|
||||
) -> tuple[str, ...]:
|
||||
"""Puts server-computed `metadata.inherited_tags` ahead of caller-supplied tags, so a caller
|
||||
can't submit e.g. `company_id:attacker-chosen` and shadow the calling key's real, same-prefix
|
||||
tag; extract_identity/entry_applies both resolve tag_id via first-match-by-prefix."""
|
||||
active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
|
||||
inherited_tags: Final = active.get("inherited_tags")
|
||||
if not isinstance(inherited_tags, (list, tuple)) or not inherited_tags:
|
||||
return tuple(tags)
|
||||
return tuple(dict.fromkeys((*inherited_tags, *tags)))
|
||||
|
||||
|
||||
def fixed_length_identity(tag_value: str) -> str:
|
||||
"""Hashes a caller-controlled, unbounded-length tag value to a fixed-length digest, bounding
|
||||
a hook's own contribution to an in-memory cache key or Redis key regardless of input size."""
|
||||
return hashlib.sha256(tag_value.encode()).hexdigest()
|
||||
|
||||
|
||||
def policy_fingerprint(entry: TagRateLimitEntry) -> str:
|
||||
"""Fixed-length digest folding every policy-distinguishing field of an entry (limit,
|
||||
period_seconds, scope_by_key_hash, enabled_for/disabled_for, apply_to_key_alias/models) into
|
||||
one bucket-key signature, so two differently-configured entries sharing a name never share
|
||||
a counter."""
|
||||
fingerprint_source: Final = (
|
||||
entry.limit,
|
||||
entry.period_seconds,
|
||||
entry.scope_by_key_hash,
|
||||
scope_signature(entry.enabled_for),
|
||||
scope_signature(entry.disabled_for),
|
||||
entry.apply_to_key_alias,
|
||||
entry.apply_to_models,
|
||||
)
|
||||
return hashlib.sha256(repr(fingerprint_source).encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
def bucket_ttl_seconds(entry: TagRateLimitEntry) -> int:
|
||||
"""Redis (and in-memory fallback) ttl for a non-concurrency bucket key: entry.key_ttl_seconds
|
||||
overrides the default of period_seconds + 3600 when set."""
|
||||
return entry.key_ttl_seconds if entry.key_ttl_seconds is not None else entry.period_seconds + 3600
|
||||
|
||||
|
||||
# None => this entry shares the hook's single default cache partition. Otherwise a value-stable
|
||||
# signature (policy_fingerprint(entry), not the override int alone) so two entries sharing a
|
||||
# max_in_memory_cache_size don't merge into one partition; max_in_memory_cache_size stays the
|
||||
# trailing element since `partition_key[-1]` reads it directly to size the partition's cache.
|
||||
PartitionKey: TypeAlias = tuple[str, str, str, int] | None
|
||||
PartitionOperations: TypeAlias = dict[PartitionKey, list[RedisPipelineIncrementOperation]]
|
||||
|
||||
|
||||
def partition_key(entry: TagRateLimitEntry) -> PartitionKey:
|
||||
if entry.max_in_memory_cache_size is None:
|
||||
return None
|
||||
return (
|
||||
entry.tag_id,
|
||||
entry.name,
|
||||
policy_fingerprint(entry),
|
||||
entry.max_in_memory_cache_size,
|
||||
)
|
||||
|
|
@ -4,6 +4,7 @@ litellm.Router Types - includes RouterConfig, UpdateRouterConfig, ModelInfo etc
|
|||
|
||||
import datetime
|
||||
import enum
|
||||
import math
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints
|
||||
|
|
@ -160,6 +161,126 @@ def _as_utc(value: datetime.datetime | None) -> datetime.datetime | None:
|
|||
return value.astimezone(datetime.timezone.utc)
|
||||
|
||||
|
||||
class TagRateLimitScope(BaseModel):
|
||||
"""A gate on a tag other than the entry's own `tag_id`, used by `TagRateLimitEntry.enabled_for`/`disabled_for`."""
|
||||
|
||||
tag_id: str
|
||||
values: tuple[str, ...]
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_tag_id(self) -> "TagRateLimitScope":
|
||||
if not self.tag_id:
|
||||
raise ValueError("tag_id must be a non-empty string")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_values(self) -> "TagRateLimitScope":
|
||||
if not self.values:
|
||||
raise ValueError("values must be a non-empty list of strings")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _normalize_values(self) -> "TagRateLimitScope":
|
||||
# sorted+deduped so dedup-signature comparisons aren't order-sensitive; __setattr__ works around frozen=True
|
||||
object.__setattr__(self, "values", tuple(sorted(set(self.values)))) # mutable-ok: frozen before escaping
|
||||
return self
|
||||
|
||||
|
||||
class TagRateLimitEntry(BaseModel):
|
||||
name: str
|
||||
tag_id: str = "end_user_id"
|
||||
limit: float
|
||||
period_seconds: int
|
||||
scope_by_key_hash: bool = False
|
||||
# overrides the default bucket ttl (period_seconds + 3600); see _PROXY_ModelBasedTagRateLimitsHook._ttl_for
|
||||
key_ttl_seconds: int | None = None
|
||||
# overrides the shared litellm.model_based_tag_rate_limits_max_in_memory_cache_size (200) with a dedicated partition
|
||||
max_in_memory_cache_size: int | None = None
|
||||
# gate this entry on another tag; disabled_for wins over enabled_for; an absent tag never matches either allowlist
|
||||
enabled_for: TagRateLimitScope | None = None
|
||||
disabled_for: TagRateLimitScope | None = None
|
||||
apply_to_key_alias: tuple[str, ...] | None = None
|
||||
apply_to_models: tuple[str, ...] | None = None
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_tag_id(self) -> "TagRateLimitEntry":
|
||||
if not self.tag_id:
|
||||
raise ValueError("tag_id must be a non-empty string")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_limit(self) -> "TagRateLimitEntry":
|
||||
# NaN compares False against every ordering operator, silently defeating whichever check gates this limit
|
||||
if math.isnan(self.limit):
|
||||
raise ValueError("limit must not be NaN")
|
||||
if math.isinf(self.limit):
|
||||
raise ValueError("limit must be finite")
|
||||
if self.limit <= 0:
|
||||
raise ValueError("limit must be a positive number")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_period_seconds(self) -> "TagRateLimitEntry":
|
||||
if self.period_seconds <= 0:
|
||||
raise ValueError("period_seconds must be a positive integer")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_key_ttl_seconds(self) -> "TagRateLimitEntry":
|
||||
if self.key_ttl_seconds is not None and self.key_ttl_seconds <= 0:
|
||||
raise ValueError("key_ttl_seconds must be a positive integer when set")
|
||||
# a shorter ttl than period_seconds resets the counter early, letting it exceed the limit
|
||||
if self.key_ttl_seconds is not None and self.key_ttl_seconds < self.period_seconds:
|
||||
raise ValueError("key_ttl_seconds must be at least period_seconds when set")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_max_in_memory_cache_size(self) -> "TagRateLimitEntry":
|
||||
if self.max_in_memory_cache_size is not None and self.max_in_memory_cache_size <= 0:
|
||||
raise ValueError("max_in_memory_cache_size must be a positive integer when set")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_apply_to_key_alias(self) -> "TagRateLimitEntry":
|
||||
if self.apply_to_key_alias is not None and not self.apply_to_key_alias:
|
||||
raise ValueError("apply_to_key_alias must be a non-empty list of strings when set")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _normalize_apply_to_key_alias(self) -> "TagRateLimitEntry":
|
||||
# same reason as TagRateLimitScope._normalize_values
|
||||
if self.apply_to_key_alias is not None:
|
||||
self.apply_to_key_alias = tuple(sorted(set(self.apply_to_key_alias))) # mutable-ok: frozen before escaping
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_apply_to_models(self) -> "TagRateLimitEntry":
|
||||
if self.apply_to_models is not None and not self.apply_to_models:
|
||||
raise ValueError("apply_to_models must be a non-empty list of strings when set")
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _normalize_apply_to_models(self) -> "TagRateLimitEntry":
|
||||
if self.apply_to_models is not None:
|
||||
self.apply_to_models = tuple(sorted(set(self.apply_to_models))) # mutable-ok: frozen before escaping
|
||||
return self
|
||||
|
||||
|
||||
class TagRateLimitGroup(BaseModel):
|
||||
limits: tuple[TagRateLimitEntry, ...] = ()
|
||||
|
||||
|
||||
class TagRateLimits(BaseModel):
|
||||
token_limits: TagRateLimitGroup | None = None
|
||||
request_limits: TagRateLimitGroup | None = None
|
||||
dollar_limits: TagRateLimitGroup | None = None
|
||||
concurrency_limits: TagRateLimitGroup | None = None
|
||||
|
||||
|
||||
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.
|
||||
|
|
@ -214,6 +335,8 @@ class ModelInfo(MirroredPricingParams):
|
|||
# in the spend log row's metadata. Set it on every deployment of the group.
|
||||
internal_router_model: bool | None = None
|
||||
|
||||
tag_rate_limits: TagRateLimits | None = None
|
||||
|
||||
def __init__(self, id: str | int | None = None, **params) -> None:
|
||||
if id is None:
|
||||
id = str(uuid.uuid4()) # Generate a UUID if id is None or not provided
|
||||
|
|
|
|||
128
tests/local_testing/test_tag_rate_limits_shared_redis.py
Normal file
128
tests/local_testing/test_tag_rate_limits_shared_redis.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
"""
|
||||
Real-Redis execution coverage for tag_rate_limits_shared's two Lua scripts.
|
||||
|
||||
Lives outside tests/test_litellm/ (which can only contain mocked tests, see
|
||||
tests/test_litellm/readme.md) because fakeredis has no EVALSHA support without
|
||||
the optional lupa dependency; these run against a throwaway local redis-server
|
||||
instead (same idiom as tests/test_litellm/proxy/hooks/test_batch_enqueued_tokens.py's
|
||||
test_redis_lua_path_full_lifecycle).
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.caching.redis_cache import RedisCache
|
||||
from litellm.proxy.hooks.tag_rate_limits_shared import (
|
||||
TAG_RL_CHECK_AND_INCR_SCRIPT,
|
||||
TAG_RL_DECR_FLOOR_ZERO_SCRIPT,
|
||||
)
|
||||
|
||||
|
||||
def _free_port() -> int:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
return sock.getsockname()[1]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def redis_port() -> Iterator[int]:
|
||||
if shutil.which("redis-server") is None:
|
||||
pytest.skip("requires a local redis-server binary to exercise the Lua script path")
|
||||
port = _free_port()
|
||||
proc = subprocess.Popen(
|
||||
["redis-server", "--port", str(port), "--save", "", "--appendonly", "no"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
try:
|
||||
deadline = time.monotonic() + 5
|
||||
while time.monotonic() < deadline:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(0.2)
|
||||
if sock.connect_ex(("127.0.0.1", port)) == 0:
|
||||
break
|
||||
else:
|
||||
proc.terminate()
|
||||
pytest.skip("local redis-server did not become ready in time")
|
||||
yield port
|
||||
finally:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
|
||||
|
||||
def _register(port: int, script: str):
|
||||
return RedisCache(host="127.0.0.1", port=port).async_register_script(script)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_and_incr_admits_under_limit_and_sets_ttl(redis_port):
|
||||
run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
admitted, new_value = await run(keys=["bucket1"], args=[5, 1, 60, 0])
|
||||
assert (admitted, new_value) == (1, 1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_and_incr_rejects_over_limit_without_incrementing(redis_port):
|
||||
run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
await run(keys=["bucket2"], args=[1, 1, 60, 0])
|
||||
rejected, current = await run(keys=["bucket2"], args=[1, 1, 60, 0])
|
||||
assert (rejected, current) == (0, 1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_and_incr_requests_ttl_is_set_once_and_not_refreshed(redis_port):
|
||||
"""refresh_ttl=0 (the `requests` fixed-window semantics): the epoch-bucketed
|
||||
TTL must be set on the first write and left alone after, or the bucket
|
||||
outlives the epoch it's meant to reset at."""
|
||||
run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
await run(keys=["bucket3"], args=[5, 1, 100, 0])
|
||||
raw = RedisCache(host="127.0.0.1", port=redis_port)
|
||||
client = raw.init_async_client()
|
||||
first_ttl = await client.ttl("bucket3")
|
||||
await run(keys=["bucket3"], args=[5, 1, 5, 0])
|
||||
second_ttl = await client.ttl("bucket3")
|
||||
assert first_ttl > 5
|
||||
assert second_ttl > 5 # unchanged by the second call's much shorter ttl arg
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_and_incr_concurrency_ttl_refreshes_on_every_admission(redis_port):
|
||||
"""refresh_ttl=1 (the `concurrency` reservation semantics): every admission
|
||||
must push the crash-safety TTL back out, or a long-lived burst of traffic
|
||||
expires the whole counter mid-flight."""
|
||||
run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
await run(keys=["bucket4"], args=[5, 1, 5, 1])
|
||||
await run(keys=["bucket4"], args=[5, 1, 100, 1])
|
||||
raw = RedisCache(host="127.0.0.1", port=redis_port)
|
||||
client = raw.init_async_client()
|
||||
ttl = await client.ttl("bucket4")
|
||||
assert ttl > 5
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_decr_floor_zero_floors_at_zero_and_deletes_the_key(redis_port):
|
||||
run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
|
||||
await run_incr(keys=["bucket5"], args=[5, 1, 60, 1])
|
||||
|
||||
floored = await run_decr(keys=["bucket5"], args=[-5])
|
||||
assert floored == 0
|
||||
|
||||
raw = RedisCache(host="127.0.0.1", port=redis_port)
|
||||
client = raw.init_async_client()
|
||||
assert await client.get("bucket5") is None # floored via DEL, not a TTL-less `SET 0`
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_decr_floor_zero_decrements_normally_when_result_stays_non_negative(redis_port):
|
||||
run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
|
||||
run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
|
||||
await run_incr(keys=["bucket6"], args=[5, 3, 60, 1])
|
||||
|
||||
remaining = await run_decr(keys=["bucket6"], args=[-1])
|
||||
assert remaining == 2
|
||||
|
|
@ -45,6 +45,39 @@ async def test_async_increment_delegates_to_locked_sync_path():
|
|||
assert cache.get_cache("counter") == 5
|
||||
|
||||
|
||||
async def test_async_increment_pipeline_applies_each_operations_own_ttl():
|
||||
"""
|
||||
Bugbot finding: async_increment_pipeline dropped each operation's own
|
||||
"ttl" field, so a counter created through it always got the cache's
|
||||
600-second default_ttl regardless of what the caller actually configured
|
||||
(e.g. an hourly or daily rate-limit window) -- the counter (and whatever
|
||||
it was tracking against a limit) silently reset mid-window.
|
||||
"""
|
||||
cache = InMemoryCache(default_ttl=600)
|
||||
await cache.async_increment_pipeline([{"key": "long-window-counter", "increment_value": 1, "ttl": 7200}])
|
||||
|
||||
ttl_remaining = await cache.async_get_ttl("long-window-counter")
|
||||
assert ttl_remaining is not None
|
||||
assert ttl_remaining > time.time() + 600
|
||||
|
||||
|
||||
async def test_async_increment_pipeline_preserves_an_existing_live_ttl_on_later_increments():
|
||||
"""
|
||||
A second increment on the same still-live counter must not reset its
|
||||
remaining ttl back up to the full window -- only the first increment
|
||||
(the one that actually creates the counter) should set it.
|
||||
"""
|
||||
cache = InMemoryCache(default_ttl=600)
|
||||
await cache.async_increment_pipeline([{"key": "counter", "increment_value": 1, "ttl": 7200}])
|
||||
ttl_after_first = cache.ttl_dict["counter"]
|
||||
|
||||
await cache.async_increment_pipeline([{"key": "counter", "increment_value": 1, "ttl": 7200}])
|
||||
ttl_after_second = cache.ttl_dict["counter"]
|
||||
|
||||
assert ttl_after_second == ttl_after_first
|
||||
assert cache.get_cache("counter") == 2
|
||||
|
||||
|
||||
def test_in_memory_openai_obj_cache():
|
||||
from openai import OpenAI
|
||||
|
||||
|
|
|
|||
446
tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
Normal file
446
tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
Normal file
|
|
@ -0,0 +1,446 @@
|
|||
"""
|
||||
Unit tests for the primitives shared by both tag-scoped rate-limit hooks
|
||||
(`model_based_tag_rate_limits_hook.py` and `global_tag_rate_limits_hook.py`).
|
||||
|
||||
These test the hook-independent logic in isolation: identity extraction,
|
||||
`entry_applies` scoping, and the partition/bucket-TTL key helpers. Each
|
||||
hook's own test file covers everything specific to how it wires these
|
||||
primitives into its own admission/accounting engine.
|
||||
"""
|
||||
|
||||
from litellm.proxy.hooks.tag_rate_limits_shared import (
|
||||
bucket_ttl_seconds,
|
||||
entry_applies,
|
||||
extract_identity,
|
||||
extract_key_alias,
|
||||
extract_key_hash,
|
||||
fixed_length_identity,
|
||||
order_tags_for_identity_resolution,
|
||||
partition_key,
|
||||
resolve_authoritative_metadata_variable_name,
|
||||
)
|
||||
from litellm.types.router import TagRateLimitEntry, TagRateLimitScope
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extract_identity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_extract_identity_matches_prefixed_tag():
|
||||
assert extract_identity(["team_id:t1", "end_user_id:u1"], "end_user_id") == "u1"
|
||||
|
||||
|
||||
def test_extract_identity_returns_none_when_absent():
|
||||
assert extract_identity(["team_id:t1"], "end_user_id") is None
|
||||
|
||||
|
||||
def test_extract_identity_skips_negation_tags():
|
||||
"""A `!end_user_id:u1` routing-negation marker must never be read as identity."""
|
||||
assert extract_identity(["!end_user_id:u1"], "end_user_id") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# order_tags_for_identity_resolution -- veria-ai finding on PR #38292: a
|
||||
# caller-supplied tag must not shadow a policy-backed (key/team/project)
|
||||
# tag sharing the same tag_id prefix
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_order_tags_for_identity_resolution_prefers_inherited_tag_over_caller_supplied():
|
||||
request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
|
||||
tags = ["company_id:attacker-chosen", "end_user_id:u1"]
|
||||
ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
|
||||
assert extract_identity(ordered, "company_id") == "real-company"
|
||||
|
||||
|
||||
def test_order_tags_for_identity_resolution_falls_back_to_caller_tags_when_nothing_inherited():
|
||||
request_kwargs = {"metadata": {}}
|
||||
tags = ["end_user_id:u1"]
|
||||
ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
|
||||
assert extract_identity(ordered, "end_user_id") == "u1"
|
||||
|
||||
|
||||
def test_order_tags_for_identity_resolution_keeps_caller_only_tags_not_shadowed_by_a_different_tag_id():
|
||||
request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
|
||||
tags = ["end_user_id:u1"]
|
||||
ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
|
||||
assert extract_identity(ordered, "end_user_id") == "u1"
|
||||
assert extract_identity(ordered, "company_id") == "real-company"
|
||||
|
||||
|
||||
def test_order_tags_for_identity_resolution_deduplicates_identical_tag_present_in_both_sources():
|
||||
request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
|
||||
tags = ["company_id:real-company"]
|
||||
ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
|
||||
assert ordered.count("company_id:real-company") == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fixed_length_identity -- tag_value is caller-controlled with no length
|
||||
# bound; a hook's own contribution to a cache key must not grow with it
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fixed_length_identity_bounds_key_contribution_regardless_of_input_size():
|
||||
"""
|
||||
A caller can submit an arbitrarily long tag value (no length or content
|
||||
bound is enforced upstream of either hook). Without hashing, that value
|
||||
would go straight into an in-memory dict key (bypassing
|
||||
max_in_memory_cache_size, which caps item *count* not key bytes) and an
|
||||
unbounded-length Redis key (Redis has no key-count or key-size cap at
|
||||
all here). A fixed-length digest bounds a hook's own contribution to
|
||||
the key regardless of input size.
|
||||
"""
|
||||
huge_value = "x" * 5_000_000
|
||||
digest = fixed_length_identity(huge_value)
|
||||
assert len(digest) == 64 # sha256 hex digest length, independent of input size
|
||||
|
||||
|
||||
def test_fixed_length_identity_preserves_distinctness():
|
||||
"""Hashing must not collapse two different tag values onto one bucket."""
|
||||
assert fixed_length_identity("user-a") != fixed_length_identity("user-b")
|
||||
assert fixed_length_identity("user-a") == fixed_length_identity("user-a")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extract_key_hash -- must read only the one field the server actually
|
||||
# authenticates into, never fall back to the other
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_extract_key_hash_ignores_a_forged_value_in_the_non_authoritative_field():
|
||||
"""
|
||||
On a route where litellm_metadata is authoritative, the server writes
|
||||
the real hash there and never touches metadata -- so a caller-supplied
|
||||
metadata.user_api_key must not be read at all, let alone win.
|
||||
"""
|
||||
request_kwargs = {
|
||||
"metadata": {"user_api_key": "forged-by-caller"},
|
||||
"litellm_metadata": {"user_api_key": "real-authenticated-hash"},
|
||||
}
|
||||
assert extract_key_hash(request_kwargs, "litellm_metadata") == "real-authenticated-hash"
|
||||
|
||||
|
||||
def test_extract_key_hash_reads_metadata_when_it_is_the_authoritative_field():
|
||||
request_kwargs = {"metadata": {"user_api_key": "real-hash"}}
|
||||
assert extract_key_hash(request_kwargs, "metadata") == "real-hash"
|
||||
|
||||
|
||||
def test_extract_key_hash_ignores_a_non_mapping_authoritative_field():
|
||||
"""Bugbot finding: metadata can arrive as an unparsed JSON string (see
|
||||
apply_client_tag_policy_pre_auth's own docstring on multipart/extra_body
|
||||
routes); a truthy non-Mapping must not reach .get() and crash."""
|
||||
request_kwargs = {"metadata": '{"user_api_key": "forged"}'}
|
||||
assert extract_key_hash(request_kwargs, "metadata") is None
|
||||
|
||||
|
||||
def test_extract_key_alias_ignores_a_non_mapping_authoritative_field():
|
||||
request_kwargs = {"metadata": '{"user_api_key_alias": "forged"}'}
|
||||
assert extract_key_alias(request_kwargs, "metadata") is None
|
||||
|
||||
|
||||
def test_extract_key_hash_finds_metadata_nested_under_litellm_params_at_log_time():
|
||||
"""Bugbot finding: by async_log_success_event/async_log_failure_event time,
|
||||
kwargs only carries metadata nested under litellm_params (see
|
||||
_active_metadata_bucket's own docstring), not at request_kwargs' own top
|
||||
level -- extract_key_hash must find it there too, or a key-hash-scoped
|
||||
bucket's accounting silently reads a different key than admission did."""
|
||||
request_kwargs = {"litellm_params": {"metadata": {"user_api_key": "real-hash"}}}
|
||||
assert extract_key_hash(request_kwargs, "metadata") == "real-hash"
|
||||
|
||||
|
||||
def test_extract_key_alias_finds_metadata_nested_under_litellm_params_at_log_time():
|
||||
request_kwargs = {"litellm_params": {"metadata": {"user_api_key_alias": "real-alias"}}}
|
||||
assert extract_key_alias(request_kwargs, "metadata") == "real-alias"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# resolve_authoritative_metadata_variable_name -- Veria AI finding: a
|
||||
# caller-supplied, non-empty litellm_metadata must not be selected over
|
||||
# metadata, the field the proxy actually wrote authenticated tags into
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_resolve_authoritative_metadata_variable_name_ignores_caller_forged_non_empty_litellm_metadata():
|
||||
"""A caller-supplied litellm_metadata with unrelated content (no
|
||||
user_api_key_auth marker) must not be picked over metadata, even though
|
||||
it is a non-empty dict."""
|
||||
metadata_source = {"litellm_metadata": {"marker": True}}
|
||||
assert resolve_authoritative_metadata_variable_name(metadata_source) == "metadata"
|
||||
|
||||
|
||||
def test_resolve_authoritative_metadata_variable_name_selects_litellm_metadata_when_server_written():
|
||||
"""A genuinely server-populated litellm_metadata (LITELLM_METADATA_ROUTES)
|
||||
always carries the user_api_key_auth marker stamped by
|
||||
add_user_api_key_auth_to_request_metadata."""
|
||||
metadata_source = {"litellm_metadata": {"user_api_key_auth": object(), "tags": ["team_id:t1"]}}
|
||||
assert resolve_authoritative_metadata_variable_name(metadata_source) == "litellm_metadata"
|
||||
|
||||
|
||||
def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_absent():
|
||||
assert resolve_authoritative_metadata_variable_name({}) == "metadata"
|
||||
|
||||
|
||||
def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_none():
|
||||
assert resolve_authoritative_metadata_variable_name({"litellm_metadata": None}) == "metadata"
|
||||
|
||||
|
||||
def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_empty():
|
||||
assert resolve_authoritative_metadata_variable_name({"litellm_metadata": {}}) == "metadata"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# entry_applies -- enabled_for / disabled_for / apply_to_key_alias / apply_to_models
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_entry_applies_with_none_of_the_scoping_fields_set():
|
||||
entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_disabled_for_on_its_own_tag_id_excludes_a_listed_value():
|
||||
"""disabled_for's `tag_id` can be set to the entry's own tag_id, gating on
|
||||
a subset of its own resolved identity rather than a second tag."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
disabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u1",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
|
||||
assert entry_applies(entry, ["end_user_id:u2"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_enabled_for_on_its_own_tag_id_restricts_to_a_listed_value():
|
||||
"""enabled_for's `tag_id` can likewise be set to the entry's own tag_id,
|
||||
admitting only a hand-picked subset of its own resolved identity."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
enabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u2", "u3")),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
|
||||
assert entry_applies(entry, ["end_user_id:u2"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_matches_an_enabled_for_gate():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_skips_when_enabled_for_gate_tag_is_absent():
|
||||
"""
|
||||
enabled_for is an allowlist gate: absence of the gate tag must not
|
||||
satisfy it, unlike disabled_for below.
|
||||
"""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
|
||||
|
||||
|
||||
def test_entry_applies_skips_when_disabled_for_gate_matches():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is False
|
||||
|
||||
|
||||
def test_entry_applies_when_disabled_for_gate_tag_is_absent():
|
||||
"""disabled_for is a denylist gate: absence of the gate tag has nothing
|
||||
to match against, so the entry still applies."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_disabled_for_overrides_a_matching_enabled_for_gate():
|
||||
"""Deny (disabled_for) takes effect independently of whether the
|
||||
enabled_for gate itself matched, even when both target the same tag."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
disabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u1",)),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is False
|
||||
|
||||
|
||||
def test_entry_applies_with_apply_to_key_alias_unset_applies_to_every_key():
|
||||
entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "any-key-alias", None) is True
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_admits_a_key_alias_on_the_allowlist():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "team-a-key", None) is True
|
||||
|
||||
|
||||
def test_entry_applies_rejects_a_key_alias_missing_from_the_allowlist():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "team-b-key", None) is False
|
||||
|
||||
|
||||
def test_entry_applies_rejects_when_key_has_no_alias_but_allowlist_is_set():
|
||||
"""apply_to_key_alias is an allowlist gate: a key with no alias at all
|
||||
never satisfies it, same as enabled_for's absent-gate-tag semantics."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
|
||||
|
||||
|
||||
def test_entry_applies_with_apply_to_models_unset_applies_to_every_model():
|
||||
entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, "opus-chain") is True
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
|
||||
|
||||
|
||||
def test_entry_applies_admits_a_model_on_the_apply_to_models_allowlist():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, "opus-chain") is True
|
||||
|
||||
|
||||
def test_entry_applies_rejects_a_model_missing_from_the_apply_to_models_allowlist():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, "sonnet-chain") is False
|
||||
|
||||
|
||||
def test_entry_applies_rejects_when_model_is_absent_but_apply_to_models_is_set():
|
||||
"""apply_to_models is an allowlist gate: a request with no model at all
|
||||
never satisfies it, same as apply_to_key_alias's absent-key semantics."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
|
||||
|
||||
|
||||
def test_entry_applies_apply_to_models_composes_with_apply_to_key_alias():
|
||||
"""Both gates must pass: a request against the listed model but a
|
||||
non-listed key alias must not apply, even though apply_to_models alone
|
||||
would have admitted it."""
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
apply_to_models=("opus-chain",),
|
||||
apply_to_key_alias=("premium-key",),
|
||||
)
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "premium-key", "opus-chain") is True
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "other-key", "opus-chain") is False
|
||||
assert entry_applies(entry, ["end_user_id:u1"], "premium-key", "sonnet-chain") is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# partition_key -- entries that share max_in_memory_cache_size but disagree
|
||||
# on any policy-fingerprinted field must never share a cache partition
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_partition_key_distinguishes_entries_that_differ_only_by_scope_by_key_hash():
|
||||
"""
|
||||
scope_by_key_hash is part of the partition-key signature: two entries
|
||||
identical in every other field but differing only on this flag are
|
||||
different rate limits (different bucket keys per each hook's own
|
||||
`_hash_tag`) and must never be routed to the same cache partition.
|
||||
"""
|
||||
unscoped = TagRateLimitEntry(
|
||||
name="per_minute", tag_id="end_user_id", limit=5, period_seconds=60, max_in_memory_cache_size=100
|
||||
)
|
||||
scoped = TagRateLimitEntry(
|
||||
name="per_minute",
|
||||
tag_id="end_user_id",
|
||||
limit=5,
|
||||
period_seconds=60,
|
||||
scope_by_key_hash=True,
|
||||
max_in_memory_cache_size=100,
|
||||
)
|
||||
assert partition_key(unscoped) != partition_key(scoped)
|
||||
|
||||
|
||||
def test_partition_key_distinguishes_entries_that_differ_only_by_scoping_fields():
|
||||
"""
|
||||
A plain, unscoped entry and a scoped override can legitimately share
|
||||
name/tag_id/limit/period_seconds/scope_by_key_hash while disagreeing on
|
||||
enabled_for/disabled_for/apply_to_key_alias/apply_to_models --
|
||||
policy_fingerprint already treats that as two distinct policies, so a
|
||||
shared max_in_memory_cache_size must not route them onto the same
|
||||
in-memory partition either, or one entry's high-cardinality traffic can
|
||||
evict the other's active counters from a cache neither entry asked to
|
||||
share.
|
||||
"""
|
||||
base_kwargs = {
|
||||
"name": "daily",
|
||||
"tag_id": "end_user_id",
|
||||
"limit": 100,
|
||||
"period_seconds": 86400,
|
||||
"max_in_memory_cache_size": 50,
|
||||
}
|
||||
unscoped = TagRateLimitEntry(**base_kwargs)
|
||||
enabled_for_scoped = TagRateLimitEntry(
|
||||
**base_kwargs, enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",))
|
||||
)
|
||||
disabled_for_scoped = TagRateLimitEntry(
|
||||
**base_kwargs, disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",))
|
||||
)
|
||||
alias_scoped = TagRateLimitEntry(**base_kwargs, apply_to_key_alias=("premium-key",))
|
||||
models_scoped = TagRateLimitEntry(**base_kwargs, apply_to_models=("opus-chain",))
|
||||
|
||||
keys = {
|
||||
partition_key(unscoped),
|
||||
partition_key(enabled_for_scoped),
|
||||
partition_key(disabled_for_scoped),
|
||||
partition_key(alias_scoped),
|
||||
partition_key(models_scoped),
|
||||
}
|
||||
assert len(keys) == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# bucket_ttl_seconds -- per-tag Redis/bucket key TTL override
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_bucket_ttl_seconds_defaults_to_period_plus_one_hour_when_unset():
|
||||
entry = TagRateLimitEntry(name="per_minute", tag_id="end_user_id", limit=1, period_seconds=60)
|
||||
assert bucket_ttl_seconds(entry) == 60 + 3600
|
||||
|
||||
|
||||
def test_bucket_ttl_seconds_honors_key_ttl_seconds_override():
|
||||
entry = TagRateLimitEntry(name="per_minute", tag_id="end_user_id", limit=1, period_seconds=60, key_ttl_seconds=120)
|
||||
assert bucket_ttl_seconds(entry) == 120
|
||||
|
|
@ -9,6 +9,10 @@ from litellm.types.router import (
|
|||
GenericLiteLLMParams,
|
||||
LiteLLM_Params,
|
||||
ModelInfo,
|
||||
TagRateLimitEntry,
|
||||
TagRateLimitGroup,
|
||||
TagRateLimits,
|
||||
TagRateLimitScope,
|
||||
)
|
||||
from litellm.types.utils import CustomPricingLiteLLMParams, MirroredPricingParams
|
||||
|
||||
|
|
@ -91,7 +95,7 @@ def test_pricing_strings_are_coerced_to_float():
|
|||
|
||||
|
||||
def test_invalid_pricing_is_rejected():
|
||||
with pytest.raises(ValueError, match='validation error for ModelInfo'):
|
||||
with pytest.raises(ValueError, match="validation error for ModelInfo"):
|
||||
ModelInfo(id="x", input_cost_per_token="free")
|
||||
|
||||
|
||||
|
|
@ -146,3 +150,156 @@ 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)
|
||||
|
||||
|
||||
# litellm/types/router.py is imported by plain SDK users, not just the proxy, so a
|
||||
# TagRateLimitEntry validation error should read as a generic config-validation
|
||||
# message and not describe the proxy rate-limit hook's internal admission mechanics.
|
||||
_INTERNAL_ENFORCEMENT_JARGON = (
|
||||
"admission",
|
||||
"tagged request",
|
||||
"tagged traffic",
|
||||
"check-and-increment",
|
||||
"read-only",
|
||||
"atomic",
|
||||
"window rolls over",
|
||||
)
|
||||
|
||||
|
||||
def _assert_message_has_no_internal_jargon(excinfo: pytest.ExceptionInfo) -> None:
|
||||
message = str(excinfo.value).lower()
|
||||
leaked = [term for term in _INTERNAL_ENFORCEMENT_JARGON if term in message]
|
||||
assert not leaked, f"validation message leaked internal enforcement jargon: {leaked}"
|
||||
|
||||
|
||||
def test_limit_infinite_rejected_without_internal_enforcement_jargon():
|
||||
with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
|
||||
TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=float("inf"), period_seconds=60)
|
||||
_assert_message_has_no_internal_jargon(excinfo)
|
||||
|
||||
|
||||
def test_limit_non_positive_rejected_without_internal_enforcement_jargon():
|
||||
with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
|
||||
TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=0, period_seconds=60)
|
||||
_assert_message_has_no_internal_jargon(excinfo)
|
||||
|
||||
|
||||
def test_key_ttl_seconds_shorter_than_period_rejected_without_internal_enforcement_jargon():
|
||||
with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
|
||||
TagRateLimitEntry(
|
||||
name="daily",
|
||||
tag_id="end_user_id",
|
||||
limit=500,
|
||||
period_seconds=86400,
|
||||
key_ttl_seconds=60,
|
||||
)
|
||||
_assert_message_has_no_internal_jargon(excinfo)
|
||||
|
||||
|
||||
def test_limit_nan_rejected():
|
||||
"""NaN compares False against every ordering operator, so it would otherwise slip
|
||||
past a `limit <= 0` style guard and silently defeat whichever check gates it."""
|
||||
with pytest.raises(ValueError, match="limit must not be NaN"):
|
||||
TagRateLimitEntry(name="daily", limit=float("nan"), period_seconds=60)
|
||||
|
||||
|
||||
def test_period_seconds_non_positive_rejected():
|
||||
with pytest.raises(ValueError, match="period_seconds must be a positive integer"):
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=0)
|
||||
|
||||
|
||||
def test_key_ttl_seconds_non_positive_rejected():
|
||||
with pytest.raises(ValueError, match="key_ttl_seconds must be a positive integer when set"):
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=60, key_ttl_seconds=0)
|
||||
|
||||
|
||||
def test_max_in_memory_cache_size_non_positive_rejected():
|
||||
with pytest.raises(ValueError, match="max_in_memory_cache_size must be a positive integer"):
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=60, max_in_memory_cache_size=0)
|
||||
|
||||
|
||||
def test_apply_to_key_alias_empty_list_rejected():
|
||||
with pytest.raises(ValueError, match="apply_to_key_alias must be a non-empty list"):
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_key_alias=())
|
||||
|
||||
|
||||
def test_apply_to_key_alias_sorted_and_deduped():
|
||||
entry = TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_key_alias=("b", "a", "a"))
|
||||
assert entry.apply_to_key_alias == ("a", "b")
|
||||
|
||||
|
||||
def test_apply_to_models_empty_list_rejected():
|
||||
with pytest.raises(ValueError, match="apply_to_models must be a non-empty list"):
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_models=())
|
||||
|
||||
|
||||
def test_apply_to_models_sorted_and_deduped():
|
||||
entry = TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_models=("gpt-4o", "claude", "claude"))
|
||||
assert entry.apply_to_models == ("claude", "gpt-4o")
|
||||
|
||||
|
||||
def test_scope_tag_id_empty_string_rejected():
|
||||
"""An empty tag_id makes identity lookup search for a bare `:` prefix, silently
|
||||
never matching instead of erroring at config load time (Greptile P1 on #39902)."""
|
||||
with pytest.raises(ValueError, match="tag_id must be a non-empty string"):
|
||||
TagRateLimitScope(tag_id="", values=("1032",))
|
||||
|
||||
|
||||
def test_scope_values_empty_list_rejected():
|
||||
with pytest.raises(ValueError, match="values must be a non-empty list"):
|
||||
TagRateLimitScope(tag_id="company_id", values=())
|
||||
|
||||
|
||||
def test_scope_values_sorted_and_deduped():
|
||||
scope = TagRateLimitScope(tag_id="company_id", values=("1032", "1001", "1001"))
|
||||
assert scope.values == ("1001", "1032")
|
||||
|
||||
|
||||
def test_scope_is_frozen():
|
||||
scope = TagRateLimitScope(tag_id="company_id", values=("1032",))
|
||||
with pytest.raises(ValidationError):
|
||||
scope.tag_id = "other_tag"
|
||||
|
||||
|
||||
def test_entry_tag_id_empty_string_rejected():
|
||||
"""Same silent-no-match failure mode as TagRateLimitScope.tag_id (Greptile P1 on
|
||||
#39902); the default is non-empty, but an explicit override could still be empty."""
|
||||
with pytest.raises(ValueError, match="tag_id must be a non-empty string"):
|
||||
TagRateLimitEntry(name="daily", tag_id="", limit=10, period_seconds=60)
|
||||
|
||||
|
||||
def test_entry_accepts_enabled_for_and_disabled_for_scopes():
|
||||
entry = TagRateLimitEntry(
|
||||
name="daily",
|
||||
limit=10,
|
||||
period_seconds=60,
|
||||
enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
|
||||
disabled_for=TagRateLimitScope(tag_id="company_id", values=("9999",)),
|
||||
)
|
||||
assert entry.enabled_for.values == ("1032",)
|
||||
assert entry.disabled_for.values == ("9999",)
|
||||
|
||||
|
||||
def test_group_defaults_to_no_limits():
|
||||
assert TagRateLimitGroup().limits == ()
|
||||
|
||||
|
||||
def test_rate_limits_group_holds_multiple_entries():
|
||||
entries = (
|
||||
TagRateLimitEntry(name="daily", limit=10, period_seconds=60),
|
||||
TagRateLimitEntry(name="weekly", limit=100, period_seconds=604800),
|
||||
)
|
||||
group = TagRateLimitGroup(limits=entries)
|
||||
assert group.limits == entries
|
||||
|
||||
|
||||
def test_model_info_tag_rate_limits_defaults_to_none():
|
||||
assert ModelInfo(id="x").tag_rate_limits is None
|
||||
|
||||
|
||||
def test_model_info_accepts_tag_rate_limits():
|
||||
limits = TagRateLimits(
|
||||
token_limits=TagRateLimitGroup(limits=(TagRateLimitEntry(name="daily", limit=10, period_seconds=60),))
|
||||
)
|
||||
info = ModelInfo(id="x", tag_rate_limits=limits)
|
||||
assert info.tag_rate_limits.token_limits.limits[0].name == "daily"
|
||||
|
|
|
|||
55
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
55
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -36665,6 +36665,60 @@ export interface components {
|
|||
/** Tpm Limit */
|
||||
tpm_limit?: number | null;
|
||||
};
|
||||
/** TagRateLimitEntry */
|
||||
TagRateLimitEntry: {
|
||||
/** Apply To Key Alias */
|
||||
apply_to_key_alias?: string[] | null;
|
||||
/** Apply To Models */
|
||||
apply_to_models?: string[] | null;
|
||||
disabled_for?: components["schemas"]["TagRateLimitScope"] | null;
|
||||
enabled_for?: components["schemas"]["TagRateLimitScope"] | null;
|
||||
/** Key Ttl Seconds */
|
||||
key_ttl_seconds?: number | null;
|
||||
/** Limit */
|
||||
limit: number;
|
||||
/** Max In Memory Cache Size */
|
||||
max_in_memory_cache_size?: number | null;
|
||||
/** Name */
|
||||
name: string;
|
||||
/** Period Seconds */
|
||||
period_seconds: number;
|
||||
/**
|
||||
* Scope By Key Hash
|
||||
* @default false
|
||||
*/
|
||||
scope_by_key_hash: boolean;
|
||||
/**
|
||||
* Tag Id
|
||||
* @default end_user_id
|
||||
*/
|
||||
tag_id: string;
|
||||
};
|
||||
/** TagRateLimitGroup */
|
||||
TagRateLimitGroup: {
|
||||
/**
|
||||
* Limits
|
||||
* @default []
|
||||
*/
|
||||
limits: components["schemas"]["TagRateLimitEntry"][];
|
||||
};
|
||||
/**
|
||||
* TagRateLimitScope
|
||||
* @description A gate on a tag other than the entry's own `tag_id`, used by `TagRateLimitEntry.enabled_for`/`disabled_for`.
|
||||
*/
|
||||
TagRateLimitScope: {
|
||||
/** Tag Id */
|
||||
tag_id: string;
|
||||
/** Values */
|
||||
values: string[];
|
||||
};
|
||||
/** TagRateLimits */
|
||||
TagRateLimits: {
|
||||
concurrency_limits?: components["schemas"]["TagRateLimitGroup"] | null;
|
||||
dollar_limits?: components["schemas"]["TagRateLimitGroup"] | null;
|
||||
request_limits?: components["schemas"]["TagRateLimitGroup"] | null;
|
||||
token_limits?: components["schemas"]["TagRateLimitGroup"] | null;
|
||||
};
|
||||
/**
|
||||
* TagSummaryMetrics
|
||||
* @description Summary metrics for a tag
|
||||
|
|
@ -39656,6 +39710,7 @@ export interface components {
|
|||
ptu_effective_from?: string | null;
|
||||
/** Ptu Effective To */
|
||||
ptu_effective_to?: string | null;
|
||||
tag_rate_limits?: components["schemas"]["TagRateLimits"] | null;
|
||||
/** Team Id */
|
||||
team_id?: string | null;
|
||||
/** Team Public Model Name */
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue