test(e2e): assert stream() holds memory constant at volume

The streaming e2e called list() on the export, which buffers every record
and so never exercised the property it appeared to cover: an implementation
that materialised the whole body would have passed it unchanged.

Add sync and async coverage that streams 20,000 operations-scoped catalog
items and asserts on the tracemalloc allocation peak, taken over three whole
reads within a run: a 2,000-record export, the 20,000-record export, and the
same export buffered with list(). Ten times the records must not cost
materially more memory, and buffering must cost far more, which is what shows
the measurement can still see buffering at this volume rather than passing
vacuously on a dataset too small to matter.

Each measured window spans a read from the request opening to the body being
exhausted, because a buffering implementation fills its list before yielding
its first record and a window starting mid-stream would measure only the walk
over an already-materialised list. Verified against a deliberately buffering
stream(), which fails both assertions.

The thresholds are ratios between peaks measured in the same run, not absolute
byte counts, so they need no per-environment calibration. tracemalloc counts
Python allocations rather than process resident size, so the result does not
move with allocator, kernel or GC timing.

MPT-24260

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: jentyk

docs/e2e_tests.md

@@ -15,7 +15,8 @@ tests/
     ├── catalog/
     ├── commerce/
     ├── helpdesk/
-    └── notifications/
+    ├── notifications/
+    └── streaming/
 ```
 
 ## Running Tests
@@ -28,6 +29,26 @@ make test args="tests/e2e/catalog"    # run a subset of E2E tests
 E2E suites use `pytest` markers and live API credentials, so they run outside the default
 `make test` invocation.
 
+## Streaming Memory Coverage
+
+`tests/e2e/streaming/` asserts the property `stream()` exists for: memory stays bounded
+however many records an export carries. It is the one suite whose result depends on the
+size of the live dataset, so it carries requirements the rest of `tests/e2e/` does not.
+
+It streams operations-scoped `catalog.items`, which holds well over the 20,000 records the
+coverage exports. The measurement is a `tracemalloc` allocation peak — Python allocations
+rather than process resident size, so it does not move with allocator or kernel behaviour —
+taken over three whole reads: a 2,000-record export, a 20,000-record export, and the same
+20,000-record export buffered with `list()`. Ten times the records must not cost materially
+more memory, and buffering the same export must cost far more, which is what shows the
+measurement is sensitive enough to see buffering reintroduced.
+
+The comparison is against the peaks measured in the same run rather than an absolute byte
+threshold, so it does not need recalibrating per environment. If the environment holds fewer
+records than the coverage exports, the fixture fails with that count rather than passing on a
+dataset too small to mean anything. `tests/e2e/streaming/memory_probe.py` documents the
+protocol and the reasoning behind each constant.
+
 ## Environment Variables
 
 | Variable                   | Description                                        |

tests/e2e/streaming/__init__.py

@@ -0,0 +1,27 @@
+import pytest
+
+from tests.e2e.streaming.memory_probe import VOLUME_RECORDS
+
+
+def _require_volume(page):
+    total = page.meta.pagination.total if page.meta else 0
+    if total < VOLUME_RECORDS:
+        pytest.fail(
+            f"Catalog items hold {total} records on this environment, fewer than the "
+            f"{VOLUME_RECORDS} this coverage streams. The constant-memory assertion is "
+            "only meaningful at volume, so this is a data problem, not a client failure."
+        )
+
+
+@pytest.fixture
+def large_collection(mpt_ops):
+    service = mpt_ops.catalog.items
+    _require_volume(service.fetch_page(limit=1))
+    return service
+
+
+@pytest.fixture
+async def async_large_collection(async_mpt_ops):
+    service = async_mpt_ops.catalog.items
+    _require_volume(await service.fetch_page(limit=1))
+    return service

tests/e2e/streaming/conftest.py

@@ -0,0 +1,137 @@
+"""Measurement protocol behind the constant-memory streaming coverage.
+
+`stream()` claims memory stays bounded no matter how many records the export carries.
+Proving that needs two comparisons on the same live data, which is what
+`profile_streaming_memory` collects in one pass:
+
+- the allocation peak of a small export against the peak of an export ten times longer,
+  so a peak that scales with the record count fails;
+- the peak of the streamed read against the peak of buffering the same export, so the
+  measurement is shown to be sensitive enough to see buffering at this volume.
+
+Every measured window covers a whole read, from the request being opened to the body being
+exhausted. That matters: an implementation that buffers does so before it yields its first
+record, so a window that started mid-stream would measure only the walk over an
+already-materialised list and see nothing wrong.
+
+Peaks come from `tracemalloc`, which counts Python allocations rather than process resident
+size, so the numbers do not move with allocator or kernel behaviour. Every peak is reported
+net of the memory already live when its window opens, and each window starts with a
+collection, so an object released by an earlier read is never charged to a later one.
+"""
+
+import gc
+import tracemalloc
+from collections.abc import Awaitable, Callable, Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Any
+
+# Records exported before the first measured window, so connection setup, lazy imports and
+# the first read buffers are paid for outside every comparison. This is a separate read,
+# not a prologue of a measured one, so nothing can hide in it.
+WARMUP_RECORDS = 500
+# Records exported by the short read that the volume read is compared against.
+SAMPLE_RECORDS = 2000
+# Records exported by the volume read: ten times the short read, and enough that buffering
+# the same export costs tens of megabytes.
+VOLUME_RECORDS = 20000
+# How much the volume read's peak may exceed the short read's. Ten times the records must
+# not cost materially more memory; a buffering read lands an order of magnitude above this.
+MAX_PEAK_GROWTH = 3.0
+# How much more buffering the export must peak than streaming it. Guards against a vacuous
+# pass: on a dataset too small to matter, the two reads would cost about the same.
+MIN_BUFFERING_CONTRAST = 20.0
+
+# One traced frame per allocation: the call site is never reported, so tracing costs less.
+TRACE_FRAMES = 1
+
+
+@dataclass(frozen=True)
+class PeakReading:
+    records: int
+    peak_bytes: int
+
+
+@dataclass(frozen=True)
+class StreamingMemoryProfile:
+    sample: PeakReading
+    volume: PeakReading
+    buffered: PeakReading
+
+
+@contextmanager
+def tracing_allocations() -> Iterator[None]:
+    gc.collect()
+    tracemalloc.start(TRACE_FRAMES)
+    try:
+        yield
+    finally:
+        tracemalloc.stop()
+
+
+def measure_peak(read: Callable[[], int]) -> PeakReading:
+    gc.collect()
+    tracemalloc.reset_peak()
+    baseline_bytes = tracemalloc.get_traced_memory()[0]
+    records = read()
+    peak_bytes = tracemalloc.get_traced_memory()[1] - baseline_bytes
+    return PeakReading(records=records, peak_bytes=peak_bytes)
+
+
+async def measure_peak_async(read: Callable[[], Awaitable[int]]) -> PeakReading:
+    gc.collect()
+    tracemalloc.reset_peak()
+    baseline_bytes = tracemalloc.get_traced_memory()[0]
+    records = await read()
+    peak_bytes = tracemalloc.get_traced_memory()[1] - baseline_bytes
+    return PeakReading(records=records, peak_bytes=peak_bytes)
+
+
+def drain_stream(service: Any, limit: int) -> int:
+    """Stream `limit` records, retaining none.
+
+    The stream is consumed to exhaustion, which is what triggers its completeness check.
+    """
+    return sum(1 for _ in service.stream(limit=limit))
+
+
+async def drain_stream_async(service: Any, limit: int) -> int:
+    """Stream `limit` records, retaining none.
+
+    The stream is consumed to exhaustion, which is what triggers its completeness check.
+    """
+    # WPS519 would have the records collected and counted, which is the buffering this
+    # measurement exists to rule out.
+    consumed = 0
+    async for _ in service.stream(limit=limit):  # noqa: WPS519
+        consumed += 1
+    return consumed
+
+
+def buffer_export(service: Any) -> int:
+    """Materialise a whole export, the buffering read the streamed read is compared with."""
+    return len(list(service.stream(limit=VOLUME_RECORDS)))
+
+
+async def buffer_export_async(service: Any) -> int:
+    """Materialise a whole export, the buffering read the streamed read is compared with."""
+    return len([record async for record in service.stream(limit=VOLUME_RECORDS)])
+
+
+def profile_streaming_memory(service: Any) -> StreamingMemoryProfile:
+    with tracing_allocations():
+        drain_stream(service, WARMUP_RECORDS)
+        sample = measure_peak(lambda: drain_stream(service, SAMPLE_RECORDS))
+        volume = measure_peak(lambda: drain_stream(service, VOLUME_RECORDS))
+        buffered = measure_peak(lambda: buffer_export(service))
+    return StreamingMemoryProfile(sample=sample, volume=volume, buffered=buffered)
+
+
+async def profile_async_streaming_memory(service: Any) -> StreamingMemoryProfile:
+    with tracing_allocations():
+        await drain_stream_async(service, WARMUP_RECORDS)
+        sample = await measure_peak_async(lambda: drain_stream_async(service, SAMPLE_RECORDS))
+        volume = await measure_peak_async(lambda: drain_stream_async(service, VOLUME_RECORDS))
+        buffered = await measure_peak_async(lambda: buffer_export_async(service))
+    return StreamingMemoryProfile(sample=sample, volume=volume, buffered=buffered)

tests/e2e/streaming/memory_probe.py

@@ -0,0 +1,17 @@
+from tests.e2e.streaming.memory_probe import (
+    MAX_PEAK_GROWTH,
+    MIN_BUFFERING_CONTRAST,
+    SAMPLE_RECORDS,
+    VOLUME_RECORDS,
+    profile_async_streaming_memory,
+)
+
+
+async def test_stream_at_volume_keeps_memory_bounded(async_large_collection):
+    result = await profile_async_streaming_memory(async_large_collection)
+
+    assert result.sample.records == SAMPLE_RECORDS
+    assert result.volume.records == VOLUME_RECORDS
+    assert result.buffered.records == VOLUME_RECORDS
+    assert result.volume.peak_bytes <= result.sample.peak_bytes * MAX_PEAK_GROWTH
+    assert result.buffered.peak_bytes >= result.volume.peak_bytes * MIN_BUFFERING_CONTRAST

tests/e2e/streaming/test_async_streaming.py

@@ -0,0 +1,17 @@
+from tests.e2e.streaming.memory_probe import (
+    MAX_PEAK_GROWTH,
+    MIN_BUFFERING_CONTRAST,
+    SAMPLE_RECORDS,
+    VOLUME_RECORDS,
+    profile_streaming_memory,
+)
+
+
+def test_stream_at_volume_keeps_memory_bounded(large_collection):
+    result = profile_streaming_memory(large_collection)
+
+    assert result.sample.records == SAMPLE_RECORDS
+    assert result.volume.records == VOLUME_RECORDS
+    assert result.buffered.records == VOLUME_RECORDS
+    assert result.volume.peak_bytes <= result.sample.peak_bytes * MAX_PEAK_GROWTH
+    assert result.buffered.peak_bytes >= result.volume.peak_bytes * MIN_BUFFERING_CONTRAST