refactor multiprocess tests

Author: cunla

scheduler/tests/test_multiprocess/test_integrity.py

@@ -25,14 +25,14 @@ def test_cancel_job_after_it_started(self):
         res = self.client.post(reverse("job_detail_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
 
         # assert
-        self.assertEqual(200, res.status_code)
-        job = JobModel.get(job.name, connection=queue.connection)
-        self.assertEqual(JobStatus.STOPPED, job.status)
-        self.assertNotIn(job.name, queue.queued_job_registry.all())
         sleep(0.2)
         process.terminate()
         process.join(2)
         process.kill()
+        self.assertEqual(200, res.status_code)
+        job = JobModel.get(job.name, connection=queue.connection)
+        self.assertEqual(JobStatus.STOPPED, job.status)
+        self.assertNotIn(job.name, queue.queued_job_registry.all())
         worker_model = WorkerModel.get(worker_name, connection=queue.connection)
         self.assertEqual(0, worker_model.completed_jobs)
         self.assertEqual(0, worker_model.failed_job_count)

scheduler/tests/test_worker/test_worker_commands_multiprocess.py

@@ -24,13 +24,11 @@ def test_kill_job_command__current_job(self):
 
         # Act
         send_command(queue.connection, StopJobCommand(worker_name=worker_name, job_name=job.name))
-
-        # Assert
-
         process.terminate()
         process.join(2)
         process.kill()
 
+        # Assert
         job = JobModel.get(job.name, connection=queue.connection)
         worker_model = WorkerModel.get(worker_name, connection=queue.connection)
         self.assertEqual(job.name, worker_model.stopped_job_name)

scheduler/tests/testtools.py

@@ -25,8 +25,8 @@ def _run_worker_process(worker: Worker, **kwargs):
 
 
 def run_worker_in_process(*args, name="test-worker") -> Tuple[multiprocessing.Process, str]:
-    worker = create_worker(*args, name=name, fork_job_execution=False)
-    process = multiprocessing.Process(target=_run_worker_process, args=(worker,), kwargs=dict(with_scheduler=False))
+    worker = create_worker(*args, name=name, fork_job_execution=False, with_scheduler=False)
+    process = multiprocessing.Process(target=_run_worker_process, args=(worker,), kwargs=dict())
     process.start()
     return process, name