Dokumentacja Celery wspomina o testowaniu Celery w Django, ale nie wyjaśnia, jak przetestować zadanie Celery, jeśli nie używasz Django. Jak Ty to robisz?

Możliwe jest synchroniczne testowanie zadań przy użyciu dowolnej biblioteki unittest. Zwykle wykonuję 2 różne sesje testowe podczas pracy z zadaniami z selera. Pierwszy (jak sugeruję poniżej) jest całkowicie synchroniczny i powinien być tym, który zapewnia, że ​​algorytm robi to, co powinien. Druga sesja wykorzystuje cały system (w tym brokera) i upewnia się, że nie mam problemów z serializacją lub inną dystrybucją, problemem z komunikacją.

Więc:

from celery import Celery

celery = Celery()

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

I twój test:

from nose.tools import eq_

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

Mam nadzieję, że to pomoże!

Używam tego:

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

Dokumenty: http://docs.celeryproject.org/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER umożliwia synchroniczne uruchamianie zadań i nie potrzebujesz serwera selera.

Zależy od tego, co dokładnie chcesz testować.

• Przetestuj kod zadania bezpośrednio. Nie wywołuj „task.delay (...)” po prostu wywołaj „task (...)” z testów jednostkowych.
• Użyj CELERY_ALWAYS_EAGER . Spowoduje to, że Twoje zadania będą wywoływane natychmiast w momencie, gdy powiesz „task.delay (...)”, więc możesz przetestować całą ścieżkę (ale nie asynchroniczne zachowanie).

unittest

import unittest

from myproject.myapp import celeryapp

class TestMyCeleryWorker(unittest.TestCase):

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

py.test armatura

# 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):
    ...

Dodatek: spraw, by send_task cieszył się szacunkiem

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

Dla tych na Seler 4 to:

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

Ponieważ nazwy ustawień zostały zmienione i wymagają aktualizacji, jeśli zdecydujesz się na aktualizację, zobacz

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

Od wersji Celery 3.0 jednym ze sposobów ustawienia CELERY_ALWAYS_EAGERw Django jest:

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())

Od wersji 4.0 Selera, urządzenia py.test są dostarczane do uruchomienia pracownika selera tylko na czas testu i są wyłączane po zakończeniu:

def test_myfunc_is_executed(celery_session_worker):
    # celery_session_worker: <Worker: [email protected] (running)>
    assert myfunc.delay().wait(3)

Wśród innych urządzeń opisanych na http://docs.celeryproject.org/en/latest/userguide/testing.html#py-test możesz zmienić domyślne opcje selera, ponownie definiując celery_configurządzenie w ten sposób:

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

Domyślnie proces roboczy testu używa brokera w pamięci i zaplecza wyników. Nie ma potrzeby korzystania z lokalnego Redis lub RabbitMQ, jeśli nie testujesz określonych funkcji.

odniesienie using pytest.

def test_add(celery_worker):
    mytask.delay()

jeśli używasz flask, ustaw plik config

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

i w 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

W moim przypadku (i zakładam wiele innych) chciałem tylko przetestować wewnętrzną logikę zadania za pomocą pytest.

TL; DR; skończyło się wyszydzaniem wszystkiego ( OPCJA 2 )

Przykład zastosowania :

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'

ale ponieważ shared_taskdekorator ma dużo wewnętrznej logiki selera, tak naprawdę nie jest to test jednostkowy.

Tak więc dla mnie były 2 opcje:

OPCJA 1: Oddzielna logika wewnętrzna

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);

Wygląda to bardzo dziwnie i poza tym, że jest mniej czytelny, wymaga ręcznego wyodrębnienia i przekazania atrybutów, które są częścią żądania, na przykład task_id w razie potrzeby, co sprawia, że ​​logika jest mniej czysta.

OPCJA 2: kpi z
wyszydzania wnętrzności selera

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()

co następnie pozwala mi na mockowanie obiektu żądania (ponownie, jeśli potrzebujesz rzeczy z żądania, takich jak identyfikator lub licznik ponownych prób.

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'

To rozwiązanie jest znacznie bardziej ręczne, ale daje mi kontrolę, której potrzebuję, aby wykonać testy jednostkowe , bez powtarzania się i bez utraty celownika.