如何使用pytest测试集成SQLAlchemy ORM与Redis的Aiogram异步机器人？

作为经常折腾Aiogram测试的开发者，我来分享一套实用的解决方案，覆盖你提到的消息模拟、依赖注入（DB/Redis/FSM）、AsyncMock的正确用法，以及代码替换的技巧。

一、先搞定消息与上下文的模拟 #

Aiogram的处理器大多依赖Message、Update或FSMContext这类对象，用AsyncMock完全能模拟出符合要求的实例，关键要把核心属性和异步方法Mock到位：

import asyncio
from unittest.mock import AsyncMock
from aiogram.types import Message

async def test_start_handler():
    # 1. 模拟Message对象，设置你需要的属性
    mock_message = AsyncMock(spec=Message)
    mock_message.text = "/start"
    mock_message.chat.id = 12345
    # 模拟异步回复方法
    mock_message.answer = AsyncMock()

    # 2. 直接调用你的处理器（假设处理器接收Message参数）
    await your_start_handler(mock_message)

    # 3. 断言行为是否符合预期
    mock_message.answer.assert_awaited_once_with("欢迎使用机器人！")

如果你的处理器依赖Update对象，同样可以嵌套模拟：

mock_update = AsyncMock()
mock_update.message = mock_message
# 然后传给需要Update参数的处理器
await your_update_based_handler(mock_update)

二、模拟SQLAlchemy与Redis依赖 #

对于DB和Redis这类外部依赖，核心思路是用AsyncMock替换真实的会话/客户端，避免测试时连接真实服务：

1. 模拟SQLAlchemy异步Session #

假设你的处理器通过依赖注入获取AsyncSession：

# 处理器代码示例
from sqlalchemy.ext.asyncio import AsyncSession
from your_models import User

async def get_user_handler(message: Message, db_session: AsyncSession):
    user = await db_session.get(User, message.chat.id)
    await message.answer(f"你好，{user.name}！")

测试时Mock整个Session：

async def test_get_user_handler():
    mock_message = AsyncMock(chat=AsyncMock(id=12345))
    mock_message.answer = AsyncMock()

    # 模拟DB Session和查询结果
    mock_db = AsyncMock()
    mock_user = User(id=12345, name="测试用户")
    # 关键：让异步方法返回可await的结果
    mock_db.get.return_value = asyncio.coroutine(lambda: mock_user)()

    # 调用处理器，传入Mock的Session
    await get_user_handler(mock_message, db_session=mock_db)

    # 断言DB操作和回复都符合预期
    mock_db.get.assert_awaited_once_with(User, 12345)
    mock_message.answer.assert_awaited_once_with("你好，测试用户！")

2. 模拟Redis客户端 #

和DB的思路一致，Mock Redis的异步方法：

async def test_redis_based_handler():
    mock_message = AsyncMock(chat=AsyncMock(id=12345))
    mock_message.answer = AsyncMock()

    # 模拟Redis客户端
    mock_redis = AsyncMock()
    mock_redis.get.return_value = asyncio.coroutine(lambda: b"已签到")()

    await your_redis_handler(mock_message, redis=mock_redis)

    mock_redis.get.assert_awaited_once_with(b"user:12345:sign")
    mock_message.answer.assert_awaited_once_with("已签到")

三、处理FSM状态的测试 #

FSM的测试有两种思路：纯Mock状态，或者用内存Storage做轻量真实测试：

1. 纯Mock FSMContext #

如果只需要验证状态的读取/设置行为，用AsyncMock模拟FSMContext即可：

from aiogram.fsm.context import FSMContext

async def test_fsm_step_handler():
    mock_message = AsyncMock()
    mock_fsm = AsyncMock(spec=FSMContext)
    # 模拟当前状态
    mock_fsm.get_state.return_value = asyncio.coroutine(lambda: "SignUp:name")()

    await your_fsm_handler(mock_message, state=mock_fsm)

    # 断言状态被更新
    mock_fsm.set_state.assert_awaited_once_with(SignUp.phone)

2. 用内存Storage做真实状态测试 #

如果需要验证状态的流转逻辑，用Aiogram自带的MemoryStorage更靠谱：

from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.context import FSMContext

async def test_real_fsm_flow():
    # 初始化临时内存Storage
    storage = MemoryStorage()
    # 创建真实的FSMContext实例
    fsm_context = FSMContext(storage=storage, chat_id=12345, user_id=12345)
    
    # 设置初始状态
    await fsm_context.set_state(SignUp.name)
    mock_message = AsyncMock(chat=AsyncMock(id=12345), text="张三")

    await your_signup_name_handler(mock_message, state=fsm_context)

    # 断言状态已流转
    current_state = await fsm_context.get_state()
    assert current_state == SignUp.phone

四、关于AsyncMock的正确性 #

你完全可以放心用AsyncMock，只要注意异步方法的返回值必须是可await的对象。除了上面用asyncio.coroutine包装的方式，还可以这样写：

# 另一种设置异步返回值的方式
mock_db.get.return_value = AsyncMock()
mock_db.get.return_value.__await__.return_value = mock_user

两种方式效果一致，只要确保处理器await Mock方法时能拿到预期结果，测试逻辑就是可靠的。

五、替换函数内部代码的技巧 #

如果处理器内部调用了外部模块的函数/方法，不想测试这些外部逻辑，可以用unittest.mock.patch替换：

1. 替换模块级函数 #

假设处理器调用了external_api.get_user_data()：

# 处理器代码
from your_utils import external_api

async def fetch_data_handler(message: Message):
    data = await external_api.get_user_data(message.chat.id)
    await message.answer(data)

测试时替换这个函数：

from unittest.mock import patch

async def test_fetch_data_handler():
    mock_message = AsyncMock(chat=AsyncMock(id=12345))
    mock_message.answer = AsyncMock()

    # 注意：patch的路径是处理器中导入该函数的路径
    with patch("your_handlers.external_api.get_user_data") as mock_api:
        mock_api.return_value = asyncio.coroutine(lambda: "模拟用户数据")()
        
        await fetch_data_handler(mock_message)
        
        mock_api.assert_awaited_once_with(12345)
        mock_message.answer.assert_awaited_once_with("模拟用户数据")

2. 替换类的方法 #

如果是调用类的异步方法，用patch.object更方便：

from your_services import UserService

async def test_service_handler():
    mock_message = AsyncMock()
    
    with patch.object(UserService, "get_user") as mock_method:
        mock_method.return_value = asyncio.coroutine(lambda: mock_user)()
        
        await your_service_based_handler(mock_message)
        
        mock_method.assert_awaited_once()

最后几个小建议 #

• 用pytest-asyncio插件来运行异步测试，直接在函数前加async即可。
• 对于核心ORM逻辑，可以偶尔用SQLite内存库做真实测试，确保你的查询语句没问题。
• 测试时尽量只关注当前处理器的逻辑，外部依赖全部Mock，让测试更快、更独立。

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