如何正确配置 pytest 以在所有测试之前和之后执行一组操作

Question

我想正确测试我的 FastAPI 应用程序。该应用程序使用具有异步连接和 alembic 的本地 postgres 数据库来进行迁移，效果很好。

现在我想使用真正的 postgres 测试数据库正确地对我的应用程序进行单元测试。所以基本上我想实现以下目标：

1. 连接到我的 postgres 数据库
2. 创建一个名为

   test_db

的新数据库
3. 运行 alembic 迁移以获得与我的真实数据库相同的架构和初始数据
4. 立即运行所有单元测试
5. 运行alembic降级基础以删除所有表并删除数据
6. 删除数据库

   test_db

问题是，当我运行

pytest

执行单元测试时，我的测试将运行，但没有创建数据库。因此实际上只运行步骤 4。但没有别的了。

我的

tests

目录中有以下结构：

.
└── 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 会话而不是真正的 prod 会话

或者有没有比我现在计划做的更好或更简单的方法来实现这一目标？

Answer 1

异步灯具不能开箱即用。您需要安装帮助程序包

pytest-asyncio

并将您的装置标记为需要异步语义。例如：

@pytest.mark.asyncio
@pytest.fixture(scope="session", autouse=True)
async def setup_test_db():
    print("Hello world")
    ...

旁注：

async_test_engine

不声明对

setup_test_db

的依赖。 Pytest 需要知道运行装置的顺序。如果没有这个，您最终可能会以错误的顺序运行装置。然而，由于

setup_test_db

具有会话范围，因此夹具设置上存在隐式排序。您不应该依赖于此，并且应该明确依赖关系。这将使代码维护更容易。