test: assert task-group cancellation directly in stdio EOF monitor tests

Author: Bortlesboat

tests/server/test_stdio.py

@@ -91,24 +91,22 @@ async def test_stdin_eof_monitor_detects_hangup():
         mock_buffer = MagicMock()
         mock_buffer.fileno.return_value = read_fd
 
-        cancelled = False
         with patch.object(sys, "platform", "linux"), patch.object(sys, "stdin", MagicMock(buffer=mock_buffer)):
-            with anyio.CancelScope() as scope:
-                async with anyio.create_task_group() as tg:
-                    monitor = _create_stdin_eof_monitor(tg)
-                    assert monitor is not None
-                    tg.start_soon(monitor)
-
-                    # Close the write end to trigger POLLHUP on read end
-                    os.close(write_fd)
-                    write_fd = -1
-
-                    # The monitor will cancel the task group scope when it
-                    # detects POLLHUP. Wait with a timeout to avoid hanging.
-                    with anyio.fail_after(5):
-                        await anyio.sleep(10)  # will be cancelled by monitor
-            cancelled = scope.cancel_called
-        assert cancelled
+            async with anyio.create_task_group() as tg:
+                monitor = _create_stdin_eof_monitor(tg)
+                assert monitor is not None
+                tg.start_soon(monitor)
+
+                # Close the write end to trigger POLLHUP on read end
+                os.close(write_fd)
+                write_fd = -1
+
+                # Wait for the monitor to cancel the task-group scope.
+                with anyio.fail_after(5):
+                    while not tg.cancel_scope.cancel_called:
+                        await anyio.sleep(0.05)
+
+                assert tg.cancel_scope.cancel_called
     finally:
         os.close(read_fd)
         if write_fd != -1:  # pragma: no cover
@@ -143,9 +141,12 @@ async def test_stdin_eof_monitor_ignores_pollin_events():
                 os.close(write_fd)
                 write_fd = -1
 
-                # Wait for the monitor to detect POLLHUP and cancel
+                # Wait for the monitor to detect POLLHUP and cancel.
                 with anyio.fail_after(5):
-                    await anyio.sleep(10)  # will be cancelled by monitor
+                    while not tg.cancel_scope.cancel_called:
+                        await anyio.sleep(0.05)
+
+                assert tg.cancel_scope.cancel_called
     finally:
         os.close(read_fd)
         if write_fd != -1:  # pragma: no cover