如何在Python中正确Mock SQLAlchemy的async_engine与async_sessionmaker？

你的问题核心是patch路径错误，加上没有正确模拟异步上下文管理器和链式调用的返回值，导致真实数据库连接依然被使用。以下是修正后的实现：

1. 关键问题分析 #

• 你patch的是db.async_ses，但如果write_to_db函数所在的模块是直接导入并使用async_ses，必须patch被测试函数实际引用的async_ses路径，而不是原始定义的路径。
• 异步Session的async with上下文需要模拟__aenter__方法，同时db.execute是异步调用，需要用AsyncMock而非普通MagicMock来处理await逻辑。
• 链式调用execute().scalars().first()需要逐层mock返回值，否则会触发真实数据库操作。

2. 修正后的测试代码 #

假设你的write_to_db函数在app/operations.py，且从app/db.py导入async_ses，测试代码如下：

import pytest
from unittest.mock import patch, AsyncMock
from app.operations import write_to_db
from sqlalchemy import select
from app.models import Item

@pytest.fixture
def mock_async_session():
    # 模拟session对象：处理execute、scalars、first的链式调用
    mock_session = AsyncMock()
    # 模拟execute的返回结果
    mock_result = AsyncMock()
    mock_scalars = AsyncMock()
    mock_scalars.first.return_value = Item(id=100, name="test_item")  # 自定义返回的测试对象
    mock_result.scalars.return_value = mock_scalars
    mock_session.execute.return_value = mock_result

    # 模拟async_ses()返回的上下文管理器
    mock_session_factory = AsyncMock()
    mock_session_factory.__aenter__.return_value = mock_session

    # 注意：这里patch的是被测试函数所在模块中的async_ses
    with patch('app.operations.async_ses', return_value=mock_session_factory):
        yield mock_session

@pytest.mark.asyncio
async def test_write_to_db(mock_async_session):
    # 执行被测试函数
    await write_to_db()

    # 验证execute是否被正确调用
    mock_async_session.execute.assert_called_once_with(
        select(Item).where(Item.id == 100)
    )
    # 验证scalars和first是否被调用
    mock_async_session.execute.return_value.scalars.assert_called_once()
    mock_async_session.execute.return_value.scalars.return_value.first.assert_called_once()

3. 核心修正点 #

• Patch路径正确：确保patch的是write_to_db所在模块中导入的async_ses，比如app.operations.async_ses，而不是定义它的db.async_ses（Python的patch是作用于引用点，而非定义点）。
• 使用AsyncMock：异步相关的方法（__aenter__、execute）必须用AsyncMock，它能正确处理await语法。
• 逐层mock链式调用：从execute到scalars再到first，每个步骤都要mock返回值，避免触发真实数据库操作。
• 模拟上下文管理器：async_ses()返回的对象需要实现__aenter__，返回我们的mock session。

4. 额外注意事项 #

• 如果你的write_to_db函数中有写入操作（比如db.add()、db.commit()），同样需要mock这些方法，比如mock_session.commit = AsyncMock()。
• 避免直接在测试中调用async_ses()，而是通过执行write_to_db()来触发，确保mock生效。

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