Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Unit testing with django-celery?

I am trying to come up with a testing methodology for our django-celery project. I have read the notes in the documentation, but it didn't give me a good idea of what to actually do. I am not worried about testing the tasks in the actual daemons, just the functionality of my code. Mainly I am wondering:

  1. How can we bypass task.delay() during the test (I tried setting CELERY_ALWAYS_EAGER = True but it made no difference)?
  2. How do we use the test settings that are recommended (if that is the best way) without actually changing our settings.py?
  3. Can we still use manage.py test or do we have to use a custom runner?

Overall any hints or tips for testing with celery would be very helpful.

like image 220
Jason Webb Avatar asked Oct 29 '10 20:10

Jason Webb


People also ask

What is Django celery used for?

Celery makes it easier to implement the task queues for many workers in a Django application. Functions of Celery: Define tasks as python functions. Listen to a message broker for new tasks.

How does Django celery beat work?

celery-beat acts as the scheduler part of celery whereas the worker executes the tasks that are either instructed from within the application or by celery-beat . The installation steps for celery in a Django application is explained in celery docs here (after pip install celery 🙃).

What is unit testing in Django?

Unit Tests are isolated tests that test one specific function. Integration Tests, meanwhile, are larger tests that focus on user behavior and testing entire applications. Put another way, integration testing combines different pieces of code functionality to make sure they behave correctly.


2 Answers

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

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

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.

like image 78
joshua Avatar answered Oct 11 '22 12:10

joshua


Try setting:

BROKER_BACKEND = 'memory' 

(Thanks to asksol's comment.)

like image 45
a paid nerd Avatar answered Oct 11 '22 14:10

a paid nerd