Как проверить, что мой Celery worker действительно работает в Django

(код внизу)

Контекст: Я работаю над проектом Django, где мне нужно предоставить пользователю обратную связь по задаче, которая занимает 15-45 секунд. На помощь приходит Celery! Я вижу, что Celery работает как ожидалось, когда я celery -A tcommerce worker -l info & python manage.py runserver.

Проблема: Я не могу понять, как запустить celery worker в моих тестах. Когда я запускаю python manage.py test, я получаю следующую ошибку:

Traceback (most recent call last):
 File "/Users/pbrockman/coding/t1v/lib/python3.8/site-packages/django/test/utils.py", line 387, in inner
   return func(*args, **kwargs)
 File "/Users/pbrockman/coding/tcommerce/tcommerce/tests.py", line 58, in test_shared_celery_task
   self.assertEqual(result.get(), 6)
 File "/Users/pbrockman/coding/t1v/lib/python3.8/site-packages/celery/result.py", line 224, in get
   return self.backend.wait_for_pending(
 File "/Users/pbrockman/coding/t1v/lib/python3.8/site-packages/celery/backends/base.py", line 756, in wait_for_pending
   meta = self.wait_for(
 File "/Users/pbrockman/coding/t1v/lib/python3.8/site-packages/celery/backends/base.py", line 1087, in _is_disabled
   raise NotImplementedError(E_NO_BACKEND.strip())
NotImplementedError: No result backend is configured.
Please see the documentation for more information.

Попытка решения:

  • Я пробовал различные комбинации @override_settings с CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPOGATES=True и BROKER_BACKEND='memory'.
  • Я пробовал как декоратор @app.task, так и декоратор @shared_task.

Как проверить, имеет ли celery ожидаемое поведение в моих тестах?

Code

Настройки Celery: my_project/celery.py

import os
from dotenv import load_dotenv
load_dotenv() 
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_project.settings')

app = Celery('my_project-{os.environ.get("ENVIRONMENT")}',
             broker=os.environ.get('REDISCLOUD_URL'),
             include=['my_project.tasks'])

from django.conf import settings
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

if __name__ == '__main__':
    app.start()

Тестирование: my_project/tests.py

from django.test import TestCase
from tcommerce.celery import app
from tcommerce.tasks import shared_add
from tcommerce.tasks import app_add

class CeleryTests(TestCase):

  def test_shared_celery_task(self):
    '@shared_task'
    result = shared_add.delay(2, 4)
    self.assertEqual(result.get(), 6)

  
  def test_app_celery_task(self):
    '@task.app'
    result = app_add.delay(2, 4)
    self.assertEqual(result.get(), 6)

Определение задач: my_project/tasks.py

from .celery import app
from celery import shared_task

@shared_task
def shared_add(x, y):
  return x + y

@app.task
def app_add(x, y):
  return x + y
Вернуться на верх