54

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?

Improve this question
1
  • 1
    have you tried setting "CELERY_ALWAYS_EAGER = True" instead of mock it? Commented Sep 5, 2013 at 12:27

3 Answers 3

Reset to default
65

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

Improve this answer
Sign up to request clarification or add additional context in comments.

6 Comments

Cheers! I've yet to try it out (project is dormant right now) but will try it out soon and mark this as answered. I seem to recall having tried a bunch of variations on the theme you're suggesting, but it's fully possible my blood sugar was low at the time... :-)
Actually, I'm doing this pretty much exactly as you suggest, as exemplified in the question code... Cannot get it to work. Oh well.
The question is patching the model. That smells wrong as I suspect you are not using "delay" in the model, but somewhere else - possibly a view, hence my patch code (above) is slightly different.
Well, I'm calling mytask.delay() inside of the overridden save() method on the model, actually. (I'm using the task to send some data from the model to an external system). Would that make a difference in how the patching works?
Went back and tried everything from scratch: this time it worked fine. No idea what I was doing wrong before, but I'm happy. :-)
|
48

The @task decorator replaces the function with a Task object (see documentation). If you mock the task itself you'll replace the (somewhat magic) Task object with a MagicMock and it won't schedule the task at all. Instead mock the Task object's run() method, like so:

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
@patch('monitor.tasks.monitor_user.run')
def test_monitor_all(self, monitor_user):
    """
    Test monitor.all task
    """

    user = ApiUserFactory()
    tasks.monitor_all.delay()
    monitor_user.assert_called_once_with(user.key)
Improve this answer

10 Comments

There a reason you posted this answer, word for word, on two questions?
@Micheled'Amico: I disagree, the answer talks to the specific problem of it not being run if you mock the task itself which this explains. Mocking the delay() method, while likely to work in a specific case is actually the wrong answer because you might later change the code to use apply_async() or other method and suddenly your tests will break for the wrong reason.
+1 @DanielleMadeley. I originally patched delay and the tests worked great until we needed to chain them as subtasks and everything went south in a hury!
In newer versions, we need to use CELERY_TASK_ALWAYS_EAGER
I think this is a way better answer because it works for chained tasks as well as tasks called with apply_async (to provide a countdown parameter).
|
3

Just patch the celery Task method

mocker.patch("celery.app.task.Task.delay", return_value=1)
Improve this answer

2 Comments

Not sure of the structure of your app but mock can't find my tasks this way
@boatcoder depends on the version of celery you are using. You can just look in the source code where that function is located.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.