Commit 061da68e authored by Philippe Pepiot's avatar Philippe Pepiot
Browse files

Rework test_multi_revoke()

We should avoid calling freeze() within a task and avoid calling freeze() twice
(once in test and once in ICeleryTask.attach_tasks()). This produce all kinds
of weird issues along celery versions, I think the API and behavior of freeze()
is still somewhat experimental...
Only assert on length of child tasks and test they are all revoked which is the
purpose of this test.
parent 98bb9bad617b
......@@ -248,27 +248,18 @@ class CeleryTaskTC(BaseCeleryTaskTC):
def test_multi_revoke(self):
with self.admin_access.cnx() as cnx:
s = celery.signature
task_ids = []
def success():
task = s("success")
return task
def group(*tasks):
g =*tasks)
return g
graph = group(
group(success(), success()),
group(success(), success()),
return celery.signature("success")
graph =, success()),, success()),
task = start_async_task(cnx, graph)
self.assertEqual(set(task_ids), set([
t.task_id for t in task.child_tasks()]))
task_ids = set([t.task_id for t in task.child_tasks()])
self.assertEqual(len(task_ids), 8)
with mock.patch('') as revoke:
self.assertEqual(revoke.call_count, 1)
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