How to test if email was sent after executing celery task

Question

I'm using Django 1.10 and Celery 4.1

I have a shared_task which sends an email to the user.

# myapp/tasks.py
@shared_task
def notify_user(user_id):
    # TODO: send email and do other stuff here
    user = get_object_or_404(User, pk=user_id)

    send_mail(
        'Subject',
        'Body',
        'from@example.com',
        [user.email],
    )

I have another file which contains a function that calls puts that tasks into the queue.

# myapp/utils.py
# ...
def update_queue(self):
    # increment no_of_used_referrals by 1
    if no_of_used_referrals == 5:
        notify_user.apply_async((self.user_id,))
    else:
        notify_user.apply_async((self.user_id,), eta=new_eta)

Now I am trying to test whether calling update_queue() (where all required checks passes) sends an email to the user when its executed.

I tried to do the following:

# myapp/tests.py
def update_queue_should_call_notify_user_immediately_after_five_referrals_were_used(self):
    with unittest.mock.patch('myapp.tasks.notify_user.apply_async') as notify_user_mock:
        # ...
        for _ in range(5):
            entry.update_queue()
        self.assertTrue(notify_user_mock.called)
        notify_user_mock.assert_called_with((user_id,))
    # TODO: check if email was sent

    # I tried using : 
    # self.assertEqual(len(mail.outbox), 1) 
    # but it fails with error saying 0 != 1
def test_notify_user_should_send_an_email(self):
    notify_user.apply_async((user_id,))

    # I tried using:
    # self.assertEqual(len(mail.outbox), 1)
    # but it fails with error saying 0 != 1

I have set EMAIL_BACKEND = 'django.core.mail.backends.locmem.EmailBackend' in my project settings.

Can someone please tell me what is wrong with what I am doing and how to correctly test this case?

EDIT I have updated my code where I excluded mocking - as suggested by @DanielRoseman.

EDIT2 Please see updated files above.

I am simulating referral system. Once 5 referral links associated with a particular user have been used, user get's some nice feature to their profile. Otherwise they have to wait for a specific time, which I set using eta argument on apply_async.

Every time I call update_queue I check if the number of referals is equal to 5(please see updated code above).

If it is - I want to call notify_user immediately, without passing eta argument value.
If it is not - I increment number of used referral links, revoke old notify_user task, create new notify_user task with new eta argument value.

In order to test that I am simulating that behaviour in for-loop, and I want to test whether after 5 iterations(equal to 5 used referral links) an email was sent to the user (for test purposes I use in-memory backend for email).

The best I can think of is you can add a field to your database for corresponding user that an email is sent to a user or not. Then you can update this field in your celery task so if sending email is successful then field is updated. This can be easily tested — Arpit Solanki, Sep 03 '17 at 14:22
You've mocked the whole task, so naturally it won't actually do anything. You shouldn't mock the code under test. — Daniel Roseman, Sep 03 '17 at 14:33
@DanielRoseman I have updated my answer. But it still does not work. — an0o0nym, Sep 03 '17 at 14:54
@ArpitSolanki thanks for your suggestion but it sounds like an unnescessary overhead just for testing purposes. If I implement that attribute to one of my database models I would never use it anywhere outside `tests.py` file — an0o0nym, Sep 03 '17 at 15:04
It's hard to know what you are doing. You should be testing these things separately; once calling `update_queue` with the task mocked to assert it is called, and once calling the task directly and then checking the [test email outbox](https://docs.djangoproject.com/en/1.11/topics/testing/tools/#email-services). Note, you should not be changing the email backend. — Daniel Roseman, Sep 03 '17 at 16:44
@DanielRoseman I have removed the line of code from settings where I have changed my email backend. I have also split the tests as you suggested. I have also updated my question content. I hope it clarify more my problem. — an0o0nym, Sep 03 '17 at 17:18

score 1 · Answer 1 · answered Jun 20 '18 at 06:50

1

I put it here for someone that will face the same problem.

I've solved it with

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

https://stackoverflow.com/a/46531472/7396169

I think this solution is suitable for unit testing.

answered Jun 20 '18 at 06:50

urDMG

428
1
6
14

score 0 · Answer 2 · answered Jun 01 '19 at 17:42

tasks.py

from django.core.mail import EmailMessage
from django.template.loader import render_to_string
from django.contrib.auth import get_user_model

from accounts.models import Token
from celery import shared_task

@shared_task(bind=True)
def send_login_email_task(self, email):
    try:
        uid = str(uuid.uuid4())
        Token.objects.create(email=email, uid=uid)
        current_site = 'localhost:8000'
        mail_subject = 'Activate your account.'
        message = render_to_string('accounts/login_activation_email.html', {
            'domain': current_site,
            'uid': uid
        })
        print('called')
        email = EmailMessage(mail_subject, message, to=[email])
        email.send()
    except Token.DoesNotExist:
        logging.warning(
            "Tried to send activation email to non-existing user '%s'", email)
    except smtplib.SMTPException as exc:
        raise self.retry(exc=exc)

test_tasks.py

from django.test import TestCase
from unittest.mock import patch
from django.contrib.auth import get_user_model
from celery.exceptions import Retry
from proj.celery import App
import smtplib
import uuid


import accounts.tasks
from accounts.models import Token
@patch('accounts.tasks.EmailMessage')
def test_send_login_email_task(self, mock_email_message):

    # call task
    token = Token.objects.get(email=self.email, uid=self.uid)
    print(token.email)
    accounts.tasks.send_login_email_task.apply_async((token.email,))
    self.assertEqual(mock_email_message.called, True)

    # patch EmailMessage
    print(mock_email_message.call_args)
    args, kwargs = mock_email_message.call_args
    subject = args[0]
    self.assertEqual(subject, 'Activate your account.')
    self.assertEqual(kwargs, {'to': ['ama@example.com']})

How to test if email was sent after executing celery task

2 Answers2

tasks.py

test_tasks.py