我正在使用Tornadoaioredis。我想在aioredis类中测试我的set实例的一些get调用(aioredis.create_redis_pooltornado.testing.AsyncHTTPTestCase等)。

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

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

预先感谢

最佳答案

我遇到了同样的问题,增加了在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())

07-28 09:25