Mock async SQLAlchemy 2.0

[tamio96]

Describe the bug Hi, I start to apply async sqlalchemy to my project and got stuck as mocking an AsyncSession To Reproduce CRUD function:

async def get_record_by_id(db: AsyncSession, user_id: int) -> Optional[sql_models.User]:
    result = await db.execute(select(sql_models.User).filter(sql_models.User.id == user_id))
    result = result.scalars().first()
    return result

Unit test function:

@pytest.mark.asyncio
async def test_get_investment_record_by_id():
    mock_result = AlchemyMagicMock()
    mock_result.scalars.return_value.first.return_value = user_obj

    mock_db = AlchemyMagicMock()
    mock_db.execute.return_value = mock_result

    user_id = 1234
    await get__record_by_id(mock_db, user_id)
    mock_db.execute.assert_called_with(select(sql_models.User).filter(sql_models.User.id == user_id))

Outcome:

E       TypeError: object AlchemyMagicMock can't be used in 'await' expression

Expected behavior

• AlchemyMagicMock support async

Desktop (please complete the following information):

• OS: Ubuntu
• Version 20.04

[edjeffery]

I would also find this feature useful 👍

[willcipriano]

I also would like this

[fabrice-toussaint]

Any updates on this?

[tamioEcoligo]

hi, I share an alternative way to mock AsyncSession

from unittest.mock import AsyncMock
from unittest.mock import MagicMock

@pytest.mark.asyncio
async def test_get_investment_record_by_id():
    mock_result = MagicMock()
    mock_result.scalars.return_value.first.return_value = user_obj

    mock_db = AsyncMock(return_value=asyncio.Future())
    mock_db.execute.return_value = mock_result

    user_id = 1234
    await get__record_by_id(mock_db, user_id)
    mock_db.execute.assert_called_with(select(sql_models.User).filter(sql_models.User.id == user_id))