如何模拟芹菜任务的绑定上下文答案

【问题标题】：How to mock the bound context of a celery task如何模拟芹菜任务的绑定上下文
【发布时间】：2019-09-13 03:59:12
【问题描述】：

如何模拟绑定上下文，或模拟 celery 任务 ID？

给定一个芹菜任务，例如：

helpers.py：

from task import some_task

def some_helper():
    some_task.delay(123)

在task.py中：

@app.task(queue="abc", bind=True)
def some_task(self, some_number: int):
    print(self.id) # how to mock this attribute access?

简单的测试用例：

from django.test.testcases import TestCase
from helpers import some_helper


class SomeTest(TestCase):

    def test_some_helper(self):
        some_helper()

我试过了：

 @patch("celery.app.base.Celery.task", return_value=lambda x: x)

我也试过了：

class MockResult(dict):
    def __getattr__(self, x):
        return self[x]

...
def test_some_task(self):
    cls = MockResult({"id": "asdf"})
    bound_some_task = some_task.__get__(cls, MockResult)
    bound_some_task(123)

相关：

【问题讨论】：

标签： django python-3.x unit-testing celery

【解决方案1】：

帮助我创建一个模拟任务的模拟类

class CeleryTaskHelper:
    """Mock task id"""

    class Request:
        @property
        def id(self):
            return ''.join(random.choices(string.ascii_uppercase + string.digits, k=20))

    @property
    def request(self):
        return self.Request()

    def revoke(self):
        return

然后

@patch('apps.orders.tasks.activate_order.apply_async', return_value=CeleryTaskHelper())

【讨论】：

我什至不需要新类，只需将.apply_async 添加到补丁字符串就可以了。谢谢！

【解决方案2】：

给定一个看起来像这样的 celery 任务：

@my_celery_app.task(bind=True)
def my_task(self):
    if self.request.retries == 1:
        my_method_to_invoke()
        # Do work for first retry
    elif self.request.retries == 2:
        # Do work for second retry
    # do work for main task

测试可以通过模拟 celery 中的基本Task.request 类属性来设置self.request.retries。

在单元测试中可以做到以下几点

@patch("path.to.task.my_method_to_invoke")
@patch("celery.app.task.Task.request")
def my_test_method(self, mock_task_request, mock_my_method_to_invoke):

    # Set the value of retries directly
    mock_task_request.retries = 1

    # Call the task and assert the inside method was
    # called
    my_task()

    mock_my_method_to_invoke.assert_called_once()

也许可以对任务上的id 做同样的事情。我被引导到这个答案，寻找如何在绑定的芹菜任务上模拟 self。

【讨论】：

【解决方案3】：

我尝试了 setattr 和补丁注释，但都没有为我工作......似乎因为“request.id”是一个嵌套对象，我无法使用 setattr（尽管应该有一种方法)。

一种选择是使用同步调用任务

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

这将分配一个唯一的任务 ID。

另一种选择是让任务成为包装方法，例如：

@app.task
def some_task(self, ...args):
    some_method(self.request.id, ...args)

然后你可以直接测试 some_method，传入你喜欢的任何 ID。

【讨论】：

【解决方案4】：

能够通过在任务方法上使用setattr 来获得一些工作，不确定是否有更好的/其他方法来做到这一点：

from django.test.testcases import TestCase
from helpers import some_helper

class SomeTest(TestCase):

    def test_some_helper(self):

        from task import some_task
        setattr(some_task, 'id', 'hello-id')

        some_helper()

除此之外，还可以像这样模拟 request.id 或“任务 id”：

@patch("task.some_task.request.id", return_value="hello-id")
def test_some_helper(...): ....

【讨论】：