Я пытаюсь выполнить интеграционное тестирование параллельной задачи сельдерея в своем приложении Django. Я хочу, чтобы задача фактически выполнялась одновременно на рабочем месте во время интеграционного теста pytest, но у меня возникли проблемы с этой работой.
Скажем, у меня есть следующая основная задача сельдерея:
@shared_task
def sleep_task(secs):
print(f"Sleeping for {secs} seconds...")
for i in range(secs):
time.sleep(i)
print(f"\t{i + 1}")
return "DONE"
Параллельный запуск задачи в скрипте работает нормально:
# script
sleep_task.delay(3)
sleep_task.delay(3)
# output
Sleeping for 3 seconds...
Sleeping for 3 seconds...
1
1
2
2
3
3
Однако я не могу воспроизвести это асинхронное поведение в модульном тесте. В conftest.py я настраиваю брокера и серверную часть результатов:
import pytest
pytest_plugins = ("celery.contrib.pytest",)
@pytest.fixture(scope="session")
def celery_config():
return {"broker_url": "memory://", "result_backend": "rpc://"}
И вот мой модульный тест. celery_session_app и celery_session_worker настраивают приложение celery и рабочий процесс, используемый для тестирования (документы):
def test_concurrent_sleep_tasks(celery_session_app, celery_session_worker):
sleep_task.delay(3)
sleep_task.delay(3)
# output
Sleeping for 3 seconds...
1
2
3
Sleeping for 3 seconds...
1
2
3
Похоже, задачи выполняются синхронно. Есть ли способ запускать задачи асинхронно?
