test_services.py 文件源码

def __init__(self, queue=None):
        self.queue = queue or list()
        redis = mock.MagicMock(aioredis.Redis)
        redis.rpush.side_effect = utils.make_coro(result=lambda key, item: self.queue.append(item))
        redis.rpop.side_effect = utils.make_coro(result=lambda key: self.queue.pop())
        redis.lpush.side_effect = utils.make_coro(result=lambda key, item: self.queue.insert(0, item))
        redis.llen.side_effect = utils.make_coro(result=lambda key: len(self.queue))
        super(MockQueue, self).__init__(redis, None)