使用pytest并发运行celery任务

Question

我正尝试在我的Django应用程序中集成测试一个并发的芹菜任务。我想让任务在pytest集成测试期间在一个worker上并发运行，但我在执行这项工作时遇到了问题。

假设我有以下基本的芹菜任务：

@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应用程序和worker (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\_workerfixture正在启动一个celery worker并发为1。这是在celery.contrib中硬编码的，当前不可配置。因为并发数是1，所以它一次只能处理一个任务。

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