Во время тестов на pytest FastAPI использует основную бд, а не тестовую

Question

Подскажите, может кто знает: Во время теста бэкенд стучится в основную базу данных, а не в тестовую, хотя команда app.dependency_overrides[get_db] = override_get_db должна это предотвращать. Причем redis прекрасно заменяется на тестовый.

conftest.py

import pytest
from httpx import AsyncClient, ASGITransport
from typing import AsyncGenerator
from database.redis_client import get_redis
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool
from src.database.db import get_db
from src.database.models import *
from src.main import app


DATABASE_URL_TEST = f"postgresql+asyncpg://admin:admin@localhost:5433/db_test"
engine_test = create_async_engine(DATABASE_URL_TEST, poolclass=NullPool)
async_session_maker = async_sessionmaker(engine_test, class_=AsyncSession, expire_on_commit=False, autoflush=False)

Base.metadata.bind = engine_test


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


def override_get_redis() -> Redis:
    return Redis(host="localhost", port=6380, db=0)


app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_redis] = override_get_redis


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

    yield

    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture(scope='session')
async def async_client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

db.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = f"postgresql+asyncpg://admin:admin@db:5432/mb_db"

engine = create_async_engine(DATABASE_URL)

async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


class Base(DeclarativeBase):
    pass


async def get_db():
    async with async_session() as session:
        yield session

Answer 1

Возможная причина проблемы с заменой зависимости базы данных на тестовую может крыться в функции override_get_db. В определении этой функции вы используете контекстный менеджер with для создания сессии асинхронного session_maker, однако в асинхронном генераторе yield следует использовать без контекстного менеджера, чтобы явно закрыть сессию после того, как тест завершен. Замените with на try/finally блок для корректного закрытия сессии:

@pytest.fixture(autouse=True, scope='function')
async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
    session: AsyncSession = async_session_maker()
    try:
        yield session
    finally:
        await session.close()

БЛОГ НА HUSL

Во время тестов на pytest FastAPI использует основную бд, а не тестовую

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