get and enforce 100% coverage

.codecov.yml

@@ -21,7 +21,9 @@ comment:
 
 coverage:
   # required range
-  range: 99.6..100
+  precision: 5
+  round: down
+  range: 100..100
   status:
     # require patches to be 100%
     patch:

newsfragments/3159.misc.rst

@@ -0,0 +1 @@
+Get more coverage

pyproject.toml

@@ -305,18 +305,16 @@ precision = 1
 skip_covered = true
 skip_empty = true
 show_missing = true
-exclude_lines = [
-    "pragma: no cover",
-    "abc.abstractmethod",
-    "if TYPE_CHECKING.*:",
-    "if _t.TYPE_CHECKING:",
-    "if t.TYPE_CHECKING:",
-    "@overload",
-    'class .*\bProtocol\b.*\):',
-    "raise NotImplementedError",
-]
 exclude_also = [
   '^\s*@pytest\.mark\.xfail',
+   "abc.abstractmethod",
+   "if TYPE_CHECKING.*:",
+   "if _t.TYPE_CHECKING:",
+   "if t.TYPE_CHECKING:",
+   "@overload",
+   'class .*\bProtocol\b.*\):',
+   "raise NotImplementedError",
+   'TODO: test this line'
 ]
 partial_branches = [
     "pragma: no branch",
@@ -326,4 +324,5 @@ partial_branches = [
     "if .* or not TYPE_CHECKING:",
     "if .* or not _t.TYPE_CHECKING:",
     "if .* or not t.TYPE_CHECKING:",
+   'TODO: test this branch',
 ]

src/trio/_core/_io_kqueue.py

@@ -98,7 +98,7 @@ def process_events(self, events: EventResult) -> None:
             if isinstance(receiver, _core.Task):
                 _core.reschedule(receiver, outcome.Value(event))
             else:
-                receiver.put_nowait(event)
+                receiver.put_nowait(event)  # TODO: test this line
 
     # kevent registration is complicated -- e.g. aio submission can
     # implicitly perform a EV_ADD, and EVFILT_PROC with NOTE_TRACK will
@@ -162,7 +162,7 @@ async def wait_kevent(
 
         def abort(raise_cancel: RaiseCancelT) -> Abort:
             r = abort_func(raise_cancel)
-            if r is _core.Abort.SUCCEEDED:
+            if r is _core.Abort.SUCCEEDED:  # TODO: test this branch
                 del self._registered[key]
             return r
 

src/trio/_core/_io_windows.py

@@ -856,9 +856,9 @@ async def wait_overlapped(
         <https://github.com/python-trio/trio/issues/52>`__.
         """
         handle = _handle(handle_)
-        if isinstance(lpOverlapped, int):
+        if isinstance(lpOverlapped, int):  # TODO: test this line
             lpOverlapped = ffi.cast("LPOVERLAPPED", lpOverlapped)
-        if lpOverlapped in self._overlapped_waiters:
+        if lpOverlapped in self._overlapped_waiters:  # TODO: test this line
             raise _core.BusyResourceError(
                 "another task is already waiting on that lpOverlapped",
             )

src/trio/_core/_run.py

@@ -115,7 +115,8 @@ def _public(fn: RetT) -> RetT:
 _r = random.Random()
 
 
-def _hypothesis_plugin_setup() -> None:
+# no cover because we don't check the hypothesis plugin works with hypothesis
+def _hypothesis_plugin_setup() -> None:  # pragma: no cover
     from hypothesis import register_random
 
     global _ALLOW_DETERMINISTIC_SCHEDULING

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 @@ def check(*, expected_readers: int, expected_writers: int) -> None:
             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,44 @@ def check(*, expected_readers: int, expected_writers: int) -> None:
         check(expected_readers=1, expected_writers=0)
 
 
+@pytest.mark.filterwarnings("ignore:.*UnboundedQueue:trio.TrioDeprecationWarning")
+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()
+
+        if sys.platform != "win32" and sys.platform != "linux":
+            # 1 for call_soon_task
+            check(expected_monitors=0, expected_readers=1, expected_writers=0)
+
+            with _core.monitor_kevent(a1.fileno(), select.KQ_FILTER_READ):
+                with (
+                    pytest.raises(_core.BusyResourceError),
+                    _core.monitor_kevent(a1.fileno(), select.KQ_FILTER_READ),
+                ):
+                    pass  # pragma: no cover
+                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/_core/_tests/test_run.py

@@ -11,6 +11,7 @@
 from contextlib import ExitStack, contextmanager, suppress
 from math import inf, nan
 from typing import TYPE_CHECKING, NoReturn, TypeVar
+from unittest import mock
 
 import outcome
 import pytest
@@ -26,7 +27,7 @@
     assert_checkpoints,
     wait_all_tasks_blocked,
 )
-from .._run import DEADLINE_HEAP_MIN_PRUNE_THRESHOLD
+from .._run import DEADLINE_HEAP_MIN_PRUNE_THRESHOLD, _count_context_run_tb_frames
 from .tutil import (
     check_sequence_matches,
     create_asyncio_future_in_new_loop,
@@ -371,6 +372,15 @@ async def test_cancel_scope_validation() -> None:
         match="^Cannot specify both a deadline and a relative deadline$",
     ):
         _core.CancelScope(deadline=7, relative_deadline=3)
+
+    with pytest.raises(ValueError, match="^deadline must not be NaN$"):
+        _core.CancelScope(deadline=nan)
+    with pytest.raises(ValueError, match="^relative deadline must not be NaN$"):
+        _core.CancelScope(relative_deadline=nan)
+
+    with pytest.raises(ValueError, match="^timeout must be non-negative$"):
+        _core.CancelScope(relative_deadline=-3)
+
     scope = _core.CancelScope()
 
     with pytest.raises(ValueError, match="^deadline must not be NaN$"):
@@ -2836,3 +2846,12 @@ async def handle_error() -> None:
 
     assert isinstance(exc, MyException)
     assert gc.get_referrers(exc) == no_other_refs()
+
+
+def test_context_run_tb_frames() -> None:
+    class Context:
+        def run(self, fn: Callable[[], object]) -> object:
+            return fn()
+
+    with mock.patch("trio._core._run.copy_context", return_value=Context()):
+        assert _count_context_run_tb_frames() == 1

src/trio/_dtls.py

@@ -58,7 +58,7 @@ def worst_case_mtu(sock: SocketType) -> int:
     if sock.family == trio.socket.AF_INET:
         return 576 - packet_header_overhead(sock)
     else:
-        return 1280 - packet_header_overhead(sock)
+        return 1280 - packet_header_overhead(sock)  # TODO: test this line
 
 
 def best_guess_mtu(sock: SocketType) -> int:
@@ -222,7 +222,7 @@ def decode_handshake_fragment_untrusted(payload: bytes) -> HandshakeFragment:
             frag_offset_bytes,
             frag_len_bytes,
         ) = HANDSHAKE_MESSAGE_HEADER.unpack_from(payload)
-    except struct.error as exc:
+    except struct.error as exc:  # TODO: test this line
         raise BadPacket("bad handshake message header") from exc
     # 'struct' doesn't have built-in support for 24-bit integers, so we
     # have to do it by hand. These can't fail.
@@ -425,14 +425,14 @@ def encode_volley(
         for message in messages:
             if isinstance(message, OpaqueHandshakeMessage):
                 encoded = encode_record(message.record)
-                if mtu - len(packet) - len(encoded) <= 0:
+                if mtu - len(packet) - len(encoded) <= 0:  # TODO: test this line
                     packets.append(packet)
                     packet = bytearray()
                 packet += encoded
                 assert len(packet) <= mtu
             elif isinstance(message, PseudoHandshakeMessage):
                 space = mtu - len(packet) - RECORD_HEADER.size - len(message.payload)
-                if space <= 0:
+                if space <= 0:  # TODO: test this line
                     packets.append(packet)
                     packet = bytearray()
                 packet += RECORD_HEADER.pack(
@@ -1326,9 +1326,8 @@ async def handler(dtls_channel):
             raise trio.BusyResourceError("another task is already listening")
         try:
             self.socket.getsockname()
-        except OSError:
-            # TODO: Write test that triggers this
-            raise RuntimeError(  # pragma: no cover
+        except OSError:  # TODO: test this line
+            raise RuntimeError(
                 "DTLS socket must be bound before it can serve",
             ) from None
         self._ensure_receive_loop()

src/trio/_tests/test_dtls.py

@@ -75,7 +75,9 @@ async def echo_handler(dtls_channel: DTLSChannel) -> None:
                     print("server starting do_handshake")
                     await dtls_channel.do_handshake()
                     print("server finished do_handshake")
-                    async for packet in dtls_channel:
+                    # no branch for leaving this for loop because we only leave
+                    # a channel by cancellation.
+                    async for packet in dtls_channel:  # pragma: no branch
                         print(f"echoing {packet!r} -> {dtls_channel.peer_address!r}")
                         await dtls_channel.send(packet)
                 except trio.BrokenResourceError:  # pragma: no cover

src/trio/_tests/test_exports.py

@@ -384,8 +384,11 @@ def lookup_symbol(symbol: str) -> dict[str, str]:
         elif tool == "mypy":
             # load the cached type information
             cached_type_info = cache_json["names"][class_name]
-            if "node" not in cached_type_info:
-                cached_type_info = lookup_symbol(cached_type_info["cross_ref"])
+            # previously this was an 'if' but it seems it's no longer possible
+            # for this cache to contain 'node', if this assert raises for you
+            # please let us know!
+            assert "node" not in cached_type_info
+            cached_type_info = lookup_symbol(cached_type_info["cross_ref"])
 
             assert "node" in cached_type_info
             node = cached_type_info["node"]

src/trio/_tests/test_socket.py

@@ -376,9 +376,7 @@ async def test_sniff_sockopts() -> None:
     from socket import AF_INET, AF_INET6, SOCK_DGRAM, SOCK_STREAM
 
     # generate the combinations of families/types we're testing:
-    families = [AF_INET]
-    if can_create_ipv6:
-        families.append(AF_INET6)
+    families = (AF_INET, AF_INET6) if can_create_ipv6 else (AF_INET,)
     sockets = [
         stdlib_socket.socket(family, type_)
         for family in families

src/trio/_tests/test_ssl.py

@@ -210,27 +210,11 @@ def __init__(
         # we still have to support versions before that, and that means we
         # need to test renegotiation support, which means we need to force this
         # to use a lower version where this test server can trigger
-        # renegotiations. Of course TLS 1.3 support isn't released yet, but
-        # I'm told that this will work once it is. (And once it is we can
-        # remove the pragma: no cover too.) Alternatively, we could switch to
-        # using TLSv1_2_METHOD.
-        #
-        # Discussion: https://github.com/pyca/pyopenssl/issues/624
-
-        # This is the right way, but we can't use it until this PR is in a
-        # released:
-        #     https://github.com/pyca/pyopenssl/pull/861
-        #
-        # if hasattr(SSL, "OP_NO_TLSv1_3"):
-        #     ctx.set_options(SSL.OP_NO_TLSv1_3)
-        #
-        # Fortunately pyopenssl uses cryptography under the hood, so we can be
-        # confident that they're using the same version of openssl
+        # renegotiations.
         from cryptography.hazmat.bindings.openssl.binding import Binding
 
         b = Binding()
-        if hasattr(b.lib, "SSL_OP_NO_TLSv1_3"):
-            ctx.set_options(b.lib.SSL_OP_NO_TLSv1_3)
+        ctx.set_options(b.lib.SSL_OP_NO_TLSv1_3)
 
         # Unfortunately there's currently no way to say "use 1.3 or worse", we
         # can only disable specific versions. And if the two sides start