如何模拟绑定上下文,或模拟芹菜任务 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)
相关:
- Python:绑定未绑定的方法?
- https://docs.python.org/2/howto/descriptor.html
- 直接调用时模拟绑定任务的芹菜"self.request"属性
给定一个芹菜任务,如下所示:
@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
该测试可以通过模拟芹菜中的基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
。
能够通过在任务方法上使用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(...): ....
帮助我的是创建了一个模拟任务的模拟类
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())
我尝试了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。