get and enforce 100% coverage

src/trio/_core/_tests/test_io.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import random
+import select
 import socket as stdlib_socket
+import sys
 from collections.abc import Awaitable, Callable
 from contextlib import suppress
 from typing import TYPE_CHECKING, TypeVar
@@ -343,6 +345,7 @@
             assert iostats.tasks_waiting_write == expected_writers
         else:
             assert iostats.backend == "kqueue"
+            assert iostats.monitors == 0
             assert iostats.tasks_waiting == expected_readers + expected_writers
 
     a1, b1 = stdlib_socket.socketpair()
@@ -381,6 +384,38 @@
         check(expected_readers=1, expected_writers=0)
 
 
+@pytest.mark.skipif(sys.platform in {"win32", "linux"}, reason="no kqueue")
+async def test_io_manager_kqueue_monitors_statistics() -> None:
+    def check(
+        *,
+        expected_monitors: int,
+        expected_readers: int,
+        expected_writers: int,
+    ) -> None:
+        statistics = _core.current_statistics()
+        print(statistics)
+        iostats = statistics.io_statistics
+        assert iostats.backend == "kqueue"
+        assert iostats.monitors == expected_monitors
+        assert iostats.tasks_waiting == expected_readers + expected_writers
+
+    a1, b1 = stdlib_socket.socketpair()
+    for sock in [a1, b1]:
+        sock.setblocking(False)
+
+    with a1, b1:
+        # let the call_soon_task settle down
+        await wait_all_tasks_blocked()
+
+        # 1 for call_soon_task
+        check(expected_monitors=0, expected_readers=1, expected_writers=0)
+
+        with trio.lowlevel.monitor_kevent(a1.fileno(), select.KQ_FILTER_READ):
+            check(expected_monitors=1, expected_readers=1, expected_writers=0)
+
+        check(expected_monitors=0, expected_readers=1, expected_writers=0)
+
+
 async def test_can_survive_unnotified_close() -> None:
     # An "unnotified" close is when the user closes an fd/socket/handle
     # directly, without calling notify_closing first. This should never happen

src/trio/_tests/test_dtls.py

@@ -117,6 +117,18 @@ async def test_smoke(ipv6: bool) -> None:
             assert client_channel.get_cleartext_mtu() == cleartext_mtu_1234
 
 
+@parametrize_ipv6
+async def test_smoke_close_immediately(ipv6: bool) -> None:
+    # used to get 100% branch coverage in this file itself
+    async with dtls_echo_server(ipv6=ipv6) as (_server_endpoint, address):
+        with endpoint(ipv6=ipv6) as client_endpoint:
+            client_channel = client_endpoint.connect(address, client_ctx)
+            with pytest.raises(trio.NeedHandshakeError):
+                client_channel.get_cleartext_mtu()
+
+            await client_channel.do_handshake()
+
+
 @slow
 async def test_handshake_over_terrible_network(
     autojump_clock: trio.testing.MockClock,

src/trio/_tests/test_subprocess.py

@@ -16,6 +16,7 @@
     Any,
     NoReturn,
 )
+from unittest import mock
 
 import pytest
 
@@ -146,6 +147,26 @@ async def test_basic(background_process: BackgroundProcessType) -> None:
     )
 
 
+@background_process_param
+async def test_basic_no_pidfd(background_process: BackgroundProcessType) -> None:
+    with mock.patch("trio._subprocess.can_try_pidfd_open", new=False):
+        async with background_process(EXIT_TRUE) as proc:
+            assert proc._pidfd is None
+            await proc.wait()
+        assert isinstance(proc, Process)
+        assert proc._pidfd is None
+        assert proc.returncode == 0
+        assert repr(proc) == f"<trio.Process {EXIT_TRUE}: exited with status 0>"
+
+        async with background_process(EXIT_FALSE) as proc:
+            await proc.wait()
+        assert proc.returncode == 1
+        assert repr(proc) == "<trio.Process {!r}: {}>".format(
+            EXIT_FALSE,
+            "exited with status 1",
+        )
+
+
 @background_process_param
 async def test_auto_update_returncode(
     background_process: BackgroundProcessType,
@@ -181,6 +202,27 @@ async def test_multi_wait(background_process: BackgroundProcessType) -> None:
             proc.kill()
 
 
+@background_process_param
+async def test_multi_wait_no_pidfd(background_process: BackgroundProcessType) -> None:
+    with mock.patch("trio._subprocess.can_try_pidfd_open", new=False):
+        async with background_process(SLEEP(10)) as proc:
+            # Check that wait (including multi-wait) tolerates being cancelled
+            async with _core.open_nursery() as nursery:
+                nursery.start_soon(proc.wait)
+                nursery.start_soon(proc.wait)
+                nursery.start_soon(proc.wait)
+                await wait_all_tasks_blocked()
+                nursery.cancel_scope.cancel()
+
+            # Now try waiting for real
+            async with _core.open_nursery() as nursery:
+                nursery.start_soon(proc.wait)
+                nursery.start_soon(proc.wait)
+                nursery.start_soon(proc.wait)
+                await wait_all_tasks_blocked()
+                proc.kill()
+
+
 COPY_STDIN_TO_STDOUT_AND_BACKWARD_TO_STDERR = python(
     "data = sys.stdin.buffer.read(); "
     "sys.stdout.buffer.write(data); "
@@ -524,6 +566,31 @@ async def test_wait_reapable_fails(background_process: BackgroundProcessType) ->
         signal.signal(signal.SIGCHLD, old_sigchld)
 
 
+@pytest.mark.skipif(not posix, reason="POSIX specific")
+@background_process_param
+async def test_wait_reapable_fails_no_pidfd(
+    background_process: BackgroundProcessType,
+) -> None:
+    if TYPE_CHECKING and sys.platform == "win32":
+        return
+    with mock.patch("trio._subprocess.can_try_pidfd_open", new=False):
+        old_sigchld = signal.signal(signal.SIGCHLD, signal.SIG_IGN)
+        try:
+            # With SIGCHLD disabled, the wait() syscall will wait for the
+            # process to exit but then fail with ECHILD. Make sure we
+            # support this case as the stdlib subprocess module does.
+            async with background_process(SLEEP(3600)) as proc:
+                async with _core.open_nursery() as nursery:
+                    nursery.start_soon(proc.wait)
+                    await wait_all_tasks_blocked()
+                    proc.kill()
+                    nursery.cancel_scope.deadline = _core.current_time() + 1.0
+                assert not nursery.cancel_scope.cancelled_caught
+                assert proc.returncode == 0  # exit status unknowable, so...
+        finally:
+            signal.signal(signal.SIGCHLD, old_sigchld)
+
+
 @slow
 def test_waitid_eintr() -> None:
     # This only matters on PyPy (where we're coding EINTR handling

src/trio/_tools/gen_exports.py

@@ -180,22 +180,13 @@ def run_linters(file: File, source: str) -> str:
       SystemExit: If either failed.
     """
 
-    success, response = run_black(file, source)
-    if not success:
-        print(response)
-        sys.exit(1)
-
-    success, response = run_ruff(file, response)
-    if not success:  # pragma: no cover  # Test for run_ruff should catch
-        print(response)
-        sys.exit(1)
-
-    success, response = run_black(file, response)
-    if not success:
-        print(response)
-        sys.exit(1)
-
-    return response
+    for fn in (run_black, run_ruff, run_black):
+        success, source = fn(file, source)
+        if not success:
+            print(source)
+            sys.exit(1)
+
+    return source
 
 
 def gen_public_wrappers_source(file: File) -> str: