rq.worker: remove useless set_state call in horse (#1618)

* rq.worker: remove useless set_state call in horse

The state should already have been set properly by the worker in
`execute_job`

`prepare_job_execution` is only called by `perform_job` which should only be
called by `main_work_horse`/`fork_work_horse` (themselves only called by `execute_job`).
Let `execute_job` do the bookkeeping.

* worker: update SimpleWorker's state in execute_job

rq/worker.py

@@ -922,7 +922,6 @@ class Worker:
         """
 
         with self.connection.pipeline() as pipeline:
-            self.set_state(WorkerStatus.BUSY, pipeline=pipeline)
             self.set_current_job_id(job.id, pipeline=pipeline)
             self.set_current_job_working_time(0, pipeline=pipeline)
 
@@ -1195,7 +1194,9 @@ class SimpleWorker(Worker):
 
     def execute_job(self, job, queue):
         """Execute job in same thread/process, do not fork()"""
-        return self.perform_job(job, queue)
+        self.set_state(WorkerStatus.BUSY)
+        self.perform_job(job, queue)
+        self.set_state(WorkerStatus.IDLE)
 
     def get_heartbeat_ttl(self, job):
         # "-1" means that jobs never timeout. In this case, we should _not_ do -1 + 60 = 59.

tests/test_worker.py

@@ -797,8 +797,7 @@ class TestWorker(RQTestCase):
         registry = StartedJobRegistry(connection=self.testconn)
         self.assertEqual(registry.get_job_ids(), [job.id])
 
-        # Updates worker statuses
+        # Updates worker's current job
-        self.assertEqual(worker.get_state(), 'busy')
         self.assertEqual(worker.get_current_job_id(), job.id)
 
         # job status is also updated