使用 FastAPI 和 SQLModel 测试应用程序

为了完成关于 FastAPI 和 SQLModel 的这一组章节，现在让我们学习如何为使用 FastAPI 和 SQLModel 的应用程序实现自动化测试。✅

包括技巧和窍门。🎁

FastAPI 应用程序

让我们使用我们在前几章中构建的 更简单 的 FastAPI 应用程序之一。

所有相同的 概念、技巧 和 窍门 也将适用于更复杂的应用程序。

我们将使用带有英雄模型但没有团队模型的应用程序，我们将使用依赖项来获取 会话。

现在我们将看到拥有这个会话依赖项是多么有用。✨

👀 完整文件预览

Python 3.8+

from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero


@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(
    *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

文件结构

现在我们将有一个包含多个文件的 Python 项目，一个文件 main.py 包含所有应用程序，一个文件 test_main.py 包含测试，与 代码结构和多个文件 中的思想相同。

文件结构是

.
├── project
    ├── __init__.py
    ├── main.py
    └── test_main.py

测试 FastAPI 应用程序

如果您还没有在 FastAPI 应用程序中进行过测试，请首先查看 FastAPI 关于测试的文档。

然后，我们可以在这里继续，第一步是安装依赖项 requests 和 pytest。

确保您创建了一个 虚拟环境，激活它，然后安装它们，例如使用

$ pip install requests pytest

---> 100%

基本测试代码

让我们从一个简单的测试开始，只包含我们需要检查 FastAPI 应用程序是否正确创建新英雄的基本测试代码。

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

from .main import app, get_session  # (1)!


def test_create_hero():
        # Some code here omitted, we will see it later 👈
        client = TestClient(app)  # (2)!

        response = client.post(  # (3)!
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        # Some code here omitted, we will see it later 👈
        data = response.json()  # (4)!

        assert response.status_code == 200  # (5)!
        assert data["name"] == "Deadpond"  # (6)!
        assert data["secret_name"] == "Dive Wilson"  # (7)!
        assert data["age"] is None  # (8)!
        assert data["id"] is not None  # (9)!

# Code below omitted 👇

1. 从 main 模块导入 app。

2. 我们为 FastAPI app 创建一个 TestClient 并将其放入变量 client 中。

3. 然后我们使用这个 client 来 与 API 对话 并发送 POST HTTP 操作，创建一个新英雄。

4. 然后我们从响应中获取 JSON 数据 并将其放入变量 data 中。

5. 接下来我们开始使用 assert 语句测试结果，我们检查响应的状态码是否为 200。

6. 我们检查创建的英雄的 name 是否为 "Deadpond"。

7. 我们检查创建的英雄的 secret_name 是否为 "Dive Wilson"。

8. 我们检查创建的英雄的 age 是否为 None，因为我们没有发送年龄。

9. 我们检查创建的英雄是否有一个由数据库创建的 id，所以它不是 None。

提示

查看数字气泡，了解每行代码的作用。

这是我们稍后所有测试所需代码的 核心。

但现在，我们需要处理一些我们尚未注意到的物流和细节。🤓

测试数据库

这个测试看起来不错，但有一个问题。

如果我们运行它，它将使用我们用来存储我们非常重要的 英雄 的相同 生产数据库，我们最终会向其中添加不必要的数据，甚至更糟的是，在未来的测试中，我们最终可能会删除生产数据。

所以，我们应该使用一个独立的 测试数据库，只用于测试。

要做到这一点，我们需要更改用于数据库的 URL。

但是当执行 API 的代码时，它会得到一个已经连接到 引擎 的 会话，并且 引擎 已经使用特定的数据库 URL。

即使我们从 main 模块导入变量并仅为测试更改其值，此时 引擎 已经使用原始值创建。

但是我们所有的 API *路径操作* 都使用 FastAPI 依赖项 获取 *会话*，我们可以在测试中覆盖依赖项。

这就是依赖项开始发挥巨大作用的地方。

覆盖依赖项

让我们为测试覆盖 get_session() 依赖项。

所有 *路径操作* 都使用此依赖项来获取 SQLModel 会话对象。

我们将覆盖它以仅用于测试使用不同的 会话 对象。

这样我们就可以保护生产数据库，并且更好地控制我们正在测试的数据。

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

from .main import app, get_session  # (1)!


def test_create_hero():
        # Some code here omitted, we will see it later 👈
        def get_session_override():  # (2)!
            return session  # (3)!

        app.dependency_overrides[get_session] = get_session_override  # (4)!

        client = TestClient(app)

        response = client.post(
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        app.dependency_overrides.clear()  # (5)!
        data = response.json()

        assert response.status_code == 200
        assert data["name"] == "Deadpond"
        assert data["secret_name"] == "Dive Wilson"
        assert data["age"] is None
        assert data["id"] is not None

# Code below omitted 👇

1. 从 main 模块导入 get_session 依赖项。

2. 定义将作为新 依赖项覆盖 的新函数。

3. 此函数将返回与原始 get_session 函数返回的 会话 不同的会话。

   我们还没有看到这个新的 会话 对象是如何创建的，但重点是它是一个与应用程序原始会话不同的会话。

   此会话附加到不同的 引擎，并且该不同的 引擎 使用不同的 URL，用于仅用于测试的数据库。

   我们还没有定义新的 URL 和新的 引擎，但在这里我们已经看到这个 session 对象将覆盖原始依赖项 get_session() 返回的对象。

4. 然后，FastAPI app 对象有一个属性 app.dependency_overrides。

   此属性是一个字典，我们可以通过将 原始依赖项函数 作为 键，将 新的覆盖依赖项函数 作为 值 来在其​​中放置依赖项覆盖。

   所以，在这里我们告诉 FastAPI 应用程序在代码中所有依赖于 get_session 的地方使用 get_session_override 而不是 get_session，也就是所有带有类似以下内容的参数

   session: Session = Depends(get_session)
   

5. 完成依赖项覆盖后，我们可以通过删除此字典 app.dependency_overrides 中的所有值来将应用程序恢复正常。

   这样，每当 *路径操作函数* 需要依赖项时，FastAPI 将使用原始依赖项而不是覆盖。

提示

查看数字气泡，了解每行代码的作用。

为测试创建引擎和会话

现在让我们创建将在测试期间使用的 会话 对象。

它将使用自己的 引擎，这个新引擎将使用一个新的 URL 用于测试数据库

sqlite:///testing.db

所以，测试数据库将位于文件 testing.db 中。

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

from .main import app, get_session  # (1)!


def test_create_hero():
    engine = create_engine(  # (2)!
        "sqlite:///testing.db", connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(engine)  # (3)!

    with Session(engine) as session:  # (4)!

        def get_session_override():
            return session  # (5)!

        app.dependency_overrides[get_session] = get_session_override  # (4)!

        client = TestClient(app)

        response = client.post(
            "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
        )
        app.dependency_overrides.clear()
        data = response.json()

        assert response.status_code == 200
        assert data["name"] == "Deadpond"
        assert data["secret_name"] == "Dive Wilson"
        assert data["age"] is None
        assert data["id"] is not None
    # (6)!

1. 这里有一个微妙之处值得注意。

   请记住 顺序很重要，我们需要确保所有 SQLModel 模型在调用 .create_all() 之前都已定义和 导入。

   在这一行，通过从 .main 导入 任何 东西，.main 中的代码将被执行，包括 表模型 的定义，这将自动将它们注册到 SQLModel.metadata 中。

2. 在这里我们创建一个新的 引擎，与 main.py 中的引擎完全不同。

   这是我们将用于测试的引擎。

   我们使用数据库的新 URL 进行测试

   sqlite:///testing.db
   

   我们再次使用连接参数 check_same_thread=False。

3. 然后我们调用

   SQLModel.metadata.create_all(engine)
   

   ...以确保我们在新的测试数据库中创建所有表。

   表模型 在 SQLModel.metadata 中注册，仅仅是因为我们从 .main 导入了 某些东西，并且 .main 中的代码被执行，创建了 表模型 的类并自动将它们注册到 SQLModel.metadata 中。

   所以，当我们调用这个方法时，表模型 已经注册在那里了。💯

4. 在这里我们创建一个自定义 会话 对象用于此测试，在一个 with 块中。

   它使用我们创建的新自定义 引擎，因此任何使用此会话的东西都将使用测试数据库。

5. 现在，回到依赖项覆盖，它只是从外部返回相同的 会话 对象，就是这样，这就是整个诀窍。

6. 此时，测试 会话 with 块结束，会话关闭，文件关闭等。

导入表模型

在这里，我们使用以下命令在测试数据库中创建所有表

SQLModel.metadata.create_all(engine)

但是请记住 顺序很重要，我们需要确保所有 SQLModel 模型在调用 .create_all() 之前都已定义和 导入。

在这种情况下，一切都因为一个需要注意的微妙之处而奏效。

因为我们从 .main 导入了 任何东西，.main 中的代码将被执行，包括 表模型 的定义，这将自动将它们注册到 SQLModel.metadata 中。

这样，当我们调用 .create_all() 时，所有 表模型 都将正确注册到 SQLModel.metadata 中，一切都将正常工作。👌

内存数据库

现在我们不再使用生产数据库。相反，我们使用带有 testing.db 文件的新 测试数据库，这非常好。

但是 SQLite 也支持拥有 内存中 数据库。这意味着所有数据库都只存在于内存中，并且永远不会保存到磁盘上的文件中。

程序终止后，内存中数据库将被删除，因此它对生产数据库没有多大帮助。

但是 它非常适合测试，因为它可以在每次测试之前快速创建，并在每次测试之后快速删除。✅

而且，由于它永远不必将任何内容写入文件，并且全部都在内存中，所以它甚至会比平时更快。🏎

其他替代方案和想法 👀

在使用 内存中数据库 的想法之前，我们可能已经探索过其他替代方案和想法。

首先是我们没有在测试完成后删除文件，因此下一个测试可能会有 残留数据。因此，正确的做法是在测试完成后立即删除文件。🔥

但是，如果每个测试都必须创建一个新文件，然后删除它，那么运行所有测试可能会 有点慢。

现在，我们有一个 testing.db 文件，所有测试都使用它（我们现在只有一个测试，但我们会有更多）。

因此，如果我们尝试同时 并行 运行测试以稍微加快速度，它们将尝试使用 相同 的 testing.db 文件而发生冲突。

当然，我们也可以通过为每个测试数据库文件使用一些 随机名称 来解决这个问题......但在 SQLite 的情况下，我们有更好的替代方案，只需使用 内存中数据库。✨

配置内存数据库

让我们更新代码以使用内存数据库。

我们只需要更改 引擎 中的几个参数。

from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool  # (1)!

from .main import app, get_session


def test_create_hero():
    engine = create_engine(
        "sqlite://",  # (2)!
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,  # (3)!
    )

# Code below omitted 👇

1. 从 sqlmodel 导入 StaticPool，我们稍后会用到它。

2. 对于 SQLite URL，不要写入任何文件名，将其留空。

   所以，不是

   sqlite:///testing.db
   

   ...只需写入

   sqlite://
   

   这足以告诉 SQLModel（实际上是 SQLAlchemy）我们想使用 内存 SQLite 数据库。

3. 还记得我们告诉负责与 SQLite 通信的 低级 库，我们希望能够通过 check_same_thread=False 从不同线程访问数据库 吗？

   现在我们使用 内存数据库，我们还需要告诉 SQLAlchemy，我们希望能够从不同线程使用 相同的内存数据库 对象。

   我们通过 poolclass=StaticPool 参数来告诉它。

   信息

   您可以在 SQLAlchemy 关于在多线程中使用内存数据库的文档 中阅读更多详细信息

提示

查看数字气泡，了解每行代码的作用。

就是这样，现在测试将使用 内存数据库 运行，这将更快，也可能更安全。

所有其他测试都可以做同样的事情。

样板代码

太棒了，这很有效，你可以在每个测试函数中复制所有这些过程。

但是我们不得不添加大量的 样板代码 来处理自定义数据库，在内存中创建它，自定义会话，以及依赖项覆盖。

我们真的必须为 每个测试 重复所有这些吗？不，我们可以做得更好！😎

我们正在使用 pytest 运行测试。Pytest 也有一个与 FastAPI 中的依赖项 非常相似的概念。

信息

事实上，pytest 是激发 FastAPI 依赖项设计的事物之一。

这是一种声明一些 应该在每个测试之前运行 的代码并为测试函数 提供一个值 的方式（这与 FastAPI 依赖项非常相似）。

事实上，它也具有相同的技巧，允许使用 yield 而不是 return 来提供值，然后 pytest 确保在测试函数完成 之后 执行 yield 之后的代码。

在 pytest 中，这些东西被称为 夹具，而不是 *依赖项*。

让我们使用这些 夹具 来改进我们的代码，并减少下一次测试的重复样板。

Pytest 夹具

你可以在 pytest 的夹具文档 中阅读更多内容，但我会给你一个我们这里需要的简短示例。

让我们看第一个带有夹具的代码示例

import pytest  # (1)!
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from .main import app, get_session


@pytest.fixture(name="session")  # (2)!
def session_fixture():  # (3)!
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session  # (4)!


def test_create_hero(session: Session):  # (5)!
    def get_session_override():
        return session  # (6)!

    app.dependency_overrides[get_session] = get_session_override

    client = TestClient(app)

    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    app.dependency_overrides.clear()
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None

1. 导入 pytest。

2. 在函数顶部使用 @pytest.fixture() 装饰器，告诉 pytest 这是一个 夹具 函数（相当于 FastAPI 依赖项）。

   我们还给它一个名称 "session"，这在测试函数中很重要。

3. 创建夹具函数。这等同于 FastAPI 依赖项函数。

   在此夹具中，我们创建自定义 引擎，带有内存数据库，我们创建表，并创建 会话。

   然后我们 yield session 对象。

4. 我们 return 或 yield 的内容将可用于测试函数，在本例中是 session 对象。

   这里我们使用 yield，这样 pytest 会在测试函数完成后，回到此函数中执行“其余代码”。

   在 yield 之后我们没有更多可见的“其余代码”，但我们有 with 块的末尾，它将关闭 会话。

   通过使用 yield，pytest 将

   • 运行第一部分
   • 创建 会话 对象
   • 将其交给测试函数
   • 运行测试函数
   • 一旦测试函数完成，它将在这里，在 yield 之后继续，并将在 with 块的末尾正确关闭 会话 对象。

5. 现在，在测试函数中，为了告诉 pytest 这个测试想要获取夹具，而不是像 FastAPI 中那样声明一些东西，例如

   session: Session = Depends(session_fixture)
   

   ...我们告诉 pytest 我们想要哪个夹具的方式是使用夹具的 确切名称。

   在这种情况下，我们将其命名为 session，因此参数必须精确命名为 session 才能工作。

   我们还添加了类型注释 session: Session，以便我们可以在编辑器中获得自动补全和内联错误检查。

6. 现在在依赖项覆盖函数中，我们只是返回从外部传入的相同 session 对象。

   session 对象来自传递给测试函数的参数，我们只是在这里在依赖项覆盖中重新使用并返回它。

提示

查看数字气泡，了解每行代码的作用。

pytest 夹具的工作方式与 FastAPI 依赖项非常相似，但有一些细微差别

• 在 pytest 夹具中，我们需要在顶部添加 @pytest.fixture() 装饰器。
• 要在函数中使用 pytest 夹具，我们必须声明参数，其 名称必须完全相同。在 FastAPI 中，我们必须 显式使用 Depends() 并在其中包含实际函数。

但除了声明它们的方式以及我们如何告诉框架我们希望在函数中拥有它们之外，它们 的工作方式非常相似。

现在我们创建了许多测试，并在所有这些测试中重用了相同的夹具，从而节省了 样板代码。

pytest 将确保在每个测试函数之前（并在之后）运行它们。因此，每个测试函数实际上都将拥有自己的数据库、引擎和会话。

客户端夹具

太棒了，这个夹具帮助我们避免了大量的重复代码。

但是目前，我们仍然必须在测试函数中编写一些对其他测试重复的代码，现在我们

• 创建 依赖项覆盖
• 将其放入 app.dependency_overrides
• 创建 TestClient
• 发出请求后清除依赖项覆盖

这在未来的其他测试中仍然会重复。我们能改进它吗？是的！🎉

每个 pytest 夹具（与 FastAPI 依赖项相同），都可以要求其他夹具。

因此，我们可以创建一个 客户端夹具，它将在所有测试中使用，并且它本身将要求 会话夹具。

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

from .main import app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")  # (1)!
def client_fixture(session: Session):  # (2)!
    def get_session_override():  # (3)!
        return session

    app.dependency_overrides[get_session] = get_session_override  # (4)!

    client = TestClient(app)  # (5)!
    yield client  # (6)!
    app.dependency_overrides.clear()  # (7)!


def test_create_hero(client: TestClient):  # (8)!
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None

1. 创建名为 "client" 的新夹具。

2. 这个 客户端夹具 反过来也需要 会话夹具。

3. 现在我们在客户端夹具中创建 依赖项覆盖。

4. 在 app.dependency_overrides 字典中设置 依赖项覆盖。

5. 使用 FastAPI app 创建 TestClient。

6. yield TestClient 实例。

   通过使用 yield，在测试函数完成之后，pytest 将返回执行 yield 之后的其余代码。

7. 这是 yield 之后和测试函数完成之后的清理代码。

   在这里，我们清除 FastAPI app 中的依赖项覆盖（这里只有一个）。

8. 现在测试函数需要 客户端夹具。

   在测试函数内部，代码非常 简单，我们只需使用 TestClient 向 API 发送请求，检查数据，仅此而已。

   夹具负责所有 设置 和 清理 代码。

提示

查看数字气泡，了解每行代码的作用。

现在我们有一个 客户端夹具，它反过来使用 会话夹具。

在实际的测试函数中，我们只需声明我们需要这个 客户端夹具。

添加更多测试

此时，所有这些可能看起来我们只是做了很多改变，却得到了 相同的结果。🤔

但通常我们会创建 许多其他测试函数。现在所有的样板和复杂性都 只编写一次，在这些两个夹具中。

让我们添加更多测试

Python 3.8+

# Code above omitted 👆

def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422

# Code below omitted 👇

👀 完整文件预览

Python 3.8+

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

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

提示

不仅要测试正常情况，还要测试 无效数据、错误 和 边缘情况 是否得到正确处理，这始终是 好主意。

这就是我们在这里添加这两个额外测试的原因。

现在，任何附加的测试函数都可以像第一个测试函数一样 简单，它们只需 声明 client 参数 即可获取包含所有数据库设置的 TestClient 夹具。太棒了！😎

为什么是两个夹具

现在，看到代码，我们可能会想，为什么我们放置 两个夹具 而不是 只有一个 包含所有代码？这完全有道理！

对于这些示例，一个夹具会更简单，没有必要将代码分成两个夹具...

但对于下一个测试函数，我们将需要 两个夹具，客户端 和 会话。

Python 3.8+

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

from .main import Hero, app, get_session

# Code here omitted 👈

def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id

# Code below omitted 👇

👀 完整文件预览

Python 3.8+

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

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

在此测试函数中，我们希望检查 读取英雄列表 的 *路径操作* 是否实际向我们发送英雄。

但是如果 数据库为空，我们将得到一个 空列表，我们将不知道英雄数据是否正确发送。

但是我们可以在发送 API 请求之前在测试数据库中 创建一些英雄。✨

而且因为我们使用的是 测试数据库，所以为测试创建英雄不会影响任何东西。

要做到这一点，我们必须

• 导入 Hero 模型
• 需要两个夹具，客户端 和 会话
• 创建一些英雄并使用 会话 将它们保存到数据库中

之后，我们可以发送请求并检查我们是否确实从数据库中正确获取了数据。💯

这里有一个重要的细节需要注意：我们可以在其他夹具中 以及 在测试函数中要求夹具。

客户端夹具 的函数和实际的测试函数将 都 接收相同的 会话。

添加其余测试

使用相同的想法，要求夹具，创建测试所需的数据等，我们现在可以添加其余的测试。它们看起来与我们目前所做的非常相似。

Python 3.8+

# Code above omitted 👆

def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

👀 完整文件预览

Python 3.8+

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

from .main import Hero, app, get_session


@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()


def test_create_hero(client: TestClient):
    response = client.post(
        "/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"}
    )
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpond"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] is not None


def test_create_hero_incomplete(client: TestClient):
    # No secret_name
    response = client.post("/heroes/", json={"name": "Deadpond"})
    assert response.status_code == 422


def test_create_hero_invalid(client: TestClient):
    # secret_name has an invalid type
    response = client.post(
        "/heroes/",
        json={
            "name": "Deadpond",
            "secret_name": {"message": "Do you wanna know my secret identity?"},
        },
    )
    assert response.status_code == 422


def test_read_heroes(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
    session.add(hero_1)
    session.add(hero_2)
    session.commit()

    response = client.get("/heroes/")
    data = response.json()

    assert response.status_code == 200

    assert len(data) == 2
    assert data[0]["name"] == hero_1.name
    assert data[0]["secret_name"] == hero_1.secret_name
    assert data[0]["age"] == hero_1.age
    assert data[0]["id"] == hero_1.id
    assert data[1]["name"] == hero_2.name
    assert data[1]["secret_name"] == hero_2.secret_name
    assert data[1]["age"] == hero_2.age
    assert data[1]["id"] == hero_2.id


def test_read_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.get(f"/heroes/{hero_1.id}")
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == hero_1.name
    assert data["secret_name"] == hero_1.secret_name
    assert data["age"] == hero_1.age
    assert data["id"] == hero_1.id


def test_update_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.patch(f"/heroes/{hero_1.id}", json={"name": "Deadpuddle"})
    data = response.json()

    assert response.status_code == 200
    assert data["name"] == "Deadpuddle"
    assert data["secret_name"] == "Dive Wilson"
    assert data["age"] is None
    assert data["id"] == hero_1.id


def test_delete_hero(session: Session, client: TestClient):
    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
    session.add(hero_1)
    session.commit()

    response = client.delete(f"/heroes/{hero_1.id}")

    hero_in_db = session.get(Hero, hero_1.id)

    assert response.status_code == 200

    assert hero_in_db is None

运行测试

现在我们可以使用 pytest 运行测试并查看结果

$ pytest

============= test session starts ==============
platform linux -- Python 3.10.0, pytest-7.4.4, pluggy-1.5.0
rootdir: /home/user/code/sqlmodel-tutorial
<b>collected 7 items                              </b>

---> 100%

project/test_main.py <font color="#A6E22E">.......         [100%]</font>

<font color="#A6E22E">============== </font><font color="#A6E22E"><b>7 passed</b></font><font color="#A6E22E"> in 0.83s ===============</font>

回顾

你读完了所有这些吗？哇，我印象深刻！😎

为应用程序添加测试将为您提供极大的 确定性，确保一切都 正常运行，正如您所期望的那样。

在 重构 代码、更改内容、添加功能 时，测试将非常有用。因为测试可以帮助捕获许多在重构时容易引入的错误。

它们将为您提供更快、更高效 地工作的信心，因为您知道您正在检查您是否 没有破坏任何东西。😅

我认为测试是那些能让您的代码和您作为开发人员达到下一个专业水平的事物之一。😎

如果您阅读并研究了所有这些，您就已经了解了许多我花了数年时间才学到的高级思想和技巧。🚀