Home | History | Annotate | Download | only in test 
      1 import os
      2 import signal
      3 import subprocess
      4 import sys
      5 import time
      6 import unittest
      7 
      8 
      9 class SIGUSR1Exception(Exception):
     10     pass
     11 
     12 
     13 class 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 
     83 if __name__ == "__main__":
     84     unittest.main()
     85