测试¶

感谢 Starlette，测试 **FastAPI** 应用程序变得轻松愉快。

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

使用它，您可以将 pytest 直接与 **FastAPI** 一起使用。

使用 `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** 提供与 fastapi.testclient 相同的 starlette.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

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

它有一个 GET 操作，可以返回错误。

它有一个 POST 操作，可以返回多个错误。

两种路径操作都需要一个 X-Token 标题。

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ 非带注解的Python 3.8+ 未标注

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

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。
对于cookie，请在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¶