Commit 0c08d44f authored by Philippe Pepiot's avatar Philippe Pepiot
Browse files

Fix running tasks out of scope of the current test

Suppose you have a test that create a task but not run it with run_all_tasks(),
then in a second test run_all_tasks() was running the task of the previous
Fix this by ensuring the task_id (an unique hash generated by celery) is
reachable within the current test database.
parent 31615252fcfa
......@@ -49,7 +49,14 @@ def run_all_tasks(cnx):
# Tasks can create other tasks, so run them until there is no one left.
while _TEST_TASKS:
task_eid = list(_TEST_TASKS)[0]
results[task_eid] = _TEST_TASKS.pop(task_eid).delay()
task = _TEST_TASKS.pop(task_eid)
# Ensure current task id is in the scope of the current test
if is not None and not cnx.execute(
'Any X WHERE X is CeleryTask, X task_id %(task_id)s',
results[task_eid] = task.delay()
if celery.current_app.conf.CELERY_ALWAYS_EAGER:
for task_eid, result in results.items():
......@@ -89,8 +89,11 @@ class CeleryTaskTC(BaseCeleryTaskTC):
# Task must run even if there is no data on current transaction
task = celery.signature("success", kwargs={"n": 10})
task_id = task.freeze().task_id
cnx.transaction_data['celerytask'] = {42: task}
cwtask = cnx.create_entity('CeleryTask', task_name=u'dummy',
cnx.transaction_data['celerytask'] = {cwtask.eid: task}
result = self.wait_async_task(cnx, task_id)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment