如何在执行 celery 任务后测试是否发送了电子邮件

Question

我正在使用 Django 1.10 和 Celery 4.1

我有一个shared_task，它会向用户发送一封电子邮件。

# 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],
    )

我有另一个文件，其中包含一个调用将任务放入队列的函数。

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

现在我正在尝试测试调用update_queue()（所有必需的检查通过）在执行时是否会向用户发送电子邮件。

我尝试执行以下操作：

# 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

我在项目设置中设置了EMAIL_BACKEND = 'django.core.mail.backends.locmem.EmailBackend'。

谁能告诉我我在做什么有什么问题以及如何正确测试这个案例？

编辑 正如@DanielRoseman 所建议的那样，我已经更新了排除模拟的代码。

EDIT2 请参阅上面的更新文件。

我正在模拟推荐系统。一旦使用了与特定用户关联的 5 个推荐链接，用户就会在他们的个人资料中获得一些不错的功能。否则他们必须等待一个特定的时间，这是我在apply_async 上使用eta 参数设置的。

每次我调用update_queue 时，我都会检查引用的数量是否等于 5（请参阅上面的更新代码）。

• 如果是 - 我想立即调用notify_user，而不传递eta 参数值。
• 如果不是 - 我增加使用的推荐链接的数量，撤销旧的 notify_user 任务，使用新的 eta 参数值创建新的 notify_user 任务。

为了测试我在 for 循环中模拟该行为，并且我想测试在 5 次迭代（等于 5 次使用的推荐链接）后是否向用户发送了一封电子邮件（出于测试目的，我在 -电子邮件的内存后端）。

Answer 1

我把它放在这里是为了那些将面临同样问题的人。

我已经解决了

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

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

我认为这个解决方案适合单元测试。

Answer 2

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']})