如何在Tornado的AsyncHTTPTestCase中测试aioredis操作？

当然有办法啦！我给你两种常用的方案，分别适配不同的测试场景，你可以按需选择：

这种方案会启动一个临时的、不持久化数据的Redis实例，测试完成后自动销毁，能最真实地模拟生产环境的Redis交互。

实现步骤： #

1. 在测试类的setUp方法中启动临时Redis服务器，指定独立端口、禁用持久化；
2. 创建aioredis连接池并注入到Tornado应用中；
3. 测试完成后在tearDown里关闭连接池、终止临时Redis进程。

代码示例： #

import subprocess
import aioredis
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application, RequestHandler
import signal

class RedisTestHandler(RequestHandler):
    async def get(self):
        # 从应用配置中获取redis连接池
        redis_pool = self.application.settings['redis_pool']
        await redis_pool.set('test_key', 'test_value')
        value = await redis_pool.get('test_key')
        self.write(value.decode('utf-8'))

class TestRedisIntegration(AsyncHTTPTestCase):
    def setUp(self):
        super().setUp()
        # 启动临时Redis服务器：指定端口6380，禁用持久化，临时目录存储数据
        self.redis_process = subprocess.Popen(
            ['redis-server', '--port', '6380', '--save', '', '--dir', '/tmp'],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        # 异步创建redis连接池，用run_sync适配同步测试环境
        self.io_loop.run_sync(self._init_redis_pool)

    async def _init_redis_pool(self):
        self.redis_pool = await aioredis.create_redis_pool(('localhost', 6380))
        # 将连接池注入应用配置
        self.app.settings['redis_pool'] = self.redis_pool

    def get_app(self):
        return Application([('/test', RedisTestHandler)], settings={})

    def tearDown(self):
        # 关闭redis连接池
        self.io_loop.run_sync(self.redis_pool.close)
        self.io_loop.run_sync(self.redis_pool.wait_closed)
        # 终止临时Redis进程
        self.redis_process.send_signal(signal.SIGTERM)
        self.redis_process.wait()
        super().tearDown()

    def test_redis_set_get(self):
        response = self.fetch('/test')
        self.assertEqual(response.body.decode('utf-8'), 'test_value')

如果只是想测试业务逻辑，不需要验证Redis本身的功能，可以用Mock来模拟aioredis的调用，这种方式更快且不依赖外部服务。

实现思路： #

• 用unittest.mock创建模拟的Redis连接池对象；
• 预设set、get等方法的返回值；
• 验证这些方法是否被正确调用。

代码示例： #

from unittest.mock import patch, MagicMock
import aioredis
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application, RequestHandler

class RedisTestHandler(RequestHandler):
    async def get(self):
        redis_pool = self.application.settings['redis_pool']
        await redis_pool.set('test_key', 'test_value')
        value = await redis_pool.get('test_key')
        self.write(value.decode('utf-8'))

class TestRedisMock(AsyncHTTPTestCase):
    @patch('aioredis.create_redis_pool')
    def test_redis_mock_interaction(self, mock_create_pool):
        # 创建模拟的Redis连接池
        mock_redis = MagicMock()
        # 预设set方法的异步返回值
        mock_redis.set.return_value = None
        # 预设get方法返回指定字节数据
        mock_redis.get.return_value = b'test_value'
        # 让create_redis_pool返回这个模拟对象
        mock_create_pool.return_value = mock_redis

        # 构建应用并注入模拟连接池
        app = Application([('/test', RedisTestHandler)], settings={'redis_pool': mock_redis})
        self.app = app

        # 发起请求并验证结果
        response = self.fetch('/test')
        self.assertEqual(response.body.decode('utf-8'), 'test_value')

        # 验证Redis方法是否被正确调用
        mock_redis.set.assert_called_once_with('test_key', 'test_value')
        mock_redis.get.assert_called_once_with('test_key')

两种方案对比： #

• 真实Redis实例：测试更贴近生产，但需要环境中有Redis服务，测试速度稍慢；
• Mock方式：速度快、无外部依赖，但只能验证业务逻辑对Redis方法的调用，无法验证Redis本身的交互正确性。

内容的提问来源于stack exchange，提问作者alauri