I have a specific test case in which I utilize the celery backend so I need the task_always_eager setting to be false or I will get a RuntimeError. The problem is however I need all other test cases ran with the setting true, so I am trying to set the celery config task_always_eager to false just for this test case using a celery mark annotation. However, It doesn't look like it is doing anything.
Here is a skeleton of my task:
@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
Error:
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: What am I doing wrong in this test case to set task_always_eager to false because it is not changing it when the test runs?
Celery task canvas Demonstration of a task which runs a startup task, then parallelizes multiple worker tasks, and then fires-off a reducer task.
Eager mode. The eager mode enabled by the task_always_eager setting is by definition not suitable for unit tests. When testing with eager mode you are only testing an emulation of what happens in a worker, and there are many discrepancies between the emulation and what happens in reality.
The "shared_task" decorator allows creation of Celery tasks for reusable apps as it doesn't need the instance of the Celery app. It is also easier way to define a task as you don't need to import the Celery app instance.
Celery tasks run asynchronously, which means that the Celery function call in the calling process returns immediately after the message request to perform the task is sent to the broker. There are two ways to get results back from your tasks.
Celery's config object is dynamically updatable, so you can update it with:
celery.conf.task_always_eager = False
or celery.conf.CELERY_ALWAYS_EAGER = False
, if you're using pre-4.0
Celery
You can do this on the fly, on a per-test basis, or during the "setUp" operation.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With