@timeout(10) decorator for unittest.TestCase methods that I never ended up using

subprocess_timeout.py

import signal
import sys
from functools import contextmanager


@contextmanager
def timeout(p, seconds):
    """Kill subprocess ``p`` if the with block times out.
    
    Usage example ::
    
        p = subprocess.Popen([...], stdout=subprocess.PIPE)
        with timeout(p, 30):
            data = p.communicate()[0]
    
    """

    finished = threading.Event()

    def killIt():
        finished.wait(seconds)
        if not finished.is_set():
            # print once to stderr, so user sees it, and once to stdout,
            # so the right doctest fails
            print("\nTimed out after {} seconds, killing process {}".format(
                    seconds, p.pid), file=sys.stderr)
            print("Timed out after {} seconds, killing process {}".format(
                    seconds, p.pid))
            p.kill()

    t = threading.Thread(name='timeout-killer', target=killIt)
    try:
        t.start()
        yield
    finally:
        finished.set()

timeout.py

import signal
import sys
from functools import wraps


class TimedOut(BaseException):
    pass


def timeout(seconds):
    """Decorator that makes a function time out.

    Because test suites that hang are no fun.  Especially on buildbots.

    Currently only implemented for Unix.
    """
    def decorator(fn):
        if hasattr(signal, 'alarm'):
            # yay, unix!
            @wraps(fn)
            def wrapper(*args, **kw):
                this_frame = sys._getframe()
                def raiseTimeOut(signal, frame):
                    # the if statement here is meant to prevent an exception in the
                    # finally: clause before clean up can take place
                    if frame is not this_frame:
                        raise TimedOut('timed out after %s seconds' % seconds)
                prev_handler = signal.signal(signal.SIGALRM, raiseTimeOut)
                try:
                    signal.alarm(seconds)
                    return fn(*args, **kw)
                finally:
                    signal.alarm(0)
                    signal.signal(signal.SIGALRM, prev_handler)
            return wrapper
        else:
            # XXX um, could someone please implement this for Windows and other
            # strange platforms?
            return fn
    return decorator

with_timeout.py

import sys
import signal
from functools import 


@contextmanager
def timeout(seconds):
    """Kill subprocess ``p`` if the with block times out.
    
    Usage example ::
    
        p = subprocess.Popen([...], stdout=subprocess.PIPE)
        with timeout(p, 30):
            data = p.communicate()[0]
    
    """
    this_frame = sys._getframe()
    def raiseTimeout(signal, frame):
        # the if statement here is meant to prevent an exception in the
        # finally: clause before clean up can take place
        if frame is not this_frame:
            raise Timeout('timed out after %s seconds' % seconds)
    prev_handler = signal.signal(signal.SIGALRM, raiseTimeout)
    try:
        signal.alarm(seconds)
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, prev_handler)

The problem with this: it's unable to interrupt a Queue.Queue().get(). To be fair, Ctrl+C is also unable to interrupt that nasty function, I end up having to hit Ctrl-\ to dump core.

Added a context manager version as well.

I can't use this one too -- turns out the function I'm trying to time out is being called in a thread. You can't use signals from a thread. :(

Added a context manager that kills a given subprocess, to avoid the thread issue.