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
, and is imported at the top of my models.py-file like so:
from .tasks import mytask
...and then runs on save()
inside the model using mytask.delay(foo, bar)
. So far so good - works out fine when I'm actually running Celeryd etc.
I want to construct a unit test that mocks the task, just to check that it gets called with the correct arguments, and doesn't actually try to run the Celery task ever.
So in the test file, I've got something like this inside of a standard TestCase:
from mock import patch # at the top of the file
# ...then later
def test_celery_task(self):
with patch('myapp.models.mytask.delay') as mock_task:
# ...create an instance of the model and save it etc
self.assertTrue(mock_task.called)
...but it never gets called/is always false. I've tried various incarnations (patching myapp.models.mytask
instead, and checking if mock_task.delay
was called instead. I've gathered from the mock docs that the import path is crucial, and googling tells me that it should be the path as it is seen inside the module under tests (which would be myapp.models.mytask.delay
rather than myapp.tasks.mytask.delay
, if I understand it correctly).
Where am I going wrong here? Is there some specific difficulties in patching Celery tasks? Could I patch celery.task
(which is used as a decorator to mytask
) instead?
The issue that you are having is unrelated to the fact that this is a Celery task. You just happen to be patching the wrong thing. ;)
Specifically, you need to find out which view or other file is importing "mytask" and patch it over there, so the relevant line would look like this:
with patch('myapp.myview.mytask.delay') as mock_task:
There is some more flavor to this here:
http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch