Исправление Django EmailMultiAlternatives send() в Celery Task так, чтобы возникало исключение

Я хочу протестировать Celery Task, вызвав SMTPException при отправке электронного письма.

С помощью следующего кода, расположенного в:

my_app.mailer.tasks

from django.core.mail import EmailMultiAlternatives

@app.task(bind=True )
def send_mail(self):
    subject, from_email, to = 'hello', 'from@example.com', 'to@example.com'
    text_content = 'This is an important message.'
    html_content = '<p>This is an <strong>important</strong> message.</p>'
    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
    msg.attach_alternative(html_content, "text/html")
    try:
        msg.send(fail_silently=False)
    except SMTPException as exc:
        print('Exception ', exc)

и затем выполнить следующий тест против него:

class SendMailTest(TestCase):

    @patch('my_app.mailer.tasks.EmailMultiAlternatives.send')
    def test_task_state(self, mock_send):
        mock_send.side_effect = SMTPException()
        task = send_mail.delay()
        results = task.get()
        self.assertEqual(task.state, 'SUCCESS')

Электронная почта отправлена без ошибок.

Однако, если я превращу задачу в стандартную функцию (my_app.mailer.views) и затем запущу следующий тест против нее:

class SendMailTest(TestCase):

    @patch('myapp.mailer.views.EmailMultiAlternatives.send')
    def test_task_state(self, mock_send):
        mock_send.side_effect = SMTPException()
        send_mail(fail_silently=False)

Отображается строка 'Exception', но нет никакой информации о том, что вызвало исключение.

Пожалуйста, что я делаю не так?

Обычно задача Celery отправляется в очередь и выполняется в отдельном процессе, поэтому вы не увидите никакого вывода в консоли. Но вы можете использовать task_always_eager, чтобы заставить задачу Celery выполняться локально. Попробуйте использовать для этого декоратор override_settings:

from django.test import TestCase, override_settings

class SendMailTest(TestCase):

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    @patch('my_app.mailer.tasks.EmailMultiAlternatives.send')
    def test_task_state(self, mock_send):
        mock_send.side_effect = SMTPException()
        task = send_mail.delay()
        results = task.get()
        self.assertEqual(task.state, 'SUCCESS')

Вот вариант тестирования задачи celery, использующий Celery Signatures.

Это позволяет нам исправить EmailMultiAlternatives.send, при этом побочный_эффект исправления будет вызывать исключение SMTPException.

Это также позволяет нам утверждать, что было предпринято необходимое количество повторных попыток.

@patch('my_app.mailer.tasks.EmailMultiAlternatives.send')
def test_smtp_exception(self, alt_send):
    with self.assertLogs(logger='celery.app.trace') as cm:
        alt_send.side_effect = SMTPException(SMTPException)
        task = send_mail.s(kwargs=self.message).apply()
        exc = cm.output

        self.assertIn('Retry in 1s', exc[0])
        self.assertIn('Retry in 2s', exc[1])
        self.assertIn('Retry in 4s', exc[2])
        self.assertIn('Retry in 8s', exc[3])

При столкновении

base_tasks.py

def backoff(attempts):
    return 2 ** attempts

class BaseTaskEmail(app.Task):
     
    abstract = True

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        super(BaseTaskEmail, self).on_retry(exc, task_id, args, kwargs, einfo)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        super(BaseTaskEmail, self).on_failure(exc, task_id, args, kwargs, einfo)

tasks.py

@app.task(bind=True,
          max_retries=4,
          base=BaseTaskEmail,
          )
def send_mail(self):
    subject, from_email, to = 'hello', 'from@example.com', 'to@example.com'
    text_content = 'This is an important message.'
    html_content = '<p>This is an <strong>important</strong> message.</p>'
    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
    msg.attach_alternative(html_content, "text/html")
    try:
        msg.send(fail_silently=False)

    except SMTPException as exc:
        self.retry(countdown=backoff(self.request.retries), exc=exc)
Вернуться на верх