__init__.py 文件源码

def _config_celery(self):
        """Disables Celery by entirely mocking it.

        Without this, actual Celery tasks will be created while the tests are running.
        """

        from celery import Celery, Task

        self.celery = unittest.mock.MagicMock(Celery)

        def fake_task(*task_args, bind=False, **task_kwargs):
            def decorator(f):
                def delay(*args, **kwargs):
                    if bind:
                        return f(decorator.sender, *args, **kwargs)
                    else:
                        return f(*args, **kwargs)

                f.delay = delay
                f.si = unittest.mock.MagicMock()
                f.s = unittest.mock.MagicMock()
                return f

            if bind:
                decorator.sender = unittest.mock.MagicMock(Task)

            return decorator

        self.celery.task = fake_task