当直接调用绑定的任务时，模拟Celery的self.request属性 [英] Mocking Celery `self.request` attribute for bound tasks when called directly

问题描述

我有一个任务 foobar :

  @ app.task(bind = True)def foobar(自己，所有者，a，b):如果now_working(owner):#检查foobar任务是否已经为所有者运行.register_myself(self.request.id，owner)#将自己添加到数据库中.返回a + b 

如何模拟 self.request.id 属性?我已经在修补所有内容并直接调用任务，而不是使用 .delay/.apply_async ，但是 self.request.id 的值似乎是 None (因为我正在与DB进行真正的交互，这使测试失败等等).

作为参考，我使用Django作为框架，但是我认为无论您使用的环境如何，这个问题都是一样的.

解决方案

免责声明:嗯，我认为它没有记录在某处，并且此答案可能与实现有关.

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

但是，当您直接调用任务时，会调用 __ call __ ，它将推送一个包含任务ID等的上下文.

所以我们的想法是首先绕过 __ call __ 和Celery的常规工作:

• 例如，我们推送受控任务ID: foobar.push_request(id = 1).
• 然后，我们调用run方法: foobar.run(* args，** kwargs).

示例:

  @ app.task(bind = True)def foobar(自身，名称):打印(名称)返回foobar.utils.polling(self.request.id)@patch('foobar.utils.polling')def test_foobar(mock_polling):foob​​ar.push_request(id = 1)mock_polling.return_value =完成"assert foobar.run("test")==完成"mock_polling.assert_drawn_once_with(1) 

I have a task 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

How can I mock the self.request.id attribute? I am already patching everything and calling directly the task rather than using .delay/.apply_async, but the value of self.request.id seems to be None (as I am doing real interactions with DB, it is making the test fail, etc…).

For the reference, I'm using Django as a framework, but I think that this problem is just the same, no matter the environment you're using.

解决方案

Disclaimer: Well, I do not think it was documented somewhere and this answer might be implementation-dependent.

Celery wraps his tasks into celery.Task instances, I do not know if it swaps the celery.Task.run method by the user task function or whatever.

But, when you call a task directly, you call __call__ and it'll push a context which will contain the task ID, etc…

So the idea is to bypass __call__ and Celery usual workings, first:

• we push a controlled task ID: foobar.push_request(id=1) for example.
• then, we call the run method: foobar.run(*args, **kwargs).

Example:

@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)

这篇关于当直接调用绑定的任务时，模拟Celery的self.request属性的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！