diff options
author | Victor Stinner <victor.stinner@gmail.com> | 2016-03-15 10:12:35 (GMT) |
---|---|---|
committer | Victor Stinner <victor.stinner@gmail.com> | 2016-03-15 10:12:35 (GMT) |
commit | 32eb840a42ec0e131daac48d43aa35290e72571e (patch) | |
tree | c945dd9b538b1e9b38d851fbdd04c4804e9f21a9 /Lib/test/signalinterproctester.py | |
parent | 58f2bd86fb0625912240f582f21678571e57bdde (diff) | |
download | cpython-32eb840a42ec0e131daac48d43aa35290e72571e.zip cpython-32eb840a42ec0e131daac48d43aa35290e72571e.tar.gz cpython-32eb840a42ec0e131daac48d43aa35290e72571e.tar.bz2 |
Issue #26566: Rewrite test_signal.InterProcessSignalTests
* Add Lib/test/signalinterproctester.py
* Don't disable the garbage collector anymore
* Don't use os.fork() with a subprocess to not inherit existing signal handlers
or threads: start from a fresh process
* Don't use UNIX kill command to send a signal but Python os.kill()
* Use a timeout of 10 seconds to wait for the signal instead of 1 second
* Always use signal.pause(), instead of time.wait(1), to wait for a signal
* Use context manager on subprocess.Popen
* remove code to retry on EINTR: it's no more needed since the PEP 475
* remove unused function exit_subprocess()
* Cleanup the code
Diffstat (limited to 'Lib/test/signalinterproctester.py')
-rw-r--r-- | Lib/test/signalinterproctester.py | 84 |
1 files changed, 84 insertions, 0 deletions
diff --git a/Lib/test/signalinterproctester.py b/Lib/test/signalinterproctester.py new file mode 100644 index 0000000..d3ae170 --- /dev/null +++ b/Lib/test/signalinterproctester.py @@ -0,0 +1,84 @@ +import os +import signal +import subprocess +import sys +import time +import unittest + + +class SIGUSR1Exception(Exception): + pass + + +class InterProcessSignalTests(unittest.TestCase): + def setUp(self): + self.got_signals = {'SIGHUP': 0, 'SIGUSR1': 0, 'SIGALRM': 0} + + def sighup_handler(self, signum, frame): + self.got_signals['SIGHUP'] += 1 + + def sigusr1_handler(self, signum, frame): + self.got_signals['SIGUSR1'] += 1 + raise SIGUSR1Exception + + def wait_signal(self, child, signame, exc_class=None): + try: + if child is not None: + # This wait should be interrupted by exc_class + # (if set) + child.wait() + + timeout = 10.0 + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + if self.got_signals[signame]: + return + signal.pause() + except BaseException as exc: + if exc_class is not None and isinstance(exc, exc_class): + # got the expected exception + return + raise + + self.fail('signal %s not received after %s seconds' + % (signame, timeout)) + + def subprocess_send_signal(self, pid, signame): + code = 'import os, signal; os.kill(%s, signal.%s)' % (pid, signame) + args = [sys.executable, '-I', '-c', code] + return subprocess.Popen(args) + + def test_interprocess_signal(self): + # Install handlers. This function runs in a sub-process, so we + # don't worry about re-setting the default handlers. + signal.signal(signal.SIGHUP, self.sighup_handler) + signal.signal(signal.SIGUSR1, self.sigusr1_handler) + signal.signal(signal.SIGUSR2, signal.SIG_IGN) + signal.signal(signal.SIGALRM, signal.default_int_handler) + + # Let the sub-processes know who to send signals to. + pid = str(os.getpid()) + + with self.subprocess_send_signal(pid, "SIGHUP") as child: + self.wait_signal(child, 'SIGHUP') + self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 0, + 'SIGALRM': 0}) + + with self.subprocess_send_signal(pid, "SIGUSR1") as child: + self.wait_signal(child, 'SIGUSR1', SIGUSR1Exception) + self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1, + 'SIGALRM': 0}) + + with self.subprocess_send_signal(pid, "SIGUSR2") as child: + # Nothing should happen: SIGUSR2 is ignored + child.wait() + + signal.alarm(1) + self.wait_signal(None, 'SIGALRM', KeyboardInterrupt) + self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1, + 'SIGALRM': 0}) + + +if __name__ == "__main__": + unittest.main() |