在 Django 单元测试中使用 mock 来修补 celery 任务

Question

我正在尝试使用 python 模拟库来修补在我的 django 应用程序中保存模型时运行的 Celery 任务，以查看它是否被正确调用。

基本上，任务在 myapp.tasks 中定义，并在我的 models.py 文件的顶部导入，如下所示：

from .tasks import mytask

…然后在模型内部使用 mytask.delay(foo, bar) save() 到目前为止一切顺利 - 当我实际运行 Celeryd 等时效果很好。

我想构建一个模拟任务的单元测试，只是为了检查它是否使用正确的参数调用，并且实际上并没有尝试运行 Celery 任务。

所以在测试文件中，我在一个标准的 TestCase 中有这样的东西：

 from mock import patch # at the top of the file

# ...then later
def test_celery_task(self):
    with patch('myapp.models.mytask.delay') as mock_task:
        # ...create an instance of the model and save it etc
        self.assertTrue(mock_task.called)

…但它永远不会被调用/总是错误的。我尝试了各种化身（改为修补 myapp.models.mytask ，并检查是否调用了 mock_task.delay 。我从模拟文档中收集到导入路径至关重要，谷歌搜索告诉我它应该是在被测模块内部看到的路径（如果我理解正确的话，它将是 myapp.models.mytask.delay 而不是 myapp.tasks.mytask.delay ）。

我哪里错了？修补 Celery 任务有什么具体困难吗？我可以修补 celery.task （用作 mytask 的装饰器）吗？

原文由 Emil 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您遇到的问题与这是 Celery 任务这一事实无关。您只是碰巧修补了错误的东西。 ;)

具体来说，您需要找出哪个视图或其他文件正在导入“mytask”并在那里修补它，因此相关行将如下所示：

 with patch('myapp.myview.mytask.delay') as mock_task:

这里有更多的味道：

http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch

原文由 Thanos Diacakis 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

@task 装饰器用 Task 对象替换函数（参见 文档）。如果您模拟任务本身，您将用 --- 替换（有点神奇的） Task MagicMock ，它根本不会安排任务。而是模拟 Task 对象的 run() 方法，如下所示：

 @override_settings(CELERY_ALWAYS_EAGER=True)
@patch('monitor.tasks.monitor_user.run')
def test_monitor_all(self, monitor_user):
    """
    Test monitor.all task
    """

    user = ApiUserFactory()
    tasks.monitor_all.delay()
    monitor_user.assert_called_once_with(user.key)

原文由 Danielle Madeley 发布，翻译遵循 CC BY-SA 3.0 许可协议