How to Test Celery Worker Task in Django

As usual, I do not have much time, so I will give the code and elaborate on it. However, there are some things to discuss beforehand.

Forewords

Of course, I will cut it short, but I am not the one who figured this out. There are many many resources on the web on testing Celery, but (including official docs) they either lack information or they are outdated. Surely, I will try to give credit to the resources that I put together (I don’t remember some of them).

Since you are here, I assume you know what Celery is, using Django, but have no clue how to test your tasks.

While I was looking up for resources on testing Celery tasks, I encountered many outdated resources. So, in this article, I use Django 2.1 and Celery 4.2.1. It is safe to use Django 2 and Celery 4, I assume. As a message broker, I use Redis, but I don’t think it is a hard dependency to cause a problem if not used, so you may use anything you like.

I also use pytest instead of classical built-in unittest library, alongside pytest-django. I think it will not cause a problem using built-in unittest and, of course, Django’s own test module, but anyway, if you have a problem testing, try out pytest-django.

About database, I almost always try to avoid using a real database in my development environment, by this I mean MySQL (and its fork, MariaDB) and PostgreSQL. SQLite can hold its data in memory while testing, which results in blazing-fast migrations and testing. That’s why I usually write my models considering the classical relational approach, rather than using PostgreSQL’s new stuff. So, if you have a PostgreSQL specific field in your models, you might want to either migrate to classical approach, or just leave this article. Seriously, migrations while testing in development environment, even if there is an option to keep the testing database as is, is painful as heck. On the other hand, SQLite uses RAM and much faster.

And, last thing to put, the topic is not Celery beat. Celery beat is a different beast. Hope you will find your answer somewhere else.

Code & Elaboration

Without further ado, this is what my test case looks like:

from django.test import TransactionTestCase
from anyapp import tasks
from celery.contrib.testing.worker import start_worker
class FooTaskTestCase(TransactionTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.celery_worker = start_worker(app)
cls.celery_worker.__enter__()
@classmethod
def tearDownClass(cls):
super().tearDownClass()
cls.celery_worker.__exit__(None, None, None)
def setUp(self):
super().setUp()
self.task = tasks.foo.delay("bar") # whatever your method and args are
self.results = self.task.get()
def test_success(self):
assert self.task.state == "SUCCESS"

So, what happens here?

As you can see, we start Celery worker process inside setUpClass and end it in tearDownClass . There’s a reason that we do this.

Starting Celery from a terminal and running tests alongside will not point to the same database. Your tests will use test database while Celery is only aware of your real database. That will run and fail probably saying “Object does not exist.” or whatever. In order to make testing and Celery to point at the same database, we need to spawn them in the same process. At first we create a Celery worker instance in setUpClass :

cls.celery_worker = start_worker(app)

We start it with __enter__ , which is, again, in setUpClass and kill it with __exit__ , which is supposed to be in tearDownClass (Do you see None, None and None?). Since setUpClass and tearDownClass are class methods, they will only called once per test, when the test starts and ends.

In setUp , we can call our tasks. It doesn’t have to be there, you can also put it in setUpClass , but this was what I needed in my case. Also mind the self.task.get() line above, this will block the thread until it gets the result from Celery, which is vital.

Did you see that our test case is not a regular TestCase . It is actually a TransactionTestCase . I put it there so that it will not run into SQLite’s lock or atomic block. If you will not change a model, you can change it to whatever you’d like to.

References

Official Celery documentation’s testing section might be good to read, though it didn’t really help me that much.

Solanki’s amazing article is much more beginner friendly and you might like to read it. However, it is so much primitive and utilizes pure Python. Django’s environment might differ a lot to that.

This one is, again, uses pure Python and the topic is HTTP.

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store