我正在使用Tornado
和aioredis
。我想在aioredis
类中测试我的set
实例的一些get
调用(aioredis.create_redis_pool
,tornado.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())