From 2d862de42d866368191bacc980e66157f81ee742 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:22:23 -0700 Subject: [PATCH] test(streaming): cover SDK 3 transport errors on finished streams and file downloads A finished chat stream whose connection drops before [DONE] should end cleanly on both the httpx and httpx2 transports, a mid-stream timeout should fall back as a Timeout, and file_content(stream=True) should raise litellm errors instead of raw transport errors --- .../test_streaming_handler.py | 127 ++++++++++++++++++ tests/unit/llms/openai/test_openai.py | 81 ++++++++++- .../test_openai_file_content_streaming.py | 88 +++++++++++- 3 files changed, 293 insertions(+), 3 deletions(-) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 68aabc3c01e..c4a55408179 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -11,6 +11,7 @@ from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, from unittest.mock import AsyncMock, MagicMock, Mock, patch import httpx +import openai import pytest from openai.types.chat import ChatCompletionChunk from pydantic import BaseModel @@ -4586,6 +4587,132 @@ async def test_transport_read_error_before_finish_reason_raises(logging_obj: Log assert fabricated_finish_reasons == [] +_SDK3_CHAT_REQUEST_URL: Final = "https://api.openai.com/v1/chat/completions" + + +def _raised_from(error: Exception, cause: Exception) -> Exception: + try: + raise error from cause + except Exception as raised: + return raised + + +def _httpx2_read_error() -> Exception: + return pytest.importorskip("httpx2").ReadError("peer closed connection") + + +def _httpx2_protocol_error() -> Exception: + return pytest.importorskip("httpx2").RemoteProtocolError("peer closed connection without sending complete body") + + +def _sdk_wrapped_read_error() -> Exception: + httpx2: Final = pytest.importorskip("httpx2") + return _raised_from( + openai.APIConnectionError(request=httpx2.Request("POST", _SDK3_CHAT_REQUEST_URL)), + httpx2.ReadError("peer closed connection"), + ) + + +def _httpx2_read_timeout() -> Exception: + return pytest.importorskip("httpx2").ReadTimeout("timed out") + + +def _sdk_wrapped_read_timeout() -> Exception: + httpx2: Final = pytest.importorskip("httpx2") + return _raised_from( + openai.APITimeoutError(request=httpx2.Request("POST", _SDK3_CHAT_REQUEST_URL)), + httpx2.ReadTimeout("timed out"), + ) + + +_SDK3_DROPPED_CONNECTIONS: Final = pytest.mark.parametrize( + "dropped_connection", + (_httpx2_read_error, _httpx2_protocol_error, _sdk_wrapped_read_error), + ids=("httpx2_read_error", "httpx2_protocol_error", "sdk_wrapped_read_error"), +) +_SYNC_AND_ASYNC: Final = pytest.mark.parametrize("sync_mode", (True, False), ids=("sync", "async")) + + +def _chunks_then_failure(chunks: Sequence[ModelResponseStream], failure: Exception) -> Iterator[ModelResponseStream]: + yield from chunks + raise failure + + +def _openai_stream_over( + chunks: Sequence[ModelResponseStream], failure: Exception, sync_mode: bool, logging_obj: Logging +) -> CustomStreamWrapper: + stream: Final = _chunks_then_failure(chunks, failure) + return CustomStreamWrapper( + completion_stream=stream if sync_mode else _as_async(stream), + model="openai/stub-model", + custom_llm_provider="openai", + logging_obj=logging_obj, + ) + + +async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelResponseStream]: + if sync_mode: + return list(wrapper) + return [chunk async for chunk in wrapper] + + +@pytest.mark.asyncio +@_SYNC_AND_ASYNC +@_SDK3_DROPPED_CONNECTIONS +async def test_sdk3_dropped_connection_after_finish_reason_ends_the_stream_cleanly( + logging_obj: Logging, sync_mode: bool, dropped_connection: Callable[[], Exception] +): + wrapper: Final = _openai_stream_over( + (_reset_test_chunk(content="Hello"), _reset_test_chunk(finish_reason="stop")), + dropped_connection(), + sync_mode, + logging_obj, + ) + + chunks: Final = await _collect(wrapper, sync_mode) + + assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "Hello" + assert [chunk.choices[0].finish_reason for chunk in chunks if chunk.choices[0].finish_reason] == ["stop"] + + +@pytest.mark.asyncio +@_SYNC_AND_ASYNC +@_SDK3_DROPPED_CONNECTIONS +async def test_sdk3_dropped_connection_before_finish_reason_raises_mid_stream_fallback( + logging_obj: Logging, sync_mode: bool, dropped_connection: Callable[[], Exception] +): + wrapper: Final = _openai_stream_over( + (_reset_test_chunk(content="Hel"),), dropped_connection(), sync_mode, logging_obj + ) + + with pytest.raises(MidStreamFallbackError) as excinfo: + await _collect(wrapper, sync_mode) + + assert excinfo.value.is_pre_first_chunk is False + assert excinfo.value.generated_content == "Hel" + assert isinstance(excinfo.value.original_exception, litellm.APIConnectionError) + + +@pytest.mark.asyncio +@_SYNC_AND_ASYNC +@pytest.mark.parametrize( + "read_timeout", (_httpx2_read_timeout, _sdk_wrapped_read_timeout), ids=("httpx2", "sdk_wrapped_httpx2") +) +@pytest.mark.parametrize("finished", (False, True), ids=("before_finish_reason", "after_finish_reason")) +async def test_sdk3_read_timeout_mid_stream_falls_back_as_a_timeout( + logging_obj: Logging, sync_mode: bool, read_timeout: Callable[[], Exception], finished: bool +): + chunks: Final = (_reset_test_chunk(content="Hel"), _reset_test_chunk(finish_reason="stop")) + wrapper: Final = _openai_stream_over(chunks if finished else chunks[:1], read_timeout(), sync_mode, logging_obj) + + with pytest.raises(MidStreamFallbackError) as excinfo: + await _collect(wrapper, sync_mode) + + assert excinfo.value.is_pre_first_chunk is False + assert isinstance(excinfo.value.original_exception, litellm.Timeout) + assert excinfo.value.original_exception.status_code == 408 + + def test_openai_custom_tool_call_stream_deltas_survive_conversion(logging_obj: Logging): """ Regression test: OpenAI chat completions custom tool calls stream as diff --git a/tests/unit/llms/openai/test_openai.py b/tests/unit/llms/openai/test_openai.py index 11d7c7d6d3c..2fab1286e51 100644 --- a/tests/unit/llms/openai/test_openai.py +++ b/tests/unit/llms/openai/test_openai.py @@ -2,7 +2,7 @@ import asyncio import importlib import json import os -from collections.abc import Mapping +from collections.abc import AsyncIterator, Iterator, Mapping from types import ModuleType from typing import Final, Literal, Protocol from unittest.mock import MagicMock, Mock, patch @@ -32,6 +32,7 @@ from litellm.llms.openai.openai import ( from litellm.types.utils import ( ImageResponse, ModelResponse, + ModelResponseStream, PromptTokensDetails, ) from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER @@ -441,6 +442,84 @@ async def test_acompletion_streams_text_deltas_over_injected_transport( assert last_finish == "stop" +def _sse_event(delta: Mapping[str, str], finish: str | None) -> bytes: + body: Final = { + "id": "chatcmpl-dropped", + "object": "chat.completion.chunk", + "created": 0, + "model": "gpt-5.6", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}], + } + return f"data: {json.dumps(body)}\n\n".encode() + + +_FINISHED_EVENTS_WITHOUT_DONE: Final = ( + _sse_event({"role": "assistant", "content": "Hel"}, None), + _sse_event({"content": "lo"}, "stop"), +) + + +def _finished_then_dropped_transport(http_backend: ModuleType) -> object: + class FinishedThenDropped(http_backend.SyncByteStream, http_backend.AsyncByteStream): + def __iter__(self) -> Iterator[bytes]: + yield from _FINISHED_EVENTS_WITHOUT_DONE + raise http_backend.ReadError("peer closed the connection before [DONE]") + + async def __aiter__(self) -> AsyncIterator[bytes]: + for event in _FINISHED_EVENTS_WITHOUT_DONE: + yield event + raise http_backend.ReadError("peer closed the connection before [DONE]") + + return http_backend.MockTransport( + lambda _request: http_backend.Response( + 200, headers={"content-type": "text/event-stream"}, stream=FinishedThenDropped() + ) + ) + + +def _streamed_text_and_finish_reasons(chunks: list[ModelResponseStream]) -> tuple[str, list[str]]: + text: Final = "".join(part.choices[0].delta.content or "" for part in chunks if part.choices) + finish_reasons: Final = [part.choices[0].finish_reason for part in chunks if part.choices and part.choices[0].finish_reason] + return text, finish_reasons + + +@pytest.mark.asyncio +async def test_acompletion_ends_a_finished_stream_cleanly_when_the_connection_drops_before_done( + http_backend: ModuleType, +): + async with http_backend.AsyncClient(transport=_finished_then_dropped_transport(http_backend)) as http_client: + stream: Final = await litellm.acompletion( + model="openai/gpt-5.6", + client=AsyncOpenAI(api_key="transport-only", http_client=http_client), + messages=[{"role": "user", "content": "dropped-stream-request"}], + stream=True, + num_retries=0, + max_retries=0, + ) + + async def drain() -> list[ModelResponseStream]: + return [part async for part in stream] + + chunks: Final = await asyncio.wait_for(drain(), timeout=10) + + assert _streamed_text_and_finish_reasons(chunks) == ("Hello", ["stop"]) + + +def test_completion_ends_a_finished_stream_cleanly_when_the_connection_drops_before_done(http_backend: ModuleType): + with http_backend.Client(transport=_finished_then_dropped_transport(http_backend)) as http_client: + stream: Final = litellm.completion( + model="openai/gpt-5.6", + client=OpenAI(api_key="transport-only", http_client=http_client), + messages=[{"role": "user", "content": "dropped-stream-request"}], + stream=True, + num_retries=0, + max_retries=0, + ) + chunks: Final = list(stream) + + assert _streamed_text_and_finish_reasons(chunks) == ("Hello", ["stop"]) + + @pytest.mark.asyncio async def test_acompletion_streams_tool_call_arguments_over_injected_transport(http_backend: ModuleType): outbound: Final = asyncio.Queue() diff --git a/tests/unit/llms/openai/test_openai_file_content_streaming.py b/tests/unit/llms/openai/test_openai_file_content_streaming.py index 466918267f6..1455fd45e67 100644 --- a/tests/unit/llms/openai/test_openai_file_content_streaming.py +++ b/tests/unit/llms/openai/test_openai_file_content_streaming.py @@ -1,6 +1,13 @@ -import pytest -from typing import AsyncIterator, Iterator, cast +import importlib +from types import ModuleType +import pytest +from typing import AsyncIterator, Final, Iterator, cast + +from openai import AsyncOpenAI, OpenAI +from openai._types import Response as SDKResponse + +import litellm from litellm.files import main as files_main from litellm.files.streaming import FileContentStreamingResponse from litellm.files.types import FileContentStreamingResult @@ -323,3 +330,80 @@ def test_file_content_streaming_passes_exception_to_context_manager_exit(): assert response_cm.exc_info[0] is RuntimeError assert response_cm.exc_info[1] is exc_info.value assert response_cm.exc_info[2] is not None + + +@pytest.fixture(params=("httpx", "sdk")) +def http_backend(request: pytest.FixtureRequest) -> ModuleType: + return importlib.import_module("httpx" if request.param == "httpx" else SDKResponse.__module__) + + +_TRANSPORT_FAILURES: Final = pytest.mark.parametrize( + ("failure_name", "public_error"), + ( + ("ReadTimeout", litellm.Timeout), + ("ReadError", litellm.APIConnectionError), + ("RemoteProtocolError", litellm.APIConnectionError), + ), +) + + +def _body_then_failure_transport(http_backend: ModuleType, failure: Exception) -> object: + class BodyThenFailure(http_backend.SyncByteStream, http_backend.AsyncByteStream): + def __iter__(self) -> Iterator[bytes]: + yield b"hello" + raise failure + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield b"hello" + raise failure + + return http_backend.MockTransport( + lambda _request: http_backend.Response( + 200, headers={"content-type": "application/octet-stream"}, stream=BodyThenFailure() + ) + ) + + +@_TRANSPORT_FAILURES +def test_file_content_stream_raises_litellm_errors_when_the_transport_fails_mid_body( + http_backend: ModuleType, failure_name: str, public_error: type[Exception] +): + failure: Final = getattr(http_backend, failure_name)("transport failed mid-body") + + with http_backend.Client(transport=_body_then_failure_transport(http_backend, failure)) as http_client: + result: Final = litellm.file_content( + file_id="file-abc123", + custom_llm_provider="openai", + client=OpenAI(api_key="transport-only", http_client=http_client), + stream=True, + chunk_size=len(b"hello"), + ) + stream: Final = cast(Iterator[bytes], result.stream_iterator) + assert next(stream) == b"hello" + with pytest.raises(public_error) as excinfo: + next(stream) + + assert excinfo.value.__cause__ is failure + + +@pytest.mark.asyncio +@_TRANSPORT_FAILURES +async def test_afile_content_stream_raises_litellm_errors_when_the_transport_fails_mid_body( + http_backend: ModuleType, failure_name: str, public_error: type[Exception] +): + failure: Final = getattr(http_backend, failure_name)("transport failed mid-body") + + async with http_backend.AsyncClient(transport=_body_then_failure_transport(http_backend, failure)) as http_client: + result: Final = await litellm.afile_content( + file_id="file-abc123", + custom_llm_provider="openai", + client=AsyncOpenAI(api_key="transport-only", http_client=http_client), + stream=True, + chunk_size=len(b"hello"), + ) + stream: Final = cast(AsyncIterator[bytes], result.stream_iterator) + assert await stream.__anext__() == b"hello" + with pytest.raises(public_error) as excinfo: + await stream.__anext__() + + assert excinfo.value.__cause__ is failure