pythongh-126434: Add test

Lib/test/multiprocessingdata/event_signal.py

@@ -0,0 +1,33 @@
+import multiprocessing
+import os
+import signal
+import concurrent.futures
+
+
+def send_sigint(pid):
+    print('send_sigint running')
+    os.kill(pid, signal.SIGINT)
+
+
+def run_signal_handler_test():
+    shutdown_event = multiprocessing.Event()
+
+    def sigterm_handler(_signo, _stack_frame):
+        try:
+            print('sigterm_handler running')
+            shutdown_event.set()
+        finally:
+            print('sigterm_handler done')
+
+    signal.signal(signal.SIGINT, sigterm_handler)
+
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        f = executor.submit(send_sigint, os.getpid())
+        while not shutdown_event.is_set():
+            pass
+        print('Received shutdown event')
+        f.result()
+
+
+if __name__ == '__main__':
+    run_signal_handler_test()

Lib/test/test_multiprocessing_event/__init__.py

@@ -0,0 +1,5 @@
+import os.path
+from test import support
+
+def load_tests(*args):
+    return support.load_package_tests(os.path.dirname(__file__), *args)

Lib/test/test_multiprocessing_event/test_event.py

@@ -0,0 +1,41 @@
+import unittest
+from test import support
+import sys
+import signal
+import os
+
+
+try:
+    import multiprocessing
+    from concurrent.futures import ProcessPoolExecutor
+    _have_multiprocessing = True
+except (NotImplementedError, ModuleNotFoundError):
+    _have_multiprocessing = False
+
+
+class TestEventSignalHandling(unittest.TestCase):
+    @unittest.skipUnless(_have_multiprocessing,
+                         "requires multiprocessing")
+    @unittest.skipUnless(hasattr(signal, 'signal'),
+                         "Requires signal.signal")
+    @unittest.skipUnless(hasattr(signal, 'SIGINT'),
+                         "Requires signal.SIGINT")
+    @unittest.skipUnless(hasattr(os, 'kill'),
+                     "Requires os.kill")
+    @unittest.skipUnless(hasattr(os, 'getppid'),
+                     "Requires os.getppid")
+    @support.requires_subprocess()
+    def test_event_signal_handling(self):
+        import subprocess
+        script = support.findfile("event_signal.py", subdir="multiprocessingdata")
+        for x in range(10):
+            try:
+                exit_code = subprocess.call([sys.executable, script], stdout=subprocess.DEVNULL, timeout=3)
+                assert exit_code == 0
+            except subprocess.TimeoutExpired:
+                assert False, 'subprocess.Timeoutexpired for event_signal.py'
+
+
+if __name__ == '__main__':
+    unittest.main()
+

Makefile.pre.in

@@ -2449,6 +2449,7 @@ TESTSUBDIRS=	idlelib/idle_test \
 		test/leakers \
 		test/libregrtest \
 		test/mathdata \
+		test/multiprocessingdata \
 		test/regrtestdata \
 		test/regrtestdata/import_from_tests \
 		test/regrtestdata/import_from_tests/test_regrtest_b \
@@ -2519,6 +2520,7 @@ TESTSUBDIRS=	idlelib/idle_test \
 		test/test_interpreters \
 		test/test_json \
 		test/test_module \
+		test/test_multiprocessing_event \
 		test/test_multiprocessing_fork \
 		test/test_multiprocessing_forkserver \
 		test/test_multiprocessing_spawn \