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

A partir de Céleri 3.0 une façon de régler CELERY_ALWAYS_EAGER sur Django est :

from django.test import TestCase, override_settings

from .foo import foo_celery_task

class MyTest(TestCase):

    @override_settings(CELERY_ALWAYS_EAGER=True)
    def test_foo(self):
        self.assertTrue(foo_celery_task.delay())

Answer 2

Depuis le céleri v4.0 Les montages py.test sont fourni par pour démarrer un travailleur de céleri juste pour le test et sont arrêtés quand ils ont terminé :

def test_myfunc_is_executed(celery_session_worker):
    # celery_session_worker: <Worker: gen93553@mymachine.local (running)>
    assert myfunc.delay().wait(3)

Parmi les autres installations décrites sur http://docs.celeryproject.org/en/latest/userguide/testing.html#py-test vous pouvez modifier les options par défaut de celery en redéfinissant le fichier celery_config fixer de cette façon :

@pytest.fixture(scope='session')
def celery_config():
    return {
        'accept_content': ['json', 'pickle'],
        'result_serializer': 'pickle',
    }

Par défaut, le travailleur de test utilise un courtier en mémoire et un backend de résultats. Il n'est pas nécessaire d'utiliser un Redis ou un RabbitMQ local si vous ne testez pas des fonctionnalités spécifiques.

Answer 3

référence en utilisant pytest.

def test_add(celery_worker):
    mytask.delay()

si vous utilisez flask, définissez la configuration de l'application

    CELERY_BROKER_URL = 'memory://'
    CELERY_RESULT_BACKEND = 'cache+memory://'

et en conftest.py

@pytest.fixture
def app():
    yield app   # Your actual Flask application

@pytest.fixture
def celery_app(app):
    from celery.contrib.testing import tasks   # need it
    yield celery_app    # Your actual Flask-Celery application

Answer 4

Dans mon cas (et je suppose beaucoup d'autres), tout ce que je voulais était de tester la logique interne d'une tâche en utilisant pytest.

TL;DR ; a fini par se moquer de tout ( OPTION 2 )

---

Exemple de cas d'utilisation :

proj/tasks.py

@shared_task(bind=True)
def add_task(self, a, b):
    return a+b;

tests/test_tasks.py

from proj import add_task

def test_add():
    assert add_task(1, 2) == 3, '1 + 2 should equal 3'

mais, comme shared_task Le décorateur fait beaucoup de logique interne à celery, il n'est pas vraiment un test unitaire.

Donc, pour moi, il y avait 2 options :

OPTION 1 : Logique interne séparée

proj/tasks_logic.py

def internal_add(a, b):
    return a + b;

proj/tasks.py

from .tasks_logic import internal_add

@shared_task(bind=True)
def add_task(self, a, b):
    return internal_add(a, b);

Cela semble très étrange, et en plus de le rendre moins lisible, cela nécessite d'extraire et de passer manuellement les attributs qui font partie de la requête, par exemple l'attribut task_id au cas où vous en auriez besoin, ce qui rend la logique moins pure.

OPTION 2 : mocks
se moquer des internes de celery

tests/__init__.py

# noinspection PyUnresolvedReferences
from celery import shared_task

from mock import patch

def mock_signature(**kwargs):
    return {}

def mocked_shared_task(*decorator_args, **decorator_kwargs):
    def mocked_shared_decorator(func):
        func.signature = func.si = func.s = mock_signature
        return func

    return mocked_shared_decorator

patch('celery.shared_task', mocked_shared_task).start()

qui me permet ensuite de simuler l'objet de la requête (encore une fois, au cas où vous auriez besoin d'éléments de la requête, comme l'identifiant ou le compteur de tentatives).

tests/test_tasks.py

from proj import add_task

class MockedRequest:
    def __init__(self, id=None):
        self.id = id or 1

class MockedTask:
    def __init__(self, id=None):
        self.request = MockedRequest(id=id)

def test_add():
    mocked_task = MockedTask(id=3)
    assert add_task(mocked_task, 1, 2) == 3, '1 + 2 should equal 3'

Cette solution est beaucoup plus manuelle, mais elle me donne le contrôle dont j'ai besoin pour réellement unité test, sans me répéter, et sans perdre la portée du céleri.

Answer 5

Je vois beaucoup de CELERY_ALWAYS_EAGER = true dans les méthodes de tests unitaires comme une solution pour les tests unitaires, mais depuis que la version 5.0.5 est disponible, il y a beaucoup de changements qui rendent la plupart des anciennes réponses dépréciées et pour moi une absurdité qui prend du temps, donc pour tous ceux qui cherchent une solution, allez à la Doc et lisez les exemples de tests unitaires bien documentés pour la nouvelle version :

https://docs.celeryproject.org/en/stable/userguide/testing.html

En ce qui concerne le mode Eager avec les tests unitaires, voici une citation de la documentation actuelle :

Le mode passionné

Le mode avide activé par le paramètre task_always_eager est par définition définition, ne convient pas aux tests unitaires.

Lorsque vous testez en mode "eager", vous ne testez qu'une émulation de ce que l'on appelle "l'émulation". ce qui se passe dans un travailleur, et il y a de nombreuses divergences entre l'émulation et ce qui se passe. émulation et ce qui se passe dans la réalité.