bpo-23395: Fix PyErr_SetInterrupt if the SIGINT signal is ignored or not handled (GH-7778)

``_thread.interrupt_main()`` now avoids setting the Python error status if the ``SIGINT`` signal is ignored or not handled by Python.

1 files changed, 35 insertions, 0 deletions

diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index 33a25f3..3bfd6fa 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -16,6 +16,7 @@ import unittest
 import weakref
 import os
 import subprocess
+import signal
 
 from test import lock_tests
 from test import support
@@ -1168,6 +1169,7 @@ class BoundedSemaphoreTests(lock_tests.BoundedSemaphoreTests):
 class BarrierTests(lock_tests.BarrierTests):
     barriertype = staticmethod(threading.Barrier)
 
+
 class MiscTestCase(unittest.TestCase):
     def test__all__(self):
         extra = {"ThreadError"}
@@ -1175,5 +1177,38 @@ class MiscTestCase(unittest.TestCase):
         support.check__all__(self, threading, ('threading', '_thread'),
                              extra=extra, blacklist=blacklist)
 
+
+class InterruptMainTests(unittest.TestCase):
+    def test_interrupt_main_subthread(self):
+        # Calling start_new_thread with a function that executes interrupt_main
+        # should raise KeyboardInterrupt upon completion.
+        def call_interrupt():
+            _thread.interrupt_main()
+        t = threading.Thread(target=call_interrupt)
+        with self.assertRaises(KeyboardInterrupt):
+            t.start()
+            t.join()
+        t.join()
+
+    def test_interrupt_main_mainthread(self):
+        # Make sure that if interrupt_main is called in main thread that
+        # KeyboardInterrupt is raised instantly.
+        with self.assertRaises(KeyboardInterrupt):
+            _thread.interrupt_main()
+
+    def test_interrupt_main_noerror(self):
+        handler = signal.getsignal(signal.SIGINT)
+        try:
+            # No exception should arise.
+            signal.signal(signal.SIGINT, signal.SIG_IGN)
+            _thread.interrupt_main()
+
+            signal.signal(signal.SIGINT, signal.SIG_DFL)
+            _thread.interrupt_main()
+        finally:
+            # Restore original handler
+            signal.signal(signal.SIGINT, handler)
+
+
 if __name__ == "__main__":
     unittest.main()