Method mock_group

t/unit/app/test_builtins.py:108–114

(self, *tasks)

        self.task.apply.assert_called()

    def mock_group(self, *tasks):
        g = group(*tasks, app=self.app)
        result = g.freeze()
        for task in g.tasks:
            task.clone = Mock(name='clone')
            task.clone.attach_mock(Mock(), 'apply_async')
        return g, result

    @patch('celery.app.base.Celery.current_worker_task')
    def test_task(self, current_worker_task):