我有一个任务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 作为框架,但我认为这个问题是一样的,不管你使用的是什么环境。
最佳答案
免责声明 :嗯,我认为它没有记录在某个地方,这个答案可能取决于实现。
Celery 将他的任务打包到 celery.Task
实例,我不知道它是否交换了 celery.Task.run
通过用户任务功能或其他方法。
但是,当你直接调用任务时,你调用 __call__
它会推送一个包含任务 ID 等的上下文……
所以想法是绕过__call__
和 celery 通常的工作原理,首先:
foobar.push_request(id=1)
例如。 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)
关于django - 直接调用绑定(bind)任务时的 Mocking Celery `self.request` 属性,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/47000371/