How do you unit test a Celery task?

python unit-testing celery
davidmytton picture davidmytton · Aug 22, 2012 · Viewed 50.8k times · Source

The Celery documentation mentions testing Celery within Django but doesn't explain how to test a Celery task if you are not using Django. How do you do this?

Answer

FlaPer87 picture FlaPer87 · Aug 22, 2012

It is possible to test tasks synchronously using any unittest lib out there. I normaly do 2 different test sessions when working with celery tasks. The first one (as I'm suggesting bellow) is completely synchronous and should be the one that makes sure the algorithm does what it should do. The second session uses the whole system (including the broker) and makes sure I'm not having serialization issues or any other distribution, comunication problem.

So:

from celery import Celery

celery = Celery()

@celery.task
def add(x, y):
    return x + y

And your test:

from nose.tools import eq_

def test_add_task():
    rst = add.apply(args=(4, 4)).get()
    eq_(rst, 8)

Hope that helps!

Related questions

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 …

Read More
Using mock to patch a celery task in Django unit tests

I'm trying to use the python mock library to patch a Celery task that is run when a model is saved in my django app, to see that it's being called correctly. Basically, the task is defined inside myapp.tasks, …

Read More
How do you test that a Python function throws an exception?

How does one write a unittest that fails only if a function doesn't throw an expected exception?

Read More