如何使用 pytest、SQLAlchemy、依赖注入通过单个数据库会话设置 fastapi 测试

Question

我正在尝试使用 Pytest 和数据库在 FastAPI 项目上设置测试。我希望数据库在每次测试时创建并在每次测试后销毁。

我的设置有效，但是当我使用factory-boy包引入工厂时，它崩溃了。我希望能够使用factory-boy包来生成对象，这些对象将被持久化到我可以在测试用例中使用的测试数据库中。

我遇到的问题是定义工厂类将使用的会话，因为工厂男孩需要

SQLAlchemy

模型的会话。

注意：我在代码中使用 dependency-injector 库将依赖项传递给项目中的类。

在我的 db.py 类中，我有这样的数据库设置，一个用于实际应用程序的

Database

类和一个将在测试中使用的

TestDatabase

类。目的是我要在测试环境中覆盖数据库。

class Database:
    def __init__(self, db_url: str) -> None:
        print("Initializing DB...")
        import pdb

        pdb.set_trace()
        engine_kwargs = {"echo": True if settings.DEBUG else False}
        self._engine = create_engine(db_url, **engine_kwargs)
        self._session_factory = scoped_session(
            sessionmaker(
                class_=Session,
                autocommit=False,
                autoflush=False,
                bind=self._engine,
            )
        )

    def session_factory(self):  # noqa
        return self._session_factory()

    @contextmanager
    def session(self) -> Callable[..., AbstractContextManager[Session]]:
        session: Session = self._session_factory()
        try:
            yield session
        except Exception as e:
            print(e)
            session.rollback()
            raise
        finally:
            session.close()


class TestDatabase(Database):
    def __init__(self, db_url: str) -> None:  # noqa
        import pdb

        pdb.set_trace()
        # super().__init__(db_url)
        print("Initializing DB...")
        engine_kwargs = {
            "echo": True,
            "connect_args": {
                "check_same_thread": False,
            },
            "poolclass": StaticPool,
        }
        self._engine = create_engine(db_url, **engine_kwargs)

        # Create database tables for for test environment.
        SQLModel.metadata.create_all(bind=self._engine)

        self._session_factory = scoped_session(
            sessionmaker(
                class_=Session,
                autocommit=False,
                autoflush=False,
                bind=self._engine,
            )
        )

我还有一个用于依赖注入设置的容器：

class AppContainer(containers.DeclarativeContainer):
    """
    DI container for the project.
    """

    # Auto-wiring
    wiring_config = containers.WiringConfiguration(packages=["app.api"])

    # Singletons
    db = providers.Singleton(Database, db_url=settings.DATABASE_URI)

    # Repositories
    book_repository = providers.Factory(
        BookRepository,
        session_factory=db.provided.session_factory,
    )

    # Services
    book_service = providers.Factory(
        BookService,
        book_repository=book_repository,
    )

在我的

conftest.py

中，我设置了这些固定装置，一个用于在返回容器之前覆盖数据库，一个用于返回会话，另一个用于测试客户端：

@pytest.fixture()
def app_container():
    container = AppContainer()

    container.db.override(
        providers.Singleton(
            TestDatabase,
            db_url="sqlite:///:memory:",
        )
    )

    return container


@pytest.fixture()
def db_session(app_container):
    return app_container.db().session_factory()


@pytest.fixture()
def test_client():
    with TestClient(application) as test_client:
        yield test_client

在我的工厂.py 中，我有一个工厂类设置：

app_container = AppContainer()
app_container.db.override(
    providers.Singleton(
        TestDatabase,
        db_url="sqlite:///:memory:",
    )
)
session = app_container.db().session_factory()


class BookFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = Book
        sqlalchemy_session = session

    name = "Test Book"
    author = factory.Faker("name")

然后在示例测试函数中，我编写了这样的测试：

def test_store_book(test_client, app_container, db_session):
    BookFactory(name="Sample Book Name 2")

    payload = {
        "name": "Sample Book Name",
        "author": "John Doe",
    }

    response = test_client.post("/books", json=payload)

    books = db_session.query(Book).all()

    assert response.status_code == 200
    assert len(books) == 2

我对这个测试的期望是图书工厂将在数据库中创建一本书，端点将在数据库中创建另一本书，这样当我从数据库中获取图书数量时，图书列表的长度应该是2但这里的情况并非如此，它返回 1。我的猜测是数据库正在被销毁并重新创建。因此，工厂存储从工厂创建的初始书籍的数据库与端点存储书籍的数据库不同。

我想要的是在单个测试用例的生命周期中拥有一个数据库，以便我可以使用工厂在数据库中存储项目，也可以通过端点存储项目，我会看到它们。

Answer 1

看起来你需要添加sqlalchemy_session_persistence 到你的元

class BookFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = Book
        sqlalchemy_session = session
        sqlalchemy_session_persistence = 'commit'  # <- Here 
        ...