feat: add pure _accumulate_capped core with property test

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

Author: lesnik512

src/httpware/client.py

@@ -43,6 +43,28 @@ def _parse_content_length(raw: str | None) -> int | None:
     return value if value >= 0 else None
 
 
+class _CapExceeded(Exception):  # noqa: N818 — internal control-flow signal, not a user-facing error
+    """Internal signal: decoded bytes crossed the cap mid-read. Carries bytes read so far."""
+
+    def __init__(self, *, read: int) -> None:
+        self.read = read
+        super().__init__(f"decoded body exceeded cap after {read} bytes")
+
+
+def _accumulate_capped(chunks: typing.Iterable[bytes], cap: int) -> bytes:
+    """Concatenate `chunks`, raising `_CapExceeded` the moment the running total exceeds `cap`.
+
+    Counts decoded bytes (the in-memory footprint). Grown in a single bytearray
+    so there is no transient list-plus-join double allocation.
+    """
+    buf = bytearray()
+    for chunk in chunks:
+        buf += chunk
+        if len(buf) > cap:
+            raise _CapExceeded(read=len(buf))
+    return bytes(buf)
+
+
 def _build_default_decoders() -> tuple[ResponseDecoder, ...]:
     """Construct the default decoder tuple based on installed extras.
 

tests/test_capped_read_props.py

@@ -0,0 +1,55 @@
+"""Hypothesis property tests for the pure _accumulate_capped core.
+
+The one subtle invariant of the response-body cap is chunk-boundary
+independence: the accumulator must behave identically no matter how the decoded
+body is split into chunks. It must raise _CapExceeded iff the total decoded
+length exceeds the cap, and otherwise return the body byte-for-byte.
+"""
+
+import pytest
+from hypothesis import given
+from hypothesis import strategies as st
+
+from httpware.client import _accumulate_capped, _CapExceeded
+
+
+def _partition(body: bytes, sizes: list[int]) -> list[bytes]:
+    """Split `body` into chunks following `sizes` (remainder becomes a final chunk)."""
+    chunks: list[bytes] = []
+    pos = 0
+    for size in sizes:
+        if pos >= len(body):
+            break
+        chunks.append(body[pos : pos + size])
+        pos += size
+    if pos < len(body):
+        chunks.append(body[pos:])
+    return chunks
+
+
+@given(
+    body=st.binary(max_size=2048),
+    sizes=st.lists(st.integers(min_value=1, max_value=64), max_size=64),
+    cap=st.integers(min_value=1, max_value=4096),
+)
+def test_accumulate_capped_chunk_boundary_independence(body: bytes, sizes: list[int], cap: int) -> None:
+    chunks = _partition(body, sizes)
+    if len(body) > cap:
+        with pytest.raises(_CapExceeded) as caught:
+            _accumulate_capped(chunks, cap)
+        assert caught.value.read > cap
+    else:
+        assert _accumulate_capped(chunks, cap) == body
+
+
+@given(body=st.binary(min_size=2, max_size=512))
+def test_accumulate_capped_trips_at_one_below_length(body: bytes) -> None:
+    cap = len(body) - 1
+    with pytest.raises(_CapExceeded):
+        _accumulate_capped([body], cap)
+
+
+@given(body=st.binary(max_size=512))
+def test_accumulate_capped_passes_at_exact_length(body: bytes) -> None:
+    cap = max(1, len(body))
+    assert _accumulate_capped([body], cap) == body