У меня есть тесты для моего бекенда на 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 бекенда внутри конфигурации теста, но это не помогло. Подскажите как правильно переопределить эту переменную чтобы бекенд вовремя прохождения тестов взаимодействовал именно с тестовой базо данных.