直接调用绑定任务时模拟 Celery `self.request` 属性答案

【问题标题】：Mocking Celery `self.request` attribute for bound tasks when called directly直接调用绑定任务时模拟 Celery `self.request` 属性
【发布时间】：2018-04-10 13:18:20
【问题描述】：

我有一个任务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（因为我正在与数据库进行真正的交互，这使得测试失败，等等......） .

作为参考，我使用Django作为框架，但我认为这个问题是一样的，无论你使用什么环境。

【问题讨论】：

这没有意义。该任务是课程的一部分吗？如果有，请出示；如果没有，它就没有self。而且在任何情况下，任务都不会作为请求的一部分运行，那么为什么会有self.request？
@DanielRoseman 你已经用过 Celery 了吗？如果没有，请事先阅读：docs.celeryproject.org/en/latest/userguide/…；装饰器将函数包装在一个类中。

标签： django unit-testing celery

【解决方案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)

【讨论】：

【解决方案2】：

您可以使用同步调用任务

task = foobar.s(<args>).apply()

这将分配一个唯一的任务 ID，因此该值不会是 None 并且您的代码将运行。然后，您可以在测试中检查结果。

【讨论】：

【解决方案3】：

patch 可能有一种方法可以做到这一点，但我无法找到分配属性的方法。最直接的方法就是模拟自己。

tasks.py：

@app.task(name='my_task')
def my_task(self, *args, **kwargs):
    *__do some thing__*

test_tasks.py：

from mock import Mock

def test_my_task():
    self = Mock()
    self.request.id = 'ci_test'
    my_task(self)

【讨论】：