Adding basic test for workers with timeout values

Author: cezarsa

pyres/worker.py

@@ -25,13 +25,14 @@ class and passes a comma-separated list of queues to listen on.::
 
     job_class = Job
 
-    def __init__(self, queues=(), server="localhost:6379", password=None):
+    def __init__(self, queues=(), server="localhost:6379", password=None, timeout=None):
         self.queues = queues
         self.validate_queues()
         self._shutdown = False
         self.child = None
         self.pid = os.getpid()
         self.hostname = os.uname()[1]
+        self.timeout = timeout
 
         if isinstance(server, basestring):
             self.resq = ResQ(server=server, password=password)
@@ -305,8 +306,7 @@ def worker_pids(self):
 
     @classmethod
     def run(cls, queues, server="localhost:6379", interval=None, timeout=None):
-        worker = cls(queues=queues, server=server)
-        worker.timeout = timeout
+        worker = cls(queues=queues, server=server, timeout=timeout)
         if interval is not None:
             worker.work(interval)
         else:

tests/__init__.py

@@ -37,6 +37,14 @@ def perform(fail_until):
         else:
             return True
 
+class TimeoutJob(object):
+    queue = 'basic'
+
+    @staticmethod
+    def perform(wait_for):
+        import time
+        time.sleep(wait_for)
+        return "Done Sleeping"
 
 class TestProcess(object):
     queue = 'high'

tests/test_worker.py

@@ -1,4 +1,4 @@
-from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob
+from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob, TimeoutJob
 from pyres import ResQ
 from pyres.job import Job
 from pyres.scheduler import Scheduler
@@ -202,6 +202,23 @@ def test_retry_on_exception(self):
         assert True == worker.process()
         assert worker.get_failed() == 0
 
+    def test_kills_stale_workers_after_timeout(self):
+        import signal
+        timeout = 1
+
+        worker = Worker(['basic'], timeout=timeout)
+        self.resq.enqueue(TimeoutJob, timeout + 1)
+
+        child = os.fork()
+        if child:
+            assert worker.get_failed() == 0
+            time.sleep(timeout + 2)
+            os.kill(child, signal.SIGKILL)
+            os.waitpid(-1, os.WNOHANG)
+            assert worker.get_failed() == 1
+        else:
+            worker.work()
+
     def test_retries_give_up_eventually(self):
         now = datetime.datetime.now()
         self.set_current_time(now)