python - 使用 pytest 同时运行 celery 任务

Question

我正在尝试在我的 Django 应用程序中集成测试并发 celery 任务。在 pytest 集成测试期间，我希望任务实际上在工作人员上同时运行，但我无法完成这项工作。

假设我有以下基本的 celery 任务:

@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 应用程序和工作程序 (docs):

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

看起来任务正在同步运行。无论如何异步运行任务？

Answer 1

celery_session_worker fixture 正在启动 celery worker并发性为 1。这是在 celery.contrib 中硬编码的，目前不可配置。由于并发度为 1，因此一次只会处理 1 个任务。

另一种解决方案是编写您自己的 fixture 以启动并发为 2 或更多的 worker。