如何Mock SQLAlchemy的异步Session？解决异步会话执行报错

原始代码 #

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(url="")
session_maker = async_sessionmaker(engine)

需求 #

需要Mock上述代码中的session_maker，让以下代码能无报错执行：

async with session_maker() as session:

无效的尝试 #

from unittest.mock import AsyncMock

session_maker.return_val = AsyncMock()

问题分析 #

这个方法无效有两个核心原因：

1. 拼写错误：return_val应为return_value
2. 逻辑错误：session_maker是async_sessionmaker的实例，本身不是Mock对象，直接给它设置return_value属性不会生效；另外async with要求对象实现异步上下文管理器方法（__aenter__和__aexit__），需确保session_maker()返回的对象具备这些方法。

正确解决方案 #

方法1：直接替换为Mock对象

from unittest.mock import AsyncMock, Mock

# 创建支持异步上下文的模拟session
mock_session = AsyncMock()
# 将session_maker替换为Mock，调用时返回模拟session
session_maker = Mock(return_value=mock_session)

方法2：测试中使用patch替换（推荐）

如果是单元测试场景，用patch替换目标模块中的session_maker：

from unittest.mock import patch, AsyncMock

@patch('你的模块路径.session_maker')
async def test_your_function(mock_session_maker):
    # 创建模拟session
    mock_session = AsyncMock()
    mock_session_maker.return_value = mock_session
    
    # 执行需要测试的代码，此时async with会使用模拟的session
    # await your_function()

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