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