Vérifiez l'état de la tâche de céleri à partir d'un autre module

Question

J'essaie de tester si une tâche de céleri a été démarrée après une demande à mon serveur Django.J'ai quelque chose comme ça:

# tasks.py
def add(x, y):
    return x + y

# views.py
def home(request): # respond to request at root url
    tasks.add.delay(1,2)
    return HttpResponse('hello world')

# tests.py
class MyTest(TestCase):
    def test_task_triggered(self):
        self.client.get('/')

        # XXXX HOW TO TEST THAT TASK HAS BEEN TRIGGERED?

Comment puis-je tester si la tâche a été démarrée dans mes tests d'unité?Évidemment, je n'ai pas d'accès direct à l'ID de tâche sinon quelque chose comme çatravail .

plus généralement, comment détecter les tâches de céleri se déclenchent de différentes fonctions, classes ou modules?

Merci pour votre aide.

Answer 1

You can use Mock for this (when unit testing). Patch out the delay attribute on your task, and then check the called attribute of the mock. In your example, try something like:

# tests.py
from mock import Mock, patch

class MyTest(TestCase):
    def test_task_triggered(self):
        with patch('views.tasks.add.delay') as patch_mock:
            self.client.get('/')
        self.assertTrue(patch_mock.called)

The caveat is that once you've done this, tasks.add won't actually be called in this test case, as we've replaced it with a mock. This test would simply assert that it's being called (and you could check patch_mock.call_args_list to assert it's being called with the correct arguments)

If that's a deal-breaker - you want to assert in one test case that a task is being called, as well as the impact/side effects of the task - you could experiment with the wraps attribute of mock, which might work:

# tests.py
from mock import Mock, patch

class MyTest(TestCase):
    def test_task_triggered(self):
        add_mock = Mock(wraps=tasks.add)
        with patch('views.tasks.add.delay', new=add_mock):
            self.client.get('/')
        self.assertTrue(add_mock.called)

But the other way is probably better, since it better isolates what's being tested.

Answer 2

I found their are two ways which can make task run synchronously in my test

1. Use CELERY_ALWAYS_EAGER = True.

In top of your test file, you can set it up like this

from django.conf import settings
settings.CELERY_ALWAYS_EAGER = True

from django.test.client import Client

class MyTest(unittest.TestCase):
    def setUp(self):
        self.client = Client()

    def test_sample(self):
        self.client = client.get('/')

2. For celery 2.5, i use mock to mock delay method

Here is my example

import mock

def mock_delay(task, *args, **kwargs):
    task.run(args, kwargs)

@mock.patch('celery.app.task.BaseTask.delay', mock_delay)
class MyTest(unittest.TestCase):
    def setUp(self):
        self.client = Client()

    def test_sample(self):
        self.client = client.get('/')

This solution work well with celery2.5.3, but when i upgrade to celery3.0.12, it doesn't work anymore.

It seems like celery2.5.3 change class BaseTask to class Task in celery.app.task module.

I modify my code to @mock.patch('celery.app.task.Task.delay', mock_delay) but can not solved it. Still try to figure it out

Answer 3

Celery does not allow such kind of interaction, but you can use cache or even session to monitor it's state. Sure, you'll need to pass some ID from outside to get that value further.

But if you need this for tests only, you can use CELERY_ALWAYS_EAGER setting which tells Celery to execute all tasks as simple procedures without even need for celeryd running.