blob: 0651cf2edeaeb75d8334b0d2b47e19a088a18036 [file] [log] [blame]
Michael Foordfa2f1cd2010-03-26 03:18:31 +00001import signal
2import weakref
3
4__unittest = True
5
6
7class _InterruptHandler(object):
8 def __init__(self, default_handler):
9 self.called = False
10 self.default_handler = default_handler
11
12 def __call__(self, signum, frame):
13 installed_handler = signal.getsignal(signal.SIGINT)
14 if installed_handler is not self:
15 # if we aren't the installed handler, then delegate immediately
16 # to the default handler
17 self.default_handler(signum, frame)
18
19 if self.called:
20 self.default_handler(signum, frame)
21 self.called = True
22 for result in _results.keys():
23 result.stop()
24
25_results = weakref.WeakKeyDictionary()
26def registerResult(result):
27 _results[result] = 1
28
29def removeResult(result):
30 return bool(_results.pop(result, None))
31
32_interrupt_handler = None
33def installHandler():
34 global _interrupt_handler
35 if _interrupt_handler is None:
36 default_handler = signal.getsignal(signal.SIGINT)
37 _interrupt_handler = _InterruptHandler(default_handler)
38 signal.signal(signal.SIGINT, _interrupt_handler)