Задать вопрос
@Maxwell012

Как запускать pytest для тестирования FastApi?

У меня есть тесты для моего бекенда на FastApi. Проблема заключается в том что при выполнении тестов переменная async_session_maker внутри тестов отличается от async_session_maker внутри проекта и так выходит что тесты работают с тестовой базой данных, а бекенд запущенный в тестах продолжает обращаться к основной базе данных.
Файл который содержит создание async_session_maker проекта:
from typing import AsyncGenerator
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker

from src.config import (
    DB_HOST,
    DB_NAME,
    DB_PASS,
    DB_PORT,
    DB_USER
)

metadata = MetaData()

DATABASE_URL = (f"postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:"
                f"{DB_PORT}/{DB_NAME}")
Base: DeclarativeMeta = declarative_base()


engine = create_async_engine(DATABASE_URL)
async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
print("Project async_session_maker:", async_session_maker)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


Файл который содержит конфигурацию тестов:
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
import asyncio
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from src.config import (
    DB_HOST_TEST,
    DB_NAME_TEST,
    DB_PASS_TEST,
    DB_PORT_TEST,
    DB_USER_TEST
)
from src.database.database import (
    metadata,
    get_async_session
)
from src.backend.main import app


DATABASE_URL = (f"postgresql+asyncpg://{DB_USER_TEST}:{DB_PASS_TEST}@{DB_HOST_TEST}:"
                f"{DB_PORT_TEST}/{DB_NAME_TEST}")

engine_test = create_async_engine(DATABASE_URL)
async_session_maker = sessionmaker(engine_test, class_=AsyncSession, expire_on_commit=False)
metadata.bind = engine_test


@pytest.fixture(scope='function')
async def override_get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session
        await session.rollback()


app.dependency_overrides[get_async_session] = override_get_async_session


# SETUP
@pytest.fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each tests case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(autouse=True, scope='session')
async def prepare_database():
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.create_all)
    yield
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.drop_all)


@pytest.fixture(scope="session")
async def ac() -> AsyncGenerator[AsyncClient, None]:
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


Пытался переопределять async_session_maker бекенда внутри конфигурации теста, но это не помогло. Подскажите как правильно переопределить эту переменную чтобы бекенд вовремя прохождения тестов взаимодействовал именно с тестовой базо данных.
  • Вопрос задан
  • 134 просмотра
Подписаться 1 Простой Комментировать
Пригласить эксперта
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Похожие вопросы