pytest：使用 SQLAlchemy 和 FastAPI 在测试之间回滚答案

【问题标题】：pytest: rollback between tests using SQLAlchemy and FastAPIpytest：使用 SQLAlchemy 和 FastAPI 在测试之间回滚
【发布时间】：2022-10-05 15:01:46
【问题描述】：

我有一个 FastAPI 应用程序，其中有几个用 pytest 编写的测试。

两个特定的测试引起了我的问题。 test_a 调用一个 post 端点，该端点在数据库中创建一个新条目。 test_b 获取这些条目。 test_b 包括从 test_a 创建的条目。这不是期望的行为.

当我单独运行测试（使用 VS Code 的测试选项卡）时，它运行良好。但是，当同时运行所有测试并且 test_a 在 test_b 之前运行时，test_b 会失败。

我的conftest.py 看起来像这样：

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from application.core.config import get_database_uri
from application.core.db import get_db
from application.main import app


@pytest.fixture(scope=\"module\", name=\"engine\")
def fixture_engine():
    engine = create_engine(
        get_database_uri(uri=\"postgresql://user:secret@localhost:5432/mydb\")
    )
    SQLModel.metadata.create_all(bind=engine)
    yield engine
    SQLModel.metadata.drop_all(bind=engine)


@pytest.fixture(scope=\"function\", name=\"db\")
def fixture_db(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope=\"function\", name=\"client\")
def fixture_client(db):
    app.dependency_overrides[get_db] = lambda: db
    with TestClient(app) as client:
        yield client

包含test_a 和test_b 的文件还有一个模块范围的pytest 固定装置，它使用engine 固定装置播种数据：

@pytest.fixture(scope=\"module\", autouse=True)
def seed(engine):
    connection = test_db_engine.connect()
    seed_data_session = Session(bind=connection)
    seed_data(seed_data_session)
    yield
    seed_data_session.rollback()

所有测试都使用client 夹具，如下所示：

def test_a(client):
    ...

SQLAlchemy 版本是 1.4.41，FastAPI 版本是 0.78.0，pytest 版本是 7.1.3。

我的观察

似乎测试本身运行良好的原因是由于在测试结束时调用了SQLModel.metadata.drop_all(bind=engine)。但是我想避免这样做，而是只在测试之间使用回滚。

标签： python postgresql sqlalchemy pytest fastapi

【解决方案1】：

对我来说真正有用的是使用测试容器：https://github.com/testcontainers/testcontainers-python。

@pytest.fixture(scope="module", name="session_for_db_in_testcontainer")
def db_engine():
    """
    Creates testcontainer with Postgres db
    """
    pg_container = PostgresContainer('postgres:latest')
    pg_container.start()

    # Fireup the SQLModel engine with the uri of the container
    db_engine = create_engine(pg_container.get_connection_url())
    sqlmodel_metadata.create_all(db_engine)

    with Session(db_engine) as session_for_db_in_testcontainer:
        # add some rows to start, for test get requests and posting existing data
        add_data_to_test_db(database_input_path, session_for_db_in_testcontainer)
        yield session_for_db_in_testcontainer

    # Will be executed after the last test
    session_for_db_in_testcontainer.close()
    pg_container.stop()

像这样在测试运行期间创建一个（Postgres）数据库，它仅在会话、模块或函数期间运行，具体取决于夹具的范围。如果需要，您可以像示例中一样将测试数据添加到数据库中。

在您的情况下，您可能希望将此夹具的范围设置为函数。比test_a 和test_b 应该独立运行。

【讨论】：