diff options
author | pulkin <gpulkin@gmail.com> | 2025-04-23 08:55:24 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2025-04-22 23:55:24 -0700 |
commit | 77605fa3bb4e380d5448665f70d73c8d3fcc2e75 (patch) | |
tree | ef374aeac9c98c0a651a98586af943cebb3f322b /Lib/test/_test_multiprocessing.py | |
parent | 862fd890361005598a6f4614ea0608c8447831c2 (diff) | |
download | cpython-77605fa3bb4e380d5448665f70d73c8d3fcc2e75.tar.gz cpython-77605fa3bb4e380d5448665f70d73c8d3fcc2e75.zip |
gh-131913: multiprocessing: add interrupt for POSIX (GH-132453)
* multiprocessing: interrupt
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>
Diffstat (limited to 'Lib/test/_test_multiprocessing.py')
-rw-r--r-- | Lib/test/_test_multiprocessing.py | 22 |
1 files changed, 20 insertions, 2 deletions
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 58d8a5eae8a..54b942e76d7 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -513,14 +513,19 @@ class _TestProcess(BaseTestCase): time.sleep(100) @classmethod + def _sleep_no_int_handler(cls): + signal.signal(signal.SIGINT, signal.SIG_DFL) + cls._sleep_some() + + @classmethod def _test_sleep(cls, delay): time.sleep(delay) - def _kill_process(self, meth): + def _kill_process(self, meth, target=None): if self.TYPE == 'threads': self.skipTest('test not appropriate for {}'.format(self.TYPE)) - p = self.Process(target=self._sleep_some) + p = self.Process(target=target or self._sleep_some) p.daemon = True p.start() @@ -567,6 +572,19 @@ class _TestProcess(BaseTestCase): return p.exitcode + @unittest.skipIf(os.name == 'nt', "POSIX only") + def test_interrupt(self): + exitcode = self._kill_process(multiprocessing.Process.interrupt) + self.assertEqual(exitcode, 1) + # exit code 1 is hard-coded for uncaught exceptions + # (KeyboardInterrupt in this case) + # in multiprocessing.BaseProcess._bootstrap + + @unittest.skipIf(os.name == 'nt', "POSIX only") + def test_interrupt_no_handler(self): + exitcode = self._kill_process(multiprocessing.Process.interrupt, target=self._sleep_no_int_handler) + self.assertEqual(exitcode, -signal.SIGINT) + def test_terminate(self): exitcode = self._kill_process(multiprocessing.Process.terminate) self.assertEqual(exitcode, -signal.SIGTERM) |