check celery task status from another module

Question

I am trying to test whether a celery task has been started after a request to my django server. I have something like this:

# 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?

How can I test whether or not the task has been started in my unit tests? Obviously, I don't have direct access to the task id otherwise something like this would work.

More generally, how can you detect celery tasks being triggered from across functions, classes, or modules?

Thanks for your help.

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.