more robust selector thread test

for thread asynciness

Author: minrk

Lib/test/test_asyncio/test_selector_thread.py

@@ -24,18 +24,30 @@ async def asyncTearDown(self):
 
     async def test_slow_reader(self):
         a, b = self.socketpair()
+        first_recv = asyncio.Future()
+
         def recv():
-            b.recv(100)
+            msg = b.recv(100)
+            if not first_recv.done():
+                first_recv.set_result(msg)
+
         mock_recv = mock.MagicMock(wraps=recv)
         # make sure select is only called once when
         # event loop thread is slow to consume events
         a.sendall(b"msg")
         with mock.patch("select.select", wraps=select.select) as mock_select:
             self.selector_thread.add_reader(b, mock_recv)
+            # ready event, but main event loop is blocked for some time
             time.sleep(0.1)
-        self.assertEqual(mock_select.call_count, 1)
-        await asyncio.sleep(0.1)
+            recvd = await asyncio.wait_for(first_recv, timeout=10)
+        self.assertEqual(recvd, b"msg")
+        # make sure recv wasn't scheduled more than once
         self.assertEqual(mock_recv.call_count, 1)
+        # 1 for add_reader
+        # 1 for finishing reader callback
+        # up to 2 more for wake FD calls if CI is slow
+        # this would be thousands if select is busy-looping while the main thread blocks
+        self.assertLessEqual(mock_select.call_count, 5)
 
     async def test_reader_error(self):
         # test error handling in callbacks doesn't break handling
@@ -44,10 +56,12 @@ async def test_reader_error(self):
 
         selector_thread = self.selector_thread
 
-        # make sure it's called a few
+        # make sure it's called a few times,
+        # and errors don't prevent rescheduling
         n_failures = 5
         counter = 0
         bad_recv_done = asyncio.Future()
+
         def bad_recv(sock):
             # fail the first n_failures calls, then succeed
             nonlocal counter
@@ -67,7 +81,9 @@ def bad_recv(sock):
 
         # make sure start_select is called
         # even when recv callback errors,
-        with mock.patch.object(selector_thread, "_start_select", wraps=selector_thread._start_select) as start_select:
+        with mock.patch.object(
+            selector_thread, "_start_select", wraps=selector_thread._start_select
+        ) as start_select:
             await asyncio.wait_for(bad_recv_done, timeout=10)
 
         # make sure recv is called N + 1 times,

Lib/test/test_asyncio/test_windows_events.py

@@ -325,7 +325,7 @@ def threadMain():
 
     def test_add_reader_invalid_argument(self):
         def assert_raises():
-            return self.assertRaisesRegex(ValueError, r'Invalid file object')
+            return self.assertRaisesRegex(ValueError, r"Invalid file object")
 
         def cb(sock):
             return None
@@ -343,6 +343,7 @@ def cb(sock):
     def test_selector_thread(self):
         assert self.loop._selector_thread is None
         a, b = socket.socketpair()
+
         async def _test():
             read_future = asyncio.Future()
             sent = b"asdf"