test_task.py 文件源码

def setUp(self):
        self.broker_url = 'redis://127.0.0.1:{port}/0'.format(port=REDIS_CONFIG['port'])
        self.tracer = get_dummy_tracer()
        self.pin = Pin(service='celery-test', tracer=self.tracer)
        patch_app(celery.Celery, pin=self.pin)
        patch_task(celery.Task, pin=self.pin)