diff --git a/litellm/utils.py b/litellm/utils.py index 5ec089a106a..2547fbd1d1b 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -8963,8 +8963,8 @@ from litellm.caching.in_memory_cache import InMemoryCache class AvailableModelsCache(InMemoryCache): - def __init__(self, ttl_seconds: int = 300, max_size: int = 1000): - super().__init__(ttl_seconds, max_size) + def __init__(self, ttl_seconds: int = 300, max_size: int = 1000, clock: Callable[[], float] | None = None): + super().__init__(max_size_in_memory=max_size, default_ttl=ttl_seconds, clock=clock) self._env_hash: str | None = None def _get_env_hash(self) -> str: diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 5eaabdd2c59..86f47aecf30 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -11200,6 +11200,33 @@ def test_get_valid_models_from_provider_cache_invalidation(monkeypatch): assert _model_cache.get_cached_model_info("openai") is None +def test_available_models_cache_expires_entries_after_ttl_seconds(): + from litellm.utils import AvailableModelsCache + + now: Final = [1_000.0] + cache: Final = AvailableModelsCache(ttl_seconds=300, max_size=1000, clock=lambda: now[0]) + cache.get_cached_model_info("openai") # first call records the env hash + cache.set_cached_model_info("openai", litellm_params=None, available_models=["gpt-5-mini"]) + + now[0] = 1_299.0 + assert cache.get_cached_model_info("openai") == ["gpt-5-mini"] + + now[0] = 1_301.0 + assert cache.get_cached_model_info("openai") is None + + +def test_available_models_cache_keeps_max_size_entries(): + from litellm.utils import AvailableModelsCache + + cache: Final = AvailableModelsCache(ttl_seconds=300, max_size=1000, clock=lambda: 1_000.0) + providers: Final = tuple(f"provider-{i}" for i in range(999)) + cache.get_cached_model_info() # first call records the env hash + for provider in providers: + cache.set_cached_model_info(provider, litellm_params=None, available_models=[provider]) + + assert all(cache.get_cached_model_info(provider) == [provider] for provider in providers) + + def test_delta_tool_calls_sequential_indices(): """ Test that multiple tool calls without explicit indices receive sequential indices.