Tests unitaires avec le django-céleri?

Question

Je suis en train d'essayer de trouver une méthodologie de test pour notre django-céleri projet. J'ai lu les notes de la documentation, mais il ne m'a pas donné une bonne idée de ce qu'il faut réellement faire. Je ne suis pas inquiet sur la façon de tester les tâches dans les démons, juste les fonctionnalités de mon code. Surtout, je me demande:

1. Comment peut-on contourner task.delay() pendant le test (j'ai essayé paramètre CELERY_ALWAYS_EAGER = True , mais il ne fait pas de différence)?
2. Comment pouvons-nous utiliser les paramètres de test sont recommandés (si c'est le meilleur moyen) sans changer notre settings.py?
3. Peut-on encore utiliser manage.py test ou devons-nous utiliser un coureur?

Général des conseils pour les tests avec le céleri serait très utile.

Answer 1

J'aime utiliser la override_settings décorateur sur des tests qui ont besoin de céleri résultats complets.

from django.test import TestCase
from django.test.utils import override_settings
from myapp.tasks import mytask

class AddTestCase(TestCase):

    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                       CELERY_ALWAYS_EAGER=True,
                       BROKER_BACKEND='memory')
    def test_mytask(self):
        result = mytask.delay()
        self.assertTrue(result.successful()

Si vous voulez l'appliquer à tous les tests, vous pouvez utiliser le céleri lanceur de test tel que décrit à http://docs.celeryproject.org/en/2.5/django/unit-testing.html ce qui en gros ensembles de ces mêmes paramètres à l'exception de (BROKER_BACKEND = 'memory').

Dans les paramètres:

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

Regardez le code source pour CeleryTestSuiteRunner et il est assez clair de ce qui se passe.

Answer 2

Essayez de régler:

 BROKER_BACKEND = 'memory'
 

(Merci au commentaire de askol .)

Answer 3

Voici un extrait de mon test de la classe de base que les talons de l' apply_async méthode et des dossiers pour les appels (qui inclut Task.delay.) C'est un peu brut, mais il a réussi à l'adapter à mes besoins au cours de la dernière quelques mois, j'ai été à l'aide.

from django.test import TestCase
from celery.task.base import Task
# For recent versions, Task has been moved to celery.task.app:
# from celery.app.task import Task
# See http://docs.celeryproject.org/en/latest/reference/celery.app.task.html

class CeleryTestCaseBase(TestCase):

    def setUp(self):
        super(CeleryTestCaseBase, self).setUp()
        self.applied_tasks = []

        self.task_apply_async_orig = Task.apply_async

        @classmethod
        def new_apply_async(task_class, args=None, kwargs=None, **options):
            self.handle_apply_async(task_class, args, kwargs, **options)

        # monkey patch the regular apply_sync with our method
        Task.apply_async = new_apply_async

    def tearDown(self):
        super(CeleryTestCaseBase, self).tearDown()

        # Reset the monkey patch to the original method
        Task.apply_async = self.task_apply_async_orig

    def handle_apply_async(self, task_class, args=None, kwargs=None, **options):
        self.applied_tasks.append((task_class, tuple(args), kwargs))

    def assert_task_sent(self, task_class, *args, **kwargs):
        was_sent = any(task_class == task[0] and args == task[1] and kwargs == task[2]
                       for task in self.applied_tasks)
        self.assertTrue(was_sent, 'Task not called w/class %s and args %s' % (task_class, args))

    def assert_task_not_sent(self, task_class):
        was_sent = any(task_class == task[0] for task in self.applied_tasks)
        self.assertFalse(was_sent, 'Task was not expected to be called, but was.  Applied tasks: %s' %                 self.applied_tasks)

Voici un "hors de la partie supérieure de la tête", exemple de comment vous pouvez l'utiliser dans vos cas de test:

mymodule.py

from my_tasks import SomeTask

def run_some_task(should_run):
    if should_run:
        SomeTask.delay(1, some_kwarg=2)

test_mymodule.py

class RunSomeTaskTest(CeleryTestCaseBase):
    def test_should_run(self):
        run_some_task(should_run=True)
        self.assert_task_sent(SomeTask, 1, some_kwarg=2)

    def test_should_not_run(self):
        run_some_task(should_run=False)
        self.assert_task_not_sent(SomeTask)

Answer 4

comme je le vois toujours dans les résultats de recherche, les paramètres sont remplacés par

 TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
 

travaillé pour moi selon Celery Docs