I have a problem testing the code where I connect to the database via SQLAlchemy using my own asynchronous context manager.
# my_module.py
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio.engine import AsyncConnection
@asynccontextmanager
async def adbcontext(url: str) -> AsyncGenerator[AsyncConnection, None]:
"""Define an asynchronous context manager for the SQLAlchemy connection."""
engine = create_async_engine(url)
conn = await engine.connect()
try:
async with conn.begin():
yield conn
finally:
await conn.close()
await engine.dispose()
async def query(url: str, sql: str) -> None:
async with adbcontext(url) as conn:
await conn.execute(sql)
# test_async.py
from unittest.mock import MagicMock
from asynctest import patch
import pytest
from my_module import query
@patch("sqlalchemy.ext.asyncio.create_async_engine")
@pytest.mark.asyncio
async def test_async_query(mock_engine: MagicMock) -> None:
"""Test asychronically execution of queries."""
async def async_func(): pass
mock_engine.return_value.__aenter__.connect = MagicMock(async_func)
await query()
I get the error: TypeError: object MagicMock can't be used in 'await' expression
Does anyone know how to deal with this?