handled unhandled exceptions in horse (#1303)

* handled unhandled exceptions in horse to prevent a job from being silently dropped without going into FailedRegistry

* changes after review

* made sure that work_horse always terminates in a proper way with tests

* minor refactoring

* fix for failing test

* fixes for the other tests

- removed exception handling (done in monitor_work_horse)
- adjusted some tests for the checks that are not relevant anymore

* review suggested changes

* cleanup

Co-authored-by: Ruslan Mullakhmetov <ruslan@twentythree.net>

1 files changed, 32 insertions, 13 deletions

diff --git a/tests/test_worker.py b/tests/test_worker.py
index 931c1fa..a5d4026 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -25,7 +25,7 @@ from tests import RQTestCase, slow
 from tests.fixtures import (
     access_self, create_file, create_file_after_timeout, div_by_zero, do_nothing,
     kill_worker, long_running_job, modify_self, modify_self_and_error,
-    run_dummy_heroku_worker, save_key_ttl, say_hello, say_pid,
+    run_dummy_heroku_worker, save_key_ttl, say_hello, say_pid, raise_exc_mock
 )
 
 from rq import Queue, SimpleWorker, Worker, get_current_connection
@@ -312,6 +312,37 @@ class TestWorker(RQTestCase):
         self.assertEqual(str(job.enqueued_at), enqueued_at_date)
         self.assertTrue(job.exc_info)  # should contain exc_info
 
+    def test_horse_fails(self):
+        """Tests that job status is set to FAILED even if horse unexpectedly fails"""
+        q = Queue()
+        self.assertEqual(q.count, 0)
+
+        # Action
+        job = q.enqueue(say_hello)
+        self.assertEqual(q.count, 1)
+
+        # keep for later
+        enqueued_at_date = str(job.enqueued_at)
+
+        w = Worker([q])
+        with mock.patch.object(w, 'perform_job', new_callable=raise_exc_mock):
+            w.work(burst=True)  # should silently pass
+
+        # Postconditions
+        self.assertEqual(q.count, 0)
+        failed_job_registry = FailedJobRegistry(queue=q)
+        self.assertTrue(job in failed_job_registry)
+        self.assertEqual(w.get_current_job_id(), None)
+
+        # Check the job
+        job = Job.fetch(job.id)
+        self.assertEqual(job.origin, q.name)
+
+        # Should be the original enqueued_at date, not the date of enqueueing
+        # to the failed queue
+        self.assertEqual(str(job.enqueued_at), enqueued_at_date)
+        self.assertTrue(job.exc_info)  # should contain exc_info
+
     def test_statistics(self):
         """Successful and failed job counts are saved properly"""
         queue = Queue()
@@ -1206,10 +1237,6 @@ class HerokuWorkerShutdownTestCase(TimeoutTestCase, RQTestCase):
         self.assertEqual(p.exitcode, 1)
         self.assertTrue(os.path.exists(os.path.join(self.sandbox, 'started')))
         self.assertFalse(os.path.exists(os.path.join(self.sandbox, 'finished')))
-        with open(os.path.join(self.sandbox, 'stderr.log')) as f:
-            stderr = f.read().strip('\n')
-            err = 'ShutDownImminentException: shut down imminent (signal: SIGRTMIN)'
-            self.assertTrue(stderr.endswith(err), stderr)
 
     @slow
     def test_1_sec_shutdown(self):
@@ -1226,10 +1253,6 @@ class HerokuWorkerShutdownTestCase(TimeoutTestCase, RQTestCase):
 
         self.assertTrue(os.path.exists(os.path.join(self.sandbox, 'started')))
         self.assertFalse(os.path.exists(os.path.join(self.sandbox, 'finished')))
-        with open(os.path.join(self.sandbox, 'stderr.log')) as f:
-            stderr = f.read().strip('\n')
-            err = 'ShutDownImminentException: shut down imminent (signal: SIGALRM)'
-            self.assertTrue(stderr.endswith(err), stderr)
 
     @slow
     def test_shutdown_double_sigrtmin(self):
@@ -1247,10 +1270,6 @@ class HerokuWorkerShutdownTestCase(TimeoutTestCase, RQTestCase):
 
         self.assertTrue(os.path.exists(os.path.join(self.sandbox, 'started')))
         self.assertFalse(os.path.exists(os.path.join(self.sandbox, 'finished')))
-        with open(os.path.join(self.sandbox, 'stderr.log')) as f:
-            stderr = f.read().strip('\n')
-            err = 'ShutDownImminentException: shut down imminent (signal: SIGRTMIN)'
-            self.assertTrue(stderr.endswith(err), stderr)
 
     @mock.patch('rq.worker.logger.info')
     def test_handle_shutdown_request(self, mock_logger_info):