From 558f803fed56e1f217af444ed409708f327e1817 Mon Sep 17 00:00:00 2001 From: Alex Date: Sat, 19 Sep 2026 14:53:18 +0100 Subject: [PATCH] test(worker): send the real startup signals, and import celery_init one way The lifecycle tests imported docsgpt.celery_init as a module next to the file's from-imports, which code scanning flags. Patch the flag by path and send the signals from celery.signals -- the objects a worker actually fires. --- tests/test_celery.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/test_celery.py b/tests/test_celery.py index 897b2623..e240871c 100644 --- a/tests/test_celery.py +++ b/tests/test_celery.py @@ -322,20 +322,20 @@ class TestInWorker: # to one greenlet, so only the worker's own startup can say this # process is a worker. The lifecycle signal records that for every # thread and greenlet in it. - import docsgpt.celery_init as celery_init + from celery.signals import worker_init - monkeypatch.setattr(celery_init, "_IS_WORKER_PROCESS", False) + monkeypatch.setattr("docsgpt.celery_init._IS_WORKER_PROCESS", False) assert self._ask_from_a_new_thread() is False - celery_init.worker_init.send(sender=None) + worker_init.send(sender=None) assert self._ask_from_a_new_thread() is True def test_prefork_children_record_it_on_their_own_start(self, monkeypatch): - import docsgpt.celery_init as celery_init + from celery.signals import worker_process_init - monkeypatch.setattr(celery_init, "_IS_WORKER_PROCESS", False) - celery_init.worker_process_init.send(sender=None) + monkeypatch.setattr("docsgpt.celery_init._IS_WORKER_PROCESS", False) + worker_process_init.send(sender=None) assert self._ask_from_a_new_thread() is True