aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/_test_multiprocessing.py
diff options
context:
space:
mode:
authorpulkin <gpulkin@gmail.com>2025-04-23 08:55:24 +0200
committerGitHub <noreply@github.com>2025-04-22 23:55:24 -0700
commit77605fa3bb4e380d5448665f70d73c8d3fcc2e75 (patch)
treeef374aeac9c98c0a651a98586af943cebb3f322b /Lib/test/_test_multiprocessing.py
parent862fd890361005598a6f4614ea0608c8447831c2 (diff)
downloadcpython-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.py22
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)