diff --git a/tests/e2e/CLAUDE.md b/tests/e2e/CLAUDE.md index 14b2b4e3299..89c04208d65 100644 --- a/tests/e2e/CLAUDE.md +++ b/tests/e2e/CLAUDE.md @@ -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 ``, 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 `` 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 diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 21c5a338dc3..09d17b9a0db 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -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 diff --git a/tests/e2e/e2e_http.py b/tests/e2e/e2e_http.py index bc76eb3ea7a..9d5f1658e91 100644 --- a/tests/e2e/e2e_http.py +++ b/tests/e2e/e2e_http.py @@ -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="", - 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( diff --git a/tests/e2e/llm_translation/test_messages_e2e.py b/tests/e2e/llm_translation/test_messages_e2e.py index a5f36a8cbdd..c731b52acd8 100644 --- a/tests/e2e/llm_translation/test_messages_e2e.py +++ b/tests/e2e/llm_translation/test_messages_e2e.py @@ -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 diff --git a/tests/e2e/llm_translation/test_together_ai_e2e.py b/tests/e2e/llm_translation/test_together_ai_e2e.py index 2c8a7a3aa20..31e74c22e17 100644 --- a/tests/e2e/llm_translation/test_together_ai_e2e.py +++ b/tests/e2e/llm_translation/test_together_ai_e2e.py @@ -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" + ) diff --git a/tests/e2e/test_e2e_http.py b/tests/e2e/test_e2e_http.py index 007801a797a..66841725d1d 100644 --- a/tests/e2e/test_e2e_http.py +++ b/tests/e2e/test_e2e_http.py @@ -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"