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.

I have seen that question and it is not what I am asking. If it isn't clear please ask more question — Ranjith Ramachandra
– Ranjith Ramachandra, Commented Jul 31, 2015 at 8:04

Sergey · Accepted Answer · 2020-06-22 14:00:32Z

19

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, 2020 at 14:00

user3064538

answered Jul 31, 2015 at 11:53

Sergey

12.5k4 gold badges43 silver badges54 bronze badges

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

3 Comments

matino Over a year ago

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.

Antoine Leclair Over a year ago

You can also use patch as a decorator for the test function (in this case, test_do_sth), which I prefer.

Marius Gedminas Over a year ago

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.)

Collectives™ on Stack Overflow

How to unit test code that runs celery tasks?

1 Answer 1

3 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related