使用mock在Django单元测试中修补芹菜任务

Question

我正在尝试使用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)吗？

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

Answer 2

该@task装饰用替换函数Task对象(见文档).如果您模拟任务本身,您将Task用a 替换(有点魔术)对象,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)