Unit testing with django-celery?

I like to use the override_settings decorator on tests which need celery results to complete.

from django.test import TestCasefrom django.test.utils import override_settingsfrom myapp.tasks import mytaskclass 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())

If you want to apply this to all tests you can use the celery test runner as described at http://docs.celeryproject.org/en/2.5/django/unit-testing.html which basically sets these same settings except (BROKER_BACKEND = 'memory').

In settings:

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

Look at the source for CeleryTestSuiteRunner and it's pretty clear what's happening.

Try setting:

BROKER_BACKEND = 'memory'

(Thanks to asksol's comment.)

Here's an excerpt from my testing base class that stubs out the apply_async method and records to the calls to it (which includes Task.delay.) It's a little gross, but it's managed to fit my needs over the past few months I've been using it.

from django.test import TestCasefrom 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.htmlclass 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)

Here's an "off the top of the head" example of how you'd use it in your test cases:

mymodule.py

from my_tasks import SomeTaskdef 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)