跳转至

测试

借助 Starlette,可以轻松、简单地测试 FastAPI 应用。

Starlette 基于 Requests,所以,大家应该对测试 FastAPI 应用并不陌生。

借助 Starlette,还可以直接使用 pytest 测试 FastAPI 应用。

使用 TestClient

说明

使用 TestClient,首先要安装 httpx

例如, pip install httpx

导入 TestClient

创建 TestClient,并把它传递给 FastAPI 应用。

创建以 test_ 开头的函数,这是 pytest 的标准惯例。

使用 TestClient 对象的方式与 requests 的方式一样。

使用 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)。

并且,调用 client 也要使用普通方式,不要使用 await

这样,直接使用 pytest 就不会变得复杂。

技术细节

您也可以使用 from starlette.testclient import TestClient

FastAPIfastapi.testclientstarlette.testclient 一样,只是为了方便开发者调用,但其实它直接继承自 Starlette。

提示

除了使用异步函数向 FastAPI 应用发送请求外(例如,异步数据库函数),如果想在测试中使用 async 异步函数,请参阅高级用户指南中的异步测试

分拆测试

实际应用中,测试会分为多个不同文件。

而且,FastAPI 应用也很有可能是由多个文件/模块组成的。

FastAPI 的 app 文件

假设要测试的项目使用与大型应用一节中相同的文件架构:

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

FastAPI 应用中的 main.py 如下所示:

from fastapi import FastAPI

app = FastAPI()


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

测试文件

创建测试文件 test_main.py。该文件可以与 main.py 在同一个 Python 包里(即包含 __init__.py 的同一个目录内)。

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

因为该文件与 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

假设 FastAPI 应用的 main.py 中还有一些其他路径操作

其中,GET 操作返回一个错误。

POST 操作返回多个错误。

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

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=400, detail="Item already exists")
    fake_db[item.id] = item
    return item
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=400, detail="Item already exists")
    fake_db[item.id] = item
    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_inexistent_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 == 400
    assert response.json() == {"detail": "Item already exists"}

如果您不知道该如何使用客户端在请求中传递信息,请去谷歌搜索如何在 httpxrequests 中执行这一操作。注意,HTTPX 是基于 Request 设计的。

执行同样的测试操作。

例如:

  • 传递路径查询参数,把它添加至 URL。
  • 传递 JSON 请求体时,需要把 dict 等 Python 对象传递给 json 参数。
  • 如需发送的不是 JSON,而是表单数据,则要使用 data 参数。
  • 传递请求头时,可在 headers 参数中使用 dict
  • 传递 cookies 时,可在 cookies 参数中使用 dict

关于如何使用 httpxTestClient 把数据传递给后端的说明,详见 HTTPX 文档

说明

注意,TestClient 接收的是可以转换为 JSON 的数据,不是 Pydantic 模型。

如果在测试中使用 Pydantic 模型,并希望在测试中把模型的数据发送给 FastAPI 应用,可以使用 JSON 编码器 中的 jsonable_encoder

运行测试

接下来,需要安装 pytest

$ pip install pytest

---> 100%

pytest 会自动检测文件,并进行测试。执行测试文件,就能生成测试报告。

运行测试:

$ 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>