From acc236e05a36ed55178bec9fc31281ce26f09a61 Mon Sep 17 00:00:00 2001 From: Yujong Lee Date: Wed, 2 Sep 2026 09:52:29 -0700 Subject: [PATCH] test(parity): strengthen core streaming contracts --- tests/route_parity/README.md | 24 +++ tests/route_parity/compare.py | 44 ++--- tests/route_parity/fixtures/test_recording.py | 158 ++++++++++++++++- tests/route_parity/inprocess.py | 21 ++- tests/route_parity/stream.py | 65 +++---- tests/route_parity/test_parity.py | 54 +++++- tests/route_parity/test_stream.py | 165 +++++++++++++++++- 7 files changed, 448 insertions(+), 83 deletions(-) diff --git a/tests/route_parity/README.md b/tests/route_parity/README.md index e8d5c62ff9b..468dc531a99 100644 --- a/tests/route_parity/README.md +++ b/tests/route_parity/README.md @@ -21,6 +21,30 @@ - Every test saves and restores the original bridge state - A small subprocess smoke test verifies environment-based startup configuration and detects fallback to the Python HTTP implementation +## Streaming execution + +The invocation callback passed to `run_in_process` must consume the stream before returning its `StreamOutcome`. +Use `consume_sync_stream` inside that callback, or await `consume_async_stream` inside the callback passed to +`run_in_process_async`. Provider requests are collected only after the callback completes. Streaming is explicit: +an iterable return value alone does not select stream consumption + +The consumers retain the wrapper type, iteration capabilities, chunk types and order, and any partial output before +an error. Errors retain their creation or iteration phase and the full public `SDKError` fields, with traceback text +removed. `capture_sync_stream` and `capture_async_stream` consume through the same helpers and then serialize the +outcome for subprocess reports. A serialization failure raises as a harness failure rather than becoming an SDK error + +Response models and stream chunks share a recursive comparator. It compares concrete model, container, and scalar +types, public fields and extras, and exact values while ignoring Pydantic private attributes at every nesting level. +An API may supply an explicit chunk normalizer for its public contract + +Shared tests exercise a local SSE provider through recording, VCR cassette storage, replay, and typed event comparison +in sync and async modes. They cover fragmented events, split UTF-8 characters, CRLF framing, coalesced events, and +application errors within a normally completed HTTP stream. HTTP byte boundaries and decoded SDK event boundaries +are checked separately + +OCR remains the only integrated LiteLLM route. These tests validate shared streaming machinery, not another route's +SDK parity. Connection interruption, early cancellation, and lifecycle timeout enforcement remain outside this coverage + ## Hypothesis and property-based testing - Hypothesis is a Python library for property-based testing diff --git a/tests/route_parity/compare.py b/tests/route_parity/compare.py index 6c0b81735be..56e539c1258 100644 --- a/tests/route_parity/compare.py +++ b/tests/route_parity/compare.py @@ -1,14 +1,12 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from typing import Final, TypeVar, cast +from typing import Final, cast from pydantic import BaseModel from tests.route_parity.models import CapturedRequest, Execution -ModelT = TypeVar("ModelT", bound=BaseModel) - def validate_harness(baseline: Execution, candidate: Execution, baseline_user_agent: str) -> None: for request in baseline.requests: @@ -29,31 +27,35 @@ def _request_after_transformation(request: CapturedRequest) -> CapturedRequest: def assert_request_parity(baseline: tuple[CapturedRequest, ...], candidate: tuple[CapturedRequest, ...]) -> None: baseline_requests: Final = tuple(_request_after_transformation(request) for request in baseline) candidate_requests: Final = tuple(_request_after_transformation(request) for request in candidate) - assert baseline_requests == candidate_requests + assert_value_parity(baseline_requests, candidate_requests) -def public_model_copy(model: ModelT) -> ModelT: - copied: Final = model.model_copy(deep=True) - object.__setattr__(copied, "__pydantic_private__", None) - return copied +def _public_model_values(model: BaseModel) -> dict[str, object]: + fields: Final = (*type(model).model_fields, *type(model).model_computed_fields) + extras: Final = cast(Mapping[str, object], model.model_extra or {}) + return { + **{name: cast(object, getattr(model, name)) for name in fields if not name.startswith("_")}, + **{name: value for name, value in extras.items() if not name.startswith("_")}, + } def assert_model_parity(baseline: BaseModel, candidate: BaseModel) -> None: - assert type(baseline) is type(candidate) - _assert_value_parity( - public_model_copy(baseline).model_dump(mode="python"), - public_model_copy(candidate).model_dump(mode="python"), - path="$", - ) + assert_value_parity(baseline, candidate) -def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> None: +def assert_value_parity(baseline: object, candidate: object, *, path: str = "$") -> None: + assert type(baseline) is type(candidate), f"type mismatch at {path}: {type(baseline)} != {type(candidate)}" + if isinstance(baseline, BaseModel) and isinstance(candidate, BaseModel): + assert_value_parity(_public_model_values(baseline), _public_model_values(candidate), path=path) + return if isinstance(baseline, Mapping) and isinstance(candidate, Mapping): baseline_mapping: Final = cast(Mapping[object, object], baseline) candidate_mapping: Final = cast(Mapping[object, object], candidate) - assert baseline_mapping.keys() == candidate_mapping.keys(), f"mapping keys differ at {path}" + assert frozenset((type(key), key) for key in baseline_mapping) == frozenset( + (type(key), key) for key in candidate_mapping + ), f"mapping keys differ at {path}" for key in baseline_mapping: - _assert_value_parity(baseline_mapping[key], candidate_mapping[key], path=f"{path}.{key}") + assert_value_parity(baseline_mapping[key], candidate_mapping[key], path=f"{path}.{key}") return if ( isinstance(baseline, Sequence) @@ -64,8 +66,10 @@ def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> N baseline_sequence: Final = cast(Sequence[object], baseline) candidate_sequence: Final = cast(Sequence[object], candidate) assert len(baseline_sequence) == len(candidate_sequence), f"sequence lengths differ at {path}" - for index, (baseline_item, candidate_item) in enumerate(zip(baseline_sequence, candidate_sequence)): - _assert_value_parity(baseline_item, candidate_item, path=f"{path}[{index}]") + for index, (baseline_item, candidate_item) in enumerate( + zip(baseline_sequence, candidate_sequence, strict=True) + ): + assert_value_parity(baseline_item, candidate_item, path=f"{path}[{index}]") return assert baseline == candidate, f"value mismatch at {path}: {baseline!r} != {candidate!r}" @@ -73,4 +77,4 @@ def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> N def assert_parity(baseline: Execution, candidate: Execution, baseline_user_agent: str) -> None: validate_harness(baseline, candidate, baseline_user_agent) assert_request_parity(baseline.requests, candidate.requests) - assert baseline.report == candidate.report + assert_value_parity(baseline.report, candidate.report) diff --git a/tests/route_parity/fixtures/test_recording.py b/tests/route_parity/fixtures/test_recording.py index 77e387e5fba..f02f84d579b 100644 --- a/tests/route_parity/fixtures/test_recording.py +++ b/tests/route_parity/fixtures/test_recording.py @@ -1,18 +1,22 @@ from __future__ import annotations +import asyncio +import queue import threading -from collections.abc import Callable, Generator +from collections.abc import AsyncIterator, Callable, Generator, Iterator from contextlib import contextmanager from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Final +from typing import Final, Literal import httpx import pytest from hypothesis import strategies as st +from openai._streaming import SSEDecoder from pydantic import BaseModel, ConfigDict +from tests.route_parity.compare import assert_request_parity from tests.route_parity.fixtures.pipeline import RecordingTarget, record_fixtures from tests.route_parity.fixtures.recording import ( UpstreamEndpoint, @@ -22,15 +26,25 @@ from tests.route_parity.fixtures.recording import ( from tests.route_parity.fixtures.store import ( FIXTURE_SCHEMA_VERSION, fixture_path, + load_fixture, recorded_fixtures, ) +from tests.route_parity.inprocess import InProcessExecution, run_in_process, run_in_process_async from tests.route_parity.recorded_http import ( HttpHeader, RecordedHttpStreamResponse, RecordedResponse, RecordedStreamChunk, ) -from tests.route_parity.replay import replay_server +from tests.route_parity.replay import ReplayServer, replay_server +from tests.route_parity.stream import ( + StreamCompleted, + StreamFailed, + StreamOutcome, + assert_stream_parity, + consume_async_stream, + consume_sync_stream, +) _SSE_CHUNKS: Final = ( b'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n', @@ -39,6 +53,75 @@ _SSE_CHUNKS: Final = ( ) +class _StreamEvent(BaseModel): + kind: Literal["delta", "done", "error"] + value: str + + +class _StreamApplicationError(Exception): + status_code: Final = 400 + code: Final = "invalid_input" + type: Final = "validation_error" + param: Final = "input" + model: Final = "fixture-model" + llm_provider: Final = "fixture-provider" + + +def _stream_event(data: str) -> _StreamEvent: + event: Final = _StreamEvent.model_validate_json(data) + if event.kind == "error": + raise _StreamApplicationError(event.value) + return event + + +def _event_chunks(failed: bool) -> tuple[bytes, ...]: + terminal: Final = ( + b'event: error\r\ndata: {"kind":"error","value":"invalid input"}\r\n\r\n' + if failed + else b'event: done\r\ndata: {"kind":"done","value":""}\r\n\r\n' + ) + return ( + b'event: delta\r\ndata: {"kind":"delta",\r\ndata: "value":"caf\xc3', + b'\xa9"}\r\n', + b'\r\nevent: delta\r\ndata: {"kind":"delta","value":"second"}\r\n\r\n' + terminal, + ) + + +def _sync_events(api_base: str, case_input: _FixtureInput) -> Iterator[_StreamEvent]: + with httpx.stream("POST", f"{api_base}/stream", json={"id": case_input.identifier}, timeout=5) as response: + response.raise_for_status() + for event in SSEDecoder().iter_bytes(response.iter_bytes()): + yield _stream_event(event.data) + + +async def _async_events(api_base: str, case_input: _FixtureInput) -> AsyncIterator[_StreamEvent]: + async with httpx.AsyncClient(timeout=5) as client: + async with client.stream("POST", f"{api_base}/stream", json={"id": case_input.identifier}) as response: + response.raise_for_status() + async for event in SSEDecoder().aiter_bytes(response.aiter_bytes()): + yield _stream_event(event.data) + + +async def _consume_async_events(api_base: str, case_input: _FixtureInput) -> StreamOutcome: + async def create() -> AsyncIterator[_StreamEvent]: + return _async_events(api_base, case_input) + + return await consume_async_stream(create) + + +async def _replay_events( + mode: Literal["sync", "async"], + provider: ReplayServer, + response: RecordedHttpStreamResponse, + case_input: _FixtureInput, +) -> InProcessExecution[StreamOutcome]: + if mode == "sync": + return run_in_process( + provider, (response,), lambda url: consume_sync_stream(lambda: _sync_events(url, case_input)) + ) + return await run_in_process_async(provider, (response,), lambda url: _consume_async_events(url, case_input)) + + class _FixtureInput(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") @@ -66,8 +149,9 @@ class _Invocation: class _ControlledUpstream(ThreadingHTTPServer): daemon_threads = True - def __init__(self) -> None: + def __init__(self, stream_chunks: tuple[bytes, ...]) -> None: super().__init__(("127.0.0.1", 0), _ControlledUpstreamHandler) + self.stream_chunks: Final = stream_chunks self.lock: Final = threading.Lock() self.two_requests_started: Final = threading.Event() self.active_requests: int = 0 @@ -116,14 +200,14 @@ class _ControlledUpstreamHandler(BaseHTTPRequestHandler): self.send_header("content-length", "0") self.end_headers() return - if self.path == "/v1/chat/completions": + if self.path in {"/v1/chat/completions", "/stream"}: with upstream.lock: upstream.request_count += 1 self.send_response(200) self.send_header("content-type", "text/event-stream") self.send_header("transfer-encoding", "chunked") self.end_headers() - for chunk in _SSE_CHUNKS: + for chunk in upstream.stream_chunks: self.wfile.write(f"{len(chunk):X}\r\n".encode("ascii")) self.wfile.write(chunk) self.wfile.write(b"\r\n") @@ -173,8 +257,8 @@ class _ControlledUpstreamHandler(BaseHTTPRequestHandler): @contextmanager -def _controlled_upstream() -> Generator[_ControlledUpstream]: - server: Final = _ControlledUpstream() +def _controlled_upstream(stream_chunks: tuple[bytes, ...] = _SSE_CHUNKS) -> Generator[_ControlledUpstream]: + server: Final = _ControlledUpstream(stream_chunks) thread: Final = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: @@ -430,3 +514,61 @@ def test_multiple_provider_calls_record_and_replay_in_order( requests: Final = provider.take_requests(2) assert len(requests) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +@pytest.mark.parametrize("failed", (False, True), ids=("completed", "application-error")) +async def test_typed_stream_recording_cassette_replay_parity( + tmp_path: Path, mode: Literal["sync", "async"], failed: bool +) -> None: + case_input: Final = _case("typed-stream") + outcomes: Final[queue.SimpleQueue[StreamOutcome]] = queue.SimpleQueue() + + def record(api_base: str, sdk_input: _FixtureInput) -> None: + outcome: Final = ( + consume_sync_stream(lambda: _sync_events(api_base, sdk_input)) + if mode == "sync" + else asyncio.run(_consume_async_events(api_base, sdk_input)) + ) + outcomes.put(outcome) + + with _controlled_upstream(_event_chunks(failed)) as upstream: + target: Final = RecordingTarget( + name="stream", + upstream=UpstreamEndpoint(upstream.url), + strategy=st.just(case_input), + invocation=_Invocation(record), + ) + summary: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase) + + assert summary.failed == () + assert len(summary.recorded) == 1 + recorded: Final = outcomes.get_nowait() + loaded: Final = load_fixture(tmp_path / "stream", case_input, _ParityCase) + assert loaded is not None + response: Final = loaded.provider_responses[0] + assert isinstance(response, RecordedHttpStreamResponse) + assert response.status_code == 200 + wire_bytes: Final = b"".join(chunk.data_bytes() for chunk in response.chunks) + assert wire_bytes == b"".join(_event_chunks(failed)) + coalesced: Final = response.model_copy(update={"chunks": (RecordedStreamChunk.from_bytes(wire_bytes),)}) + + with replay_server() as provider: + first: Final = await _replay_events(mode, provider, response, case_input) + second: Final = await _replay_events(mode, provider, coalesced, case_input) + assert_request_parity(first.requests, second.requests) + assert len(first.requests) == 1 + assert first.requests[0].body == {"id": case_input.identifier} + assert_stream_parity(recorded, first.response) + assert_stream_parity(first.response, second.response) + expected: Final = (_StreamEvent(kind="delta", value="café"), _StreamEvent(kind="delta", value="second")) + assert first.response.chunks == (expected if failed else (*expected, _StreamEvent(kind="done", value=""))) + if failed: + assert isinstance(first.response.terminal, StreamFailed) + assert first.response.terminal.phase == "iteration" + assert first.response.terminal.exception_type is _StreamApplicationError + assert first.response.terminal.error.code == "invalid_input" + assert first.response.terminal.error.message == "invalid input" + else: + assert first.response.terminal == StreamCompleted() diff --git a/tests/route_parity/inprocess.py b/tests/route_parity/inprocess.py index 51744ae4e94..6a11fd6a88d 100644 --- a/tests/route_parity/inprocess.py +++ b/tests/route_parity/inprocess.py @@ -1,8 +1,8 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import Generic, TypeVar +from typing import Final, Generic, TypeVar from tests.route_parity.models import CapturedRequest from tests.route_parity.recorded_http import RecordedResponse @@ -25,7 +25,22 @@ def run_in_process( for recorded_response in recorded_responses: provider.enqueue_response(recorded_response) try: - response = call(provider.url) + response: Final = call(provider.url) + return InProcessExecution(requests=provider.take_requests(len(recorded_responses)), response=response) + except Exception: + provider.reset() + raise + + +async def run_in_process_async( + provider: ReplayServer, + recorded_responses: tuple[RecordedResponse, ...], + call: Callable[[str], Awaitable[ResponseT]], +) -> InProcessExecution[ResponseT]: + for recorded_response in recorded_responses: + provider.enqueue_response(recorded_response) + try: + response: Final = await call(provider.url) return InProcessExecution(requests=provider.take_requests(len(recorded_responses)), response=response) except Exception: provider.reset() diff --git a/tests/route_parity/stream.py b/tests/route_parity/stream.py index a85ff2dbf67..e1bc7855f3d 100644 --- a/tests/route_parity/stream.py +++ b/tests/route_parity/stream.py @@ -4,11 +4,9 @@ from collections.abc import AsyncIterable, Awaitable, Callable, Iterable from dataclasses import dataclass from typing import Final, Literal, TypeAlias -from pydantic import BaseModel - -from tests.route_parity.compare import public_model_copy +from tests.route_parity.compare import assert_value_parity from tests.route_parity.models import ( - SDKChunk, + SDKError, SDKReport, SDKStreamCompleted, SDKStreamFailed, @@ -27,9 +25,7 @@ class StreamCompleted: class StreamFailed: phase: Literal["creation", "iteration"] exception_type: type[BaseException] - status_code: int | None - llm_provider: str | None - model: str | None + error: SDKError kind: Literal["failed"] = "failed" @@ -60,51 +56,34 @@ async def drain_async_stream(stream: AsyncIterable[object]) -> None: def capture_sync_stream(create: Callable[[], Iterable[object]]) -> SDKReport: - try: - stream: Final = create() - except Exception as error: - return sdk_error_report(error) - - chunks: list[SDKChunk] = [] # mutable-ok: partial chunks must survive an iteration failure - try: - for chunk in stream: - chunks.append(sdk_chunk(chunk)) - except Exception as error: - return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamFailed(error=sdk_error_report(error))) - return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamCompleted()) + return _stream_report(consume_sync_stream(create)) async def capture_async_stream(create: Callable[[], Awaitable[AsyncIterable[object]]]) -> SDKReport: - try: - stream: Final = await create() - except Exception as error: - return sdk_error_report(error) - - chunks: list[SDKChunk] = [] # mutable-ok: partial chunks must survive an iteration failure - try: - async for chunk in stream: - chunks.append(sdk_chunk(chunk)) - except Exception as error: - return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamFailed(error=sdk_error_report(error))) - return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamCompleted()) + return _stream_report(await consume_async_stream(create)) -def _failed(phase: Literal["creation", "iteration"], error: BaseException) -> StreamFailed: - status_code: Final = getattr(error, "status_code", None) - llm_provider: Final = getattr(error, "llm_provider", None) - model: Final = getattr(error, "model", None) +def _stream_report(outcome: StreamOutcome) -> SDKReport: + terminal: Final = outcome.terminal + if isinstance(terminal, StreamFailed) and terminal.phase == "creation": + return terminal.error + return SDKStreamReport( + chunks=tuple(sdk_chunk(chunk) for chunk in outcome.chunks), + terminal=SDKStreamFailed(error=terminal.error) if isinstance(terminal, StreamFailed) else SDKStreamCompleted(), + ) + + +def _failed(phase: Literal["creation", "iteration"], error: Exception) -> StreamFailed: return StreamFailed( phase=phase, exception_type=type(error), - status_code=status_code if isinstance(status_code, int) else None, - llm_provider=llm_provider if isinstance(llm_provider, str) else None, - model=model if isinstance(model, str) else None, + error=sdk_error_report(error), ) def consume_sync_stream(create: Callable[[], Iterable[object]]) -> StreamOutcome: try: - stream = create() + stream: Final = create() except Exception as error: return StreamOutcome( wrapper_type=None, @@ -142,7 +121,7 @@ def consume_sync_stream(create: Callable[[], Iterable[object]]) -> StreamOutcome async def consume_async_stream(create: Callable[[], Awaitable[AsyncIterable[object]]]) -> StreamOutcome: try: - stream = await create() + stream: Final = await create() except Exception as error: return StreamOutcome( wrapper_type=None, @@ -179,8 +158,6 @@ async def consume_async_stream(create: Callable[[], Awaitable[AsyncIterable[obje def normalize_chunk(chunk: object) -> object: - if isinstance(chunk, BaseModel): - return public_model_copy(chunk) return chunk @@ -195,6 +172,6 @@ def assert_stream_parity( assert baseline.supports_async_iteration is candidate.supports_async_iteration assert baseline.chunk_types == candidate.chunk_types assert len(baseline.chunks) == len(candidate.chunks) - for baseline_chunk, candidate_chunk in zip(baseline.chunks, candidate.chunks, strict=True): - assert normalize(baseline_chunk) == normalize(candidate_chunk) + for index, (baseline_chunk, candidate_chunk) in enumerate(zip(baseline.chunks, candidate.chunks, strict=True)): + assert_value_parity(normalize(baseline_chunk), normalize(candidate_chunk), path=f"$.chunks[{index}]") assert baseline.terminal == candidate.terminal diff --git a/tests/route_parity/test_parity.py b/tests/route_parity/test_parity.py index b95cccbb9e9..55ebdc654be 100644 --- a/tests/route_parity/test_parity.py +++ b/tests/route_parity/test_parity.py @@ -1,9 +1,10 @@ from __future__ import annotations +from types import MappingProxyType from typing import Final import pytest -from pydantic import BaseModel, JsonValue, PrivateAttr +from pydantic import BaseModel, ConfigDict, JsonValue, PrivateAttr from tests.route_parity.compare import assert_model_parity, assert_parity from tests.route_parity.models import CapturedRequest, Execution, SDKError, SDKSuccess, sdk_error_report @@ -27,6 +28,12 @@ class _FloatResponse(BaseModel): values: list[float] +class _PublicValue(BaseModel): + model_config = ConfigDict(extra="allow") + + value: object + + class _PublicError(ValueError): status_code: Final = 400 @@ -136,3 +143,48 @@ def test_model_parity_rejects_meaningful_float_difference() -> None: _FloatResponse(values=[0.22590550796036835]), _FloatResponse(values=[0.2259]), ) + + +@pytest.mark.parametrize( + ("baseline", "candidate"), + ( + (_ComparableResponse(value="same"), {"value": "same"}), + (_ComparableResponse(value="same"), _DifferentResponse(value="same")), + (True, 1), + (1, 1.0), + (["same"], ("same",)), + ({"value": "same"}, MappingProxyType({"value": "same"})), + ({True: "same"}, {1: "same"}), + ), + ids=("model-dict", "model-class", "bool-int", "int-float", "list-tuple", "mapping-class", "key-type"), +) +def test_model_parity_rejects_nested_type_changes(baseline: object, candidate: object) -> None: + with pytest.raises(AssertionError, match=r"\$\.value\[0\]"): + assert_model_parity(_PublicValue(value=[baseline]), _PublicValue(value=[candidate])) + + +def test_model_parity_ignores_nested_private_attributes() -> None: + baseline: Final = _ComparableResponse(value="same") + candidate: Final = _ComparableResponse(value="same") + baseline.set_hidden_param("request_id", "baseline") + candidate.set_hidden_param("request_id", "candidate") + + assert_model_parity(_PublicValue(value={"nested": [baseline]}), _PublicValue(value={"nested": [candidate]})) + + +@pytest.mark.parametrize("extras", ({"provider_value": "changed"}, {}, {"provider_value": {"value": "same"}})) +def test_model_parity_compares_public_extras(extras: dict[str, object]) -> None: + baseline: Final = _PublicValue.model_validate({"value": None, "provider_value": _ComparableResponse(value="same")}) + candidate: Final = _PublicValue.model_validate({"value": None, **extras}) + + with pytest.raises(AssertionError): + assert_model_parity(baseline, candidate) + + +def test_serialized_parity_rejects_boolean_integer_substitution() -> None: + with pytest.raises(AssertionError, match="type mismatch"): + assert_parity( + _execution(body={"enabled": True}, user_agent=SENTINEL), + _execution(body={"enabled": 1}, user_agent="candidate"), + SENTINEL, + ) diff --git a/tests/route_parity/test_stream.py b/tests/route_parity/test_stream.py index 503c4b3d93b..cf39d8f557c 100644 --- a/tests/route_parity/test_stream.py +++ b/tests/route_parity/test_stream.py @@ -2,14 +2,25 @@ from __future__ import annotations import queue from collections.abc import AsyncIterator, Iterator -from typing import Final +from typing import Final, Literal, NoReturn import pytest -from pydantic import BaseModel, PrivateAttr +from pydantic import BaseModel, PrivateAttr, ValidationError -from tests.route_parity.models import SDKBytesChunk, SDKJsonChunk, SDKStreamFailed, SDKStreamReport +from tests.route_parity.models import ( + SDKBytesChunk, + SDKError, + SDKJsonChunk, + SDKReport, + SDKStreamCompleted, + SDKStreamFailed, + SDKStreamReport, + sdk_error_report, +) from tests.route_parity.stream import ( + StreamCompleted, StreamFailed, + StreamOutcome, assert_stream_parity, capture_async_stream, capture_sync_stream, @@ -28,6 +39,10 @@ class _Chunk(BaseModel): self._hidden_params[key] = value +class _NestedChunk(BaseModel): + value: object + + class _SyncStream: def __init__(self, chunks: tuple[object, ...], error: BaseException | None = None) -> None: self.chunks: Final = chunks @@ -53,15 +68,26 @@ class _AsyncStream: class _PublicStreamError(Exception): def __init__( - self, message: str = "runtime-specific message", *, status_code: int, llm_provider: str, model: str + self, + message: str = "invalid input", + *, + status_code: int = 400, + llm_provider: str = "test", + model: str = "test-model", + code: str = "invalid_input", + error_type: str = "validation_error", + param: str = "input", ) -> None: super().__init__(message) self.status_code: Final = status_code self.llm_provider: Final = llm_provider self.model: Final = model + self.code: Final = code + self.type: Final = error_type + self.param: Final = param -def _creation_error() -> _SyncStream: +def _creation_error() -> NoReturn: raise _PublicStreamError(status_code=429, llm_provider="test", model="test-model") @@ -69,6 +95,22 @@ async def _async_stream(chunks: tuple[object, ...], error: BaseException | None return _AsyncStream(chunks, error) +async def _consume( + mode: Literal["sync", "async"], chunks: tuple[object, ...], error: Exception | None = None +) -> StreamOutcome: + if mode == "sync": + return consume_sync_stream(lambda: _SyncStream(chunks, error)) + return await consume_async_stream(lambda: _async_stream(chunks, error)) + + +async def _capture( + mode: Literal["sync", "async"], chunks: tuple[object, ...], error: Exception | None = None +) -> SDKReport: + if mode == "sync": + return capture_sync_stream(lambda: _SyncStream(chunks, error)) + return await capture_async_stream(lambda: _async_stream(chunks, error)) + + def test_sync_stream_parity_compares_chunks_and_ignores_private_attrs() -> None: python_chunk: Final = _Chunk(value="same") accelerated_chunk: Final = _Chunk(value="same") @@ -120,10 +162,10 @@ def test_stream_outcome_distinguishes_creation_and_iteration_errors() -> None: @pytest.mark.asyncio async def test_async_stream_uses_same_trace_contract() -> None: python_error: Final = _PublicStreamError( - "python runtime detail", status_code=500, llm_provider="test", model="test-model" + "invalid input\nTraceback (most recent call last):\npython detail", status_code=500 ) accelerated_error: Final = _PublicStreamError( - "rust runtime detail", status_code=500, llm_provider="test", model="test-model" + "invalid input\nTraceback (most recent call last):\nrust detail", status_code=500 ) python: Final = await consume_async_stream(lambda: _async_stream((_Chunk(value="same"),), python_error)) accelerated: Final = await consume_async_stream(lambda: _async_stream((_Chunk(value="same"),), accelerated_error)) @@ -191,3 +233,112 @@ async def test_capture_async_stream_serializes_message_bytes_in_order() -> None: b"first", b"second", ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +@pytest.mark.parametrize( + "candidate_error", + ( + ValueError("invalid input"), + _PublicStreamError("changed message"), + _PublicStreamError(status_code=429), + _PublicStreamError(code="changed_code"), + _PublicStreamError(error_type="changed_type"), + _PublicStreamError(param="changed_param"), + _PublicStreamError(model="changed_model"), + _PublicStreamError(llm_provider="changed_provider"), + ), + ids=("exception", "message", "status", "code", "type", "param", "model", "provider"), +) +async def test_stream_parity_rejects_public_error_changes( + mode: Literal["sync", "async"], candidate_error: Exception +) -> None: + chunks: Final = (_Chunk(value="partial"),) + baseline: Final = await _consume(mode, chunks, _PublicStreamError()) + candidate: Final = await _consume(mode, chunks, candidate_error) + + with pytest.raises(AssertionError): + assert_stream_parity(baseline, candidate) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +@pytest.mark.parametrize("candidate", (("one",), ("one", "two", "three"), ("two", "one"), ("one", "changed"))) +async def test_stream_parity_checks_event_sequence(mode: Literal["sync", "async"], candidate: tuple[str, ...]) -> None: + baseline: Final = await _consume(mode, (_Chunk(value="one"), _Chunk(value="two"))) + changed: Final = await _consume(mode, tuple(_Chunk(value=value) for value in candidate)) + + with pytest.raises(AssertionError): + assert_stream_parity(baseline, changed) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +async def test_stream_parity_preserves_nested_types_and_ignores_private_fields(mode: Literal["sync", "async"]) -> None: + first: Final = _Chunk(value="same") + second: Final = _Chunk(value="same") + first.set_hidden_param("request_id", "first") + second.set_hidden_param("request_id", "second") + baseline: Final = await _consume(mode, (_NestedChunk(value=[first]),)) + candidate: Final = await _consume(mode, (_NestedChunk(value=[second]),)) + + assert_stream_parity(baseline, candidate) + changed: Final = await _consume(mode, (_NestedChunk(value=[{"value": "same"}]),)) + with pytest.raises(AssertionError, match=r"\$\.chunks\[0\]\.value\[0\]"): + assert_stream_parity(baseline, changed) + + +def test_stream_parity_rejects_wrapper_and_chunk_type_changes() -> None: + baseline: Final = consume_sync_stream(lambda: _SyncStream((_Chunk(value="same"),))) + different_wrapper: Final = consume_sync_stream(lambda: iter((_Chunk(value="same"),))) + different_chunk: Final = consume_sync_stream(lambda: _SyncStream(({"value": "same"},))) + + with pytest.raises(AssertionError): + assert_stream_parity(baseline, different_wrapper) + with pytest.raises(AssertionError): + assert_stream_parity(baseline, different_chunk) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +@pytest.mark.parametrize("error", (None, _PublicStreamError())) +async def test_capture_keeps_serialization_failures_out_of_sdk_errors( + mode: Literal["sync", "async"], error: Exception | None +) -> None: + with pytest.raises(ValidationError): + await _capture(mode, (_Chunk(value="valid"), object()), error) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +async def test_empty_stream_completes(mode: Literal["sync", "async"]) -> None: + outcome: Final = await _consume(mode, ()) + assert outcome.chunks == () + assert outcome.terminal == StreamCompleted() + assert await _capture(mode, ()) == SDKStreamReport(chunks=(), terminal=SDKStreamCompleted()) + + +@pytest.mark.asyncio +async def test_async_creation_error_matches_sync_capture() -> None: + async def create() -> _AsyncStream: + _creation_error() + + sync: Final = consume_sync_stream(_creation_error) + asynchronous: Final = await consume_async_stream(create) + assert_stream_parity(sync, asynchronous) + assert isinstance(sync.terminal, StreamFailed) + assert sync.terminal.phase == "creation" + assert isinstance(capture_sync_stream(_creation_error), SDKError) + assert capture_sync_stream(_creation_error) == await capture_async_stream(create) == sync.terminal.error + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ("sync", "async")) +async def test_capture_preserves_partial_output_and_complete_error(mode: Literal["sync", "async"]) -> None: + error: Final = _PublicStreamError() + report: Final = await _capture(mode, (_Chunk(value="partial"),), error) + assert report == SDKStreamReport( + chunks=(SDKJsonChunk(value={"value": "partial"}),), + terminal=SDKStreamFailed(error=sdk_error_report(error)), + )