Макет async_task из Django-q

Я использую django-q и в настоящее время работаю над добавлением тестов с помощью mock для моих существующих задач. Я мог бы легко создать тесты для каждой задачи, не завися от django-q, но одна из моих задач вызывает другую 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 возвращает идентификатор задачи, отсюда и строка mock_async_task.return_value = "12345". Однако, после выполнения теста, mock_async_task возвращает False и задача добавляется в очередь (я видел кучу 01:42:59 [Q] INFO Enqueued 1 от сервера), чего я и пытаюсь избежать. Есть ли способ сделать это?

Вернуться на верх