From 0b43d058d1b944b772f415091ca915742825384a Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 10 Nov 2026 13:04:28 -0800 Subject: [PATCH] test(speech): make the speech callback tests order-independent and cover the Vertex AI await The sync callback test now starts from a fresh callback registration, so it passes after other speech tests in the same run. The internal sub-call cache test records real callback events instead of mocking the logging object, and a Vertex AI test fails when that branch passes aspeech=False --- tests/unit/caching/test_caching_handler.py | 42 +++++++++++++++---- .../text_to_speech/test_transformation.py | 40 ++++++++++++++++++ tests/unit/test_main.py | 4 +- 3 files changed, 76 insertions(+), 10 deletions(-) diff --git a/tests/unit/caching/test_caching_handler.py b/tests/unit/caching/test_caching_handler.py index e12c8800dc9..1ff368ae010 100644 --- a/tests/unit/caching/test_caching_handler.py +++ b/tests/unit/caching/test_caching_handler.py @@ -3095,9 +3095,29 @@ async def test_async_get_cache_answers_400_for_a_single_object_embedding_input() @pytest.mark.parametrize("internal", [False, True]) async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its_own( monkeypatch: pytest.MonkeyPatch, internal: bool -): - cache: Final = Cache(type="local") +) -> None: + from litellm.integrations.custom_logger import CustomLogger + + logged: Final[list[tuple[str, object]]] = [] + + class Recorder(CustomLogger): + def log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + logged.append(("sync", kwargs.get("litellm_call_id"))) + + async def async_log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + logged.append(("async", kwargs.get("litellm_call_id"))) + + recorder: Final = Recorder() + cache: Final = Cache(type=LiteLLMCacheType.LOCAL) monkeypatch.setattr(litellm, "cache", cache) + monkeypatch.setattr(litellm, "callbacks", []) + monkeypatch.setattr(litellm, "input_callback", []) + monkeypatch.setattr(litellm, "success_callback", [recorder]) + monkeypatch.setattr(litellm, "_async_success_callback", [recorder]) kwargs: Final = {"model": "gpt-5.4", "messages": [{"role": "user", "content": "hello"}], "caching": True} cache.add_cache( result=litellm.ModelResponse( @@ -3108,10 +3128,7 @@ async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its **kwargs, ) async_logging: Final = _build_logging_obj(CallTypes.acompletion.value, stream=False) - async_logging.async_success_handler = AsyncMock() - async_logging.handle_sync_success_callbacks_for_async_calls = MagicMock() sync_logging: Final = _build_logging_obj(CallTypes.completion.value, stream=False) - sync_logging.handle_sync_cache_hit_success_callbacks = MagicMock() with internal_sub_call() if internal else nullcontext(): async_hit: Final = await LLMCachingHandler( @@ -3136,8 +3153,17 @@ async def test_a_cache_hit_inside_an_internal_sub_call_fires_no_callbacks_of_its kwargs=kwargs, ) + expected: Final = ( + Counter() + if internal + else Counter({("async", async_logging.litellm_call_id): 1, ("sync", sync_logging.litellm_call_id): 1}) + ) + for _ in range(100): + if len(logged) >= expected.total(): + break + await asyncio.sleep(0.05) + await asyncio.sleep(0.5) + assert async_hit is not None and async_hit.cached_result is not None assert sync_hit.cached_result is not None - assert async_logging.async_success_handler.called is not internal - assert async_logging.handle_sync_success_callbacks_for_async_calls.called is not internal - assert sync_logging.handle_sync_cache_hit_success_callbacks.called is not internal + assert Counter(logged) == expected, logged diff --git a/tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py b/tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py index a3a11b2a66b..5713cf9ddf0 100644 --- a/tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py +++ b/tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py @@ -1,7 +1,9 @@ +import asyncio import base64 from typing import Final import json from collections.abc import Iterator, Mapping +from concurrent.futures import ThreadPoolExecutor from unittest.mock import MagicMock, Mock, patch import httpx @@ -1169,3 +1171,41 @@ async def test_aspeech_vertex_ai_gemini_tts_no_cache_request_reaches_vertex( await litellm.aspeech(**speech_kwargs, cache={"no-cache": True}) assert vertex_route.call_count == 2 + + +@pytest.mark.asyncio +async def test_aspeech_vertex_ai_gemini_tts_waits_for_a_stalled_upstream_without_holding_a_worker_thread( + respx_mock: respx.MockRouter, google_token_endpoint: responses.RequestsMock +) -> None: + upstream_reached: Final = asyncio.Event() + release_upstream: Final = asyncio.Event() + + async def stalled_upstream(request: httpx.Request) -> httpx.Response: + upstream_reached.set() + await release_upstream.wait() + return httpx.Response(200, json=_GENERATE_CONTENT_RESPONSE) + + respx_mock.post( + url__regex=r"https://aiplatform\.googleapis\.com/.*/models/gemini-3\.8-flash-tts:generateContent$" + ).mock(side_effect=stalled_upstream) + loop: Final = asyncio.get_running_loop() + with ThreadPoolExecutor(max_workers=1) as single_worker: + loop.set_default_executor(single_worker) + speech_task: Final = asyncio.create_task( + litellm.aspeech( + model="vertex_ai/gemini-3.8-flash-tts", + input="Say hello", + voice="Kore", + vertex_credentials=_AUTHORIZED_USER, + vertex_location="global", + ) + ) + await asyncio.wait_for(upstream_reached.wait(), timeout=10) + + worker_answer: Final = await asyncio.wait_for(loop.run_in_executor(None, lambda: "free"), timeout=10) + + release_upstream.set() + response: Final = await asyncio.wait_for(speech_task, timeout=10) + + assert worker_answer == "free" + assert _PCM_AUDIO in response.content diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index ac22723de8f..51ce69ec326 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -4547,7 +4547,6 @@ async def test_aspeech_gemini_bridge_logs_only_the_speech_call_under_its_call_id _assert_logged_once_as_speech(events) - @pytest.mark.asyncio async def test_aspeech_gemini_bridge_runs_sync_function_callbacks_once_with_the_speech_response( respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch @@ -4560,6 +4559,7 @@ async def test_aspeech_gemini_bridge_runs_sync_function_callbacks_once_with_the_ def sync_callback(kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object) -> None: logged.append((type(response_obj), kwargs.get("litellm_call_id"))) + monkeypatch.setattr(litellm.utils, "callback_list", []) monkeypatch.setattr(litellm, "success_callback", [sync_callback]) respx_mock.post( url__regex=r"https://generativelanguage\.googleapis\.com/v1beta/models/gemini-2\.5-flash-preview-tts:generateContent.*" @@ -5097,7 +5097,7 @@ def test_speech_gemini_tts_keeps_cached_audio_only_for_the_request_ttl( async def test_aspeech_gemini_tts_waits_for_a_stalled_upstream_without_holding_a_worker_thread( respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch -): +) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) upstream_reached: Final = asyncio.Event() release_upstream: Final = asyncio.Event()