Python Watchdog with gevent

gevent_watchdog.py

""" Use watchdogs to make your code safer. (http://en.wikipedia.org/wiki/Watchdog_timer)

Usage:

    with watchdog.Watchdog(duration = 1.0):
        # Some code which takes less than 1s
        # If it takes more, the exception Watchdog.Reset will be raised
    
        # To notify the watchdog the section is still alive, call the method
        # 'tick' to reset its internal timer.

Example:

    with watchdog.Watchdog(duration = 1.0):
        print "Hello, World! :-)"

will print "Hello, World! :-)".

    with watchdog.Watchdog(duration = 1.0):
        time.sleep(2.0)

will raises the exception Watchdog.Reset

    with watchdog.Watchdog(duration = 1.0) as w:
        for i in range(10):
            time.sleep(0.9)
            w.tick()

won't raise any exception.

"""

import gevent

class Watchdog(object):
    class Reset(Exception):
        pass

    def __init__(self, duration):
        super(Watchdog, self).__init__()

        self.duration = duration
        self.timeout = None

    def start(self):
        assert not self.timeout, "The watchdog is already started."

        self.timeout = gevent.Timeout(self.duration, Watchdog.Reset)
        self.timeout.start()
        
    def tick(self):
        self.stop()
        self.start()

    def stop(self):
        if self.timeout:
            self.timeout.cancel()
            self.timeout = None

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, *exc):
        self.stop()

test_gevent_watchdog.py

""" nosetests for gevent_watchdog.py """

import nose
import gevent_watchdog as watchdog

import contextlib
import time

import gevent

@contextlib.contextmanager
def assert_exception(exc = None):
    try:
        yield

    except BaseException, e:
        if exc and not isinstance(e, exc):
            raise

    else:
        raise AssertionError("No exception had been raised.")

def test_init():
    with watchdog.Watchdog(0.1):
        pass

def test_fail():
    with assert_exception(watchdog.Watchdog.Reset):
        with watchdog.Watchdog(0.1):
            gevent.sleep(0.2)

def test_tick():
    with watchdog.Watchdog(0.1) as w:
        for i in range(4):
            gevent.sleep(0.09)
            w.tick()

I see, did not notice tick method. You can probably emulate it though, by cancelling/starting the timeout.

I fixed this. The code is indeed lighter.