I want to test a Celery Task by raising an SMTPException when sending an email.
With the following code, located in:
my_app.mailer.tasks
from django.core.mail import EmailMultiAlternatives
@app.task(bind=True )
def send_mail(self):
subject, from_email, to = 'hello', '[email protected]', '[email protected]'
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)
and then running the following test against it:
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')
The email is sent without error.
However, if I turn the task into a standard function (my_app.mailer.views) and then run the following test against it:
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)
The string 'Exception' is displayed, but there is no exc information as to what caused the exception.
Please, what am I doing wrong?
!!UPDATE!!
I now understand why no exc info was printed for the function version of the code. This can be achieved by changing;
mock_send.side_effect = SMTPException()
to;
mock_send.side_effect = Exception(SMTPException)
resulting in;
Exception <class 'smtplib.SMTPException'>
However, the issue of how to raise the same exception in the Celery Task in the first part of this post remains.
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', '[email protected]', '[email protected]'
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)