Oct 13, 2023 · Oct 12, 2023 · Oct 13, 2023 · Oct 13, 2023 · Oct 13, 2023 · Oct 13, 2023
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
        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.
Original file line number	Diff line number	Diff line change
Expand Up		@@ -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.
Expand Down