mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-24 18:55:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

bpo-43356: Allow passing a signal number to interrupt_main() (GH-24755)

Browse source 
Also introduce a new C API ``PyErr_SetInterruptEx(int signum)``.
This commit is contained in:
Antoine Pitrou 2021-03-11 23:35:45 +01:00 • committed by GitHub
parent b4fc44bb2d
commit ba251c2ae6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 211 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

45
Lib/test/test_threading.py
View file

@ -1489,6 +1489,29 @@ class MiscTestCase(unittest.TestCase):
class InterruptMainTests(unittest.TestCase):
def check_interrupt_main_with_signal_handler(self, signum):
def handler(signum, frame):
1/0
old_handler = signal.signal(signum, handler)
self.addCleanup(signal.signal, signum, old_handler)
with self.assertRaises(ZeroDivisionError):
_thread.interrupt_main()
def check_interrupt_main_noerror(self, signum):
handler = signal.getsignal(signum)
try:
# No exception should arise.
signal.signal(signum, signal.SIG_IGN)
_thread.interrupt_main(signum)
signal.signal(signum, signal.SIG_DFL)
_thread.interrupt_main(signum)
finally:
# Restore original handler
signal.signal(signum, handler)
def test_interrupt_main_subthread(self):
# Calling start_new_thread with a function that executes interrupt_main
# should raise KeyboardInterrupt upon completion.
@ -1506,18 +1529,18 @@ class InterruptMainTests(unittest.TestCase):
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()
def test_interrupt_main_with_signal_handler(self):
self.check_interrupt_main_with_signal_handler(signal.SIGINT)
self.check_interrupt_main_with_signal_handler(signal.SIGTERM)
signal.signal(signal.SIGINT, signal.SIG_DFL)
_thread.interrupt_main()
finally:
# Restore original handler
signal.signal(signal.SIGINT, handler)
def test_interrupt_main_noerror(self):
self.check_interrupt_main_noerror(signal.SIGINT)
self.check_interrupt_main_noerror(signal.SIGTERM)
def test_interrupt_main_invalid_signal(self):
self.assertRaises(ValueError, _thread.interrupt_main, -1)
self.assertRaises(ValueError, _thread.interrupt_main, signal.NSIG)
self.assertRaises(ValueError, _thread.interrupt_main, 1000000)
class AtexitTests(unittest.TestCase):