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