[Answered ]-Patch Django EmailMultiAlternatives send() in a Celery Task so that an exception is raised

by

0👍

✅

A solution for testing a celery task is by utilising Celery Signatures.

This allows us to patch EmailMultiAlternatives.send, with a patch side_effect to raise an SMTPException.

It also allows us to assert that the required number of retries have been attempted.

@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])

When run against

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)

my_app.mailer.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)
👤Radial

1👍

Normally Celery task is being send to queue and run in separate process so you won’t see any output in your console. But you can use task_always_eager to force celery task to be executed locally. Try to use override_settings decorator for this:

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')
👤neverwalkaloner

Leave a comment