Adding test to check whether pyres identifies a crashed process as a

failure (currently failing).

The biggest problem shown by this test is the fact that, even after the child
process has crashed and the worker is ready to process new jobs, the entry
associating the job with the worker remain on Redis.

It also refactors the test for timeout errors on staled workers, making it simpler.

Author: cezarsa

tests/__init__.py

@@ -46,6 +46,16 @@ def perform(wait_for):
         time.sleep(wait_for)
         return "Done Sleeping"
 
+class CrashJob(object):
+    queue = 'basic'
+
+    @staticmethod
+    def perform():
+        # Dangerous, this will cause a hard crash of the python process
+        import ctypes
+        ctypes.string_at(1)
+        return "Never got here"
+
 class TestProcess(object):
     queue = 'high'
 

tests/test_worker.py

@@ -1,4 +1,4 @@
-from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob, TimeoutJob
+from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob, TimeoutJob, CrashJob
 from pyres import ResQ
 from pyres.job import Job
 from pyres.scheduler import Scheduler
@@ -203,21 +203,26 @@ def test_retry_on_exception(self):
         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()
+        assert worker.get_failed() == 0
+        worker.fork_worker(worker.reserve())
+        assert worker.get_failed() == 1
+
+    def test_detect_crashed_workers_as_failures(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(CrashJob)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 1
 
     def test_retries_give_up_eventually(self):
         now = datetime.datetime.now()