tests: avoid abandoned-async-generator warnings under the trio backend

The session-level timeout test now runs on trio's virtual clock. At teardown
the streamable-HTTP client abandons its httpx/httpx-sse response generators;
trio's asyncgen finalizer warns about each one (asyncio finalizes abandoned
generators silently at loop shutdown), and filterwarnings=error turns that
into a test failure. Scope-ignore the two third-party generator signatures,
and make the bridge's response body delegate to the memory stream's own
iterator so the harness itself leaves no abandoned generator.

Author: maxisbey

tests/interaction/conftest.py

@@ -23,6 +23,19 @@ def pytest_configure(config: pytest.Config) -> None:
         "filterwarnings", "ignore:.*MemoryObject(Send|Receive)Stream:pytest.PytestUnraisableExceptionWarning"
     )
     config.addinivalue_line("filterwarnings", "ignore:.*MemoryObject(Send|Receive)Stream:ResourceWarning")
+    # The trio-mockclock leg of the session-level timeout test (test_timeouts.py) is the suite's
+    # only test on the trio backend. v1's streamable-HTTP client abandons its httpx/httpx-sse
+    # response generators when the session task group is cancelled at teardown; asyncio finalizes
+    # abandoned async generators silently at loop shutdown, but trio's finalizer warns about each
+    # one (`Async generator ... was garbage collected before it had been exhausted`). The fixes
+    # live in `src/` on `main` and are out of scope for this tests-only backport. The filters are
+    # scoped to the two known httpx generator signatures so an unrelated leak still fails the suite.
+    config.addinivalue_line("filterwarnings", "ignore:Async generator 'httpx:ResourceWarning")
+    config.addinivalue_line(
+        "filterwarnings",
+        "ignore:.*async_generator object (Response.aiter_text|EventSource.aiter_sse)"
+        ":pytest.PytestUnraisableExceptionWarning",
+    )
 
 
 _FACTORIES: dict[str, Connect] = {

tests/interaction/transports/_bridge.py

@@ -47,9 +47,14 @@ def __init__(self, chunks: MemoryObjectReceiveStream[bytes], client_disconnected
         self._chunks = chunks
         self._client_disconnected = client_disconnected
 
-    async def __aiter__(self) -> AsyncIterator[bytes]:
-        async for chunk in self._chunks:
-            yield chunk
+    def __aiter__(self) -> AsyncIterator[bytes]:
+        # Delegate to the memory stream's own async iterator instead of wrapping it in an async
+        # generator. httpx abandons the iterator without closing it when a streamed response is
+        # closed mid-stream; trio's asyncgen finalizer warns about abandoned generators (asyncio
+        # finalizes them silently at loop shutdown), which would fail the suite's one trio-backend
+        # test. The memory stream is a plain async iterator with the same EndOfStream ->
+        # StopAsyncIteration semantics and is not tracked by that machinery.
+        return self._chunks
 
     async def aclose(self) -> None:
         self._client_disconnected.set()