测试¶

得益于 Starlette，测试 FastAPI 应用程序变得轻松愉快。

它基于 HTTPX，而 HTTPX 又基于 Requests 设计，因此非常熟悉且直观。

有了它，你可以直接将 pytest 与 FastAPI 配合使用。

使用 `TestClient`¶

信息

要使用 TestClient，首先安装 httpx。

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

$ pip install httpx

导入 TestClient。

将你的 FastAPI 应用程序传给 TestClient 来创建它。

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

像使用 httpx 一样使用 TestClient 对象。

使用你需要的标准 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"}

提示

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

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

这允许你直接使用 pytest，没有复杂性。

技术细节

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

FastAPI 提供了与 starlette.testclient 相同的 fastapi.testclient，仅为开发者提供便利。但它直接来源于 Starlette。

提示

如果你想在测试中除了向 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+ 非 AnnotatedPython 3.8+（非 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

提示

如果可能，请优先使用 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

提示

如果可能，请优先使用 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"}

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

然后你只需要在你的测试中做同样的事情。

例如：

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

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

信息

请注意，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`¶