• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import gc
2import os
3import signal
4import subprocess
5import sys
6import time
7import unittest
8from test import support
9
10
11class SIGUSR1Exception(Exception):
12    pass
13
14
15class InterProcessSignalTests(unittest.TestCase):
16    def setUp(self):
17        self.got_signals = {'SIGHUP': 0, 'SIGUSR1': 0, 'SIGALRM': 0}
18
19    def sighup_handler(self, signum, frame):
20        self.got_signals['SIGHUP'] += 1
21
22    def sigusr1_handler(self, signum, frame):
23        self.got_signals['SIGUSR1'] += 1
24        raise SIGUSR1Exception
25
26    def wait_signal(self, child, signame):
27        if child is not None:
28            # This wait should be interrupted by exc_class
29            # (if set)
30            child.wait()
31
32        start_time = time.monotonic()
33        for _ in support.busy_retry(support.SHORT_TIMEOUT, error=False):
34            if self.got_signals[signame]:
35                return
36            signal.pause()
37        else:
38            dt = time.monotonic() - start_time
39            self.fail('signal %s not received after %.1f seconds'
40                      % (signame, dt))
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        # gh-110033: Make sure that the subprocess.Popen is deleted before
64        # the next test which raises an exception. Otherwise, the exception
65        # may be raised when Popen.__del__() is executed and so be logged
66        # as "Exception ignored in: <function Popen.__del__ at ...>".
67        child = None
68        gc.collect()
69
70        with self.assertRaises(SIGUSR1Exception):
71            with self.subprocess_send_signal(pid, "SIGUSR1") as child:
72                self.wait_signal(child, 'SIGUSR1')
73        self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
74                                            'SIGALRM': 0})
75
76        with self.subprocess_send_signal(pid, "SIGUSR2") as child:
77            # Nothing should happen: SIGUSR2 is ignored
78            child.wait()
79
80        try:
81            with self.assertRaises(KeyboardInterrupt):
82                signal.alarm(1)
83                self.wait_signal(None, 'SIGALRM')
84            self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
85                                                'SIGALRM': 0})
86        finally:
87            signal.alarm(0)
88
89
90if __name__ == "__main__":
91    unittest.main()
92