https://github.com/python/cpython/commit/6c48ed7d62c6ca0eb24935b0e612f9e4a1a3b1bc
commit: 6c48ed7d62c6ca0eb24935b0e612f9e4a1a3b1bc
branch: main
author: Victor Stinner <[email protected]>
committer: vstinner <[email protected]>
date: 2025-03-04T11:16:07Z
summary:
gh-130736: Fix asyncio test_shutdown_default_executor_timeout() (#130800)
Replace time.sleep() with threading.Event.
files:
M Lib/test/test_asyncio/test_base_events.py
diff --git a/Lib/test/test_asyncio/test_base_events.py
b/Lib/test/test_asyncio/test_base_events.py
index 8cf1f6891faf97..7838d1fccb88e1 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -233,20 +233,25 @@ def test_set_default_executor_error(self):
self.assertIsNone(self.loop._default_executor)
def test_shutdown_default_executor_timeout(self):
+ event = threading.Event()
+
class DummyExecutor(concurrent.futures.ThreadPoolExecutor):
def shutdown(self, wait=True, *, cancel_futures=False):
if wait:
- time.sleep(0.1)
+ event.wait()
self.loop._process_events = mock.Mock()
self.loop._write_to_self = mock.Mock()
executor = DummyExecutor()
self.loop.set_default_executor(executor)
- with self.assertWarnsRegex(RuntimeWarning,
- "The executor did not finishing joining"):
- self.loop.run_until_complete(
- self.loop.shutdown_default_executor(timeout=0.01))
+ try:
+ with self.assertWarnsRegex(RuntimeWarning,
+ "The executor did not finishing
joining"):
+ self.loop.run_until_complete(
+ self.loop.shutdown_default_executor(timeout=0.01))
+ finally:
+ event.set()
def test_call_soon(self):
def cb():
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: [email protected]
