From 4dba2f785ad4d73241965e4973a537b98ef7856a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:00:38 +0000 Subject: [PATCH] test(redis): satisfy final lint and formatting checks --- litellm/caching/_redis_capabilities.py | 4 +- litellm/caching/redis_cache.py | 5 +- .../auth_cache_invalidation_pubsub.py | 10 +++- .../proxy/common_utils/config_sync_pubsub.py | 8 +++- .../db_transaction_queue/pod_lock_manager.py | 6 ++- .../redis_update_buffer.py | 9 ++-- litellm/proxy/hooks/batch_enqueued_tokens.py | 18 ++++--- .../hooks/parallel_request_limiter_v3.py | 47 +++++++++++++------ .../unit/caching/test_redis_cluster_cache.py | 5 +- .../test_redis_distributed_lock.py | 6 ++- tests/unit/proxy/auth/test_login_utils.py | 8 ++-- 11 files changed, 86 insertions(+), 40 deletions(-) diff --git a/litellm/caching/_redis_capabilities.py b/litellm/caching/_redis_capabilities.py index d70decc49c1..b2218221282 100644 --- a/litellm/caching/_redis_capabilities.py +++ b/litellm/caching/_redis_capabilities.py @@ -66,7 +66,9 @@ def failure(error: BaseException) -> CacheFailure: reason: Final = ( "circuit_open" if isinstance(error, RedisCircuitBreakerOpenError) - else "timeout" if is_redis_timeout_failure(error) else "other" + else "timeout" + if is_redis_timeout_failure(error) + else "other" ) return CacheFailure( kind="unavailable", message=f"Cache operation failed ({type(error).__name__}): {error}", reason=reason diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index c39a79b56d7..a59848a28b0 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -544,7 +544,9 @@ def log_redis_failure( logger: logging.Logger, level: int, message: str, exc: BaseException | CacheFailure, with_traceback: bool = False ) -> None: detail: Final = exc.message if isinstance(exc, CacheFailure) else exc - circuit_open: Final = exc.reason == "circuit_open" if isinstance(exc, CacheFailure) else isinstance(exc, RedisCircuitBreakerOpenError) + circuit_open: Final = ( + exc.reason == "circuit_open" if isinstance(exc, CacheFailure) else isinstance(exc, RedisCircuitBreakerOpenError) + ) if circuit_open: logger.debug("%s: %s", message, detail, stacklevel=2) return @@ -2332,7 +2334,6 @@ class RedisCache(BaseCache): if len(increment_list) == 0: return None - _redis_client: Final = cast( # cast-ok: RedisCluster stubs omit the shared Redis commands used here "Redis[bytes]", self._get_async_client() ) diff --git a/litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py b/litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py index 522c70d17b6..ba51de9c439 100644 --- a/litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py +++ b/litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py @@ -45,7 +45,11 @@ _DECODED_JSON: Final = TypeAdapter(object) def auth_cache_invalidation_channel(redis_cache: "RedisCache") -> str: - return AUTH_CACHE_INVALIDATION_CHANNEL if redis_cache.namespace is None else f"{redis_cache.namespace}:{AUTH_CACHE_INVALIDATION_CHANNEL}" + return ( + AUTH_CACHE_INVALIDATION_CHANNEL + if redis_cache.namespace is None + else f"{redis_cache.namespace}:{AUTH_CACHE_INVALIDATION_CHANNEL}" + ) @dataclass(frozen=True, slots=True) @@ -296,7 +300,9 @@ class AuthCacheInvalidationSubscriber: return CacheFailure("unavailable", str(error)) async def _consume(self, subscription: CacheSubscription) -> CacheFailure: - while not isinstance(message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure): + while not isinstance( + message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure + ): if message is None: continue self._apply_message(message) diff --git a/litellm/proxy/common_utils/config_sync_pubsub.py b/litellm/proxy/common_utils/config_sync_pubsub.py index 7d74afc852a..a8dd0d9d42f 100644 --- a/litellm/proxy/common_utils/config_sync_pubsub.py +++ b/litellm/proxy/common_utils/config_sync_pubsub.py @@ -251,7 +251,9 @@ class ConfigSyncSubscriber: return CacheFailure("unavailable", str(error)) async def _consume(self, subscription: CacheSubscription) -> CacheFailure: - while not isinstance(message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure): + while not isinstance( + message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure + ): if message is None: continue if (failure := await self._debounce_and_resync(subscription)) is not None: @@ -271,7 +273,9 @@ class ConfigSyncSubscriber: async def _wait_for_min_resync_interval(self) -> None: if self._last_resync_at is None: return - seconds_until_next_resync: Final = self._min_resync_interval_seconds - (self._monotonic() - self._last_resync_at) + seconds_until_next_resync: Final = self._min_resync_interval_seconds - ( + self._monotonic() - self._last_resync_at + ) if seconds_until_next_resync <= 0: return verbose_proxy_logger.debug( diff --git a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py b/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py index 3fc42e22291..95ba17b5edf 100644 --- a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py +++ b/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py @@ -160,8 +160,10 @@ class PodLockManager: if not isinstance(released, CacheFailure): return int(released) log_redis_failure( - verbose_proxy_logger, logging.WARNING, - f"Atomic release failed for lock_key={lock_key}, falling back to GET+DEL", released + verbose_proxy_logger, + logging.WARNING, + f"Atomic release failed for lock_key={lock_key}, falling back to GET+DEL", + released, ) current_value = await self.redis_cache.async_get_cache(lock_key) diff --git a/litellm/proxy/db/db_transaction_queue/redis_update_buffer.py b/litellm/proxy/db/db_transaction_queue/redis_update_buffer.py index 0bb3a000e35..98c72fe5dc5 100644 --- a/litellm/proxy/db/db_transaction_queue/redis_update_buffer.py +++ b/litellm/proxy/db/db_transaction_queue/redis_update_buffer.py @@ -561,11 +561,13 @@ class RedisUpdateBuffer: if isinstance(result, CacheFailure): verbose_proxy_logger.error( "Spend tracking - failed to restore uncommitted transactions to Redis. " - "These spend updates are lost. Error: %s", result.message, + "These spend updates are lost. Error: %s", + result.message, ) return verbose_proxy_logger.info( - "Spend tracking - restored %d uncommitted transaction set(s) to Redis for retry on next tick.", len(rpush_list) + "Spend tracking - restored %d uncommitted transaction set(s) to Redis for retry on next tick.", + len(rpush_list), ) @with_service_target(SPEND_QUEUE_TARGET) @@ -582,7 +584,8 @@ class RedisUpdateBuffer: if isinstance(result, CacheFailure): verbose_proxy_logger.error( "Tool discovery - failed to push %d tool rows to Redis, restoring them for the next tick. Error: %s", - len(items), result.message, + len(items), + result.message, ) for item in items: tool_discovery_queue.add_update(item) diff --git a/litellm/proxy/hooks/batch_enqueued_tokens.py b/litellm/proxy/hooks/batch_enqueued_tokens.py index 56d989ec99a..6cb07847434 100644 --- a/litellm/proxy/hooks/batch_enqueued_tokens.py +++ b/litellm/proxy/hooks/batch_enqueued_tokens.py @@ -187,8 +187,10 @@ class BatchEnqueuedTokenStore: if not isinstance(result, CacheFailure): return result log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Redis enqueued-token reserve failed, falling back to in-memory", result + verbose_proxy_logger, + logging.WARNING, + "Redis enqueued-token reserve failed, falling back to in-memory", + result, ) return await self._reserve_in_memory(tokens=tokens, scopes=scopes, span=litellm_parent_otel_span) @@ -325,8 +327,10 @@ class BatchEnqueuedTokenStore: if not isinstance(result, CacheFailure): return log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Redis enqueued-token reservation save failed, falling back to in-memory", result + verbose_proxy_logger, + logging.WARNING, + "Redis enqueued-token reservation save failed, falling back to in-memory", + result, ) await self.internal_usage_cache.async_set_cache( key=self._record_key(batch_id), @@ -371,8 +375,10 @@ class BatchEnqueuedTokenStore: if not isinstance(result, CacheFailure): return result log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Redis enqueued-token reservation pop failed, falling back to in-memory", result + verbose_proxy_logger, + logging.WARNING, + "Redis enqueued-token reservation pop failed, falling back to in-memory", + result, ) return None diff --git a/litellm/proxy/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index 2bdad8a26de..d1ef1163187 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -1254,7 +1254,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if not isinstance(result, CacheFailure): return result self._reject_if_rate_limit_unverifiable("atomic counter read", result) - log_redis_failure(verbose_proxy_logger, logging.WARNING, "Atomic counter read failed, using local mirror", result) + log_redis_failure( + verbose_proxy_logger, logging.WARNING, "Atomic counter read failed, using local mirror", result + ) return await self._batch_get_counter_values(keys=keys, parent_otel_span=parent_otel_span, local_only=True) def _reject_if_rate_limit_unverifiable(self, failed_operation: str, error: CacheFailure) -> None: @@ -1312,8 +1314,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): await self._refund_later_pipelined_groups(key_groups[index + 1 :], operations[index + 1 :]) self._reject_if_rate_limit_unverifiable("window counter increment", result) log_redis_failure( - verbose_proxy_logger, logging.WARNING, - f"Window counter increment failed for hash tag {hash_tag}", result + verbose_proxy_logger, + logging.WARNING, + f"Window counter increment failed for hash tag {hash_tag}", + result, ) all_cache_values.extend( await self.in_memory_cache_sliding_window( @@ -1609,8 +1613,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if isinstance(result, CacheFailure): self._reject_if_rate_limit_unverifiable("concurrency acquire", result) log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Concurrency acquire failed, falling back to in-memory gauge", result + verbose_proxy_logger, + logging.WARNING, + "Concurrency acquire failed, falling back to in-memory gauge", + result, ) async with self._check_and_increment_lock: return await self._acquire_parallel_slots_in_memory(gauges, slot_id, parent_otel_span) @@ -1642,7 +1648,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if not isinstance(result, CacheFailure): return tuple(max(0, count) for count in result) self._reject_if_rate_limit_unverifiable("concurrency count", result) - log_redis_failure(verbose_proxy_logger, logging.WARNING, "Concurrency count failed, using local mirror", result) + log_redis_failure( + verbose_proxy_logger, logging.WARNING, "Concurrency count failed, using local mirror", result + ) return await self._read_local_gauge_counts(gauge_keys, parent_otel_span) async def _read_local_gauge_counts( @@ -1760,8 +1768,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): await self._mirror_released_parallel_slots(counter_keys, result, parent_otel_span) return log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Concurrency release failed, falling back to in-memory release", result + verbose_proxy_logger, + logging.WARNING, + "Concurrency release failed, falling back to in-memory release", + result, ) await self._release_parallel_request_slots_in_memory(counter_keys, slot_id, parent_otel_span) @@ -1789,8 +1799,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): result: Final = future.result() if isinstance(result, CacheFailure): log_redis_failure( - verbose_proxy_logger, logging.WARNING, - "Concurrency release failed, the slot stays released in memory only", result + verbose_proxy_logger, + logging.WARNING, + "Concurrency release failed, the slot stays released in memory only", + result, ) redis_cache.async_release_concurrency( @@ -1975,7 +1987,8 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): await self._refund_applied_descriptor_groups(applied) self._reject_if_rate_limit_unverifiable("window capacity reservation", result) log_redis_failure( - verbose_proxy_logger, logging.ERROR, + verbose_proxy_logger, + logging.ERROR, f"Window capacity reservation failed, refunded {len(applied)} prior descriptors and falling back to memory", result, ) @@ -2033,7 +2046,8 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): await self._refund_applied_descriptor_groups(applied) self._reject_if_rate_limit_unverifiable("window capacity reservation", failure) log_redis_failure( - verbose_proxy_logger, logging.ERROR, + verbose_proxy_logger, + logging.ERROR, f"Window capacity reservation failed, refunded {len(applied)} batched descriptors and falling back to memory", failure, ) @@ -2135,7 +2149,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): ), CacheFailure, ): - log_redis_failure(verbose_proxy_logger, logging.WARNING, f"Failed to refund {refund.counter_key} on rollback", result) + log_redis_failure( + verbose_proxy_logger, logging.WARNING, f"Failed to refund {refund.counter_key} on rollback", result + ) def _build_atomic_response( self, @@ -4244,7 +4260,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if not isinstance(result, CacheFailure): return log_redis_failure( - verbose_proxy_logger, logging.WARNING, "TTL preservation failed, falling back to regular pipeline", result + verbose_proxy_logger, + logging.WARNING, + "TTL preservation failed, falling back to regular pipeline", + result, ) await self.internal_usage_cache.dual_cache.async_increment_cache_pipeline(increment_list=group_operations) diff --git a/tests/unit/caching/test_redis_cluster_cache.py b/tests/unit/caching/test_redis_cluster_cache.py index ceb00f8844c..c1dd02d16ed 100644 --- a/tests/unit/caching/test_redis_cluster_cache.py +++ b/tests/unit/caching/test_redis_cluster_cache.py @@ -661,9 +661,10 @@ async def test_cancelled_typed_cluster_read_releases_its_connection_admission() startup_nodes=[{"host": "typed-cluster-cancellation", "port": 7000}], connection_gate=ClusterConnectionGate(limit=1, wait_timeout=60), ) + first: Final = asyncio.create_task(cache.async_read_values(keys=("cancelled",))) with pytest.warns(FutureWarning, match="async_read_values is experimental and may change"): - first: Final = asyncio.create_task(cache.async_read_values(keys=("cancelled",))) - assert await asyncio.wait_for(entered.get(), timeout=1) == "cancelled" + entered_key: Final = await asyncio.wait_for(entered.get(), timeout=1) + assert entered_key == "cancelled" first.cancel() with pytest.raises(asyncio.CancelledError): await first diff --git a/tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_redis_distributed_lock.py b/tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_redis_distributed_lock.py index 7771bdc2732..f0e2f738be5 100644 --- a/tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_redis_distributed_lock.py +++ b/tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_redis_distributed_lock.py @@ -133,7 +133,8 @@ async def test_lease_cancellation_is_not_translated_to_an_outage( ) -> None: lock, transport = backend transport.side_effect = asyncio.CancelledError - with pytest.raises(asyncio.CancelledError): + + async def invoke() -> None: match operation: case "acquire": await lock.acquire("k", "tok", 10.0) @@ -144,6 +145,9 @@ async def test_lease_cancellation_is_not_translated_to_an_outage( case "is_held": await lock.is_held("k") + with pytest.raises(asyncio.CancelledError): + await invoke() + @pytest.mark.asyncio async def test_lease_apis_warn_as_experimental_at_the_lock_callsite( diff --git a/tests/unit/proxy/auth/test_login_utils.py b/tests/unit/proxy/auth/test_login_utils.py index 5b30fc72042..70bdc76ce6c 100644 --- a/tests/unit/proxy/auth/test_login_utils.py +++ b/tests/unit/proxy/auth/test_login_utils.py @@ -1509,11 +1509,9 @@ async def test_invalid_shared_throttle_replies_do_not_allow_unchecked_logins( redis, _ = login_redis_backend throttle: Final = _throttle(user_limit=1, redis_cache=redis) with patch("redis.asyncio.client.Redis.register_script", return_value=AsyncMock(return_value=reply)): - with pytest.raises(ValueError): - if phase == "attempt": - await throttle.attempt("admin") - else: - await throttle.record_failure("admin") + operation: Final = throttle.attempt if phase == "attempt" else throttle.record_failure + with pytest.raises(ValueError, match="validation error"): + await operation("admin") assert not throttle.counters.cache_dict assert not throttle.blocks.cache_dict