python - 如何用 Tornado 测试 aioredis

Question

我正在使用 Tornado 和 aioredis。我想测试我的 aioredis.create_redis_pool 实例的一些 aioredis 调用(set、get 等)在 tornado.testing.AsyncHTTPTestCase 类中。

我试过上网，但我还没找到方法。

有没有办法在我的 Tornado 测试中模拟 aioredis 对临时 Redis 数据库的调用。

提前致谢

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())