41

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

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

from .tasks import mytask

...然后save()使用mytask.delay(foo, bar). 到目前为止一切顺利 - 当我实际运行 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）吗？

4

3 回答 3

55

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

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

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

这里有更多的味道：

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

于 2013-11-08T20:04:30.993 回答
36

装饰器将@task函数替换为Task对象（参见文档）。如果您模拟任务本身，您将用Taska 替换（有点神奇的）对象，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)
于 2015-03-26T00:46:45.140 回答
0

只需修补 celery Task 方法

mocker.patch("celery.app.task.Task.delay", return_value=1)
于 2022-02-24T00:42:37.280 回答