[Python] AsyncMock not working when mocked using SQLAlchemy in FastAPI

I'm testing a FastAPI service that queries a database using SQLAlchemy's async execution. However, when I mock the database query, execute().scalars().all() returns an empty list instead of the expected [1, 2, 3].

async def retrieve_relevant_docs(session: AsyncSession, query: str, top_k: int):
results = await session.execute(select(Document.id).limit(top_k))
print("debug results", vars(results)) # Debugging
document_ids = list(await results.scalars())
print("document_ids", document_ids) # Debugging
return document_ids


import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_retrieve_relevant_docs():
mock_session = AsyncMock()
mock_execute = AsyncMock()

# Attempt to mock scalars() to return values
mock_scalars_selected = AsyncMock()
mock_scalars_selected.scalars.return_value.all.return_value = [1, 2, 3]

mock_execute.side_effect = [mock_scalars_selected, mock_scalars_selected]
mock_session.execute = mock_execute # Mocking session.execute

# Calling the function
document_ids = await retrieve_relevant_docs(mock_session, "test query", top_k=3)

# Debugging
print("Final document_ids:", document_ids)

assert document_ids == [1, 2, 3] # This fails because document_ids is []



Observed Behavior: results.scalars().all() unexpectedly returns [], even though I attempted to mock it. Debugging vars(results) shows _mock_side_effect = None, suggesting the mock isn't working as expected. Expected Behavior: document_ids should contain [1, 2, 3], matching the mocked return value

What I've Tried:

• Explicitly setting scalars().all().return_value = [1, 2, 3].
• Checking vars(results) for missing attributes.
• Ensuring mock_execute.side_effect is properly assigned.
• Calling await session.execute(...).scalars().all() instead of wrapping it in list().

Extra Debug logs:

selected_ids_result debug results {'_mock_return_value': sentinel.DEFAULT, '_mock_parent': None, '_mock_name': None, '_mock_new_name': '()', '_mock_new_parent': , '_mock_sealed': False, '_spec_class': None, '_spec_set': None, '_spec_signature': None, '_mock_methods': None, '_spec_asyncs': [], '_mock_children': {'scalars': , 'str': }, '_mock_wraps': None, '_mock_delegate': None, '_mock_called': False, '_mock_call_args': None, '_mock_call_count': 0, '_mock_call_args_list': [], '_mock_mock_calls': [call.str(), call.scalars(), call.scalars().all()], 'method_calls': [call.scalars()], '_mock_unsafe': False, '_mock_side_effect': None, '_is_coroutine': <object object at 0x0000029C06E16A30>, '_mock_await_count': 0, '_mock_await_args': None, '_mock_await_args_list': [], 'code': , 'str': } document_ids [] AssertionError: assert [] == [1, 2, 3]
​

What is the correct way to mock SQLAlchemy's async execution (session.execute().scalars().all()) in a FastAPI test using AsyncMock?

Continue reading...