测试¶

得益于 Starlette，测试 FastAPI 应用变得简单而愉快。

它基于 HTTPX，而 HTTPX 的设计又借鉴了 Requests，因此非常熟悉和直观。

通过它，你可以直接使用 pytest 来测试 FastAPI。

使用 `TestClient`¶

Info

要使用 TestClient，请先安装 httpx。

请确保创建一个虚拟环境，激活它，然后安装，例如：

$ pip install httpx

导入 TestClient。

通过将你的 FastAPI 应用传递给 TestClient 来创建实例。

创建以 test_ 开头的函数（这是标准的 pytest 约定）。

使用 TestClient 对象的方式与使用 httpx 相同。

使用需要检查的标准 Python 表达式编写简单的 assert 语句（同样是标准的 pytest）。

Python 3.8+

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

Tip

请注意，测试函数是普通的 def，而不是 async def。

并且对客户端的调用也是普通调用，不使用 await。

这允许你直接使用 pytest 而不会复杂化。

技术细节

你也可以使用 from starlette.testclient import TestClient。

FastAPI 提供与 starlette.testclient 相同的 fastapi.testclient，只是为了方便开发者。但它直接来自 Starlette。

Tip

如果除了向 FastAPI 应用发送请求外，你还想在测试中调用 async 函数（例如异步数据库函数），请查看高级教程中的异步测试。

分离测试¶

在实际应用中，你可能会将测试放在不同的文件中。

而你的 FastAPI 应用也可能由多个文件/模块等组成。

FastAPI 应用文件¶

假设你有一个如大型应用中描述的文件结构：

.
├── app
│   ├── __init__.py
│   └── main.py

在 main.py 文件中，你有你的 FastAPI 应用：

Python 3.8+

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

测试文件¶

然后你可以有一个包含测试的 test_main.py 文件。它可以位于同一个 Python 包中（同一目录下带有 __init__.py 文件）：

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

由于此文件在同一包中，你可以使用相对导入从 main 模块（main.py）导入 app 对象：

Python 3.8+

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

...并保持与之前相同的测试代码。

测试：扩展示例¶

现在让我们扩展这个示例并添加更多细节，看看如何测试不同的部分。

扩展的 FastAPI 应用文件¶

让我们继续使用之前的相同文件结构：

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

假设现在包含你的 FastAPI 应用的 main.py 文件有一些其他的路径操作。

它有一个可能返回错误的 GET 操作。

它有一个可能返回多个错误的 POST 操作。

两个路径操作都需要 X-Token 请求头。

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.8+ non-Annotated

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

from typing import Annotated, Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from typing_extensions import Annotated

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

Tip

如果可能，建议使用 Annotated 版本。

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

Tip

如果可能，建议使用 Annotated 版本。

from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

扩展的测试文件¶

然后你可以用扩展的测试更新 test_main.py：

Python 3.8+

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 200
    assert response.json() == {
        "id": "foo",
        "title": "Foo",
        "description": "There goes my hero",
    }


def test_read_item_bad_token():
    response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_read_nonexistent_item():
    response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_create_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }


def test_create_item_bad_token():
    response = client.post(
        "/items/",
        headers={"X-Token": "hailhydra"},
        json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
    )
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_create_existing_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={
            "id": "foo",
            "title": "The Foo ID Stealers",
            "description": "There goes my stealer",
        },
    )
    assert response.status_code == 409
    assert response.json() == {"detail": "Item already exists"}

每当需要客户端在请求中传递信息但不知道如何操作时，你可以搜索（Google）如何在 httpx 中实现，甚至如何在 requests 中实现，因为 HTTPX 的设计基于 Requests 的设计。

然后在测试中做同样的事情。

例如：

要传递路径或查询参数，将其添加到 URL 本身。
要传递 JSON 请求体，将 Python 对象（例如 dict）传递给参数 json。
如果需要发送表单数据而不是 JSON，请改用 data 参数。
要传递请求头，在 headers 参数中使用 dict。
对于cookies，在 cookies 参数中使用 dict。

有关如何将数据传递到后端（使用 httpx 或 TestClient）的更多信息，请查看 HTTPX 文档。

Info

请注意，TestClient 接收可转换为 JSON 的数据，而不是 Pydantic 模型。

如果测试中有 Pydantic 模型并希望在测试期间将其数据发送到应用，可以使用 JSON 兼容编码器中描述的 jsonable_encoder。

运行测试¶

之后，你只需要安装 pytest。

请确保创建一个虚拟环境，激活它，然后安装，例如：

$ pip install pytest

---> 100%

它将自动检测文件和测试，执行它们，并将结果报告给你。

运行测试：

$ pytest

================ test session starts ================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /home/user/code/superawesome-cli/app
plugins: forked-1.1.3, xdist-1.31.0, cov-2.8.1
collected 6 items

---> 100%

test_main.py <span style="color: green; white-space: pre;">......                            [100%]</span>

<span style="color: green;">================= 1 passed in 0.03s =================</span>

测试¶

使用 TestClient¶

分离测试¶

FastAPI 应用文件¶

测试文件¶

测试：扩展示例¶

扩展的 FastAPI 应用文件¶

扩展的测试文件¶

运行测试¶

使用 `TestClient`¶