test_write_timeout(trio) is failing on python3.14

[cclauss]

To get GitHub Actions to run tests on Python 3.14, we needed to skip one pytest on Trio.

• #3645

tests/test_timeouts.py

@pytest.mark.xfail(sys.version_info >= (3, 14), reason="Fix trio on Python >= 3.14")
async def test_write_timeout(server):

Please try to determine why the write timeout test fails on Trio but not on asyncio, and create a pull request that allows us to remove the xfail.

[augustogoulart]

I can see this in python3.13.3 already.

=============================================================================================== test session starts ================================================================================================
platform darwin -- Python 3.13.3, pytest-8.4.1, pluggy-1.6.0
rootdir: /Users/augustogoulart/Workspace/httpx
configfile: pyproject.toml
plugins: anyio-4.11.0
collected 2 items

tests/test_timeouts.py .F                                                                                                                                                                                    [100%]

===================================================================================================== FAILURES =====================================================================================================
_____________________________________________________________________________________________ test_write_timeout[trio] _____________________________________________________________________________________________

agen = <async_generator object ByteStream.__aiter__ at 0x108b8d840>

    @_core.enable_ki_protection
    def finalizer(agen: AsyncGeneratorType[object, NoReturn]) -> None:
        try:
            self.foreign.remove(id(agen))
        except KeyError:
            is_ours = True
        else:
            is_ours = False

        agen_name = name_asyncgen(agen)
        if is_ours:
            runner.entry_queue.run_sync_soon(
                finalize_in_trio_context,
                agen,
                agen_name,
            )

            # Do this last, because it might raise an exception
            # depending on the user's warnings filter. (That
            # exception will be printed to the terminal and
            # ignored, since we're running in GC context.)
>           warnings.warn(
                f"Async generator {agen_name!r} was garbage collected before it "
                "had been exhausted. Surround its use in 'async with "
                "aclosing(...):' to ensure that it gets cleaned up as soon as "
                "you're done using it.",
                ResourceWarning,
                stacklevel=2,
                source=agen,
            )
E           ResourceWarning: Async generator 'httpx._content.ByteStream.__aiter__' was garbage collected before it had been exhausted. Surround its use in 'async with aclosing(...):' to ensure that it gets cleaned up as soon as you're done using it.

venv/lib/python3.13/site-packages/trio/_core/_asyncgens.py:122: ResourceWarning

[augustogoulart]

AFAIK asyncio is more lenient (although providing shutdown_asyncgens() for such cases) and doesn't monitor async generators for resource leaks, while Trio does.

If an async generator gets garbage collected before being exhausted, asyncio will do nothing while Trio will raise the ResourceWarning shown above.

In the patch below you can see that if we replace the generator returned by ByteStream.__aiter__ with an async iterator (ByteStreamAsyncIterator), Trio will have nothing to watch for and the tests will pass.

Not saying this is a solution (maybe?) but it helps illustrating the issue.

diff --git a/httpx/_content.py b/httpx/_content.py
index 6f479a0..ae64717 100644
--- a/httpx/_content.py
+++ b/httpx/_content.py
@@ -28,6 +28,24 @@ from ._utils import peek_filelike_length, primitive_value_to_str
 __all__ = ["ByteStream"]
 
 
+class ByteStreamAsyncIterator:
+    def __init__(self, stream: bytes) -> None:
+        self._stream = stream
+        self._yielded = False
+
+    def __aiter__(self) -> AsyncIterator[bytes]:
+        return self
+
+    async def __anext__(self) -> bytes:
+        if self._yielded:
+            raise StopAsyncIteration
+        self._yielded = True
+        return self._stream
+
+    async def aclose(self) -> None:
+        pass
+
+
 class ByteStream(AsyncByteStream, SyncByteStream):
     def __init__(self, stream: bytes) -> None:
         self._stream = stream
@@ -35,8 +53,8 @@ class ByteStream(AsyncByteStream, SyncByteStream):
     def __iter__(self) -> Iterator[bytes]:
         yield self._stream
 
-    async def __aiter__(self) -> AsyncIterator[bytes]:
-        yield self._stream
+    def __aiter__(self) -> AsyncIterator[bytes]:
+        return ByteStreamAsyncIterator(self._stream)
 
 
 class IteratorByteStream(SyncByteStream):