测试¶
感谢 Starlette，测试 FastAPI 应用程序变得简单而愉快。
它基于 HTTPX，而 HTTPX 的设计又基于 Requests，因此它非常熟悉且直观。
通过它，你可以直接使用 pytest 与 FastAPI 进行测试。
使用 TestClient¶
导入 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"}
Tip
注意，测试函数是普通的 def，而不是 async def。
并且对客户端的调用也是普通的调用，而不是使用 await。
这允许你直接使用 pytest 而不会产生复杂性。
"技术细节"
你也可以使用 from starlette.testclient import TestClient。
FastAPI 提供了与 starlette.testclient 相同的 fastapi.testclient，只是为了方便你，开发者。但它直接来自 Starlette。
Tip
如果你想在测试中调用 async 函数，除了向你的 FastAPI 应用程序发送请求（例如异步数据库函数），请查看高级教程中的 异步测试。
分离测试¶
在一个实际的应用程序中，你可能会将测试放在一个单独的文件中。
而你的 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"}
测试文件¶
然后你可以在同一个 Python 包（同一个目录，包含一个 __init__.py 文件）中有一个 test_main.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/", 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 文件，添加扩展的测试：
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
================ 测试会话开始 ================
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>