From f8e2fcdbbf25e0e20a35c27e1e825a03f357b193 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:52:25 -0700 Subject: [PATCH] test(caching): run loop-reuse tests until a loop address repeats The collected-loop tests ran a fixed 50 loops and passed with the fix reverted whenever no loop reused an earlier address. They now stop at the first reuse and skip if none happens, through a shared helper. --- tests/unit/caching/_event_loop_reuse.py | 20 +++++++++++++ .../caching/test_evicted_client_closer.py | 30 +++++++------------ .../unit/caching/test_llm_caching_handler.py | 18 ++++++----- 3 files changed, 42 insertions(+), 26 deletions(-) create mode 100644 tests/unit/caching/_event_loop_reuse.py diff --git a/tests/unit/caching/_event_loop_reuse.py b/tests/unit/caching/_event_loop_reuse.py new file mode 100644 index 00000000000..97cf8fe31f0 --- /dev/null +++ b/tests/unit/caching/_event_loop_reuse.py @@ -0,0 +1,20 @@ +from collections.abc import Callable +from typing import Final, TypeVar + +import pytest + +LoopResult = TypeVar("LoopResult") + + +def runs_until_a_loop_address_repeats( + run_on_new_loop: Callable[[], tuple[int, LoopResult]], + earlier: tuple[tuple[int, LoopResult], ...] = (), + limit: int = 50, +) -> tuple[tuple[int, LoopResult], ...]: + if len(earlier) == limit: + pytest.skip(f"none of {limit} collected loops had its address handed to a later loop") + latest: Final = run_on_new_loop() + runs: Final = (*earlier, latest) + if latest[0] in {loop_id for loop_id, _ in earlier}: + return runs + return runs_until_a_loop_address_repeats(run_on_new_loop, runs, limit) diff --git a/tests/unit/caching/test_evicted_client_closer.py b/tests/unit/caching/test_evicted_client_closer.py index c69c6e8b523..936eb37a733 100644 --- a/tests/unit/caching/test_evicted_client_closer.py +++ b/tests/unit/caching/test_evicted_client_closer.py @@ -19,6 +19,7 @@ from redis.asyncio import ConnectionPool, Redis from litellm.caching.evicted_client_closer import EvictedClientCloser from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler +from tests.unit.caching._event_loop_reuse import runs_until_a_loop_address_repeats class FakeClock: @@ -278,37 +279,28 @@ def test_sync_client_evicted_outside_an_event_loop_is_still_closed(): def test_async_client_from_collected_loop_is_not_closed_by_fresh_loops() -> None: clock: Final = FakeClock() closer: Final = EvictedClientCloser(grace_seconds=0, clock=clock) - client: Final = RecordingAsyncClient() - closer.mark_owned(client) - loop_a: Final = asyncio.new_event_loop() - async def schedule() -> None: - closer.schedule(client) - - try: - loop_a.run_until_complete(schedule()) - finally: - loop_a.close() - del loop_a - gc.collect() - - async def reap() -> None: + async def reap_then_schedule(client: RecordingAsyncClient) -> None: closer.reap() await asyncio.sleep(0) + closer.schedule(client) - def reap_on_new_loop() -> None: + def run_on_new_loop() -> tuple[int, RecordingAsyncClient]: + client: Final = RecordingAsyncClient() + closer.mark_owned(client) loop: Final = asyncio.new_event_loop() + loop_id: Final = id(loop) try: - loop.run_until_complete(reap()) + loop.run_until_complete(reap_then_schedule(client)) finally: loop.close() del loop gc.collect() + return loop_id, client - for _ in range(50): - reap_on_new_loop() + runs: Final = runs_until_a_loop_address_repeats(run_on_new_loop) - assert client.close_calls == 0 + assert all(client.close_calls == 0 for _, client in runs) def test_async_client_queued_on_a_live_loop_is_closed_only_by_that_loop() -> None: diff --git a/tests/unit/caching/test_llm_caching_handler.py b/tests/unit/caching/test_llm_caching_handler.py index 7812ba254de..c81b4ebd9f4 100644 --- a/tests/unit/caching/test_llm_caching_handler.py +++ b/tests/unit/caching/test_llm_caching_handler.py @@ -30,6 +30,7 @@ from litellm.llms.custom_httpx.http_handler import( from litellm.types.utils import LlmProviders from litellm.utils import _invalidate_model_cost_lowercase_map from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome +from tests.unit.caching._event_loop_reuse import runs_until_a_loop_address_repeats async def _get_and_set_cache(cache: LLMClientCache) -> object | None: @@ -38,30 +39,32 @@ async def _get_and_set_cache(cache: LLMClientCache) -> object | None: return cached -def _cache_result_on_new_loop(cache: LLMClientCache) -> object | None: +def _cache_result_on_new_loop(cache: LLMClientCache) -> tuple[int, object | None]: loop: Final = asyncio.new_event_loop() + loop_id: Final = id(loop) try: result: Final = loop.run_until_complete(_get_and_set_cache(cache)) finally: loop.close() del loop gc.collect() - return result + return loop_id, result async def _get_cache_key(cache: LLMClientCache) -> str: return cache.update_cache_key_with_event_loop("client") -def _cache_key_on_new_loop(cache: LLMClientCache) -> str: +def _cache_key_on_new_loop(cache: LLMClientCache) -> tuple[int, str]: loop: Final = asyncio.new_event_loop() + loop_id: Final = id(loop) try: key: Final = loop.run_until_complete(_get_cache_key(cache)) finally: loop.close() del loop gc.collect() - return key + return loop_id, key class MockAsyncClient: @@ -86,14 +89,15 @@ class MockSyncClient: def test_async_client_cache_does_not_reuse_clients_between_collected_loops() -> None: cache: Final = LLMClientCache() - results: Final = tuple(_cache_result_on_new_loop(cache) for _ in range(50)) + runs: Final = runs_until_a_loop_address_repeats(lambda: _cache_result_on_new_loop(cache)) - assert all(result is None for result in results) + assert all(result is None for _, result in runs) def test_async_client_cache_keys_differ_for_every_collected_loop() -> None: cache: Final = LLMClientCache() - keys: Final = tuple(_cache_key_on_new_loop(cache) for _ in range(50)) + runs: Final = runs_until_a_loop_address_repeats(lambda: _cache_key_on_new_loop(cache)) + keys: Final = tuple(key for _, key in runs) assert len(set(keys)) == len(keys)