scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 85

Merge pull request #39953 from BerriAI/litellm_/litellm-e2e-flaky-test-2159ae

Browse source 
test(e2e): judge /v1/messages streaming on the clock, not on the provider's delta count
This commit is contained in:
yuneng-jiang 2026-09-05 16:04:45 -07:00 • committed by GitHub
commit 1b25132863
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 167 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/e2e/CLAUDE.md
View file

@ -77,7 +77,7 @@ Mark live tests with `@pytest.mark.e2e` (on the class or the module). Pure cover
The seam is `provider_edge.py`: `start_provider_edge` boots an in-process HTTP server (one shared instance per pytest process, `e2e_config.provider_edge_base` is the accessor) that mounts each supported provider under a path prefix (`EDGE_MOUNTS`: `/openai` -> `https://api.openai.com`, `/anthropic` -> `https://api.anthropic.com`). A test participates by registering its deployment with `api_base=provider_edge_base("openai")` plus the provider's path suffix; `quota_management/spend_tracking/test_provider_edge_spend_e2e.py` is the reference. In live mode the accessor returns None and the deployment defaults to the real provider, so an edge-wired test runs in all three modes unchanged. Non-wired tests hit their providers live in every mode. The edge binds `E2E_PROVIDER_EDGE_BIND_HOST` (default 127.0.0.1) and advertises `E2E_PROVIDER_EDGE_ADVERTISE_HOST` in the api_base it hands out, for proxies running in containers
A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per provider call in call order (`0000-post-openai-v1-chat-completions.json`). Request headers are never stored (provider credentials never touch disk), non-JSON request bodies store a canonicalized sha256 digest instead of the bytes, `multipart/form-data` bodies store their ordinary fields plus a JSON list of the uploaded parts' `[field, filename, content-type]` triples and a digest of their content, so the per-request random boundary and the envelope never reach the key, and responses store status, filtered headers, and the verbatim body base64-encoded, which is part of why bundles are gitignored. Responses come in two shapes told apart by a `kind` tag: an ordinary one holding a single base64 body, and, for a response the provider streamed (`content-type: text/event-stream`), one holding its transfer chunks in order plus why the stream ended early if it did, so replay reproduces the split points the provider chose instead of one coalesced body. `fixture_bundle.py` owns the format, and `BUNDLE_FORMAT_VERSION` is checked on load, so a bundle recorded under older rules is refused by name rather than partially read. Record serves the proxy the same filtered stored response replay will serve later, chunk for chunk on a stream, so the two modes are byte-identical from the proxy's side of the socket
A bundle (default `tests/e2e/.fixtures`, override with `E2E_FIXTURE_DIR`) is a directory: `manifest.json` carries the record timestamp, harness git version, and format version, and each test gets a subdirectory holding one JSON file per provider call in call order (`0000-post-openai-v1-chat-completions.json`). Request headers are never stored (provider credentials never touch disk), non-JSON request bodies store a canonicalized sha256 digest instead of the bytes, `multipart/form-data` bodies store their ordinary fields plus a JSON list of the uploaded parts' `[field, filename, content-type]` triples and a digest of their content, so the per-request random boundary and the envelope never reach the key, and responses store status, filtered headers, and the verbatim body base64-encoded, which is part of why bundles are gitignored. Responses come in two shapes told apart by a `kind` tag: an ordinary one holding a single base64 body, and, for a response the provider streamed (`content-type: text/event-stream`), one holding its transfer chunks in order plus why the stream ended early if it did, so replay reproduces the split points the provider chose instead of one coalesced body. `fixture_bundle.py` owns the format, and `BUNDLE_FORMAT_VERSION` is checked on load, so a bundle recorded under older rules is refused by name rather than partially read. Record serves the proxy the same filtered stored response replay will serve later, chunk for chunk on a stream, so the two modes are byte-identical from the proxy's side of the socket. Replay does not reproduce the provider's inter-chunk timing (chunks go out as fast as the socket takes them), so a test that judges streaming on the clock, such as the `stream_event_arrivals` lead between the first content delta and `message_stop`, gates that assertion on `provider_paces_stream()` and proves only the event grammar in replay
Multipart identity is the fiddly corner, and the rules exist because each one had a collision behind it. A part counts as an upload when it carries a filename or declares its own content type, and everything else is an ordinary field. Field names get a `name[n]` suffix on repeats, with a literal `[` doubled first, so a form that repeats `purpose` never keys the same as one that literally sends `purpose[1]`. A field whose name reads as a credential is stored as `<secret>`, which stays key-preserving because the key is recomputed from the stored request rather than saved alongside it, so the live request carrying the real value still matches its redacted fixture. A field value that is not UTF-8 is stored as a base64 sha256 digest, base64 and not hex because the canonicalizer rewrites any 64-character hex run to `<sha256>` and would fold every binary value onto one key. The uploaded parts contribute a JSON list rather than a `field:filename` string, so a separator inside a filename cannot impersonate a field boundary, and their byte length is stored for a reader's benefit but deliberately left out of the key, since the canonicalizer absorbs timestamp and id drift inside a file that changes its length

8
tests/e2e/e2e_config.py
View file

@ -10,6 +10,7 @@ import os
import time
import uuid
from pathlib import Path
from typing import Final
from dotenv import load_dotenv
@ -192,6 +193,13 @@ def provider_edge_base(mount: str) -> str | None:
)
STREAM_MIN_LEAD_SECONDS: Final = 1.0
def provider_paces_stream() -> bool:
return parse_fixture_mode(FIXTURE_MODE_RAW) != "replay"
def unique_marker() -> str:
"""A short unique token per call/run, so concurrent runs and the shared
response cache never collide on prompts, tags, or customer ids. In record

92
tests/e2e/e2e_http.py
View file

@ -16,9 +16,9 @@ requests itself imports.
from __future__ import annotations
import time
from collections.abc import Callable
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Generator, Generic, Iterator, Literal, NewType, Protocol, TypeVar, cast
from typing import Final, Generator, Generic, Iterator, Literal, NewType, Protocol, TypeVar, cast
import pytest
import requests
@ -142,6 +142,7 @@ class StreamingResponse(BaseModel):
body: str
chunks: int = 0 # streamed events (0 for non-streaming)
stream_events: list[str] = []
stream_event_arrivals: list[float] = []
# First in-stream error event, if any. A streamed call commits its HTTP 200
# before the upstream completes, so upstream failures (e.g. insufficient
# quota) arrive as SSE error events inside an otherwise-successful response;
@ -184,7 +185,20 @@ class BinaryStream(BaseModel):
return "chunked" in (self.transfer_encoding or "")
def _hdr(resp: requests.Response, name: str) -> str | None:
class SseResponse(Protocol):
@property
def status_code(self) -> int: ...
@property
def headers(self) -> Mapping[str, str]: ...
@property
def text(self) -> str: ...
def iter_lines(self) -> Iterator[bytes]: ...
def _hdr(resp: SseResponse, name: str) -> str | None:
value = resp.headers.get(name)
return value if isinstance(value, str) else None
@ -457,7 +471,7 @@ def probe(
return ProbeResult(status_code=resp.status_code, body=resp.text)
def _parse_response_cost(resp: requests.Response) -> float | None:
def _parse_response_cost(resp: SseResponse) -> float | None:
raw = _hdr(resp, "x-litellm-response-cost")
if raw is None or raw == "":
return None
@ -467,11 +481,26 @@ def _parse_response_cost(resp: requests.Response) -> float | None:
return None
def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingResponse:
call_id = _hdr(resp, "x-litellm-call-id")
response_cost = _parse_response_cost(resp)
content_type = _hdr(resp, "content-type")
headers = {name.lower(): value for name, value in resp.headers.items()}
_SSE_DATA_PREFIX: Final = b"data: "
_SSE_DONE: Final = "[DONE]"
def _is_stream_error_line(line: bytes) -> bool:
return (
line.startswith(b"event: error")
or b'"type":"error"' in line
or b'"type": "error"' in line
or line.startswith(b'data: {"error"')
)
def streaming_outcome(
resp: SseResponse, stream: bool, *, sent_at: float, clock: Callable[[], float] = time.monotonic
) -> StreamingResponse:
call_id: Final = _hdr(resp, "x-litellm-call-id")
response_cost: Final = _parse_response_cost(resp)
content_type: Final = _hdr(resp, "content-type")
headers: Final = {name.lower(): value for name, value in resp.headers.items()}
if not stream or not (200 <= resp.status_code < 300):
return StreamingResponse(
status_code=resp.status_code,
@ -481,29 +510,13 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon
headers=headers,
body=resp.text,
)
lines = cast("Iterator[bytes]", resp.iter_lines())
chunks = 0
stream_error: str | None = None
stream_events: list[str] = []
stream_done = False
for line in lines:
if not line:
continue
chunks += 1
decoded_line = line.decode(errors="replace")
if decoded_line.startswith("data: "):
payload = decoded_line.removeprefix("data: ")
if payload == "[DONE]":
stream_done = True
else:
stream_events.append(payload)
if stream_error is None and (
line.startswith(b"event: error")
or b'"type":"error"' in line
or b'"type": "error"' in line
or line.startswith(b'data: {"error"')
):
stream_error = line.decode(errors="replace")[:300]
stamped: Final = tuple((line, clock() - sent_at) for line in resp.iter_lines() if line)
payloads: Final = tuple(
(line.removeprefix(_SSE_DATA_PREFIX).decode(errors="replace"), arrived)
for line, arrived in stamped
if line.startswith(_SSE_DATA_PREFIX)
)
events: Final = tuple((payload, arrived) for payload, arrived in payloads if payload != _SSE_DONE)
return StreamingResponse(
status_code=resp.status_code,
call_id=call_id,
@ -511,10 +524,14 @@ def _streaming_outcome(resp: requests.Response, stream: bool) -> StreamingRespon
content_type=content_type,
headers=headers,
body="<streamed>",
chunks=chunks,
stream_events=stream_events,
stream_done=stream_done,
stream_error=stream_error,
chunks=len(stamped),
stream_events=[payload for payload, _ in events],
stream_event_arrivals=[arrived for _, arrived in events],
stream_done=any(payload == _SSE_DONE for payload, _ in payloads),
stream_error=next(
(line.decode(errors="replace")[:300] for line, _ in stamped if _is_stream_error_line(line)),
None,
),
)
@ -531,6 +548,7 @@ def send(
x-litellm-call-id header. For native/passthrough bodies and for calls judged by
status rather than a typed JSON model (e.g. a budget block is a non-2xx). With
``stream=True`` the SSE body is consumed and its events counted instead."""
sent_at: Final = time.monotonic()
try:
resp = request_with_retry(
lambda: requests.post(
@ -544,7 +562,7 @@ def send(
)
except requests.RequestException as exc:
return StreamingResponse(status_code=-1, body=str(exc))
return _streaming_outcome(resp, stream)
return streaming_outcome(resp, stream, sent_at=sent_at)
def stream(

37
tests/e2e/llm_translation/test_messages_e2e.py
View file

@ -8,8 +8,15 @@ litellm-regression-tests/tests/test_inference_endpoints.py.
from __future__ import annotations
from typing import Final
import pytest
from e2e_config import provider_edge_base, unique_marker
from e2e_config import (
STREAM_MIN_LEAD_SECONDS,
provider_edge_base,
provider_paces_stream,
unique_marker,
)
from e2e_http import assert_client_error, require_successful_call, unwrap
from endpoints_client import EndpointsClient, MessagesResult
from lifecycle import ResourceManager
@ -159,19 +166,21 @@ class TestAnthropicMessages:
"""Edge-wired like its non-streaming siblings, so record and replay both
carry the streamed response.
Asserts the shape of the event sequence, not just that deltas and a stop
appeared somewhere in it: the answer arrives across several deltas, and the
usage event sits between the last of them and ``message_stop``. A replay that
coalesced the response into one buffered body could not satisfy either."""
Asserts what the proxy controls: the event grammar (usage between the last
content delta and ``message_stop``) and, on the clock, that the relay is
incremental. How many deltas a reply is split into is the provider's choice, so
the first content delta must instead reach the client well before
``message_stop``, which a buffered response cannot do. Replay serves chunks back
to back, so only live and record runs judge the timing."""
model, key = self._register(endpoints_client, resources)
result = endpoints_client.proxy.messages_stream(
key,
AnthropicMessagesBody(
model=model,
max_tokens=400,
max_tokens=800,
stream=True,
messages=[ChatMessage(role="user", content="Count from 1 to 100, one number per line.")],
messages=[ChatMessage(role="user", content="Count from 1 to 200, one number per line.")],
),
)
require_successful_call(result)
@ -186,10 +195,7 @@ class TestAnthropicMessages:
delta_positions = [
index for index, event in enumerate(events) if event.type == "content_block_delta"
]
assert len(delta_positions) >= 2, (
f"stream carried {len(delta_positions)} content deltas, so it was not "
f"incremental: {types}"
)
assert delta_positions, f"stream carried no content deltas: {types}"
text = "".join(
event.delta.text
for event in events
@ -209,6 +215,15 @@ class TestAnthropicMessages:
f"usage did not land between the last content delta and message_stop: {types}"
)
first_delta_at: Final = result.stream_event_arrivals[delta_positions[0]]
stop_at: Final = result.stream_event_arrivals[stop_position]
if provider_paces_stream():
assert stop_at - first_delta_at >= STREAM_MIN_LEAD_SECONDS, (
f"first content delta reached the client {first_delta_at:.2f}s after the request "
f"and message_stop {stop_at:.2f}s after it; a relayed stream shows the first delta "
f"at least {STREAM_MIN_LEAD_SECONDS}s before the end, so the response was buffered"
)
@pytest.mark.covers("llm.messages.anthropic.tool_use.nonstream.works")
def test_messages_tool_use(
self, endpoints_client: EndpointsClient, resources: ResourceManager

29
tests/e2e/llm_translation/test_together_ai_e2e.py
View file

@ -23,7 +23,7 @@ from datetime import date
from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_config import STREAM_MIN_LEAD_SECONDS, provider_paces_stream, unique_marker
from e2e_http import StreamingResponse, require_successful_call, unwrap
from lifecycle import ResourceManager
from models import (
@ -81,7 +81,7 @@ PERSON_RESPONSE_FORMAT: dict[str, object] = {
}
WEATHER_PROMPT = "What is the weather in Paris? Use the tool."
WEATHER_REPORT = "Paris: 22 degrees Celsius, clear skies, wind from the northwest at 9 km/h"
COUNTING_PROMPT = "Count from 1 to 20, one number per line."
COUNTING_PROMPT = "Count from 1 to 200, one number per line."
WEATHER_TOOL = ChatTool(
function=ChatToolFunction(
@ -753,7 +753,7 @@ class TestTogetherMessages:
key,
AnthropicMessagesBody(
model=model,
max_tokens=512,
max_tokens=2048,
stream=True,
messages=[ChatMessage(role="user", content=COUNTING_PROMPT)],
),
@ -763,11 +763,24 @@ class TestTogetherMessages:
assert not result.stream_error, f"stream errored: {result.stream_error}"
events = [_MessagesStreamEvent.model_validate_json(event) for event in result.stream_events]
types = [event.type for event in events]
text_deltas = [
delta_positions = [
index for index, event in enumerate(events) if event.type == "content_block_delta"
]
assert delta_positions, f"stream carried no content deltas: {types}"
text = "".join(
event.delta.text
for event in events
if event.type == "content_block_delta" and event.delta is not None and event.delta.text
]
assert len(text_deltas) >= 2, f"stream was not incremental: {types}"
assert "20" in "".join(text_deltas), f"streamed text lost the answer: {text_deltas}"
if event.type == "content_block_delta" and event.delta is not None
)
assert "200" in text, f"streamed text lost the answer: {text[:300]!r}"
assert "message_stop" in types, f"stream never reached message_stop: {types}"
stop_position: Final = types.index("message_stop")
first_delta_at: Final = result.stream_event_arrivals[delta_positions[0]]
stop_at: Final = result.stream_event_arrivals[stop_position]
if provider_paces_stream():
assert stop_at - first_delta_at >= STREAM_MIN_LEAD_SECONDS, (
f"first content delta reached the client {first_delta_at:.2f}s after the request "
f"and message_stop {stop_at:.2f}s after it; a relayed stream shows the first delta "
f"at least {STREAM_MIN_LEAD_SECONDS}s before the end, so the response was buffered"
)

58
tests/e2e/test_e2e_http.py
View file

@ -12,12 +12,14 @@ monkeypatches anything.
from __future__ import annotations
from collections.abc import Callable, Sequence
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Final
import pytest
from e2e_http import RETRY_ATTEMPTS, TRANSIENT_STATUSES, request_with_retry
from e2e_http import RETRY_ATTEMPTS, TRANSIENT_STATUSES, request_with_retry, streaming_outcome
@dataclass
@ -80,3 +82,55 @@ class TestTransientRetryPolicy:
assert result is responses[RETRY_ATTEMPTS - 1]
assert sleep.delays == [0.5, 1.0]
assert [r.close_calls for r in responses] == [1, 1, 0, 0]
@dataclass(frozen=True, slots=True)
class FakeSseResponse:
lines: Sequence[bytes]
status_code: int = 200
headers: Mapping[str, str] = MappingProxyType({"content-type": "text/event-stream"})
text: str = ""
def iter_lines(self) -> Iterator[bytes]:
return iter(self.lines)
def _ticking_clock(start: float, step: float) -> Callable[[], float]:
ticks: Final = iter(range(10_000))
return lambda: start + step * next(ticks)
class TestStreamEventArrivals:
def test_each_event_is_stamped_at_the_moment_its_line_arrives(self) -> None:
resp: Final = FakeSseResponse(
lines=(
b"event: message_start",
b'data: {"type":"message_start"}',
b"",
b"event: ping",
b'data: {"type":"ping"}',
b"event: content_block_delta",
b'data: {"type":"content_block_delta"}',
b"data: [DONE]",
)
)
result: Final = streaming_outcome(resp, True, sent_at=100.0, clock=_ticking_clock(start=100.0, step=0.5))
assert result.stream_events == [
'{"type":"message_start"}',
'{"type":"ping"}',
'{"type":"content_block_delta"}',
]
assert result.stream_event_arrivals == [0.5, 1.5, 2.5]
assert result.stream_done
assert result.chunks == 7
def test_a_non_streaming_outcome_carries_no_arrivals(self) -> None:
resp: Final = FakeSseResponse(lines=(), status_code=400, text="bad request")
result: Final = streaming_outcome(resp, True, sent_at=0.0, clock=_ticking_clock(start=0.0, step=1.0))
assert result.stream_events == []
assert result.stream_event_arrivals == []
assert result.body == "bad request"