Add type annotations for windows_pipes and its test

pyproject.toml

@@ -83,9 +83,6 @@ disallow_untyped_calls = false
 # files not yet fully typed
 [[tool.mypy.overrides]]
 module = [
-# internal
-"trio/_windows_pipes",
-
 # tests
 "trio/testing/_fake_net",
 "trio/_core/_tests/test_guest_mode",
@@ -116,7 +113,6 @@ module = [
 "trio/_tests/test_tracing",
 "trio/_tests/test_util",
 "trio/_tests/test_wait_for_object",
-"trio/_tests/test_windows_pipes",
 "trio/_tests/tools/test_gen_exports",
 ]
 check_untyped_defs = false

trio/_tests/test_windows_pipes.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import sys
-from typing import Any, Tuple
+from typing import Any
 
 import pytest
 
@@ -14,24 +16,51 @@
 else:
     pytestmark = pytest.mark.skip(reason="windows only")
     pipe: Any = None
-    PipeSendStream: Any = None
-    PipeReceiveStream: Any = None
+    _handle: Any = None
+    kernel32: Any = None
+    from .._abc import ReceiveStream, SendStream
+
+    class _fake_recieve(ReceiveStream):
+        def __init__(self, _: int) -> None:
+            ...
+
+        async def receive_some(self, max_bytes: int | None = None) -> bytes | bytearray:
+            return b""
+
+        async def aclose(self) -> None:
+            ...
+
+    class _fake_send(SendStream):
+        def __init__(self, _: int) -> None:
+            ...
+
+        async def send_all(self, data: bytes | bytearray | memoryview) -> None:
+            ...
+
+        async def wait_send_all_might_not_block(self) -> None:
+            ...
+
+        async def aclose(self) -> None:
+            ...
+
+    PipeReceiveStream = _fake_recieve
+    PipeSendStream = _fake_send
 
 
-async def make_pipe() -> Tuple[PipeSendStream, PipeReceiveStream]:
+async def make_pipe() -> tuple[PipeSendStream, PipeReceiveStream]:
     """Makes a new pair of pipes."""
     (r, w) = pipe()
     return PipeSendStream(w), PipeReceiveStream(r)
 
 
-async def test_pipe_typecheck():
+async def test_pipe_typecheck() -> None:
     with pytest.raises(TypeError):
-        PipeSendStream(1.0)
+        PipeSendStream(1.0)  # type: ignore[arg-type]
     with pytest.raises(TypeError):
-        PipeReceiveStream(None)
+        PipeReceiveStream(None)  # type: ignore[arg-type]
 
 
-async def test_pipe_error_on_close():
+async def test_pipe_error_on_close() -> None:
     # Make sure we correctly handle a failure from kernel32.CloseHandle
     r, w = pipe()
 
@@ -47,18 +76,18 @@
         await receive_stream.aclose()
 
 
-async def test_pipes_combined():
+async def test_pipes_combined() -> None:
     write, read = await make_pipe()
     count = 2**20
     replicas = 3
 
-    async def sender():
+    async def sender() -> None:
         async with write:
             big = bytearray(count)
             for _ in range(replicas):
                 await write.send_all(big)
 
-    async def reader():
+    async def reader() -> None:
         async with read:
             await wait_all_tasks_blocked()
             total_received = 0
@@ -76,7 +105,7 @@
         n.start_soon(reader)
 
 
-async def test_async_with():
+async def test_async_with() -> None:
     w, r = await make_pipe()
     async with w, r:
         pass
@@ -87,11 +116,11 @@
         await r.receive_some(10)
 
 
-async def test_close_during_write():
+async def test_close_during_write() -> None:
     w, r = await make_pipe()
     async with _core.open_nursery() as nursery:
 
-        async def write_forever():
+        async def write_forever() -> None:
             with pytest.raises(_core.ClosedResourceError) as excinfo:
                 while True:
                     await w.send_all(b"x" * 4096)
@@ -102,7 +131,7 @@
         await w.aclose()
 
 
-async def test_pipe_fully():
+async def test_pipe_fully() -> None:
     # passing make_clogged_pipe tests wait_send_all_might_not_block, and we
     # can't implement that on Windows
     await check_one_way_stream(make_pipe, None)

trio/_windows_pipes.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import sys
 from typing import TYPE_CHECKING
 
@@ -23,18 +25,18 @@ def __init__(self, handle: int) -> None:
         _core.register_with_iocp(self.handle)
 
     @property
-    def closed(self):
+    def closed(self) -> bool:
         return self.handle == -1
 
-    def close(self):
+    def close(self) -> None:
         if self.closed:
             return
         handle = self.handle
         self.handle = -1
         if not kernel32.CloseHandle(_handle(handle)):
             raise_winerror()
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.close()
 
 
@@ -50,7 +52,7 @@ def __init__(self, handle: int) -> None:
             "another task is currently using this pipe"
         )
 
-    async def send_all(self, data: bytes):
+    async def send_all(self, data: bytes) -> None:
         with self._conflict_detector:
             if self._handle_holder.closed:
                 raise _core.ClosedResourceError("this pipe is already closed")
@@ -76,10 +78,10 @@ async def wait_send_all_might_not_block(self) -> None:
             # not implemented yet, and probably not needed
             await _core.checkpoint()
 
-    def close(self):
+    def close(self) -> None:
         self._handle_holder.close()
 
-    async def aclose(self):
+    async def aclose(self) -> None:
         self.close()
         await _core.checkpoint()
 
@@ -94,7 +96,7 @@ def __init__(self, handle: int) -> None:
             "another task is currently using this pipe"
         )
 
-    async def receive_some(self, max_bytes=None) -> bytes:
+    async def receive_some(self, max_bytes: int | None = None) -> bytes:
         with self._conflict_detector:
             if self._handle_holder.closed:
                 raise _core.ClosedResourceError("this pipe is already closed")
@@ -133,9 +135,9 @@ async def receive_some(self, max_bytes=None) -> bytes:
                 del buffer[size:]
                 return buffer
 
-    def close(self):
+    def close(self) -> None:
         self._handle_holder.close()
 
-    async def aclose(self):
+    async def aclose(self) -> None:
         self.close()
         await _core.checkpoint()