# tests.py
import celery
import celeryconfig
from my_app.tasks import test_1, test_2, test_3
assert celeryconfig.task_ignore_result is True
def run_test(func, kwargs, *, ready_expected, task_expected):
print(func.__name__, kwargs)
task = func.apply_async(**kwargs)
try:
task_result = task.get(timeout=1)
except celery.exceptions.TimeoutError:
task_result = 'TIMEOUT'
print(' TIMEOUT')
ready_result = task.ready()
if ready_expected == ready_result:
print(' PASS ', end='')
else:
print(' FAIL ', end='')
print('Ready expected %r, got %r' % (ready_expected, ready_result))
if task_expected == task_result:
print(' PASS ', end='')
else:
print(' FAIL ', end='')
print('Result expected %r, got %r' % (task_expected, task_result))
try:
task.forget()
except FileNotFoundError:
pass
print()
if __name__ == '__main__':
print()
run_test(test_1, dict(), ready_expected=False, task_expected=None)
run_test(test_2, dict(), ready_expected=True, task_expected=1 )
# The following test fails.
run_test(test_3, dict(), ready_expected=False, task_expected=None)
run_test(test_1, dict(ignore_result=False), ready_expected=True, task_expected=1 )
run_test(test_2, dict(ignore_result=False), ready_expected=True, task_expected=2 )
# The following test fails.
run_test(test_3, dict(ignore_result=False), ready_expected=True, task_expected=3 )
run_test(test_1, dict(ignore_result=True), ready_expected=False, task_expected=None)
run_test(test_2, dict(ignore_result=True), ready_expected=False, task_expected=None)
run_test(test_3, dict(ignore_result=True), ready_expected=False, task_expected=None)
Checklist
mainbranch of Celery.contribution guide
on reporting bugs.
for similar or identical bug reports.
for existing proposed fixes.
to find out if the bug was already fixed in the main branch.
in this issue (If there are none, check this box anyway).
Mandatory Debugging Information
celery -A proj reportin the issue.(if you are not able to do this, then at least specify the Celery
version affected).
mainbranch of Celery.pip freezein the issue.to reproduce this bug.
Optional Debugging Information
and/or implementation.
result backend.
broker and/or result backend.
ETA/Countdown & rate limits disabled.
and/or upgrading Celery and its dependencies.
Related Issues and Possible Duplicates
Related Issues
Possible Duplicates
Environment & Settings
Celery version: 5.5.0 (immunity)
celery reportOutput:Steps to Reproduce
Required Dependencies
Python Packages
pip freezeOutput:Other Dependencies
Details
N/A
Minimally Reproducible Test Case
Details
Expected Behavior
I expect that when a task is called with
apply_async(ignore_result=False)I expect to be able to.get()the result regardless of whethertask_ignore_result = Trueinceleryconfig.pyorignore_resultis set on the task.Actual Behavior
Task results are not available when
apply_async(ignore_result=False)andtask_ignore_result = Trueinceleryconfig.py