feat: bound stream() error pre-read via _read_capped

Both stream() error branches route the 4xx/5xx pre-read through the shared
accumulator when a cap is set, so chunked/compression-bombed error bodies are
caught instead of read unbounded; exc.response.content stays populated.
User-driven success streaming is never capped.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lesnik512

src/httpware/client.py

@@ -1118,17 +1118,13 @@ async def stream(  # noqa: PLR0913, C901 — mirrors httpx2 per-method signature
 
         async with _httpx2_exception_mapper(), self._httpx2_client.stream(method, url, **kwargs) as response:
             if HTTPStatus.BAD_REQUEST <= response.status_code < 600:  # noqa: PLR2004 — 600 is the synthetic upper bound for 5xx
-                if self._max_response_body_bytes is not None:
-                    content_length = _parse_content_length(response.headers.get("content-length"))
-                    if content_length is not None and content_length > self._max_response_body_bytes:
-                        raise ResponseTooLargeError(
-                            status_code=response.status_code,
-                            limit=self._max_response_body_bytes,
-                            content_length=content_length,
-                            reason="declared",
-                        )
-                await response.aread()  # pre-read body so exc.response.content works
-                _raise_on_status_error(response)
+                cap = self._max_response_body_bytes
+                if cap is None:
+                    await response.aread()  # pre-read body so exc.response.content works
+                    _raise_on_status_error(response)
+                else:
+                    # Bound the error pre-read; raises ResponseTooLargeError when over cap.
+                    _raise_on_status_error(await _read_capped_async(response, cap, response.request))
             yield response
 
     async def __aenter__(self) -> typing.Self:
@@ -2116,15 +2112,11 @@ def stream(  # noqa: PLR0913, C901 — mirrors httpx2 per-method signatures; kwa
 
         with _httpx2_exception_mapper_sync(), self._httpx2_client.stream(method, url, **kwargs) as response:
             if HTTPStatus.BAD_REQUEST <= response.status_code < 600:  # noqa: PLR2004 — 600 is the synthetic upper bound for 5xx
-                if self._max_response_body_bytes is not None:
-                    content_length = _parse_content_length(response.headers.get("content-length"))
-                    if content_length is not None and content_length > self._max_response_body_bytes:
-                        raise ResponseTooLargeError(
-                            status_code=response.status_code,
-                            limit=self._max_response_body_bytes,
-                            content_length=content_length,
-                            reason="declared",
-                        )
-                response.read()  # pre-read body so exc.response.content works
-                _raise_on_status_error(response)
+                cap = self._max_response_body_bytes
+                if cap is None:
+                    response.read()  # pre-read body so exc.response.content works
+                    _raise_on_status_error(response)
+                else:
+                    # Bound the error pre-read; raises ResponseTooLargeError when over cap.
+                    _raise_on_status_error(_read_capped(response, cap, response.request))
             yield response

tests/test_client_stream.py

@@ -387,6 +387,40 @@ def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
     await client.aclose()
 
 
+async def test_stream_error_pre_read_streamed_over_cap() -> None:
+    async def body() -> typing.AsyncIterator[bytes]:
+        yield b"a" * 50
+        yield b"b" * 50
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(500, content=body())  # chunked: no Content-Length
+
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)), max_response_body_bytes=70
+    )
+    with pytest.raises(ResponseTooLargeError) as caught:
+        async with client.stream("GET", "https://example.test/x"):
+            pytest.fail("unreachable")
+    assert caught.value.reason == "streamed"
+    assert caught.value.content_length is None
+    await client.aclose()
+
+
+async def test_stream_user_driven_success_body_not_capped() -> None:
+    body = b"x" * 100_000
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body)
+
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)), max_response_body_bytes=10
+    )
+    async with client.stream("GET", "https://example.test/x") as response:
+        chunks = [chunk async for chunk in response.aiter_bytes()]
+    assert b"".join(chunks) == body  # user-driven streaming is never capped
+    await client.aclose()
+
+
 @pytest.mark.parametrize(
     ("raw", "expected"),
     [(None, None), ("123", 123), ("abc", None), ("-5", None), ("0", 0)],

tests/test_client_stream_sync.py

@@ -346,3 +346,28 @@ def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         pytest.fail("unreachable")
     assert caught.value.response.content == body
     client.close()
+
+
+def test_stream_error_pre_read_streamed_over_cap_sync() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(500, content=(c for c in (b"a" * 50, b"b" * 50)))  # chunked: no Content-Length
+
+    client = Client(httpx2_client=httpx2.Client(transport=httpx2.MockTransport(handler)), max_response_body_bytes=70)
+    with pytest.raises(ResponseTooLargeError) as caught, client.stream("GET", "https://example.test/x"):
+        pytest.fail("unreachable")
+    assert caught.value.reason == "streamed"
+    assert caught.value.content_length is None
+    client.close()
+
+
+def test_stream_user_driven_success_body_not_capped_sync() -> None:
+    body = b"x" * 100_000
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body)
+
+    client = Client(httpx2_client=httpx2.Client(transport=httpx2.MockTransport(handler)), max_response_body_bytes=10)
+    with client.stream("GET", "https://example.test/x") as response:
+        chunks = list(response.iter_bytes())
+    assert b"".join(chunks) == body  # user-driven streaming is never capped
+    client.close()