1import os
2import signal
3import subprocess
4import sys
5import time
6import unittest
7from test import support
8
9
10class SIGUSR1Exception(Exception):
11    pass
12
13
14class InterProcessSignalTests(unittest.TestCase):
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        timeout = support.SHORT_TIMEOUT
32        deadline = time.monotonic() + timeout
33
34        while time.monotonic() < deadline:
35            if self.got_signals[signame]:
36                return
37            signal.pause()
38
39        self.fail('signal %s not received after %s seconds'
40                  % (signame, timeout))
41
42    def subprocess_send_signal(self, pid, signame):
43        code = 'import os, signal; os.kill(%s, signal.%s)' % (pid, signame)
44        args = [sys.executable, '-I', '-c', code]
45        return subprocess.Popen(args)
46
47    def test_interprocess_signal(self):
48        # Install handlers. This function runs in a sub-process, so we
49        # don't worry about re-setting the default handlers.
50        signal.signal(signal.SIGHUP, self.sighup_handler)
51        signal.signal(signal.SIGUSR1, self.sigusr1_handler)
52        signal.signal(signal.SIGUSR2, signal.SIG_IGN)
53        signal.signal(signal.SIGALRM, signal.default_int_handler)
54
55        # Let the sub-processes know who to send signals to.
56        pid = str(os.getpid())
57
58        with self.subprocess_send_signal(pid, "SIGHUP") as child:
59            self.wait_signal(child, 'SIGHUP')
60        self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 0,
61                                            'SIGALRM': 0})
62
63        with self.assertRaises(SIGUSR1Exception):
64            with self.subprocess_send_signal(pid, "SIGUSR1") as child:
65                self.wait_signal(child, 'SIGUSR1')
66        self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
67                                            'SIGALRM': 0})
68
69        with self.subprocess_send_signal(pid, "SIGUSR2") as child:
70            # Nothing should happen: SIGUSR2 is ignored
71            child.wait()
72
73        try:
74            with self.assertRaises(KeyboardInterrupt):
75                signal.alarm(1)
76                self.wait_signal(None, 'SIGALRM')
77            self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
78                                                'SIGALRM': 0})
79        finally:
80            signal.alarm(0)
81
82
83if __name__ == "__main__":
84    unittest.main()
85