scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-07 07:58:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 150

test(streaming): cover SDK 3 transport errors on finished streams and file downloads

Browse source 
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
This commit is contained in:
mateo-berri 2026-10-20 19:22:23 -07:00
parent e333ccabdb
commit 2d862de42d
3 changed files with 293 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

127
tests/unit/litellm_core_utils/test_streaming_handler.py
View file

@ -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

81
tests/unit/llms/openai/test_openai.py
View file

@ -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()

88
tests/unit/llms/openai/test_openai_file_content_streaming.py
View file

@ -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