测试¶
得益于 Starlette,测试 FastAPI 应用变得既轻松又愉快。
它基于 HTTPX,而 HTTPX 的设计又是基于 Requests 的,因此它非常直观且易于上手。
使用它,你可以直接在 FastAPI 中使用 pytest。
使用 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"}
提示
请注意,测试函数是普通的 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。
有关如何将数据传递到后端的更多信息(使用 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>