Add tests for custom event loop implementation.

python · gvanrossum · Oct 13, 2023 · Oct 12, 2023 · Oct 13, 2023 · Oct 13, 2023
commit 6672f6875eee9d281889f1fc34180c913f4faed2
@@ -922,6 +922,43 @@ def test_run_forever_pre_stopped(self):
         self.loop.run_forever()
         self.loop._selector.select.assert_called_once_with(0)
 
+    def test_custom_run_forever_integration(self):
+        # Test that the run_forever_setup() and run_forever_cleanup() primitives
+        # can be used to implement a custom run_forever loop.
+        self.loop._process_events = mock.Mock()
+
+        count = 0
+
+        def callback():
+            nonlocal count
+            count += 1
+
+        self.loop.call_soon(callback)
+
+        # Set up the custom event loop
+        orig_state = self.loop.run_forever_setup()
+
+        # Confirm the loop has been started
+        self.assertEqual(asyncio.get_running_loop(), self.loop)
+        self.assertTrue(self.loop.is_running())
+
+        # Our custom "event loop" just iterates 10 times before exiting.
+        for i in range(10):
+            self.loop._run_once()
+
+        # Clean up the event loop
+        self.loop.run_forever_cleanup(orig_state)
+
+        # Confirm the loop has been cleaned up
+        with self.assertRaises(RuntimeError):
+            asyncio.get_running_loop()
+        self.assertFalse(self.loop.is_running())
+
+        # Confirm the loop actually did run, processing events 10 times,
+        # and invoking the callback once.
+        self.assertEqual(self.loop._process_events.call_count, 10)
+        self.assertEqual(count, 1)
+
     async def leave_unfinalized_asyncgen(self):
         # Create an async generator, iterate it partially, and leave it
         # to be garbage collected.