Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 38 additions & 95 deletions .circleci/config.yml

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion .circleci/scripts/run_integration.sh
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ if [ "$suite" = management ] || [ "$suite" = mcp ]; then
fi

if [ "$suite" = providers ]; then
INTEGRATION_RUN_ID="$integration_identity" .venv/bin/python -m pytest --noconftest -o addopts= \
INTEGRATION_RUN_ID="$integration_identity" .venv/bin/python -m pytest --tb=short --noconftest -o addopts= \
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
--strict-markers --strict-config -p no:pytest-retry -p no:rerunfailures --timeout=30 \
tests/e2e/test_provider_edge.py::TestReplayMode::test_content_drift_returns_the_miss_status_naming_both_keys \
tests/e2e/test_provider_edge.py::TestReplayMode::test_exhausted_key_returns_the_miss_status \
Expand Down
Original file line number Diff line number Diff line change
@@ -1,51 +1,40 @@
"""Least-busy routing keeps its in-flight counters in Redis, and the clamp at zero plus the
create-once TTL both live inside a Lua script. Nothing but a real Redis runs that script, so
these are the only tests that fail when the script itself is wrong."""

import os
import uuid
from collections.abc import Iterator
from typing import Final

import pytest
from dotenv import load_dotenv

load_dotenv()

from litellm.caching.redis_cache import RedisCache

TTL: Final = 600


@pytest.fixture
def counter():
cache: Final = RedisCache(host=os.getenv("REDIS_HOST"), port=os.getenv("REDIS_PORT"))
key: Final = f"lit7039-{uuid.uuid4()}"
def counter() -> Iterator[tuple[RedisCache, str, str]]:
cache: Final = RedisCache(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"]))
key: Final = f"increment-with-floor-{uuid.uuid4()}"
yield cache, key, cache.check_and_fix_namespace(key=key)
cache.delete_cache(key)


def test_a_counter_adds_every_increment_and_reads_back_what_it_holds(counter):
def test_a_counter_adds_every_increment_and_reads_back_what_it_holds(counter: tuple[RedisCache, str, str]) -> None:
cache, key, _ = counter

assert cache.increment_with_floor(key, 3, TTL) == 3
assert cache.increment_with_floor(key, 2, TTL) == 5
assert cache.batch_get_counts([key]) == (5,)


def test_a_decrement_past_zero_leaves_the_counter_at_zero(counter):
"""A worker whose counter expired mid-request decrements a key that is no longer there.
Without the clamp that deployment reads negative, and least-busy pins every later request
on it until the count climbs back to zero."""
def test_a_decrement_past_zero_leaves_the_counter_at_zero(counter: tuple[RedisCache, str, str]) -> None:
cache, key, _ = counter

assert cache.increment_with_floor(key, 1, TTL) == 1
assert cache.increment_with_floor(key, -5, TTL) == 0
assert cache.batch_get_counts([key]) == (0,)


def test_traffic_never_pushes_a_counters_expiry_back_out(counter):
"""The TTL is what releases a count whose worker died mid-request. Rewriting it on every
touch would keep that stuck count alive for as long as the group takes traffic."""
def test_traffic_never_pushes_a_counters_expiry_back_out(counter: tuple[RedisCache, str, str]) -> None:
cache, key, namespaced_key = counter

cache.increment_with_floor(key, 1, TTL)
Expand All @@ -57,7 +46,7 @@ def test_traffic_never_pushes_a_counters_expiry_back_out(counter):
assert cache.redis_client.ttl(namespaced_key) <= 30


def test_clamping_to_zero_keeps_the_expiry_it_already_had(counter):
def test_clamping_to_zero_keeps_the_expiry_it_already_had(counter: tuple[RedisCache, str, str]) -> None:
cache, key, namespaced_key = counter

cache.increment_with_floor(key, 1, TTL)
Expand All @@ -68,7 +57,7 @@ def test_clamping_to_zero_keeps_the_expiry_it_already_had(counter):


@pytest.mark.asyncio
async def test_the_async_counter_behaves_the_same_way(counter):
async def test_the_async_counter_behaves_the_same_way(counter: tuple[RedisCache, str, str]) -> None:
cache, key, namespaced_key = counter

assert await cache.async_increment_with_floor(key, 2, TTL) == 2
Expand Down
94 changes: 94 additions & 0 deletions tests/integration/sdk/test_dual_cache_redis.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
import asyncio
import os
import uuid
from typing import Final
from unittest.mock import patch

import pytest

from litellm.caching.dual_cache import DualCache
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.caching.redis_cache import RedisCache


def _redis_cache() -> RedisCache:
return RedisCache(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"]))
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved


@pytest.mark.asyncio
async def test_a_value_only_in_redis_is_read_once_from_redis_then_from_memory() -> None:
redis_cache: Final = _redis_cache()
dual_cache: Final = DualCache(in_memory_cache=InMemoryCache(), redis_cache=redis_cache)
sync_key: Final = f"redis-only-sync-{uuid.uuid4()}"
async_key: Final = f"redis-only-async-{uuid.uuid4()}"
redis_cache.set_cache(sync_key, {"v": "sync"})
await redis_cache.async_set_cache(async_key, {"v": "async"})

assert dual_cache.get_cache(sync_key) == {"v": "sync"}
assert await dual_cache.async_get_cache(async_key) == {"v": "async"}

with (
patch.object(redis_cache, "get_cache") as sync_redis_read,
patch.object(redis_cache, "async_get_cache") as async_redis_read,
):
assert dual_cache.get_cache(sync_key) == {"v": "sync"}
assert await dual_cache.async_get_cache(async_key) == {"v": "async"}
sync_redis_read.assert_not_called()
async_redis_read.assert_not_called()


@pytest.mark.asyncio
async def test_a_deleted_key_is_gone_from_both_memory_and_redis() -> None:
redis_cache: Final = _redis_cache()
dual_cache: Final = DualCache(in_memory_cache=InMemoryCache(), redis_cache=redis_cache)
sync_key: Final = f"deleted-sync-{uuid.uuid4()}"
async_key: Final = f"deleted-async-{uuid.uuid4()}"
dual_cache.set_cache(sync_key, {"v": "sync"})
await dual_cache.async_set_cache(async_key, {"v": "async"})

dual_cache.delete_cache(sync_key)
await dual_cache.async_delete_cache(async_key)

assert dual_cache.get_cache(sync_key) is None
assert await dual_cache.async_get_cache(async_key) is None
assert redis_cache.get_cache(sync_key) is None
assert await redis_cache.async_get_cache(async_key) is None


@pytest.mark.asyncio
async def test_a_batch_read_without_an_in_memory_cache_reads_redis() -> None:
redis_cache: Final = _redis_cache()
dual_cache: Final = DualCache(in_memory_cache=None, redis_cache=redis_cache)
key: Final = f"no-memory-{uuid.uuid4()}"
await redis_cache.async_set_cache(key, {"v": "from-redis"})

assert await dual_cache.async_batch_get_cache([key]) == [{"v": "from-redis"}]


@pytest.mark.asyncio
async def test_sync_and_async_batch_reads_share_one_redis_without_sync_reads_going_async() -> None:
redis_cache: Final = _redis_cache()
dual_cache: Final = DualCache(redis_cache=redis_cache)
run_id: Final = uuid.uuid4().hex
sync_keys: Final = [f"sync_{run_id}_{index}" for index in range(5)]
async_keys: Final = [f"async_{run_id}_{index}" for index in range(5)]
in_loop_keys: Final = [f"in_loop_{run_id}_{index}" for index in range(3)]
survivor_key: Final = f"survivor_{run_id}"
expected: Final = {key: {"key": key} for key in [*sync_keys, *async_keys, *in_loop_keys, survivor_key]}
await asyncio.gather(*(redis_cache.async_set_cache(key, value, ttl=60) for key, value in expected.items()))

concurrent_results: Final = await asyncio.gather(
*(asyncio.to_thread(dual_cache.batch_get_cache, keys=[key]) for key in sync_keys),
*(dual_cache.async_batch_get_cache(keys=[key]) for key in async_keys),
)
assert list(concurrent_results) == [[expected[key]] for key in [*sync_keys, *async_keys]]

with patch.object(
redis_cache,
"async_batch_get_cache",
side_effect=AssertionError("sync batch reads must not call async Redis"),
):
in_loop_results: Final = [dual_cache.batch_get_cache(keys=[key]) for key in in_loop_keys]

assert in_loop_results == [[expected[key]] for key in in_loop_keys]
assert await dual_cache.async_batch_get_cache(keys=[survivor_key]) == [expected[survivor_key]]
Loading
Loading