From 9e9c2bde576b8d446f3c5061041976c3ac2554df Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 7 Jun 2026 12:58:54 +0200 Subject: [PATCH] fix: buffer-based SSE parser for all supported providers Replace the fragile `re.split(b"data: ")` streaming parser with a buffered, event-delimited parser in both the chat-completion and Responses-API streamers. Events are accumulated until the SSE blank-line delimiter, so parsing is independent of network chunk boundaries. Fixes: - OpenRouter `: OPENROUTER PROCESSING` keepalive comments no longer leak to clients as `data: : ...` (the `Unexpected token ':'` client crash). - JSON payloads split across TCP reads are reassembled before parsing. - CRLF framing (Gemini native alt=sse) handled. - Combined content+usage chunks (Gemini thinking models over the OpenAI-compat endpoint) forward content once and still report usage in the cost trailer, instead of dropping the assistant message. - Multi-line non-JSON `data` blocks are re-prefixed per line so they stay valid SSE framing for the client. Adds tests/unit/test_streaming_sse_providers.py driving the real generator against per-provider on-the-wire framing, and switches the integration token-mint fallback to secrets.token_hex to kill a PRNG/clock collision flake. --- routstr/upstream/base.py | 316 +++++++++++++------ tests/integration/conftest.py | 11 +- tests/unit/test_streaming_sse_providers.py | 339 +++++++++++++++++++++ 3 files changed, 566 insertions(+), 100 deletions(-) create mode 100644 tests/unit/test_streaming_sse_providers.py diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index bc267dc2..d440509c 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -2,10 +2,9 @@ from __future__ import annotations import asyncio import json -import re import traceback import uuid -from collections.abc import AsyncGenerator, AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator, Iterator from typing import Any, Mapping, cast import httpx @@ -728,56 +727,140 @@ class BaseUpstreamProvider: except Exception: pass + def _process_event(raw_event: bytes) -> Iterator[bytes]: + """Process one complete SSE event block (lines up to a blank line). + + Handles arbitrary upstream framing across every supported + provider: + + * ``data:`` lines are gathered and concatenated per the SSE + spec, so a payload split across network chunks is reassembled + before parsing. + * Comment/keepalive lines (those beginning with ``:`` such as + OpenRouter's ``: OPENROUTER PROCESSING``) are dropped. They + carry no JSON and forwarding them downstream breaks naive SSE + clients; the keepalive only matters for the upstream hop. + * Other SSE fields (``event:``/``id:``/``retry:``) are preserved + and kept attached to the event's ``data:`` line, which the + OpenAI Responses API and Anthropic-style streams rely on. + * ``[DONE]`` is swallowed so it can be re-emitted exactly once at + end of stream. + """ + nonlocal last_model_seen, usage_chunk_data, done_seen + + event = raw_event.strip(b"\r\n") + if not event: + return + + field_lines: list[bytes] = [] + data_lines: list[bytes] = [] + for line in event.split(b"\n"): + line = line.rstrip(b"\r") + if line.startswith(b"data:"): + # Strip the field name and a single optional leading space. + data_lines.append(line[len(b"data:") :].lstrip(b" ")) + elif line.startswith(b":"): + # SSE comment / keepalive - drop. + continue + elif line: + # Other SSE field (event:/id:/retry:) - preserve in order. + field_lines.append(line) + + if not data_lines: + return + + data = b"\n".join(data_lines) + if not data.strip(): + return + + # Re-emit preserved SSE fields immediately before the data line so + # event/data framing stays intact (single trailing newline each; + # the blank-line terminator is appended to the data line below). + prefix = b"".join(fl + b"\n" for fl in field_lines) + + if data.strip() == b"[DONE]": + done_seen = True + return + + try: + obj = json.loads(data) + except Exception: + obj = None + + if isinstance(obj, dict): + self._apply_provider_field(obj) + if obj.get("model"): + last_model_seen = str(obj.get("model")) + if requested_model: + obj["model"] = requested_model + if ( + "id" not in obj + or not isinstance(obj["id"], str) + or obj["id"] == "existing-id" + ): + if not hasattr(self, "_current_stream_id"): + self._current_stream_id = f"chatcmpl-{uuid.uuid4()}" + obj["id"] = self._current_stream_id + if isinstance(obj.get("usage"), dict): + # Capture usage for end-of-stream cost reconciliation. + # Some models (e.g. Gemini thinking models over the + # OpenAI-compat endpoint) attach ``usage`` to the SAME + # chunk that carries the final content/finish_reason + # rather than sending a separate ``choices: []`` usage + # chunk. Only swallow the chunk when it is a pure usage + # chunk (no choices); otherwise the content would be + # silently dropped and the client would receive no + # assistant message at all. + if obj.get("choices"): + # Capture usage (with model) for the cost trailer, + # but with choices stripped so the trailer never + # re-emits this chunk's content. + usage_chunk_data = { + k: v for k, v in obj.items() if k != "choices" + } + usage_chunk_data["choices"] = [] + # Forward the content now, without usage, so token + # usage is reported exactly once (in the trailer). + forward = {k: v for k, v in obj.items() if k != "usage"} + yield ( + prefix + + b"data: " + + json.dumps(forward).encode() + + b"\n\n" + ) + return + usage_chunk_data = obj + return + yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n" + else: + # Non-JSON data payload (partial fragment already reassembled + # by buffering, or a provider control string). Re-prefix each + # line so multi-line ``data`` stays valid SSE framing - a bare + # second line would otherwise reach the client without its + # ``data:`` field and break naive parsers. + body = b"".join( + b"data: " + ln + b"\n" for ln in data.split(b"\n") + ) + yield prefix + body + b"\n" + try: + # Buffer bytes across network chunks and dispatch only on the SSE + # event delimiter (a blank line). ``aiter_bytes`` yields arbitrary + # byte boundaries, so a single event's JSON can span chunks and + # multiple events can arrive together; buffering makes parsing + # boundary-independent for every provider. + buffer = b"" async for chunk in response.aiter_bytes(): - # Split chunk into SSE events - parts = re.split(b"data: ", chunk) - for i, part in enumerate(parts): - if not part: - continue + buffer += chunk.replace(b"\r\n", b"\n") + while b"\n\n" in buffer: + raw_event, buffer = buffer.split(b"\n\n", 1) + for out in _process_event(raw_event): + yield out - stripped_part = part.strip() - if not stripped_part: - continue - - if stripped_part == b"[DONE]": - done_seen = True - continue - - try: - # Only parse if it looks like a JSON object to avoid SSE control messages or partials - if part.strip().startswith(b"{") and part.strip().endswith( - b"}" - ): - obj = json.loads(part) - if isinstance(obj, dict): - self._apply_provider_field(obj) - if obj.get("model"): - last_model_seen = str(obj.get("model")) - if requested_model: - obj["model"] = requested_model - if ( - "id" not in obj - or not isinstance(obj["id"], str) - or obj["id"] == "existing-id" - ): - if not hasattr(self, "_current_stream_id"): - self._current_stream_id = ( - f"chatcmpl-{uuid.uuid4()}" - ) - obj["id"] = self._current_stream_id - if isinstance(obj.get("usage"), dict): - usage_chunk_data = obj - continue - yield b"data: " + json.dumps(obj).encode() + b"\n\n" - continue - except Exception: - pass - - prefix = ( - b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b"" - ) - yield prefix + part + # Flush any trailing event that lacked a final blank line. + if buffer.strip(): + for out in _process_event(buffer): + yield out async with create_session() as session: fresh_key = await session.get(key.__class__, key.hashed_key) @@ -1079,56 +1162,97 @@ class BaseUpstreamProvider: except Exception: pass + def _process_event(raw_event: bytes) -> Iterator[bytes]: + """Process one complete SSE event block for the Responses API. + + Buffers full events (delimited by a blank line) so parsing is + boundary-independent, gathers ``data:`` lines, drops comment/ + keepalive lines (e.g. OpenRouter's ``: OPENROUTER PROCESSING``), + and preserves ``event:``/``id:`` fields attached to their data + line so Responses API event framing stays intact. + """ + nonlocal last_model_seen, usage_chunk_data, done_seen + nonlocal reasoning_tokens + + event = raw_event.strip(b"\r\n") + if not event: + return + + field_lines: list[bytes] = [] + data_lines: list[bytes] = [] + for line in event.split(b"\n"): + line = line.rstrip(b"\r") + if line.startswith(b"data:"): + data_lines.append(line[len(b"data:") :].lstrip(b" ")) + elif line.startswith(b":"): + # SSE comment / keepalive - drop. + continue + elif line: + # Preserve event:/id:/retry: (Responses API event names). + field_lines.append(line) + + if not data_lines: + return + + data = b"\n".join(data_lines) + if not data.strip(): + return + + prefix = b"".join(fl + b"\n" for fl in field_lines) + + if data.strip() == b"[DONE]": + done_seen = True + return + + try: + obj = json.loads(data) + except json.JSONDecodeError: + obj = None + + if isinstance(obj, dict): + self._apply_provider_field(obj) + if obj.get("model"): + last_model_seen = str(obj.get("model")) + if requested_model: + obj["model"] = requested_model + + # Track reasoning tokens for Responses API + if usage := obj.get("usage", {}): + if isinstance(usage, dict) and "reasoning_tokens" in usage: + reasoning_tokens += usage.get("reasoning_tokens", 0) + + # Responses API usage is in response.completed/incomplete events + chunk_type = obj.get("type", "") + if chunk_type in ( + "response.completed", + "response.incomplete", + ): + usage_chunk_data = obj + return + + yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n" + else: + # Re-prefix each line so multi-line ``data`` stays valid SSE + # framing for the client. + body = b"".join( + b"data: " + ln + b"\n" for ln in data.split(b"\n") + ) + yield prefix + body + b"\n" + try: + # Buffer across network chunks; dispatch only on the SSE event + # delimiter so parsing is independent of byte boundaries. + buffer = b"" async for chunk in response.aiter_bytes(): - # Split chunk into SSE events - parts = re.split(b"data: ", chunk) - for i, part in enumerate(parts): - if not part: - continue + buffer += chunk.replace(b"\r\n", b"\n") + while b"\n\n" in buffer: + raw_event, buffer = buffer.split(b"\n\n", 1) + for out in _process_event(raw_event): + yield out - stripped_part = part.strip() - if not stripped_part: - continue - - if stripped_part == b"[DONE]": - done_seen = True - continue - - try: - obj = json.loads(part) - if isinstance(obj, dict): - self._apply_provider_field(obj) - if obj.get("model"): - last_model_seen = str(obj.get("model")) - if requested_model: - obj["model"] = requested_model - - # Track reasoning tokens for Responses API - if usage := obj.get("usage", {}): - if ( - isinstance(usage, dict) - and "reasoning_tokens" in usage - ): - reasoning_tokens += usage.get( - "reasoning_tokens", 0 - ) - - # Responses API usage is in response.completed/incomplete events - chunk_type = obj.get("type", "") - if chunk_type in ( - "response.completed", - "response.incomplete", - ): - usage_chunk_data = obj - continue - except json.JSONDecodeError: - pass - - prefix = ( - b"data: " if (i > 0 or chunk.startswith(b"data: ")) else b"" - ) - yield prefix + part + if buffer.strip(): + for out in _process_event(buffer): + yield out # Always emit a cost-bearing data chunk async with create_session() as session: diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 40ab9d53..b9d1c25c 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -174,17 +174,20 @@ class TestmintWallet: """Fallback method to create a basic test token""" import base64 import json - import random - import time + import secrets - unique_id = int(time.time() * 1000000) + random.randint(1000, 9999) + # Use a cryptographically random id/secret so every minted token is + # guaranteed unique. Time/PRNG-based ids can collide on hosts with + # coarse clock resolution, producing byte-identical tokens that hash to + # the same api_key and silently dedupe (flaky concurrency tests). + unique_id = secrets.token_hex(16) token_data = { "token": [ { "mint": self.mint_url, "proofs": [ { - "id": f"009a1f293253e41e{unique_id % 100000000:08d}", + "id": f"009a1f293253e41e{unique_id[:8]}", "amount": amount, "secret": f"test-secret-{amount}-{unique_id}", "C": "02194603ffa36356f4a56b7df9371fc3192472351453ec7398b8da8117e7c3e104", diff --git a/tests/unit/test_streaming_sse_providers.py b/tests/unit/test_streaming_sse_providers.py new file mode 100644 index 00000000..77d02903 --- /dev/null +++ b/tests/unit/test_streaming_sse_providers.py @@ -0,0 +1,339 @@ +"""Battle-test the streaming SSE parser against real per-provider framing. + +Each test drives the *actual* ``handle_streaming_chat_completion`` generator +with a mock upstream response whose ``aiter_bytes`` emits byte sequences that +mirror what each supported provider sends on the wire (captured from the +providers' own streaming docs): + +* OpenAI / Groq / Fireworks / xAI / Perplexity / Azure - plain + ``data: {json}\\n\\n`` + ``data: [DONE]``. +* OpenRouter - same, but with ``: OPENROUTER PROCESSING`` keepalive comments + interleaved (the framing that produced the original + ``Unexpected token ':'`` client crash). +* Gemini (native ``alt=sse``) - ``data:`` payloads framed with CRLF. + +The invariant every provider must satisfy: every line the proxy emits that +starts with ``data: `` either equals ``[DONE]`` or is valid JSON, and no SSE +comment ever reaches the client. That invariant is exactly what the buggy +``re.split(b"data: ")`` parser violated for OpenRouter. +""" + +import json +from collections.abc import AsyncGenerator +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from routstr.core.db import ApiKey +from routstr.upstream import base +from routstr.upstream.base import BaseUpstreamProvider + + +def _make_response(chunks: list[bytes]) -> MagicMock: + async def aiter_bytes() -> AsyncGenerator[bytes, None]: + for chunk in chunks: + yield chunk + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": "text/event-stream"} + mock_response.aiter_bytes = aiter_bytes + return mock_response + + +async def _drive(chunks: list[bytes], requested_model: str | None = None) -> list[bytes]: + """Run the real streaming generator over ``chunks`` and collect output bytes.""" + provider = BaseUpstreamProvider( + base_url="https://api.example.com", api_key="test_key" + ) + + key = MagicMock(spec=ApiKey) + key.hashed_key = "test_hash" + key.balance = 1000 + + base.adjust_payment_for_tokens = AsyncMock( + return_value={"total_usd": 0.1, "total_msats": 100} + ) + mock_session = MagicMock() + mock_session.get = AsyncMock(return_value=key) + mock_ctx = MagicMock() + mock_ctx.__aenter__ = AsyncMock(return_value=mock_session) + mock_ctx.__aexit__ = AsyncMock(return_value=None) + base.create_session = MagicMock(return_value=mock_ctx) + + streaming_response = await provider.handle_streaming_chat_completion( + response=_make_response(chunks), + key=key, + max_cost_for_model=100, + background_tasks=MagicMock(), + requested_model=requested_model, + ) + + out: list[bytes] = [] + async for chunk in streaming_response.body_iterator: + if isinstance(chunk, str): + out.append(chunk.encode()) + else: + out.append(bytes(chunk)) + return out + + +def _data_payloads(out: list[bytes]) -> list[bytes]: + """Return the raw payload of every ``data: `` line across all emitted bytes.""" + payloads: list[bytes] = [] + for chunk in out: + for line in chunk.split(b"\n"): + if line.startswith(b"data: "): + payloads.append(line[len(b"data: ") :]) + return payloads + + +def _assert_clean(out: list[bytes]) -> list[dict]: + """Core invariant: every data line is [DONE] or valid JSON; no comments leak.""" + blob = b"".join(out) + # No SSE comment line must ever reach the client. + for line in blob.split(b"\n"): + assert not line.startswith(b":"), f"comment leaked to client: {line!r}" + # The original bug signature: a data line whose value is itself a comment. + assert not line.startswith(b"data: :"), f"mangled comment frame: {line!r}" + + objs: list[dict] = [] + for payload in _data_payloads(out): + stripped = payload.strip() + if stripped == b"[DONE]": + continue + obj = json.loads(stripped) # raises if the proxy emitted non-JSON data + objs.append(obj) + return objs + + +@pytest.mark.asyncio +async def test_openai_style_plain_stream() -> None: + """OpenAI / Groq / Fireworks / xAI / Perplexity: plain data + [DONE].""" + chunks = [ + b'data: {"id":"x","choices":[{"delta":{"content":"Hello"}}]}\n\n', + b'data: {"id":"x","choices":[{"delta":{"content":" world"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + assert any(o.get("choices") for o in objs) + assert b"data: [DONE]\n\n" in b"".join(out) + + +@pytest.mark.asyncio +async def test_openrouter_keepalive_comments() -> None: + """OpenRouter ``: OPENROUTER PROCESSING`` keepalives must never crash clients. + + This is the exact regression: the old parser emitted + ``data: : OPENROUTER PROCESSING`` which made downstream + ``JSON.parse`` throw ``Unexpected token ':'``. + """ + chunks = [ + b": OPENROUTER PROCESSING\n\n", + b": OPENROUTER PROCESSING\n\n", + b'data: {"id":"x","choices":[{"delta":{"content":"Hi"}}]}\n\n', + b": OPENROUTER PROCESSING\n\n", + b'data: {"id":"x","choices":[{"delta":{"content":"!"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + # The keepalive must be gone entirely. + assert b"OPENROUTER PROCESSING" not in b"".join(out) + # Real content survived. + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + assert "Hi" in contents and "!" in contents + + +@pytest.mark.asyncio +async def test_openrouter_comment_glued_to_data_chunk() -> None: + """Keepalive packed into the same TCP chunk as data (the harder case).""" + chunks = [ + b'data: {"id":"x","choices":[{"delta":{"content":"a"}}]}\n\n' + b": OPENROUTER PROCESSING\n\n" + b'data: {"id":"x","choices":[{"delta":{"content":"b"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + assert contents == ["a", "b"] + + +@pytest.mark.asyncio +async def test_json_split_across_chunk_boundary() -> None: + """A single event's JSON arriving in two TCP reads must reassemble.""" + chunks = [ + b'data: {"id":"x","choices":[{"delta":{"con', + b'tent":"split"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + assert contents == ["split"] + + +@pytest.mark.asyncio +async def test_byte_by_byte_fragmentation() -> None: + """Pathological framing: one byte per chunk. Must still parse cleanly.""" + raw = ( + b'data: {"id":"x","choices":[{"delta":{"content":"drip"}}]}\n\n' + b": OPENROUTER PROCESSING\n\n" + b"data: [DONE]\n\n" + ) + chunks = [raw[i : i + 1] for i in range(len(raw))] + out = await _drive(chunks) + objs = _assert_clean(out) + assert objs and objs[0]["choices"][0]["delta"]["content"] == "drip" + + +@pytest.mark.asyncio +async def test_gemini_crlf_framing() -> None: + """Gemini native (alt=sse) frames events with CRLF.""" + chunks = [ + b'data: {"id":"g","choices":[{"delta":{"content":"hej"}}]}\r\n\r\n', + b'data: {"id":"g","choices":[{"delta":{"content":"!"}}]}\r\n\r\n', + ] + out = await _drive(chunks) + objs = _assert_clean(out) + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + assert contents == ["hej", "!"] + + +@pytest.mark.asyncio +async def test_azure_leading_role_chunk() -> None: + """Azure OpenAI opens with a content-filter / role-only chunk.""" + chunks = [ + b'data: {"id":"az","choices":[],"prompt_filter_results":[]}\n\n', + b'data: {"id":"az","choices":[{"delta":{"role":"assistant"}}]}\n\n', + b'data: {"id":"az","choices":[{"delta":{"content":"ok"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + _assert_clean(out) + + +@pytest.mark.asyncio +async def test_openrouter_mid_stream_error_event() -> None: + """OpenRouter mid-stream errors arrive as a normal data JSON event.""" + err = { + "id": "x", + "object": "chat.completion.chunk", + "model": "openai/gpt-4o", + "error": {"code": "server_error", "message": "Provider disconnected"}, + "choices": [{"index": 0, "delta": {"content": ""}, "finish_reason": "error"}], + } + chunks = [ + b'data: {"id":"x","choices":[{"delta":{"content":"partial"}}]}\n\n', + b"data: " + json.dumps(err).encode() + b"\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + assert any("error" in o for o in objs), "error event must be forwarded intact" + + +@pytest.mark.asyncio +async def test_gemini_combined_content_and_usage_chunk() -> None: + """Gemini thinking models pack usage into the final *content* chunk. + + Regression: the parser swallowed any chunk carrying a ``usage`` dict, so + when content + usage arrived together the assistant text was dropped and + the client saw "no assistant messages" despite a 200 + token accounting. + """ + chunks = [ + b'data: {"id":"g","choices":[{"delta":{"content":"the answer"},' + b'"finish_reason":"stop"}],"usage":{"prompt_tokens":3,' + b'"completion_tokens":2,"total_tokens":5}}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + # Content delivered exactly once (not dropped, not duplicated by the trailer). + assert contents == ["the answer"] + + +@pytest.mark.asyncio +async def test_separate_usage_chunk_not_forwarded_as_content() -> None: + """A pure usage chunk (choices: []) is still swallowed, content intact.""" + chunks = [ + b'data: {"id":"x","choices":[{"delta":{"content":"hello"}}]}\n\n', + b'data: {"id":"x","choices":[],"usage":{"total_tokens":4}}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + objs = _assert_clean(out) + contents = [ + c["delta"]["content"] + for o in objs + for c in o.get("choices", []) + if "delta" in c + ] + assert contents == ["hello"] + + +@pytest.mark.asyncio +async def test_requested_model_override_applied() -> None: + """Model rewriting still works through the buffered parser.""" + chunks = [ + b'data: {"id":"x","model":"upstream-model","choices":[{"delta":{"content":"hi"}}]}\n\n', + b"data: [DONE]\n\n", + ] + out = await _drive(chunks, requested_model="routstr-model") + objs = _assert_clean(out) + # The upstream content chunk carried model "upstream-model"; the parser must + # rewrite it to the requested model. (The trailing routstr-generated usage + # chunk is excluded - it is not an upstream-forwarded chunk.) + content_chunks = [o for o in objs if o.get("choices")] + assert content_chunks, "expected at least one forwarded content chunk" + assert all(o.get("model") == "routstr-model" for o in content_chunks) + + +@pytest.mark.asyncio +async def test_multiline_non_json_data_each_line_prefixed() -> None: + """A multi-line non-JSON ``data`` block must keep a ``data:`` prefix per line. + + Two ``data:`` lines in one event reassemble to ``line one\\nline two``, which + is not JSON, so it takes the raw-forward path. The parser must re-prefix each + line; a bare second line would reach the client without its ``data:`` field + and break naive SSE parsers. + """ + chunks = [ + b"data: line one\ndata: line two\n\n", + b"data: [DONE]\n\n", + ] + out = await _drive(chunks) + blob = b"".join(out) + for line in blob.split(b"\n"): + stripped = line.strip() + if not stripped or stripped == b"[DONE]": + continue + assert line.startswith(b"data: "), f"bare line leaked to client: {line!r}" + assert b"data: line one" in blob and b"data: line two" in blob