import signal | |
import weakref | |
from functools import wraps | |
__unittest = True | |
class _InterruptHandler(object): | |
def __init__(self, default_handler): | |
self.called = False | |
self.default_handler = default_handler | |
def __call__(self, signum, frame): | |
installed_handler = signal.getsignal(signal.SIGINT) | |
if installed_handler is not self: | |
# if we aren't the installed handler, then delegate immediately | |
# to the default handler | |
self.default_handler(signum, frame) | |
if self.called: | |
self.default_handler(signum, frame) | |
self.called = True | |
for result in _results.keys(): | |
result.stop() | |
_results = weakref.WeakKeyDictionary() | |
def registerResult(result): | |
_results[result] = 1 | |
def removeResult(result): | |
return bool(_results.pop(result, None)) | |
_interrupt_handler = None | |
def installHandler(): | |
global _interrupt_handler | |
if _interrupt_handler is None: | |
default_handler = signal.getsignal(signal.SIGINT) | |
_interrupt_handler = _InterruptHandler(default_handler) | |
signal.signal(signal.SIGINT, _interrupt_handler) | |
def removeHandler(method=None): | |
if method is not None: | |
@wraps(method) | |
def inner(*args, **kwargs): | |
initial = signal.getsignal(signal.SIGINT) | |
removeHandler() | |
try: | |
return method(*args, **kwargs) | |
finally: | |
signal.signal(signal.SIGINT, initial) | |
return inner | |
global _interrupt_handler | |
if _interrupt_handler is not None: | |
signal.signal(signal.SIGINT, _interrupt_handler.default_handler) |