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