(self, current_worker_task)
| 115 | |
| 116 | @patch('celery.app.base.Celery.current_worker_task') |
| 117 | def test_task(self, current_worker_task): |
| 118 | g, result = self.mock_group(self.add.s(2), self.add.s(4)) |
| 119 | self.task(g.tasks, result, result.id, (2,)).results |
| 120 | g.tasks[0].clone().apply_async.assert_called_with( |
| 121 | group_id=result.id, producer=self.app.producer_or_acquire(), |
| 122 | add_to_parent=False, |
| 123 | ) |
| 124 | current_worker_task.add_trail.assert_called_with(result) |
| 125 | |
| 126 | @patch('celery.app.base.Celery.current_worker_task') |
| 127 | def test_task__disable_add_to_parent(self, current_worker_task): |
nothing calls this directly
no test coverage detected