如何模拟sqlalchemy的异步会话？

Question

代码片段

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

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

如何模拟会话生成器，以便以下执行没有任何错误：

async with session_maker() as session:

我尝试执行以下操作，但不起作用

from unittest.mock import AsyncMock

session_maker.return_val = AsyncMock()

Answer 1

在mock-alchemy项目中 tamioEcoligo 发现像这个例子这样的代码运行得还不错：

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