docs: clarify wrap_file ownership and lifecycle

newsfragments/3379.doc.rst

@@ -0,0 +1 @@
+Clarified :func:`trio.wrap_file` ownership and lifecycle behavior, including that closing the async wrapper closes the underlying file, garbage collection does not close it automatically, and the original synchronous file object should not be used concurrently.

src/trio/_file_io.py

@@ -493,6 +493,19 @@ def wrap_file(file: FileT) -> AsyncIOWrapper[FileT]:
     Returns:
         An :term:`asynchronous file object` that wraps ``file``
 
+    The returned wrapper object shares the underlying file with the original
+    ``file`` object; it does not copy it. Closing the wrapper (via
+    :meth:`~trio.abc.AsyncResource.aclose` or ``async with``) will close the
+    underlying file. However, if the wrapper is garbage collected without
+    being explicitly closed, the underlying file is *not* closed
+    automatically, so you should always close it explicitly.
+
+    The original synchronous file object should not be used directly while
+    the wrapper exists, as the wrapper may call file methods in a worker
+    thread, and concurrent access from multiple threads is not safe for most
+    file objects. If you need synchronous access, use the
+    :attr:`~trio._file_io.AsyncIOWrapper.wrapped` attribute.
+
     Example::
 
         async_file = trio.wrap_file(StringIO('asdf'))

src/trio/_tests/test_file_io.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import gc
 import importlib
 import io
 import os
@@ -230,6 +231,35 @@ async def test_open_context_manager(path: pathlib.Path) -> None:
     assert f.closed
 
 
+async def test_wrap_file_aclose_closes_underlying_file() -> None:
+    wrapped = io.StringIO("test")
+    async_file = trio.wrap_file(wrapped)
+
+    await async_file.aclose()
+
+    assert wrapped.closed
+
+
+async def test_wrap_file_context_manager_closes_underlying_file() -> None:
+    wrapped = io.StringIO("test")
+
+    async with trio.wrap_file(wrapped) as async_file:
+        assert async_file.wrapped is wrapped
+        assert not wrapped.closed
+
+    assert wrapped.closed
+
+
+def test_wrap_file_garbage_collection_does_not_close_underlying_file() -> None:
+    wrapped = io.StringIO("test")
+    trio.wrap_file(wrapped)
+
+    gc.collect()
+
+    assert not wrapped.closed
+    wrapped.close()
+
+
 async def test_async_iter() -> None:
     async_file = trio.wrap_file(io.StringIO("test\nfoo\nbar"))
     expected = list(async_file.wrapped)