0001 """PyUnit testing that threads honor our signal semantics""" 0002 0003 import unittest 0004 import thread 0005 import signal 0006 import os 0007 import sys 0008 from test.test_support import run_unittest, TestSkipped 0009 0010 if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos': 0011 raise TestSkipped, "Can't test signal on %s" % sys.platform 0012 0013 process_pid = os.getpid() 0014 signalled_all=thread.allocate_lock() 0015 0016 0017 def registerSignals((for_usr1, for_usr2, for_alrm)): 0018 usr1 = signal.signal(signal.SIGUSR1, for_usr1) 0019 usr2 = signal.signal(signal.SIGUSR2, for_usr2) 0020 alrm = signal.signal(signal.SIGALRM, for_alrm) 0021 return usr1, usr2, alrm 0022 0023 0024 # The signal handler. Just note that the signal occured and 0025 # from who. 0026 def handle_signals(sig,frame): 0027 signal_blackboard[sig]['tripped'] += 1 0028 signal_blackboard[sig]['tripped_by'] = thread.get_ident() 0029 0030 # a function that will be spawned as a separate thread. 0031 def send_signals(): 0032 os.kill(process_pid, signal.SIGUSR1) 0033 os.kill(process_pid, signal.SIGUSR2) 0034 signalled_all.release() 0035 0036 class ThreadSignals(unittest.TestCase): 0037 """Test signal handling semantics of threads. 0038 We spawn a thread, have the thread send two signals, and 0039 wait for it to finish. Check that we got both signals 0040 and that they were run by the main thread. 0041 """ 0042 def test_signals(self): 0043 signalled_all.acquire() 0044 self.spawnSignallingThread() 0045 signalled_all.acquire() 0046 # the signals that we asked the kernel to send 0047 # will come back, but we don't know when. 0048 # (it might even be after the thread exits 0049 # and might be out of order.) If we haven't seen 0050 # the signals yet, send yet another signal and 0051 # wait for it return. 0052 if signal_blackboard[signal.SIGUSR2]['tripped'] == 0 \ 0053 or signal_blackboard[signal.SIGUSR2]['tripped'] == 0: 0054 signal.alarm(1) 0055 signal.pause() 0056 signal.alarm(0) 0057 0058 self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1) 0059 self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'], 0060 thread.get_ident()) 0061 self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1) 0062 self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'], 0063 thread.get_ident()) 0064 signalled_all.release() 0065 0066 def spawnSignallingThread(self): 0067 thread.start_new_thread(send_signals, ()) 0068 0069 0070 def test_main(): 0071 global signal_blackboard 0072 0073 signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 }, 0074 signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 }, 0075 signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } } 0076 0077 oldsigs = registerSignals((handle_signals, handle_signals, handle_signals)) 0078 try: 0079 run_unittest(ThreadSignals) 0080 finally: 0081 registerSignals(oldsigs) 0082 0083 if __name__ == '__main__': 0084 test_main() 0085
Generated by PyXR 0.9.4