diff --git a/docs/sse.md b/docs/sse.md index a1a9310d..cd250543 100644 --- a/docs/sse.md +++ b/docs/sse.md @@ -22,6 +22,8 @@ It works the same way with the async client, using `async with` and `async for`: ... print(event.event, event.data) ``` +Event streams are decoded as UTF-8 regardless of the response's charset or default encoding. One leading UTF-8 byte-order mark is ignored; later U+FEFF characters are left intact. + `sse()` issues a `GET` request by default. Some APIs stream events in response to a `POST` - pass `method="POST"` along with the usual request arguments: ```pycon diff --git a/src/httpx2/httpx2/_sse.py b/src/httpx2/httpx2/_sse.py index df2be6aa..3995f751 100644 --- a/src/httpx2/httpx2/_sse.py +++ b/src/httpx2/httpx2/_sse.py @@ -33,6 +33,7 @@ from dataclasses import dataclass from ._config import DEFAULT_MAX_EVENT_SIZE_BYTES +from ._decoders import TextDecoder from ._exceptions import TransportError, request_context from ._models import Response @@ -176,14 +177,17 @@ def _consume_pending(self) -> str: class _SSEParser: def __init__(self, max_event_size: int | None = None) -> None: + self._text_decoder = TextDecoder("utf-8-sig") self._event_decoder = _SSEEventDecoder(max_event_size) self._line_decoder = _SSELineDecoder() - def decode(self, text: str) -> Iterator[ServerSentEvent]: + def decode(self, content: bytes) -> Iterator[ServerSentEvent]: + text = self._text_decoder.decode(content) yield from self._decode_lines(self._line_decoder.decode(text)) self._event_decoder.check_pending(self._line_decoder.pending_size) def flush(self) -> Iterator[ServerSentEvent]: + yield from self._decode_lines(self._line_decoder.decode(self._text_decoder.flush())) yield from self._decode_lines(self._line_decoder.flush()) def _decode_lines(self, lines: list[str]) -> Iterator[ServerSentEvent]: @@ -211,7 +215,7 @@ def __iter__(self) -> Iterator[ServerSentEvent]: with request_context(request=self._response.request): self._check_content_type() parser = _SSEParser(self._max_event_size) - for chunk in self._response.iter_text(): + for chunk in self._response.iter_bytes(): yield from parser.decode(chunk) yield from parser.flush() @@ -219,7 +223,7 @@ async def __aiter__(self) -> AsyncIterator[ServerSentEvent]: with request_context(request=self._response.request): self._check_content_type() parser = _SSEParser(self._max_event_size) - async for chunk in self._response.aiter_text(): + async for chunk in self._response.aiter_bytes(): for sse in parser.decode(chunk): yield sse for sse in parser.flush(): diff --git a/tests/httpx2/test_sse.py b/tests/httpx2/test_sse.py index 7aaad5b0..4b6d3593 100644 --- a/tests/httpx2/test_sse.py +++ b/tests/httpx2/test_sse.py @@ -8,6 +8,131 @@ SSE_BODY = b": this is a comment\nevent: ping\ndata: hello\nid: 1\nretry: 500\n\ndata: first\ndata: second\n\n" +BOM_CASES = [ + pytest.param( + [b"\xef\xbb\xbfdata: first\n\ndata: second\n\n"], + 1024, + [httpx2.ServerSentEvent(data="first"), httpx2.ServerSentEvent(data="second")], + id="first-data-field", + ), + pytest.param( + [b"", b"\xef", b"\xbb", b"\xbf", b"data: first", b"\n\n"], + 1024, + [httpx2.ServerSentEvent(data="first")], + id="split-bom-and-first-line", + ), + pytest.param( + [b"\xef", b"\xbb\xbfdata: caf\xc3", b"\xa9\n\n"], + 1024, + [httpx2.ServerSentEvent(data="café")], + id="split-bom-and-utf8-character", + ), + pytest.param( + [b"\xef\xbb\xbfevent: ping\ndata: first\n\n"], + 1024, + [httpx2.ServerSentEvent(event="ping", data="first")], + id="first-event-field", + ), + pytest.param( + [b"\xef\xbb\xbfid: 1\ndata: first\n\ndata: second\n\n"], + 1024, + [httpx2.ServerSentEvent(data="first", id="1"), httpx2.ServerSentEvent(data="second", id="1")], + id="first-id-persists", + ), + pytest.param( + [b"\xef\xbb\xbf", b"data: hi", b"\n\n"], + 8, + [httpx2.ServerSentEvent(data="hi")], + id="bom-does-not-count-towards-pending-line-limit", + ), + pytest.param( + [b"\xef\xbb\xbf: keepalive comment\n", b"data: hi\n\n"], + 8, + [httpx2.ServerSentEvent(data="hi")], + id="first-comment", + ), + pytest.param( + [b"data: \xef\xbb\xbfinside\n\n\xef\xbb\xbfdata: ignored\n\ndata: tail\n\n"], + 1024, + [httpx2.ServerSentEvent(data="\ufeffinside"), httpx2.ServerSentEvent(data="tail")], + id="later-boms-are-not-stripped", + ), + pytest.param( + [b"\xef\xbb\xbf\xef\xbb\xbfdata: ignored\n\ndata: tail\n\n"], + 1024, + [httpx2.ServerSentEvent(data="tail")], + id="strip-only-one-leading-bom", + ), + pytest.param( + [b"\n", b"\xef\xbb\xbfdata: ignored\n\ndata: tail\n\n"], + 1024, + [httpx2.ServerSentEvent(data="tail")], + id="blank-line-consumes-stream-start", + ), + pytest.param( + [b"data: plain\n\n"], + 1024, + [httpx2.ServerSentEvent(data="plain")], + id="no-bom", + ), +] + + +@pytest.mark.parametrize(("chunks", "max_event_size", "expected"), BOM_CASES) +def test_sse_leading_bom_sync(chunks: list[bytes], max_event_size: int, expected: list[httpx2.ServerSentEvent]) -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=iter(chunks), headers={"Content-Type": "text/event-stream"}) + + with httpx2.Client(transport=httpx2.MockTransport(handler)) as client: + with client.sse("http://testserver/sse", max_event_size=max_event_size) as source: + assert list(source) == expected + + +@pytest.mark.parametrize(("chunks", "max_event_size", "expected"), BOM_CASES) +@pytest.mark.anyio +async def test_sse_leading_bom_async( + chunks: list[bytes], max_event_size: int, expected: list[httpx2.ServerSentEvent] +) -> None: + async def body() -> AsyncIterator[bytes]: + for chunk in chunks: + yield chunk + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=body(), headers={"Content-Type": "text/event-stream"}) + + async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as client: + async with client.sse("http://testserver/sse", max_event_size=max_event_size) as source: + assert [event async for event in source] == expected + + +@pytest.mark.parametrize("charset", [None, "latin-1", "utf-8-sig"]) +def test_sse_utf8_decoding_sync(charset: str | None) -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + content_type = "text/event-stream" if charset is None else f"text/event-stream; charset={charset}" + body = b"\xef\xbb\xbf\xef\xbb\xbfdata: ignored\n\ndata: caf\xc3\xa9\n\n" + return httpx2.Response(200, content=body, headers={"Content-Type": content_type}) + + with httpx2.Client(transport=httpx2.MockTransport(handler), default_encoding="latin-1") as client: + with client.sse("http://testserver/sse") as source: + assert source.response.encoding == (charset or "latin-1") + assert list(source) == [httpx2.ServerSentEvent(data="café")] + assert source.response.encoding == (charset or "latin-1") + + +@pytest.mark.parametrize("charset", [None, "latin-1", "utf-8-sig"]) +@pytest.mark.anyio +async def test_sse_utf8_decoding_async(charset: str | None) -> None: + def handler(request: httpx2.Request) -> httpx2.Response: + content_type = "text/event-stream" if charset is None else f"text/event-stream; charset={charset}" + body = b"\xef\xbb\xbf\xef\xbb\xbfdata: ignored\n\ndata: caf\xc3\xa9\n\n" + return httpx2.Response(200, content=body, headers={"Content-Type": content_type}) + + async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler), default_encoding="latin-1") as client: + async with client.sse("http://testserver/sse") as source: + assert source.response.encoding == (charset or "latin-1") + assert [event async for event in source] == [httpx2.ServerSentEvent(data="café")] + assert source.response.encoding == (charset or "latin-1") + def sse_handler(request: httpx2.Request) -> httpx2.Response: return httpx2.Response(200, content=SSE_BODY, headers={"Content-Type": "text/event-stream"})