Как правильно настроить pytest для выполнения набора действий до и после всех моих тестов ⇐ Python

[Anonymous]

Я хочу правильно протестировать свое приложение FastAPI. Приложение использует локальную базу данных Postgres с асинхронным соединением и перегонным кубом для выполнения миграции, которая работает нормально.
Теперь я хочу правильно выполнить модульное тестирование своего приложения с помощью настоящей тестовой базы данных Postgres. Итак, в основном я хочу добиться следующего:

• Подключиться к моей базе данных Postgres
• Создать новую базу данных с именем test_db< /code>
• Запустите перегонную миграцию, чтобы получить ту же схему и исходные данные, что и у моей реальной базы данных
• Запустите все модульные тесты сейчас
• Запустите базу для понижения версии Alembic, чтобы удалить все таблицы и данные
• Удалить базу данных test_db

Проблема в том, что когда я запускаю pytest для выполнения модульных тестов, мой тест запускается, но база данных не создается. Таким образом, эффективно выполняется только шаг 4. Но больше ничего.
В моих тестах dir есть следующая структура:

Код: Выделить всё

.
└── tests/
├── __init__.py
├── conftest.py
├── test_app.py
└── test_user.py

Я создал файл conftest.py, который включает код для шагов 1–3 и 5–6:

Код: Выделить всё

import pytest
import asyncpg
from sqlmodel import SQLModel, create_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
# from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.exc import ProgrammingError
from alembic.config import Config
from alembic import command
from config import config

TEST_DATABASE_URI = config.SQLALCHEMY_DATABASE_URI_UNIT_TEST

# 1. Create the test database
async def create_test_db():
print("Creating test database")
conn = await asyncpg.connect(
user=config.DB_USER, password=config.DB_PASSWD, database=config.DB_NAME, host=config.DB_HOST
)
try:
await conn.execute(f"CREATE DATABASE {config.DB_NAME}_test;")
print("Database created successfully")
except asyncpg.exceptions.DuplicateDatabaseError as e:
print(e)
finally:
await conn.close()

# 2. Drop the test database
async def drop_test_db():
conn = await asyncpg.connect(
user=config.DB_USER, password=config.DB_PASSWD, database=config.DB_NAME, host=config.DB_HOST
)
try:
await conn.execute(f"DROP DATABASE IF EXISTS {config.DB_NAME}_test;")
finally:
await conn.close()

# 3. Run Alembic migrations
def run_migrations(db_uri, direction="upgrade", revision="head"):
alembic_cfg = Config("alembic.ini")
alembic_cfg.set_main_option("sqlalchemy.url", db_uri)
if direction == "upgrade":
command.upgrade(alembic_cfg, revision)
elif direction == "downgrade":
command.downgrade(alembic_cfg, revision)

# 4. Fixture to manage the test database lifecycle
@pytest.fixture(scope="session", autouse=True)
async def setup_test_db():
print("Hello world")
# Create test database
await drop_test_db()
await create_test_db()

# Run migrations
run_migrations(TEST_DATABASE_URI, "upgrade", "head")

yield  # All tests execute here

# Clean up: Downgrade and drop test database
run_migrations(TEST_DATABASE_URI, "downgrade", "base")
drop_test_db()

# 5.  Fixture for async test engine
@pytest.fixture(scope="function")
async def async_test_engine():
engine = create_async_engine(url=TEST_DATABASE_URI, echo=False)
yield engine
await engine.dispose()

Мой test_app.py выглядит так:

Код: Выделить всё

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
from httpx import AsyncClient
from httpx._transports.asgi import ASGITransport
from app import app
from config import config

@pytest.mark.usefixtures("setup_test_db")  # Ensure setup_test_db fixture is used
@pytest.mark.asyncio
async def test_root_route():
# Use ASGITransport explicitly
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Perform GET request
response = await client.get("/")
assert response.status_code == 200
assert response.json() == {"message": config.WELCOME_MESSAGE}

Идея состоит в том, что когда начинается сеанс pytest (запуск модульных тестов), он должен сначала настроить мою test_db, затем запустить все модульные тесты, используя эту test_db, и снова удалить test_db. Однако функция setup_test_db() никогда не запускается. Оператор «Hello World» никогда не печатается в терминале, и в это время база данных не выполняла никаких операций чтения/записи.
Как я могу это правильно настроить, чтобы:

• test_db настраивается один раз в начале сеанса.
• Маршруты, вызываемые в модульном тесте, фактически используют сеанс test_db. вместо настоящей сессии

Или есть лучший или более простой способ добиться этого, чем тот, который я планирую сделать прямо сейчас?

Подробнее здесь: https://stackoverflow.com/questions/791 ... d-after-al