Decode server-sent events as UTF-8 with BOM handling

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

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,15 +215,15 @@ 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()
 
     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():

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"})