使用补丁在 Python 中模拟 Celery 任务调用答案

【问题标题】：Mocking a Celery task call in Python with patch使用补丁在 Python 中模拟 Celery 任务调用
【发布时间】：2014-05-06 03:57:47
【问题描述】：

使用模拟返回值修补 Celery 任务调用会返回 <Mock name='mock().get()' ...>，而不是 mock_task.get.return_value = "value" 定义的预期 return_value。但是，模拟任务在我的单元测试中正常运行。

这是我正在修补 Celery 任务的单元测试：

def test_foo(self):

    mock_task = Mock()
    mock_task.get = Mock(return_value={'success': True})

    print mock_task.get() # outputs {'success': True}

    with patch('app.tasks.my_task.delay', new=mock_task) as mocked_task:
        foo()  # this calls the mocked task with an argument, 'input from foo'
        mock_tasked.assert_called_with('input from foo')  # works

这是正在测试的功能：

def foo():
    print tasks.my_task.delay  # shows a Mock object, as expected
    # now let's call get() on the mocked task:
    task_result = tasks.my_task.delay('input from foo').get()
    print task_result  # => <Mock name='mock().get()' id='122741648'>
    # unexpectedly, this does not return {'success': True}
    if task_result['success']:
        ...

最后一行引发TypeError: 'Mock' object has no attribute '__getitem__'

为什么我可以在单元测试中调用 mock_task.get()，但从 foo 调用它会返回 <Mock ...> 而不是预期的返回值？

【问题讨论】：

标签： python mocking celery celery-task python-mock

【解决方案1】：

不幸的是，我对 Celery 几乎一无所知，但问题似乎在于嘲笑。

你有：

tasks.my_task.delay('input from foo').get()

patch('app.tasks.my_task.delay', new=mock_task) 之后变成：

mock_task('input from foo').get()

不一样的是：

mock_task.get()

您应该将您的模拟创建更改为：

mock_task().get = Mock(return_value={'success': True})

当您访问现有的 Mock 属性或调用它时，默认会创建新的 Mock 实例。所以我们可以稍微简化一下：

mock_task().get.return_value = {'success': True}

【讨论】：