Comment tester unitairement une tâche Celery ?

Question

La documentation sur le céleri mentionne les tests de Celery dans Django mais n'explique pas comment tester une tâche Celery si vous n'utilisez pas Django. Comment procéder ?

Answer 1

Il est possible de tester des tâches de manière synchrone en utilisant n'importe quelle librairie unittest existante. Je fais normalement 2 sessions de test différentes lorsque je travaille avec des tâches celery. La première (comme je le suggère ci-dessous) est complètement synchrone et devrait être celle qui permet de s'assurer que l'algorithme fait ce qu'il doit faire. La seconde session utilise l'ensemble du système (y compris le broker) et s'assure que je n'ai pas de problèmes de sérialisation ou tout autre problème de distribution ou de communication.

Donc :

from celery import Celery

celery = Celery()

@celery.task
def add(x, y):
    return x + y

Et votre test :

from nose.tools import eq_

def test_add_task():
    rst = add.apply(args=(4, 4)).get()
    eq_(rst, 8)

J'espère que cela vous aidera !

Answer 2

Une mise à jour de ma réponse vieille de sept ans :

Vous pouvez exécuter un worker dans un thread séparé via un fixture pytest :

https://docs.celeryq.dev/en/v5.2.6/userguide/testing.html#celery-worker-embed-live-worker

Selon la documentation, vous ne devez pas utiliser "always_eager" (voir le haut de la page du lien ci-dessus).

---

Vieille réponse :

J'utilise ceci :

with mock.patch('celeryconfig.CELERY_ALWAYS_EAGER', True, create=True):
    ...

Docs : https://docs.celeryq.dev/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER vous permet d'exécuter votre tâche de manière synchrone, et vous n'avez pas besoin d'un serveur celery.

Answer 3

Cela dépend de ce que vous voulez tester exactement.

• Testez directement le code de la tâche. N'appelez pas "task.delay(...)" mais "task(...)" à partir de vos tests unitaires.
• Utilice CÉLERI_TOUJOURS_VIEUX . Ainsi, vos tâches seront appelées immédiatement au moment où vous dites "task.delay(...)", ce qui vous permettra de tester l'ensemble du chemin (mais pas le comportement asynchrone).

Answer 4

Unittest

import unittest

from myproject.myapp import celeryapp

class TestMyCeleryWorker(unittest.TestCase):

  def setUp(self):
      celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)

py.test fixtures

# conftest.py
from myproject.myapp import celeryapp

@pytest.fixture(scope='module')
def celery_app(request):
    celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)
    return celeryapp

# test_tasks.py
def test_some_task(celery_app):
    ...

Addendum : faire respecter send_task eager

from celery import current_app

def send_task(name, args=(), kwargs={}, **opts):
    # https://github.com/celery/celery/issues/581
    task = current_app.tasks[name]
    return task.apply(args, kwargs, **opts)

current_app.send_task = send_task

Answer 5

Pour ceux qui sont sur Celery 4, c'est :

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

Les noms des paramètres ont été modifiés et doivent être mis à jour si vous choisissez de procéder à une mise à niveau, voir

https://docs.celeryproject.org/en/latest/history/whatsnew-4.0.html?highlight=what%20is%20new#lowercase-setting-names