(self, celery_setup: CeleryTestSetup)
| 71 | return app |
| 72 | |
| 73 | def test_warm_shutdown(self, celery_setup: CeleryTestSetup): |
| 74 | queue = celery_setup.worker.worker_queue |
| 75 | worker = celery_setup.worker |
| 76 | sig = long_running_task.si(5, verbose=True).set(queue=queue) |
| 77 | res = sig.delay() |
| 78 | |
| 79 | worker.assert_log_exists("Starting long running task") |
| 80 | self.kill_worker(worker, WorkerKill.Method.SIGTERM) |
| 81 | worker.assert_log_exists("worker: Warm shutdown (MainProcess)") |
| 82 | |
| 83 | assert_container_exited(worker) |
| 84 | assert res.get(RESULT_TIMEOUT) |
| 85 | |
| 86 | def test_multiple_warm_shutdown_does_nothing(self, celery_setup: CeleryTestSetup): |
| 87 | queue = celery_setup.worker.worker_queue |
nothing calls this directly
no test coverage detected