def test_create_chord_exclude_body(self):
"""If the body task of a chord is not a UserTask, it should be cleanly omitted from the status."""
chord([
sample_task.s(self.user.id, '1', user_task_name='Chord: 1 & 2'),
sample_task.s(self.user.id, '2', user_task_name='I should be ignored')
])(normal_task.s('3'))
assert UserTaskStatus.objects.count() == 4
chord_status = UserTaskStatus.objects.get(task_class='celery.chord')
assert chord_status.task_id
assert chord_status.parent is None
assert chord_status.is_container
assert chord_status.name == 'Chord: 1 & 2'
assert chord_status.total_steps == 2
verify_state(chord_status, False)
group_status = UserTaskStatus.objects.get(task_class='celery.group')
assert group_status.task_id
assert group_status.parent_id == chord_status.id
assert group_status.is_container
assert group_status.name == 'Chord: 1 & 2'
assert group_status.total_steps == 2
verify_state(group_status, False)
header_tasks = UserTaskStatus.objects.filter(parent=group_status)
assert len(header_tasks) == 2
for status in header_tasks:
assert status.task_id
assert status.parent_id == group_status.id
assert not status.is_container
assert status.name in ['SampleTask: 1', 'SampleTask: 2']
assert status.total_steps == 1
verify_state(status, False)
评论列表
文章目录