Michael W. Hudson | 43220ea | 2004-08-03 14:37:14 +0000 | [diff] [blame] | 1 | """PyUnit testing that threads honor our signal semantics""" |
| 2 | |
| 3 | import unittest |
| 4 | import thread |
| 5 | import signal |
| 6 | import os |
| 7 | from test import test_support |
| 8 | |
| 9 | signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 }, |
| 10 | signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 }, |
| 11 | signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } } |
| 12 | |
| 13 | process_pid = os.getpid() |
| 14 | signalled_all=thread.allocate_lock() |
| 15 | |
| 16 | |
| 17 | def registerSignals((for_usr1, for_usr2, for_alrm)): |
| 18 | usr1 = signal.signal(signal.SIGUSR1, for_usr1) |
| 19 | usr2 = signal.signal(signal.SIGUSR2, for_usr2) |
| 20 | alrm = signal.signal(signal.SIGALRM, for_alrm) |
| 21 | return usr1, usr2, alrm |
| 22 | |
| 23 | |
| 24 | # The signal handler. Just note that the signal occured and |
| 25 | # from who. |
| 26 | def handle_signals(sig,frame): |
| 27 | signal_blackboard[sig]['tripped'] += 1 |
| 28 | signal_blackboard[sig]['tripped_by'] = thread.get_ident() |
| 29 | |
| 30 | # a function that will be spawned as a separate thread. |
| 31 | def send_signals(): |
| 32 | os.kill(process_pid, signal.SIGUSR1) |
| 33 | os.kill(process_pid, signal.SIGUSR2) |
| 34 | signalled_all.release() |
| 35 | |
| 36 | class ThreadSignals(unittest.TestCase): |
| 37 | """Test signal handling semantics of threads. |
| 38 | We spawn a thread, have the thread send two signals, and |
| 39 | wait for it to finish. Check that we got both signals |
| 40 | and that they were run by the main thread. |
| 41 | """ |
| 42 | def test_signals(self): |
| 43 | signalled_all.acquire() |
| 44 | self.spawnSignallingThread() |
| 45 | signalled_all.acquire() |
| 46 | # the signals that we asked the kernel to send |
| 47 | # will come back, but we don't know when. |
| 48 | # (it might even be after the thread exits |
| 49 | # and might be out of order.) If we haven't seen |
| 50 | # the signals yet, send yet another signal and |
| 51 | # wait for it return. |
| 52 | if signal_blackboard[signal.SIGUSR2]['tripped'] == 0 \ |
| 53 | or signal_blackboard[signal.SIGUSR2]['tripped'] == 0: |
| 54 | signal.alarm(1) |
| 55 | signal.pause() |
| 56 | signal.alarm(0) |
| 57 | |
| 58 | self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1) |
| 59 | self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'], |
| 60 | thread.get_ident()) |
| 61 | self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1) |
| 62 | self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'], |
| 63 | thread.get_ident()) |
| 64 | |
| 65 | def spawnSignallingThread(self): |
| 66 | thread.start_new_thread(send_signals, ()) |
| 67 | |
| 68 | |
| 69 | def test_main(): |
| 70 | oldsigs = registerSignals((handle_signals, handle_signals, handle_signals)) |
| 71 | try: |
| 72 | test_support.run_unittest(ThreadSignals) |
| 73 | finally: |
| 74 | registerSignals(oldsigs) |
| 75 | |
| 76 | if __name__ == '__main__': |
| 77 | test_main() |