小编dyn*_*set的帖子

我正在对 celery 任务进行单元测试。我的连锁任务也有组，所以产生了一个和弦。

测试应该如下所示：

• 运行 celery 任务（延迟）
• 等待任务和所有子任务
• 断言

我尝试了以下方法：

def wait_for_result(result):
    result.get()
    for child in result.children or list():
        if isinstance(child, GroupResult):
           # tried looping over task result in group
           # until tasks are ready, but without success 
           pass
        wait_for_result(child)

这会造成死锁，chord_unlock 会永远重试。我对任务结果不感兴趣。如何等待所有子任务完成？