test_redis.py 文件源码

def test_queue(loop):
    config = MergeDict(key=str(uuid.uuid4()))
    config['app.redis_pool'] = await aioredis.create_pool(
        ('localhost', 6379), loop=loop)
    context = config
    q = RedisQueue(config, context=context, loop=loop)
    await q.init()
    await q.put(3)
    assert 1 == await q.length()
    assert [b'3'] == await q.list()
    assert b'3' == await q.get()
    await q.put(3)
    assert 1 == await q.length()
    await q.clear()
    assert not await q.length()