Pytest + FastApi + SQLAlchemy + asyncpg: Ошибка asyncio

Question

Пытаюсь протестировать приложение FastApi с использованием SQLAlchemy. Первый тест проходит хорошо, а на втором выскакивает ошибка asyncio:

E               AttributeError: 'NoneType' object has no attribute 'send'

C:\Python310\lib\asyncio\proactor_events.py:397: AttributeError

database.py

import sys

from sqlalchemy.orm import declarative_base
from config import Config, TestConfig
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

Base = declarative_base()
config_class = TestConfig if 'pytest' in sys.modules else Config


engine = create_async_engine(config_class.DB_CONFIG, echo=True, future=True)
LocalAsyncSession = async_sessionmaker(engine, expire_on_commit=False)


async def init_models():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)


async def get_session() -> AsyncSession:
    if config_class is TestConfig and not config_class.DB_CREATED:
        await init_models()
        TestConfig.DB_CREATED = True
    async with LocalAsyncSession() as session:
        yield session

account_test.py

import pytest
from httpx import AsyncClient
from main import app
from pytest import fixture


@pytest.fixture
def anyio_backend():
    return 'asyncio'


@fixture(scope='session')
def client():
    client = AsyncClient(app=app, base_url='http://127.0.0.1:8000')
    return client


@pytest.mark.anyio
async def test_create_user(client):
    response = await client.post(
        '/users/create/',
        json={'email': '[email protected]', 'password': '123456'}
    )
    assert response.status_code == 201, response.text

    data = response.json()
    assert 'id' in data
    assert 'email' in data


@pytest.mark.anyio
async def test_get_user(client):
    auth = await client.post('/token/', data={'username': '[email protected]', 'password': '123456'})
    assert auth.json() == {'detail': 'Incorrect email or password'}
    assert auth.status_code == 401

Как я понимаю, на первом тесте все идет хорошо, а потом event loop FastApi приложения почему то ломается, и при обращении к методам engine происходит ошибка. Если на каждый тест создавать свой отдельный engine то тесты проходят успешно, но такой подход сликшом неоптимален. При ручном тестировании через Postman все сценарии обрабатываются успешно.

БЛОГ НА HUSL

Pytest + FastApi + SQLAlchemy + asyncpg: Ошибка asyncio

Ответы (0 шт):