Fix Python – How do you unit test a Celery task?


Asked By – davidmytton

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?

Now we will see solution for issue: How do you unit test a Celery task?


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.


from celery import Celery

celery = Celery()

def add(x, y):
    return x + y

And your test:

from import eq_

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

Hope that helps!

This question is answered By – FlaPer87

This answer is collected from stackoverflow and reviewed by FixPython community admins, is licensed under cc by-sa 2.5 , cc by-sa 3.0 and cc by-sa 4.0