Python: Mock doesn't work inside celery task

Question

I want to use python mock library for testing that my Django application sends email.

Test code:

# tests.py
from django.test import TestCase

class MyTestCase(TestCase):

    @mock.patch('django.core.mail.mail_managers')
    def test_canceled_wo_claiming(self, mocked_mail_managers):
        client = Client()
        client.get('/')
        print(mocked_mail_managers.called)
        mocked_mail_managers.assert_called_with('Hi, managers!', 'Message Body')

First example - without tasks

# views.py
from django.views.generic import View
from django.core.mail import mail_managers

class MyView(View):

    def get(self, request):
        mail_managers('Hi, managers!', 'Message Body')
        return HttpResponse('Hello!')

Second example - with tasks

# views.py
from django.views.generic import View
from . import tasks

class MyView(View):
    def get(self, request):
        tasks.notify.apply_async()
        return HttpResponse('Hello!')


# tasks.py
from celery import shared_task
from django.core.mail import mail_managers

@shared_task
def notify():
    mail_managers('Hi, managers!', 'Message Body')

The first example works normal, second example fails, with Not called exception.

My settings:

# Celery
BROKEN_URL = 'memory://'
BROKER_BACKEND = 'memory'

CELERY_ALWAYS_EAGER = True
CELERY_EAGER_PROPAGATES_EXCEPTIONS = True
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

Is it possible to perform such integrated test or the only way to solve this problem is split test into two?

score 7 · Accepted Answer · answered May 29 '15 at 14:51

7

I found a problem and it was pretty silly. Described here and Here:

The basic principle is that you patch where an object is looked up, which is not necessarily the same place as where it is defined.

I need to change:

@mock.patch('django.core.mail.mail_managers')

with

@mock.patch('path.to.tasks.mail_managers')

answered May 29 '15 at 14:51

Nikolai Golub

3,327
4
31
61

And another bit - I called task with `task_name.delay(args)`, so I had to patch `task_caller_location.task_name.delay`, not just `task_caller_location.task_name`. – egor83 Apr 24 '21 at 17:02

score 4 · Answer 2 · answered May 29 '15 at 14:32

Testing the behavior of asynchronous tasks directly from the code that uses them can be tricky. One reason is that the test may execute assertions even before the task actually runs, which will likely give you false positives. What I do in cases like this is to break the testing into two steps:

Mocking the task and testing that it is called when it has to be called and with the expected arguments.
Test the task as a standalone function and by executing it as a normal function, i.e. without the need of the celery server.

To illustrate, this could be like:

# views.py
from path.to.tasks import my_task


def my_view(requtest):
    # do stuff
    my_task.delay('foo', 'bar')
    return HttpResponse('whatever')


# test_my_task.py
from views import my_view
from path.to.tasks import my_task


class MyTest(TestCase):
    @mock.patch('path.to.tasks.my_task')
    def test_my_task_is_called(self, mocked_task):
        client = Client()
        client.get('/')
        my_task.assert_called_with('foo', 'bar')

    def test_my_task_works(self):
        my_task('foo', 'bar')  # note I don't use .delay(...), .apply_async(...), etc
        assert 'my task did what I expected it to do'

This way you can test that your implementing code behaves correctly in regards to your task and that the task behaves correctly once it's called as expected.

I hope it's useful! :)

Gerard, thanks for your writeup! Could not figure out the official docs, but your approach worked just fine! — egor83, Apr 24 '21 at 18:52

Python: Mock doesn't work inside celery task

2 Answers2