How to test aioredis with Tornado

user_4173278 0
Viewed 915

I'm using Tornado and aioredis. I'd like to test some aioredis calls (set, get, etc) of my aioredis.create_redis_pool instance within tornado.testing.AsyncHTTPTestCase class.

I've tried to go through the Internet, but I haven't found how to do that.

Is there a way to simulate aioredis calls to a temporary Redis database in my Tornado tests.

Thank in advance

tornado redis testing python
2 Answers

I had the same problem, with the added twist of creating my redis connection pool before the Application instance, so that it can be shared between requests. I successfully used testig.redis, which creates a redis instance in a temporary directory. The library is old and not much has happened in it for years, but it seems to work. Anyway, the test looks like this:

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

For completion, the usual (non test) app init looks like this:

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

I've used the library mockaioredis which worked very well in my case.

To use it, first of all I've "mock" the aioredis with sys.modules['aioredis'] = mockaioredis in <project>/tests/__ini__.py file.

Secondly I've extended the tornado.web.Application and set up a method which initialise a Redis pool:

class Application(tornado.web.Application):

    async def _conf_redis(self) -> None:
        loop = asyncio.get_event_loop()
        self.redis = await aioredis.create_redis_pool(address='address',
                                                      password='password',
                                                      loop=loop)

Then I've set up the test class like below:

import my.web


class TestHandler(AsyncHTTPTestCase):

    @tornado.testing.gen_test
    async def get_app(self):

        app = my.web.Application([...])

        await app._conf_redis()
        self.redis = app.redis

        return app

Now it is possible to use Redis normally in any test by using, for example, self.redis._redis.set(...) or self.redis._redis.delete(...) or self.redis._redis.hsetnx(...) or self.redis._redis.flushdb().

user_41732780
Related