diff -r 36af3566b67a Lib/test/test_thread.py --- a/Lib/test/test_thread.py Thu Nov 03 15:38:58 2016 +0200 +++ b/Lib/test/test_thread.py Fri Nov 04 14:35:45 2016 -0400 @@ -1,9 +1,11 @@ +from contextlib import contextmanager import os import unittest import random from test import support thread = support.import_module('_thread') import time +import signal import sys import weakref @@ -147,6 +149,45 @@ time.sleep(0.01) self.assertIn("Traceback", stderr.getvalue()) + @contextmanager + def expect_sigint(self): + + # fail after 0.1s: expected SIGINT was not received while spinning + signal.signal(signal.SIGALRM, lambda *args: self.fail("timeout")) + signal.setitimer(signal.ITIMER_REAL, 0.1) + + def cleanup(): + signal.signal(signal.SIGINT, signal.SIG_DFL) + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, signal.SIG_DFL) + self.addCleanup(cleanup) + + # lock: held before yielding, released when interrupt received + lock = thread.allocate_lock() + signal.signal(signal.SIGINT, lambda *args: lock.release()) + + # will complete (w/o timeout) if SIGINTs received to release the lock + lock.acquire() + yield + while lock.locked(): + time.sleep(0.01) + + @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()") + def test_interrupt_main(self): + + # in this context, "main" is the thread running this test, + # so instead of catching KeyboardInterrupt, we assert receipt of SIGINT + with self.expect_sigint(): + thread.interrupt_main() + + @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()") + def test_interrupt_main_from_subthread(self): + + # same as test_interrupt_main(), except now we test that + # interrupt_main() works from a subthread + with self.expect_sigint(): + thread.start_new_thread(thread.interrupt_main, ()) + class Barrier: def __init__(self, num_threads):