Add a test case.

Also fix the existing test cases that were broken.

Author: ZeroIntensity

Lib/test/test_interpreters/test_api.py

@@ -1855,6 +1855,7 @@ def test_new_config(self):
                 allow_threads=True,
                 allow_daemon_threads=False,
                 check_multi_interp_extensions=True,
+                can_handle_signals=True,
                 gil='own',
             ),
             'legacy': types.SimpleNamespace(
@@ -1864,6 +1865,7 @@ def test_new_config(self):
                 allow_threads=True,
                 allow_daemon_threads=True,
                 check_multi_interp_extensions=bool(Py_GIL_DISABLED),
+                can_handle_signals=False,
                 gil='shared',
             ),
             'empty': types.SimpleNamespace(
@@ -1873,6 +1875,7 @@ def test_new_config(self):
                 allow_threads=False,
                 allow_daemon_threads=False,
                 check_multi_interp_extensions=False,
+                can_handle_signals=False,
                 gil='default',
             ),
         }
@@ -2134,6 +2137,7 @@ def test_get_config(self):
         with self.subTest('main'):
             expected = _interpreters.new_config('legacy')
             expected.gil = 'own'
+            expected.can_handle_signals = True
             if Py_GIL_DISABLED:
                 expected.check_multi_interp_extensions = False
             interpid, *_ = _interpreters.get_main()
@@ -2360,6 +2364,37 @@ def test_set___main___attrs(self):
                 )
             self.assertEqual(rc, 0)
 
+    def test_interpreter_handles_signals(self):
+        import subprocess
+        import sys
+        import signal
+
+        interp_source = """if True:
+        import sys
+        import time
+
+        sys.stdout.write('x')
+        sys.stdout.flush()
+        time.sleep(10)
+        print("should never happen", flush=True)
+        """
+
+        source = f"""if True:
+        from concurrent import interpreters
+
+        interp = interpreters.create()
+        interp.exec('''{interp_source}''')
+        """
+
+        proc = subprocess.Popen([sys.executable, '-c', source],
+                                stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE, close_fds=True)
+        self.assertEqual(proc.stdout.read(1), b'x')
+        proc.send_signal(signal.SIGINT)
+        stdout, stderr = proc.communicate()
+        self.assertEqual(stdout, b"")
+        self.assertIn(b"KeyboardInterrupt", stderr)
+
 
 if __name__ == '__main__':
     # Test needs to be a package, so we can do relative imports.

Python/interpconfig.c

@@ -262,6 +262,7 @@ _PyInterpreterConfig_InitFromState(PyInterpreterConfig *config,
         .allow_threads = FLAG(THREADS),
         .allow_daemon_threads = FLAG(DAEMON_THREADS),
         .check_multi_interp_extensions = FLAG(MULTI_INTERP_EXTENSIONS),
+        .can_handle_signals = FLAG(CAN_HANDLE_SIGNALS),
 #undef FLAG
         .gil = interp->ceval.own_gil
             ? PyInterpreterConfig_OWN_GIL