How to mock sqlalchemy's async session?

Question

Code snippet

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(url="")
session_maker = async_sessionmaker(engine)

How can I mock the session maker so that the following executes without any error:

async with session_maker() as session:

I tried doing the following and it doesn't work

from unittest.mock import AsyncMock

session_maker.return_val = AsyncMock()

Answer 1

In the mock-alchemy project tamioEcoligo found that code like this example worked tolerably well:

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))

---

Maybe you don't really need a mock? For fast reproducible runs, it can be helpful to point sqlalchemy at a local throw-away sqlite DB file. I have seen this work well for many tests even when the production target is MariaDB or postgres.

Answer 2

AsyncSession.scalars returns an instance of ScalarResult. Mock async session with AsyncMock and mock return value of AsyncSession.scalars with a mocked instance of ScalarResult by autospeccing it.

from unittest import mock

async with session_maker() as session:
    session_mock = mock.AsyncMock(spec=session,
                                  **{'scalars.return_value': mock.create_autospec(
                                      spec=ScalarResult, instance=True)})
    await session_mock.scalars(...)

Mocking the return value of AsyncSession.scalars is not enough. What you are interested in mocking the final result to get the expected value. So, you have to mock the methods of ScalarResult as well.

Mock ScalarResult.all

expected_value = ['result_one', 'result_two', 'result_three']
session_mock = mock.AsyncMock(
    spec=session, **{'scalars.return_value': mock.create_autospec(spec=ScalarResult,
                                                                  instance=True),
                     'scalars.return_value.all.return_value': expected_value})

result = await session_mock.scalars(...)
assert result.all == expected_value

Mock ScalarResult.one_or_none

session_mock = mock.AsyncMock(
    spec=session, **{'scalars.return_value': mock.create_autospec(spec=ScalarResult,
                                                                  instance=True),
                     'scalars.return_value.one_or_none.return_value': None})

result = await session_mock.scalars(...)
assert result.one_or_none is None