I am using Python 3.13.2
I am trying a sample use case of using AsyncMock to simulate fetching of mock rows from DB to test my endpoint (using pytest-asyncio).
I have also included a non-mock endpoint in the code which is working for me.
However, for the mock endpoint, I am getting empty list.
Sample code,
# test_main.py
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
import aiosqlite
app = FastAPI()
client = TestClient(app)
@app.get("/users")
async def get_users() -> list[dict[str, int | str]]:
try:
conn = await aiosqlite.connect('test.db')
conn.row_factory = aiosqlite.Row
cursor = await conn.execute('SELECT id, name FROM users')
rows: list[dict[str, str | int]] = []
async for row in cursor:
rows.append(dict(row))
return rows # Getting [{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]
finally:
await cursor.close()
await conn.close()
# this test is passing for me
def test_users_no_mock():
expected_data = [{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]
response = client.get('/users')
assert response.status_code == 200
data = response.json()
assert data == expected_data
# this test is failing for me
@pytest.mark.asyncio
async def test_get_users_mock():
expected_data = [{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]
mock_cursor = AsyncMock()
async def async_row_generator():
for row in expected_data:
yield row
mock_cursor.__aiter__.return_value = async_row_generator()
mock_conn = AsyncMock()
mock_conn.execute.return_value = mock_cursor()
with patch('aiosqlite.connect', return_value=mock_conn()):
response = client.get('/users')
data = response.json()
assert response.status_code == 200
assert data == expected_data
When I run using pytest, here is what I am getting,
============================================================================================ test session starts =============================================================================================
platform darwin -- Python 3.13.2, pytest-9.0.1, pluggy-1.6.0
rootdir: /Users/amit_tendulkar/quest/experiment
plugins: mock-3.15.1, langsmith-0.4.11, anyio-4.10.0, asyncio-1.3.0
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 2 items
test_main.py .F [100%]
================================================================================================== FAILURES ==================================================================================================
____________________________________________________________________________________________ test_get_users_mock _____________________________________________________________________________________________
@pytest.mark.asyncio
async def test_get_users_mock():
expected_data = [{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]
mock_cursor = AsyncMock()
async def async_row_generator():
for row in expected_data:
yield row
mock_cursor.__aiter__.return_value = async_row_generator()
mock_conn = AsyncMock()
mock_conn.execute.return_value = mock_cursor()
with patch('aiosqlite.connect', return_value=mock_conn()):
response = client.get('/users')
data = response.json()
assert response.status_code == 200
> assert data == expected_data
E AssertionError: assert [] == [{'id': 1, 'n...name': 'Bob'}]
E
E Right contains 2 more items, first extra item: {'id': 1, 'name': 'Alice'}
E Use -v to get more diff
test_main.py:46: AssertionError
========================================================================================== short test summary info ===========================================================================================
FAILED test_main.py::test_get_users_mock - AssertionError: assert [] == [{'id': 1, 'n...name': 'Bob'}]
======================================================================================== 1 failed, 1 passed in 0.15s =========================================================================================
<sys>:0: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited
The below code worked for me,
@pytest.mark.asyncio
async def test_get_users_mock():
expected: list[dict[str, int | str]] = [{"userid":1,"fullname":"Alice"},{"userid":2,"fullname":"Bob"}]
fake_rows = [MagicMock(**row, keys=lambda: ['userid', 'fullname']) for row in expected]
fake_rows[0].__getitem__.side_effect = lambda arg: expected[0][arg]
fake_rows[1].__getitem__.side_effect = lambda arg: expected[1][arg]
cursor = AsyncMock()
cursor.__aiter__.return_value = iter(fake_rows)
conn = AsyncMock(execute=AsyncMock(return_value=cursor))
async def fake_connect(*args, **kwargs):
return conn
with patch('aiosqlite.connect', side_effect=fake_connect):
response = client.get('/users')
data = response.json()
assert response.status_code == 200
assert data == expected
While patching aiosqlite.connect I needed to provide an async def function returning AsyncMock as it was awaited.
I had to add the side_effect for __getitem__ as the dict call in the system under test was internally calling the same.
I can't add the lambda expression in the loop as the first argument to the expected will take the final value of the variable (applying the rules of closure). For e.g., this won't work,
for i, row in enumerate(expected):
fake_rows[i].__getitem__.side_effect = lambda arg: expected[i][arg]
in place of,
fake_rows[0].__getitem__.side_effect = lambda arg: expected[0][arg]
fake_rows[1].__getitem__.side_effect = lambda arg: expected[1][arg]