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.
This commit is contained in:
Alex committed 2026-09-19 14:53:18 +01:00
1 parent 5e06470f9e
commit 558f803fed
1 file changed
+6 -6
+6 -6
View File
@@ -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