我对芹菜任务的第一次测试

Question

我是新手，我不懂，我需要你的帮助。我有一部分 API 代码可以保存对象并在完成后发送短信。它是如何测试的？我将不胜感激你的帮助。谢谢！

看起来像这样：

tasks.py/

@shared_task
send_sms(addr, text):
    ...


views.py/

def mydef():
     my_objs_list = MyModel.values().filter(my_flag=False) # addr list
     for item in my_objs_list:
         send_sms.delay(str(item), text)
     my_objs_list.update(my_flag=True)

Answer 1

您需要模拟 delay 方法并对传递的任何参数进行断言：

@override_settings(CELERY_ALWAYS_EAGER=True)
@mock.patch('project.tasks.send_sms.delay')
def test_view(self, send_sms):
    ...
    send_sms.assert_called_with('item-value-1', 'text-value-1')
    send_sms.assert_called_with('item-value-2', 'text-value-2')