Python: Mock не работает внутри задачи celery

Question

Я хочу использовать библиотеку python mock для тестирования того, что приложение Django отправляет электронную почту.

код тест:

# 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') 

Первого пример - без задач

# 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!') 

Второго пример - с задачами

# 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') 

Первый пример работает нормальный, второй е xample failed, с исключением Not called.

Мои настройки:

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

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

Можно ли выполнять такие интегрированный тест или единственный способ решить эту проблему раскола тест на два?

Answer 1

Я нашел проблему, и это было довольно глупо. Described here и Here:

Основной принцип заключается в том, что вы патч, где объект ищется, который не обязательно то же место, где она определена.

мне нужно изменить:

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

с

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

Answer 2

Тестирование поведения асинхронных задач непосредственно из кода, который их использует, может быть сложным. Одна из причин заключается в том, что тест может выполнять утверждения даже до того, как действительно выполняется задача, что, скорее всего, даст вам ложные срабатывания. То, что я делаю в таких случаях, это разбить тестирование на два этапа:

1. Смещение задачи и тестирование, которое она вызывается, когда она должна быть вызвана и с ожидаемыми аргументами.
2. Проверьте задачу как автономную функцию и выполнив ее как обычную функцию, то есть без использования сервера сельдерея.

Чтобы проиллюстрировать, что это может быть, как:

# 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' 

Таким образом, вы можете проверить, что ваш внедрения кода ведет себя правильно по отношению к вашей задаче и что задача правильно ведет себя, как только это называется, как и ожидалось.

Надеюсь, это полезно! :)