使用 FastAPI 和 SQLModel 测试应用程序

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

包括提示和技巧。 🎁

FastAPI 应用程序

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

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

我们将使用带有 hero 模型但不带 team 模型的应用程序，并且我们将使用依赖项来获取 会话。

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

👀 完整文件预览

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)
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    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 应用程序是否正确创建新 hero 的基本测试代码。

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 操作，创建一个新的 hero。

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

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

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

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

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

9. 我们检查创建的 hero 是否具有数据库创建的 id，因此它不是 None。

提示

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

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

但是现在，我们需要处理一些我们尚未关注的后勤和细节。 🤓

测试数据库

此测试看起来不错，但存在问题。

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

因此，我们应该使用独立的 测试数据库，仅用于测试。

为此，我们需要更改用于数据库的 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. 这里有一个微妙的事情需要注意。

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

   在此行中，通过从 .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)

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

在这种情况下，一切都正常工作，因为一个值得关注的细微之处。

因为我们从 .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 中，这些东西称为 fixture 而不是 依赖项。

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

Pytest Fixture

您可以在 pytest fixture 文档 中阅读有关它们的更多信息，但我将为您提供一个简短的示例，说明我们在此处需要什么。

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

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 这是一个 fixture 函数（等效于 FastAPI 依赖项）。

   我们还为其命名为 "session"，这在测试函数中将很重要。

3. 创建 fixture 函数。这等效于 FastAPI 依赖项函数。

   在此 fixture 中，我们创建自定义 引擎，使用内存数据库，创建表，并创建 会话。

   然后我们 yield session 对象。

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

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

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

   通过使用 yield，pytest 将

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

5. 现在，在测试函数中，要告诉 pytest 此测试想要获取 fixture，而不是像 FastAPI 中那样声明类似的内容

   session: Session = Depends(session_fixture)
   

   ...我们告诉 pytest 我们想要获取的 fixture 是什么的方式是使用 fixture 的 完全相同的名称。

   在本例中，我们将其命名为 session，因此参数必须完全命名为 session 才能使其工作。

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

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

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

提示

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

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

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

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

现在我们创建了大量测试并在所有测试中重用相同的 fixture，从而为我们节省了 样板代码。

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

客户端 Fixture

太棒了，该 fixture 帮助我们防止了大量重复代码。

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

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

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

每个 pytest fixture（与 FastAPI 依赖项相同的方式）都可以需要其他 fixture。

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

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" 的新 fixture。

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

3. 现在我们在客户端 fixture 内创建 依赖项覆盖。

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

5. 使用 FastAPI app 创建 TestClient。

6. yield TestClient 实例。

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

7. 这是清理代码，在 yield 之后，以及在测试函数完成后。

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

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

   在测试函数内部，代码非常 简单，我们只需使用 TestClient 向 API 发出请求，检查数据，就是这样。

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

提示

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

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

在实际的测试函数中，我们只需要声明我们需要此 客户端 fixture。

添加更多测试

此时，所有这些似乎只是我们做了很多更改，但实际上 结果相同。 🤔

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

让我们添加更多测试

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 fixture。太棒了！ 😎

为什么使用两个 Fixture

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

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

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

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

在此测试函数中，我们要检查 读取 hero 列表 的 *路径操作* 实际上是否向我们发送了 hero。

但是，如果 数据库为空，我们将获得一个 空列表，并且我们不知道 hero 数据是否被正确发送。

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

并且因为我们正在使用 测试数据库，所以我们不会因创建用于测试的 hero 而影响任何内容。

为此，我们必须

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

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

这是需要注意的重要细节：我们可以在其他 fixture 以及 测试函数中需要 fixture。

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

添加其余测试

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

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.7.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
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>

回顾

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

向您的应用程序添加测试将使您更加 确信 一切都 正常工作，正如您预期的那样。

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

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

我认为测试是将您的代码和您作为开发人员提升到下一个专业水平的事物之一。 😎

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