python - 为单个单元测试用例更改 celery 设置 task_always_eager

Question

我有一个使用 celery 后端的特定测试用例，因此我需要将 task_always_eager 设置为 false，否则我将收到 RuntimeError。但是问题是我需要所有其他测试用例都以设置为 true 运行，所以我尝试使用 celery mark 注释将 celery 配置 task_always_eager 设置为 false 仅针对此测试用例。但是，它看起来并没有做任何事情。

这是我的任务的框架:

@pytest.mark.celery(task_always_eager=False)
def test_task(self):
   # do some stuff to start a task
   do_stuff()
   # do some stuff to get info on a task
   get_info()
   # assertions

错误:

    def _ensure_not_eager(self):
        if self.app.conf.task_always_eager:
            raise RuntimeError(
               "Cannot retrieve result with task_always_eager enabled")
E          RuntimeError: Cannot retrieve result with task_always_eager enabled

/usr/local/lib/python3.6/site-packages/celery/backends/base.py:334: RuntimeError

TLDR:我在这个测试用例中将 task_always_eager 设置为 false 做错了什么，因为它在测试运行时没有更改它？

Answer 1

Celery 的配置对象是动态更新的，所以你可以更新它:

celery.conf.task_always_eager = False

或 celery.conf.CELERY_ALWAYS_EAGER = False ，如果您使用的是 4.0 之前的 Celery

您可以在运行中、在每次测试的基础上或在“设置”操作期间执行此操作。