跳到内容

测试

得益于 Starlette,测试 FastAPI 应用变得既轻松又愉快。

它基于 HTTPX,而 HTTPX 的设计又是基于 Requests 的,因此它非常直观且易于上手。

使用它,你可以直接在 FastAPI 中使用 pytest

使用 TestClient

信息

要使用 TestClient,请先安装 httpx

确保你创建了虚拟环境,激活它,然后进行安装,例如:

$ pip install httpx

导入 TestClient

通过传入你的 FastAPI 应用来创建一个 TestClient

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

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

使用标准的 Python 表达式编写简单的 assert 语句来进行验证(同样是标准的 pytest 用法)。

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 应用。

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 对象。

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

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

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

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

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

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/")
async def create_item(item: Item, x_token: Annotated[str, Header()]) -> Item:
    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.model_dump()
    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/")
async def create_item(item: Item, x_token: str = Header()) -> Item:
    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.model_dump()
    return item

扩展的测试文件

然后你可以更新 test_main.py 来包含这些扩展测试。

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"}
🤓 其他版本和变体

提示

如果可能,请优先使用 Annotated 版本。

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),在 headers 参数中使用一个 dict
  • 对于 Cookie,在 cookies 参数中使用一个 dict

有关如何将数据传递到后端的更多信息(使用 httpxTestClient),请查看 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>