У меня есть API REST, написанный на Django, с конечной точкой и конечной точкой, которая ставит задачу по сельдеру при отправке на нее. Ответ содержит идентификатор задачи, который я хотел бы использовать для проверки того, что задача создана и получить результат. Итак, я хотел бы сделать что-то вроде:
def test_async_job():
response = self.client.post("/api/jobs/", some_test_data, format="json")
task_id = response.data['task_id']
result = my_task.AsyncResult(task_id).get()
self.assertEquals(result, ...)
Я, очевидно, не хочу запускать работника сельдерея для запуска модульных тестов, я ожидаю, что он каким-то образом издевается над ним. Я не могу использовать CELERY_ALWAYS_EAGER, потому что это, похоже, вообще обходит брокера, не позволяя мне использовать AsyncResult, чтобы получить задание по его идентификатору (как указано здесь).
Пройдя через сельдерей и kombu docs, я обнаружил, что в модульных тестах есть транспорт в памяти, который будет делать то, что Я ищу. Я попытался переопределить параметр BROKER_URL
, чтобы использовать его в тестах:
@override_settings(BROKER_URL='memory://')
def test_async_job():
Но поведение такое же, как и с ampq broker: он блокирует тест, ожидающий результата. Любая идея, как я должен настроить этого брокера, чтобы заставить его работать в тестах?