FastAPI中Mock SQLAlchemy异步执行时AsyncMock返回空列表

问题根源 #

1. 异步调用链模拟错误：SQLAlchemy异步API中，execute()返回的AsyncResult对象的scalars()是异步方法（需要await），你之前的同步mock配置无法匹配这个异步调用流程。
2. Patch目标错误：你直接patch了AsyncSession.execute类方法，但实际代码中使用的是AsyncSessionLocal()生成的实例，类方法的patch不会作用到实例上。

正确的模拟方案 #

步骤1：模拟异步调用链 #

每个数据库查询的返回结果需要按以下结构配置：

• execute() 返回一个AsyncMock对象
• 该对象的scalars()方法是AsyncMock，调用后返回的对象的all()方法返回预期数据

步骤2：Patch正确的会话源 #

直接patchAsyncSessionLocal，让它返回预先配置好的mock会话实例，这样代码中的async with AsyncSessionLocal() as db会拿到mock对象。

修正后的测试代码 #

import pytest
from unittest.mock import patch, AsyncMock
from src.services.retrieval_service.retrieval import RetrievalService

@pytest.mark.asyncio
async def test_retrieve_relevant_docs_valid_query():
    service = RetrievalService()
    query = "What is AI?"
    top_k = 3

    # 配置第一个查询的返回结果（selected_ids）
    mock_selected_result = AsyncMock()
    mock_selected_scalars = AsyncMock()
    mock_selected_scalars.all.return_value = [1, 2, 3]
    mock_selected_result.scalars.return_value = mock_selected_scalars

    # 配置第二个查询的返回结果（搜索到的document_ids）
    mock_search_result = AsyncMock()
    mock_search_scalars = AsyncMock()
    mock_search_scalars.all.return_value = [1, 2, 3]
    mock_search_result.scalars.return_value = mock_search_scalars

    # 创建mock会话，execute按顺序返回两个结果
    mock_db = AsyncMock()
    mock_db.execute.side_effect = [mock_selected_result, mock_search_result]
    # 模拟begin()异步上下文管理器
    mock_db.begin.return_value.__aenter__.return_value = None

    with patch.object(service.embedding_generator, 'generate_embedding', new_callable=AsyncMock) as mock_generate_embedding, \
         patch('src.backend.database.config.AsyncSessionLocal', return_value=mock_db):
        mock_generate_embedding.return_value = [0.1] * 384
        
        document_ids = await service.retrieve_relevant_docs(query, top_k)
        assert document_ids == [1, 2, 3], f"Expected [1, 2, 3] but got {document_ids}"

额外说明 #

• 必须模拟db.begin()的异步上下文：因为代码中使用了async with db.begin()，所以需要让mock_db.begin().__aenter__()返回有效值。
• 如果需要测试get_document_texts方法，可采用同样的模式：配置mock会话的execute返回对应结果，确保scalars().all()返回预期的文本列表。

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