我正在对芹菜任务进行单元测试。我有链式任务也有组,所以产生了和弦。
测试应如下所示:
- 运行 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。我对任务结果不感兴趣。如何等待所有子任务完成?