How to Test Celery Worker Task in Django
This article of mine has been taken from Medium. Article is old but still gives proper basis on how to start Celery worker in a test. I, from time to time, refer to it and thought that Hashnode community might refer to it as well.
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.
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): def setUpClass(cls): super().setUpClass() cls.celery_worker = start_worker(app) cls.celery_worker.__enter__() 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
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
tearDownClass are class methods, they will only called once per test, when the test starts and ends.
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.
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.