traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

bpo-35426: Eliminate race condition in test_interprocess_signal (GH-11087) 

The test only except SIGUSR1Exception inside wait_signal(), but the signal can be sent during subprocess_send_signal() call.
This commit is contained in:
Pablo Galindo 2018-12-11 11:32:12 +00:00 committed by GitHub
parent a932d0b496
commit 2ab2afd387
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 16 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
Lib/test/signalinterproctester.py
View File

@ -21,25 +21,19 @@ class InterProcessSignalTests(unittest.TestCase):
self.got_signals['SIGUSR1'] += 1
raise SIGUSR1Exception
def wait_signal(self, child, signame, exc_class=None):
try:
if child is not None:
# This wait should be interrupted by exc_class
# (if set)
child.wait()
def wait_signal(self, child, signame):
if child is not None:
# This wait should be interrupted by exc_class
# (if set)
child.wait()
timeout = 10.0
deadline = time.monotonic() + timeout
timeout = 10.0
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if self.got_signals[signame]:
return
signal.pause()
except BaseException as exc:
if exc_class is not None and isinstance(exc, exc_class):
# got the expected exception
while time.monotonic() < deadline:
if self.got_signals[signame]:
return
raise
signal.pause()
self.fail('signal %s not received after %s seconds'
% (signame, timeout))
@ -65,8 +59,9 @@ class InterProcessSignalTests(unittest.TestCase):
self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 0,
'SIGALRM': 0})
with self.subprocess_send_signal(pid, "SIGUSR1") as child:
self.wait_signal(child, 'SIGUSR1', SIGUSR1Exception)
with self.assertRaises(SIGUSR1Exception):
with self.subprocess_send_signal(pid, "SIGUSR1") as child:
self.wait_signal(child, 'SIGUSR1')
self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
'SIGALRM': 0})
@ -75,8 +70,9 @@ class InterProcessSignalTests(unittest.TestCase):
child.wait()
try:
signal.alarm(1)
self.wait_signal(None, 'SIGALRM', KeyboardInterrupt)
with self.assertRaises(KeyboardInterrupt):
signal.alarm(1)
self.wait_signal(None, 'SIGALRM')
self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
'SIGALRM': 0})
finally: