Python 单元测试 task.apply_async()

Question

用coverage看有什么要测试的，coverage显示next要测试的是： send_alert.apply_async()

我知道是celery任务，但是有什么办法可以测试这行代码吗？

用于理解测试逻辑的其余代码：

class SomeView(GenericAPIView)
    def post(self, request, *args, **kwargs):
        #some code
        if not value:
           send_alert.apply_async()
           # response
        # not if Response

根据第一个答案编写测试

    @patch('event.views.send_alert')
    def test_send_alert(self, mock_send):
        data = {some data for post}
        response = self.client.post('/api/{0}/mettings/'.format(self.user), data)
        print(response.json())
        self.assertTrue(mock_send.called)

我还查看了任务 print('Tasks Passed') 之后的消息，我看到任务已通过的消息，但测试因错误 AssertionError: False is not true

而失败

Answer 1

我假设您使用的是 Django，因为它是这样标记的。

用以下方式装饰你的测试：

@override_settings(CELERY_EAGER=True)
def test_foo(self):
    # Make your post

或者如果您只是想测试它是否被调用。

from mock import patch

@patch('your_module.your_file.send_alert.apply_async')
def test_foo(self, mock_send):
    # make your post with "value" set to True
    self.assertTrue(mock_send.called)

Python 单元测试 task.apply_async()

Python unittest task.apply_async()

python

django-testing

celery

python-unittest

django-tests