如何使用龙卷风测试aioredis

Question

我正在使用Tornado和aioredis。我想测试aioredis类中set实例的某些get调用（aioredis.create_redis_pool，tornado.testing.AsyncHTTPTestCase等）。

我曾尝试通过Internet进行访问，但尚未找到如何执行此操作的方法。

在我的aioredis测试中，有没有一种方法可以模拟对临时Redis数据库的Tornado调用。

提前感谢

Answer 1

我遇到了同样的问题，增加了在Application实例之前创建我的redis连接池的麻烦，以便可以在请求之间共享它。我成功使用了testig.redis，它在一个临时目录中创建了一个Redis实例。该图书馆很旧，多年来没有发生太多变化，但是似乎可以正常工作。无论如何，测试看起来像这样：

import functools

import aioredis
import testing.redis
import redis
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application

from myapp.base import MyApplication


class TestHandler(AsyncHTTPTestCase):

    def setUp(self) -> None:
        self.redis_server = testing.redis.RedisServer()
        self.redis_client = redis.Redis(**self.redis_server.dsn())
        super().setUp()

    def tearDown(self) -> None:
        self.redis_server.stop()

    def get_app(self) -> Application:
        redis_dsn = self.redis_server.dsn()
        redis = self.io_loop.run_sync(functools.partial(
            aioredis.create_redis_pool, f'redis://{redis_dsn["host"]}:{redis_dsn["port"]}/{redis_dsn["db"]}'
        ))
        return MyApplication(redis)

    def test_client_handler_should_return_200(self):
        self.redis_client.set('val', 'a')
        response = self.fetch('/get-some-redis-data/')
        self.assertEqual(response.code, 200)
        self.assertEqual(response.body, 'a')

为了完成，通常的（非测试）应用程序初始化看起来像这样：

class MyApplication(Application):
    def __init__(self, redis_connection, *args, **kwargs):
        self.redis_connection = redis_connection
        super().__init__(url_patterns, *args, **kwargs)

async def main():
    redis_connection = await aioredis.create_redis_pool(
        f'redis://{options.redis_host}:{options.redis_port}/{options.redis_db}'
    )
    app = MyApplication(redis_connection)
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port, address=options.listen_ips)
    event = tornado.locks.Event()
    await event.wait()


if __name__ == "__main__":
    asyncio.run(main())