How to unit test code that runs celery tasks?

Question

The app I am working on is heavily asynchronous. The web application runs a lot of tasks through celery depending on user actions. The celery tasks themselves are capable of launching further tasks.

Code such as the one shown below occurs in our code base quite frequently.

def do_sth():
    logic();
    if condition:
         function1.apply_async(*args)
    else:
         function2.apply_asynch(*args)

Now we want to start unit testing any new code that we write and we are not sure how to do this. What we would like to assert in our pytest unit tests is that we want to see if function1 actually got called. We do not want to necessarily run function1 itself as we will have a unit test for the function1.

I do not want to be running celery as a process, neither do I want to run any AMQP broker during the unit test session.

Is this achievable?

Edit

It was pointed out that this is a duplicate of How do you unit test a Celery task?

It is not. Think about it. What I am asking is how to test if function has called function1 through apply_async. That question is about how do I test function1 itself. There is a major difference. I did hit that question before framing this one.

possible duplicate of [How do you unit test a Celery task?](http://stackoverflow.com/questions/12078667/how-do-you-unit-test-a-celery-task) — olofom, Jul 31 '15 at 08:00
I have seen that question and it is not what I am asking. If it isn't clear please ask more question — Ranjith Ramachandra, Jul 31 '15 at 08:04

score 18 · Accepted Answer · edited Jun 22 '20 at 14:00

18

Have a look at the unittest.mock module which allows replacing functions you don't want to be invoked with "mocks" which look sufficiently similar to the original functions to convince the calling code that it's invoking the "real thing". You can then check that the mock was actually invoked and with what parameters. Example:

from unittest.mock import patch

def test_do_sth():
    with patch('function1.apply_async') as function1_mock:
        do_sth()
        assert function1_mock.called

edited Jun 22 '20 at 14:00

Boris Verkhovskiy

14,854
11
100
103

answered Jul 31 '15 at 11:53

Sergey

11,892
2
41
52

1

Note that you can also use `function1_mock.assert_called_once_with(*args)` to test the number of calls and if was called with correct arguments. – matino Jul 31 '15 at 13:07
You can also use patch as a decorator for the test function (in this case, test_do_sth), which I prefer. – Antoine Leclair Jul 31 '15 at 17:33
Does this still work? I'm getting an AttributeError: apply_async raised in celery.local when mock tries to undo the patch after the test. (That might be because I'm using @shared_task on my tasks.) – Marius Gedminas Dec 19 '22 at 12:38

How to unit test code that runs celery tasks?

1 Answers1

Linked