Issue #8407: pthread_sigmask() checks immediatly if signal handlers have been

called. The test checks that SIG_UNBLOCK calls immediatly the signal handler of
the pending SIGUSR1. Improve also the tests using an exception (division by
zero) instead of a flag (a function attribute).
This commit is contained in:
Victor Stinner 2011-05-03 14:57:12 +02:00
parent 2d4a91e0d0
commit d0e516db50
2 changed files with 20 additions and 14 deletions

View File

@ -498,8 +498,7 @@ class PthreadSigmaskTests(unittest.TestCase):
signum = signal.SIGUSR1 signum = signal.SIGUSR1
def handler(signum, frame): def handler(signum, frame):
handler.tripped = True 1/0
handler.tripped = False
def read_sigmask(): def read_sigmask():
return signal.pthread_sigmask(signal.SIG_BLOCK, []) return signal.pthread_sigmask(signal.SIG_BLOCK, [])
@ -519,36 +518,39 @@ class PthreadSigmaskTests(unittest.TestCase):
# function. # function.
faulthandler.cancel_dump_tracebacks_later() faulthandler.cancel_dump_tracebacks_later()
# Install our signal handler
old_handler = signal.signal(signum, handler) old_handler = signal.signal(signum, handler)
self.addCleanup(signal.signal, signum, old_handler) self.addCleanup(signal.signal, signum, old_handler)
# unblock SIGUSR1, copy the old mask and test our signal handler # Unblock SIGUSR1 (and copy the old mask) to test our signal handler
old_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) old_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
self.addCleanup(signal.pthread_sigmask, signal.SIG_SETMASK, old_mask) self.addCleanup(signal.pthread_sigmask, signal.SIG_SETMASK, old_mask)
os.kill(pid, signum) with self.assertRaises(ZeroDivisionError):
self.assertTrue(handler.tripped) os.kill(pid, signum)
# block SIGUSR1 # Block and then raise SIGUSR1. The signal is blocked: the signal
handler.tripped = False # handler is not called, and the signal is now pending
signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
os.kill(pid, signum) os.kill(pid, signum)
self.assertFalse(handler.tripped)
# check the mask # Check the new mask
blocked = read_sigmask() blocked = read_sigmask()
self.assertIn(signum, blocked) self.assertIn(signum, blocked)
self.assertEqual(set(old_mask) ^ set(blocked), {signum}) self.assertEqual(set(old_mask) ^ set(blocked), {signum})
# unblock SIGUSR1 # Unblock SIGUSR1
signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) with self.assertRaises(ZeroDivisionError):
os.kill(pid, signum) # unblock the pending signal calls immediatly the signal handler
self.assertTrue(handler.tripped) signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
with self.assertRaises(ZeroDivisionError):
os.kill(pid, signum)
# check the mask # Check the new mask
unblocked = read_sigmask() unblocked = read_sigmask()
self.assertNotIn(signum, unblocked) self.assertNotIn(signum, unblocked)
self.assertEqual(set(blocked) ^ set(unblocked), {signum}) self.assertEqual(set(blocked) ^ set(unblocked), {signum})
self.assertSequenceEqual(old_mask, unblocked) self.assertSequenceEqual(old_mask, unblocked)
# Finally, restore the previous signal handler and the signal mask
def test_main(): def test_main():

View File

@ -573,6 +573,10 @@ signal_pthread_sigmask(PyObject *self, PyObject *args)
return NULL; return NULL;
} }
/* if signals was unblocked, signal handlers have been called */
if (PyErr_CheckSignals())
return NULL;
result = PyList_New(0); result = PyList_New(0);
if (result == NULL) if (result == NULL)
return NULL; return NULL;