add some tests for kqueue

python-trio · Dec 23, 2024 · 009cbb5 · 009cbb5
1 parent 31891d1
commit 009cbb5
Showing 1 changed file with 9 additions and 1 deletion.
diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py
@@ -410,13 +410,21 @@ def check(
             # 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 _core.monitor_kevent(a1.fileno(), select.KQ_FILTER_READ) as q:
                 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)
+                b1.send(b"\x00")
+                with trio.fail_after(1):
+                    await q.receive()
+
+                _core.notify_closing(a1)
+                a1.close()
+                with trio.fail_after(1):
+                    assert len([v async for v in q]) == 0
 
             check(expected_monitors=0, expected_readers=1, expected_writers=0)