feat: add shared _read_capped streaming accumulator

_read_capped / _read_capped_async wrap the pure core with the Content-Length
early-reject and the buffered Response rebuild; _safe_extensions drops the
stale network_stream. Caller owns the stream lifecycle.

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

Author: lesnik512

src/httpware/client.py

@@ -2,7 +2,7 @@
 
 import contextlib
 import typing
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from http import HTTPStatus
 
 import httpx2
@@ -65,6 +65,69 @@ def _accumulate_capped(chunks: typing.Iterable[bytes], cap: int) -> bytes:
     return bytes(buf)
 
 
+def _safe_extensions(extensions: Mapping[str, typing.Any]) -> dict[str, typing.Any]:
+    """Copy response extensions, dropping the now-stale `network_stream`.
+
+    The rebuilt buffered Response never touches its network stream, so carrying a
+    consumed/closed one wholesale is sloppy. `http_version`/`reason_phrase` and
+    any other keys are preserved.
+    """
+    return {key: value for key, value in extensions.items() if key != "network_stream"}
+
+
+def _read_capped(response: httpx2.Response, cap: int, request: httpx2.Request) -> httpx2.Response:
+    """Buffer a streaming sync `response` under `cap` decoded bytes; return a buffered Response.
+
+    Raises `ResponseTooLargeError` (reason="declared") if the declared
+    Content-Length already exceeds `cap` — before any byte is read — and
+    (reason="streamed") if the decoded body crosses `cap` mid-read. Does not
+    close `response`; the caller owns the stream lifecycle.
+    """
+    content_length = _parse_content_length(response.headers.get("content-length"))
+    if content_length is not None and content_length > cap:
+        raise ResponseTooLargeError(
+            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
+        )
+    try:
+        content = _accumulate_capped(response.iter_bytes(), cap)
+    except _CapExceeded:
+        raise ResponseTooLargeError(
+            status_code=response.status_code, limit=cap, content_length=content_length, reason="streamed"
+        ) from None
+    return httpx2.Response(
+        status_code=response.status_code,
+        headers=response.headers,
+        content=content,
+        request=request,
+        extensions=_safe_extensions(response.extensions),
+        history=response.history,
+    )
+
+
+async def _read_capped_async(response: httpx2.Response, cap: int, request: httpx2.Request) -> httpx2.Response:
+    """Async mirror of `_read_capped` (counts decoded bytes from `aiter_bytes`)."""
+    content_length = _parse_content_length(response.headers.get("content-length"))
+    if content_length is not None and content_length > cap:
+        raise ResponseTooLargeError(
+            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
+        )
+    buf = bytearray()
+    async for chunk in response.aiter_bytes():
+        buf += chunk
+        if len(buf) > cap:
+            raise ResponseTooLargeError(
+                status_code=response.status_code, limit=cap, content_length=content_length, reason="streamed"
+            )
+    return httpx2.Response(
+        status_code=response.status_code,
+        headers=response.headers,
+        content=bytes(buf),
+        request=request,
+        extensions=_safe_extensions(response.extensions),
+        history=response.history,
+    )
+
+
 def _build_default_decoders() -> tuple[ResponseDecoder, ...]:
     """Construct the default decoder tuple based on installed extras.
 

tests/test_capped_read.py

@@ -0,0 +1,177 @@
+"""Unit tests for the shared _read_capped wrappers (sync + async).
+
+Drive real streaming responses through MockTransport, then hand the streaming
+Response to _read_capped / _read_capped_async directly — exercising the
+Content-Length early reject, the decoded-byte accumulator, the rebuilt Response,
+and extension sanitisation, independent of client wiring.
+"""
+
+import gzip
+from collections.abc import AsyncIterator
+
+import httpx2
+import pytest
+
+from httpware.client import _read_capped, _read_capped_async
+from httpware.errors import ResponseTooLargeError
+
+
+def _sync_stream(handler: object) -> tuple[httpx2.Client, httpx2.Response]:
+    client = httpx2.Client(transport=httpx2.MockTransport(handler))  # ty: ignore[invalid-argument-type]
+    request = client.build_request("GET", "https://example.test/x")
+    return client, client.send(request, stream=True)
+
+
+async def _async_stream(handler: object) -> tuple[httpx2.AsyncClient, httpx2.Response]:
+    client = httpx2.AsyncClient(transport=httpx2.MockTransport(handler))  # ty: ignore[invalid-argument-type]
+    request = client.build_request("GET", "https://example.test/x")
+    return client, await client.send(request, stream=True)
+
+
+# ---- sync ----
+
+
+def test_read_capped_returns_buffered_response_within_cap() -> None:
+    body = b"hello world"
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body)
+
+    client, resp = _sync_stream(handler)
+    try:
+        out = _read_capped(resp, 1000, resp.request)
+        assert out.content == body
+        assert out.status_code == 200  # noqa: PLR2004 — mirrors handler
+        assert "network_stream" not in out.extensions
+    finally:
+        resp.close()
+        client.close()
+
+
+def test_read_capped_declared_content_length_over_cap() -> None:
+    body = b"x" * 200
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(500, content=body)
+
+    client, resp = _sync_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            _read_capped(resp, 10, resp.request)
+        assert caught.value.reason == "declared"
+        assert caught.value.content_length == 200  # noqa: PLR2004 — len(body)
+        assert caught.value.limit == 10  # noqa: PLR2004 — cap above
+    finally:
+        resp.close()
+        client.close()
+
+
+def test_read_capped_streamed_over_cap_chunked_no_content_length() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=(c for c in (b"a" * 50, b"b" * 50)))
+
+    client, resp = _sync_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            _read_capped(resp, 10, resp.request)
+        assert caught.value.reason == "streamed"
+        assert caught.value.content_length is None
+    finally:
+        resp.close()
+        client.close()
+
+
+def test_read_capped_gzip_bomb_trips_on_decoded_bytes() -> None:
+    raw = gzip.compress(b"A" * 100_000)
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, headers={"content-encoding": "gzip"}, content=raw)
+
+    client, resp = _sync_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            _read_capped(resp, 1000, resp.request)
+        assert caught.value.reason == "streamed"  # compressed CL (small) passed; decoded tripped
+    finally:
+        resp.close()
+        client.close()
+
+
+def test_read_capped_exact_cap_passes() -> None:
+    body = b"x" * 10
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body)
+
+    client, resp = _sync_stream(handler)
+    try:
+        assert _read_capped(resp, 10, resp.request).content == body
+    finally:
+        resp.close()
+        client.close()
+
+
+def test_read_capped_empty_body_passes() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(204)
+
+    client, resp = _sync_stream(handler)
+    try:
+        assert _read_capped(resp, 1, resp.request).content == b""
+    finally:
+        resp.close()
+        client.close()
+
+
+# ---- async ----
+
+
+async def test_read_capped_async_returns_buffered_response_within_cap() -> None:
+    body = b"hello world"
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body)
+
+    client, resp = await _async_stream(handler)
+    try:
+        out = await _read_capped_async(resp, 1000, resp.request)
+        assert out.content == body
+        assert "network_stream" not in out.extensions
+    finally:
+        await resp.aclose()
+        await client.aclose()
+
+
+async def test_read_capped_async_declared_over_cap() -> None:
+    body = b"x" * 200
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(500, content=body)
+
+    client, resp = await _async_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            await _read_capped_async(resp, 10, resp.request)
+        assert caught.value.reason == "declared"
+        assert caught.value.content_length == 200  # noqa: PLR2004 — len(body)
+    finally:
+        await resp.aclose()
+        await client.aclose()
+
+
+async def test_read_capped_async_streamed_over_cap() -> None:
+    async def body() -> AsyncIterator[bytes]:
+        yield b"a" * 50
+        yield b"b" * 50
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, content=body())
+
+    client, resp = await _async_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            await _read_capped_async(resp, 70, resp.request)  # trips on the second 50-byte chunk
+        assert caught.value.reason == "streamed"
+    finally:
+        await resp.aclose()
+        await client.aclose()