如何使用具有参数的函数/固定装置覆盖 FastApi app.dependency_overrides？

Question

我使用 deb 连接的固定装置设置了测试（基于 SQLAlchemy）。这里的主要特点是它正在回滚事务（测试没有更改数据库）：

@pytest.fixture(scope="session")
def connection():
    SQLALCHEMY_DATABASE_URL = "postgresql://{0}:{1}@{2}:{3}/{4}".format(
        settings.DB_username, settings.DB_password, settings.DB_hostname, settings.DB_port, settings.DB_name)

    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        connect_args={"options": "-c timezone=utc"}
    )

    return engine.connect()


@pytest.fixture(scope="session")
def db_session(connection):

    transaction = connection.begin()
    yield scoped_session(
        sessionmaker(autocommit=False, autoflush=False, bind=connection)
    )
    transaction.rollback()

另一方面，我有一个提供测试客户端的装置。我用它来进行从 api 到数据库的集成测试：

@pytest.fixture(scope='module')
def client() -> Generator:
    with TestClient(app) as c:
        yield c

一切正常。问题是应用程序和所有端点不是直接注入数据库，而是最后通过

db: Session = Depends(get_db)

注入数据库。因此，客户端使用的是不会回滚的“真实”连接，并且数据库更改将被持久化。

现在我尝试用以下方法覆盖

get_db

：

@pytest.fixture(scope='module')
def client() -> Generator:
    app.dependency_overrides[get_db] = db_session
    with TestClient(app) as c:
        yield c

问题是：这会产生 422 这是由于我的设置中缺少 db_session 需要的参数引起的。一旦我删除覆盖线，测试就会再次正常工作。

那么如何用我的

get_db

覆盖

db_session(connection)

或者一般如何拥有一个使用与直接依赖的其他测试相同的 db_session 的 TestClient。

Answer 1

get_db

将返回数据库会话而不是数据库会话固定装置。因此，我们需要使用从

get_db

夹具生成的函数来覆盖

db_session

依赖项。

首先，我们需要使用

session.remove()

清除测试后留下的任何会话，如下所示：

@pytest.fixture(scope="session")
def db_session(connection):
    transaction = connection.begin()
    session_factory = sessionmaker(autocommit=False, autoflush=False, bind=connection)
    session = scoped_session(session_factory)
    yield session
    session.remove()
    transaction.rollback()

我们重写

get_db

方法来使用 db_session 固定装置：

def override_get_db():
    try:
        db = next(iter(db_session(connection)))
        yield db
    finally:
        db.close()

生成测试客户端后，清除依赖项覆盖以避免后续测试出现任何潜在问题。

@pytest.fixture(scope='module')
def client() -> Generator:
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()