django - 直接调用绑定(bind)任务时的 Mocking Celery `self.request` 属性

Question

我有一个任务foobar :

@app.task(bind=True)
def foobar(self, owner, a, b):
   if already_working(owner): # check if a foobar task is already running for owner.
       register_myself(self.request.id, owner) # add myself in the DB.
   return a + b

我如何模拟 self.request.id属性？我已经在修补所有内容并直接调用任务而不是使用 .delay/.apply_async ，但 self.request.id 的值好像是 None (当我与 DB 进行真正的交互时，它会使测试失败，等等……)。

作为引用，我使用 Django 作为框架，但我认为这个问题是一样的，不管你使用的是什么环境。

Answer 1

免责声明 :嗯，我认为它没有记录在某个地方，这个答案可能取决于实现。

Celery 将他的任务打包到 celery.Task实例，我不知道它是否交换了 celery.Task.run通过用户任务功能或其他方法。

但是，当你直接调用任务时，你调用 __call__它会推送一个包含任务 ID 等的上下文……

所以想法是绕过__call__和 celery 通常的工作原理，首先:

我们推送一个受控任务 ID:foobar.push_request(id=1)例如。

然后，我们调用run方法:foobar.run(*args, **kwargs) .

例子:

@app.task(bind=True)
def foobar(self, name):
    print(name)
    return foobar.utils.polling(self.request.id)

@patch('foobar.utils.polling')
def test_foobar(mock_polling):
    foobar.push_request(id=1)
    mock_polling.return_value = "done"
    assert foobar.run("test") == "done"
    mock_polling.assert_called_once_with(1)