Django-q的模拟async_task

Question

我正在使用django-q，目前正在为我现有的任务添加使用mock的测试。我可以轻松地为每个任务创建测试，而不依赖于django，但我的任务之一是调用另一个async_task。下面是一个例子：

import requests
from django_q.tasks import async_task

task_a():
    response = requests.get(url)
    # process response here

    if condition:
        async_task('task_b')

task_b():
    response = requests.get(another_url)

下面是我测试它们的方法：

import requests

from .tasks import task_a
from .mock_responses import task_a_response

@mock.patch.object(requests, "get")
@mock.patch("django_q.tasks.async_task")
def test_async_task(self, mock_async_task, mock_task_a):
    mock_task_a.return_value.status_code = 200
    mock_task_a.return_value.json.return_value = task_a_response

    mock_async_task.return_value = "12345"

    # execute the task
    task_a()

    self.assertTrue(mock_task_a.called)
    self.assertTrue(mock_async_task.called)

我知道的一个事实是，async_task返回任务ID，因此行mock_async_task.return_value = "12345"。但是，在运行测试之后，mock_async_task返回False，任务将被添加到队列中(我可以从服务器看到一堆01:42:59 [Q] INFO Enqueued 1 )，这正是我想要避免的。有什么办法可以做到这一点吗？

Answer 1

为了防止任务被添加到队列中，需要在测试运行时将配置sync更改为True。您可以找到有关配置这里的更多信息。