mirrors
/
cpython
mirror of https://github.com/python/cpython.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[3.13] gh-130737: Fix multiprocessing test_notify() (GH-130797) (#130802) 

gh-130737: Fix multiprocessing test_notify() (GH-130797)

Replace hardcoded delay (100 ms) with a loop awaiting until a
condition is true: replace assertReturnsIfImplemented() with
assertReachesEventually().

Use sleeping_retry() in assertReachesEventually() to tolerate slow
buildbots and raise an exception on timeout (30 seconds).
(cherry picked from commit 8a64a62002)

Co-authored-by: Victor Stinner <vstinner@python.org>
This commit is contained in:
Miss Islington (bot) 2025-03-03 20:09:34 +01:00 committed by GitHub
parent 4c6318e24d
commit d00e92bc04
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 5 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
Lib/test/_test_multiprocessing.py
View File

@ -1577,14 +1577,13 @@ def f(cls, cond, sleeping, woken, timeout=None):
cond.release()
def assertReachesEventually(self, func, value):
for i in range(10):
for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
try:
if func() == value:
break
except NotImplementedError:
break
time.sleep(DELTA)
time.sleep(DELTA)
self.assertReturnsIfImplemented(value, func)
def check_invariant(self, cond):
@ -1618,8 +1617,7 @@ def test_notify(self):
sleeping.acquire()
# check no process/thread has woken up
time.sleep(DELTA)
self.assertReturnsIfImplemented(0, get_value, woken)
self.assertReachesEventually(lambda: get_value(woken), 0)
# wake up one process/thread
cond.acquire()
@ -1627,8 +1625,7 @@ def test_notify(self):
cond.release()
# check one process/thread has woken up
time.sleep(DELTA)
self.assertReturnsIfImplemented(1, get_value, woken)
self.assertReachesEventually(lambda: get_value(woken), 1)
# wake up another
cond.acquire()
@ -1636,8 +1633,7 @@ def test_notify(self):
cond.release()
# check other has woken up
time.sleep(DELTA)
self.assertReturnsIfImplemented(2, get_value, woken)
self.assertReachesEventually(lambda: get_value(woken), 2)
# check state is not mucked up
self.check_invariant(cond)