python之如何在执行 celery 任务后测试是否发送了电子邮件_编程设计

我正在使用 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 所建议的那样。

编辑2 请参阅上面的更新文件。

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

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

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

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

请您参考如下方法:

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

我已经解决了

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

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

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