src/third_party/WebKit/Tools/Scripts/webkitpy/thirdparty/unittest2/signals.py

   1 import signal
   2 import weakref
   3
   4 from unittest2.compatibility import wraps
   5
   6 __unittest = True
   7
   8
   9 class _InterruptHandler(object):
  10     def __init__(self, default_handler):
  11         self.called = False
  12         self.default_handler = default_handler
  13
  14     def __call__(self, signum, frame):
  15         installed_handler = signal.getsignal(signal.SIGINT)
  16         if installed_handler is not self:
  17             # if we aren't the installed handler, then delegate immediately
  18             # to the default handler
  19             self.default_handler(signum, frame)
  20
  21         if self.called:
  22             self.default_handler(signum, frame)
  23         self.called = True
  24         for result in _results.keys():
  25             result.stop()
  26
  27 _results = weakref.WeakKeyDictionary()
  28 def registerResult(result):
  29     _results[result] = 1
  30
  31 def removeResult(result):
  32     return bool(_results.pop(result, None))
  33
  34 _interrupt_handler = None
  35 def installHandler():
  36     global _interrupt_handler
  37     if _interrupt_handler is None:
  38         default_handler = signal.getsignal(signal.SIGINT)
  39         _interrupt_handler = _InterruptHandler(default_handler)
  40         signal.signal(signal.SIGINT, _interrupt_handler)
  41
  42
  43 def removeHandler(method=None):
  44     if method is not None:
  45         @wraps(method)
  46         def inner(*args, **kwargs):
  47             initial = signal.getsignal(signal.SIGINT)
  48             removeHandler()
  49             try:
  50                 return method(*args, **kwargs)
  51             finally:
  52                 signal.signal(signal.SIGINT, initial)
  53         return inner
  54
  55     global _interrupt_handler
  56     if _interrupt_handler is not None:
  57         signal.signal(signal.SIGINT, _interrupt_handler.default_handler)