烧瓶，芹菜，pytest和apply_async倒计时混乱

Question

我敢肯定这是在某个地方注明的，但是我在旋转我的车轮，所以希望这里有人可以提供帮助。

我正在使用一个使用芹菜的烧瓶应用程序，并且正在使用Pytest进行测试。这是两个测试：

def test_celery_tasks_countdown_apply_async(celery_app, celery_worker):
    r1 = add.apply_async((4, 4), countdown=2)
    r2 = multiply.apply_async((4, 4), countdown=2)
    assert r1.get(timeout=10) == 8
    assert r2.get(timeout=10) == 16

def test_celery_tasks_countdown_signatures(celery_app, celery_worker):
    r1 = add.s(4, 4).apply_async(countdown=2)
    r2 = multiply.s(4, 4).apply_async(countdown=2)
    assert r1.get(timeout=10) == 8
    assert r2.get(timeout=10) == 16

实际任务是这样：

@shared_task
def add(x, y):
    return x + y

@shared_task()
def multiply(x, y):
    return x * y

如果我一个一个人运行它们，则通过它们。

pytest tests/test_tasks.py::test_celery_tasks_countdown_apply_async
pytest tests/test_tasks.py::test_celery_tasks_countdown_signatures

但是，如果我一起运行它们（通过调用整个 test_tasks.pys.py 文件），它们都失败了。

pytest tests/test_tasks.py

我还有其他一些工作（例如，对于 delay ）。而且，如果我从中删除 Countdown 选项，则它们都会通过，如果一起运行。

为什么使用 Countdown 选项并将这些测试一起运行会导致故障？

目前，我在 conftest.py 看起来像这样的固定装置：

pytest_plugins = ('celery.contrib.pytest', )

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'redis://localhost:8001',
        'result_backend': 'redis://localhost:8001',
        'task_always_eager': False,
    }

Update

我将这个问题留下来，因为我相信这是一个需要一些文档的有效问题。虽然我仍然不了解问题，也不知道如何在自动化测试的范围内解决该问题，但我成功地将我的任务在Pytest之外运行。

Answer 1

这不是一个完整的答案，而是更多的解决方法，可以使结果保存到后端并完成测试。但是，实际任务未发送到队列，并且倒计时选项似乎被忽略了，这两者都不是我想要的。

@pytest.fixture()
def celery_config():
    return {
        'broker_url': 'redis://localhost:8001',
        'result_backend': 'redis://localhost:8001',
        'task_always_eager': True,
        'task_ignore_result': False,
        'task_store_eager_result': True,
    }

参考在这里。

我不确定我是否想在生产中使用它们，所以这只是使我的测试通过。我仍然很想知道为什么这一切都会发生。