mirror of
https://github.com/python/cpython.git
synced 2025-07-25 12:14:38 +00:00
1742f3331f
appears to be utterly insane. Plug some theoretical insecurities in the test script: - Verify that the SIGALRM handler was actually installed. - Don't call alarm() before the handler is installed. - Move everything that can fail inside the try/finally, so the test cleans up after itself more often. - Try sending all the expected signals in force_test_exit(), not just SIGALRM. Since that was fixed to actually send SIGALRM (instead of invisibly dying with an AttributeError), we've seen that sending SIGALRM alone does not stop this from hanging. - Move the "kill the child" business into the finally clause, so the child doesn't survive test failure to send SIGALRM to other tests later (there are also baffling SIGALRM-related failures in test_socket). - Cancel the alarm in the finally clause -- if the test dies early, we again don't want SIGALRM showing up to confuse a later test. Alas, this still relies on timing luck wrt the spawned script that sends the test signals, but it's hard to see how waiting for seconds can so often be so unlucky. test_threadedsignals: curiously, this test never fails on Tru64, but doesn't normally signal SIGALRM. Anyway, fixed an obvious (but probably inconsequential) logic error.
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""PyUnit testing that threads honor our signal semantics"""
|
|
|
|
import unittest
|
|
import thread
|
|
import signal
|
|
import os
|
|
import sys
|
|
from test.test_support import run_unittest, TestSkipped
|
|
|
|
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
|
|
raise TestSkipped, "Can't test signal on %s" % sys.platform
|
|
|
|
process_pid = os.getpid()
|
|
signalled_all=thread.allocate_lock()
|
|
|
|
|
|
def registerSignals((for_usr1, for_usr2, for_alrm)):
|
|
usr1 = signal.signal(signal.SIGUSR1, for_usr1)
|
|
usr2 = signal.signal(signal.SIGUSR2, for_usr2)
|
|
alrm = signal.signal(signal.SIGALRM, for_alrm)
|
|
return usr1, usr2, alrm
|
|
|
|
|
|
# The signal handler. Just note that the signal occurred and
|
|
# from who.
|
|
def handle_signals(sig,frame):
|
|
signal_blackboard[sig]['tripped'] += 1
|
|
signal_blackboard[sig]['tripped_by'] = thread.get_ident()
|
|
|
|
# a function that will be spawned as a separate thread.
|
|
def send_signals():
|
|
os.kill(process_pid, signal.SIGUSR1)
|
|
os.kill(process_pid, signal.SIGUSR2)
|
|
signalled_all.release()
|
|
|
|
class ThreadSignals(unittest.TestCase):
|
|
"""Test signal handling semantics of threads.
|
|
We spawn a thread, have the thread send two signals, and
|
|
wait for it to finish. Check that we got both signals
|
|
and that they were run by the main thread.
|
|
"""
|
|
def test_signals(self):
|
|
signalled_all.acquire()
|
|
self.spawnSignallingThread()
|
|
signalled_all.acquire()
|
|
# the signals that we asked the kernel to send
|
|
# will come back, but we don't know when.
|
|
# (it might even be after the thread exits
|
|
# and might be out of order.) If we haven't seen
|
|
# the signals yet, send yet another signal and
|
|
# wait for it return.
|
|
if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
|
|
or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
|
|
signal.alarm(1)
|
|
signal.pause()
|
|
signal.alarm(0)
|
|
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
|
|
thread.get_ident())
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
|
|
thread.get_ident())
|
|
signalled_all.release()
|
|
|
|
def spawnSignallingThread(self):
|
|
thread.start_new_thread(send_signals, ())
|
|
|
|
|
|
def test_main():
|
|
global signal_blackboard
|
|
|
|
signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },
|
|
signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },
|
|
signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }
|
|
|
|
oldsigs = registerSignals((handle_signals, handle_signals, handle_signals))
|
|
try:
|
|
run_unittest(ThreadSignals)
|
|
finally:
|
|
registerSignals(oldsigs)
|
|
|
|
if __name__ == '__main__':
|
|
test_main()
|