1import os 2import signal 3import subprocess 4import sys 5import time 6import unittest 7 8 9class SIGUSR1Exception(Exception): 10 pass 11 12 13class InterProcessSignalTests(unittest.TestCase): 14 def setUp(self): 15 self.got_signals = {'SIGHUP': 0, 'SIGUSR1': 0, 'SIGALRM': 0} 16 17 def sighup_handler(self, signum, frame): 18 self.got_signals['SIGHUP'] += 1 19 20 def sigusr1_handler(self, signum, frame): 21 self.got_signals['SIGUSR1'] += 1 22 raise SIGUSR1Exception 23 24 def wait_signal(self, child, signame, exc_class=None): 25 try: 26 if child is not None: 27 # This wait should be interrupted by exc_class 28 # (if set) 29 child.wait() 30 31 timeout = 10.0 32 deadline = time.monotonic() + timeout 33 34 while time.monotonic() < deadline: 35 if self.got_signals[signame]: 36 return 37 signal.pause() 38 except BaseException as exc: 39 if exc_class is not None and isinstance(exc, exc_class): 40 # got the expected exception 41 return 42 raise 43 44 self.fail('signal %s not received after %s seconds' 45 % (signame, timeout)) 46 47 def subprocess_send_signal(self, pid, signame): 48 code = 'import os, signal; os.kill(%s, signal.%s)' % (pid, signame) 49 args = [sys.executable, '-I', '-c', code] 50 return subprocess.Popen(args) 51 52 def test_interprocess_signal(self): 53 # Install handlers. This function runs in a sub-process, so we 54 # don't worry about re-setting the default handlers. 55 signal.signal(signal.SIGHUP, self.sighup_handler) 56 signal.signal(signal.SIGUSR1, self.sigusr1_handler) 57 signal.signal(signal.SIGUSR2, signal.SIG_IGN) 58 signal.signal(signal.SIGALRM, signal.default_int_handler) 59 60 # Let the sub-processes know who to send signals to. 61 pid = str(os.getpid()) 62 63 with self.subprocess_send_signal(pid, "SIGHUP") as child: 64 self.wait_signal(child, 'SIGHUP') 65 self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 0, 66 'SIGALRM': 0}) 67 68 with self.subprocess_send_signal(pid, "SIGUSR1") as child: 69 self.wait_signal(child, 'SIGUSR1', SIGUSR1Exception) 70 self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1, 71 'SIGALRM': 0}) 72 73 with self.subprocess_send_signal(pid, "SIGUSR2") as child: 74 # Nothing should happen: SIGUSR2 is ignored 75 child.wait() 76 77 signal.alarm(1) 78 self.wait_signal(None, 'SIGALRM', KeyboardInterrupt) 79 self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1, 80 'SIGALRM': 0}) 81 82 83if __name__ == "__main__": 84 unittest.main() 85