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