测试¶
感谢 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。
还有client的调用也是普通的调用，不是用 await。
这让你可以直接使用 pytest 而不会遇到麻烦。
技术细节
你也可以用 from starlette.testclient import TestClient。
FastAPI 提供了和 starlette.testclient 一样的 fastapi.testclient，只是为了方便开发者。但它直接来自Starlette。
提示
除了发送请求之外，如果你还想测试时在FastAPI应用中调用 async 函数（例如异步数据库函数）， 可以在高级教程中看下 Async Tests 。
分离测试¶
在实际应用中，你可能会把你的测试放在另一个文件里。
您的FastAPI应用程序也可能由一些文件/模块组成等等。
FastAPI app 文件¶
假设你有一个像 更大的应用 中所描述的文件结构:
.
├── app
│ ├── __init__.py
│ └── main.py
在 main.py 文件中你有一个 FastAPI app:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_main():
return {"msg": "Hello World"}
测试文件¶
然后你会有一个包含测试的文件 test_main.py 。app可以像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 文件¶
让我们继续之前的文件结构：
.
├── app
│ ├── __init__.py
│ ├── main.py
│ └── test_main.py
假设现在包含FastAPI app的文件 main.py 有些其他路径操作。
有个 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=400, 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=400, 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=400, detail="Item already exists")
fake_db[item.id] = item
return item
提示
Prefer to use the Annotated version if possible.
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
提示
Prefer to use the Annotated version if possible.
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"}
每当你需要客户端在请求中传递信息，但你不知道如何传递时，你可以通过搜索（谷歌）如何用 httpx做，或者是用 requests 做，毕竟HTTPX的设计是基于Requests的设计的。
接着只需在测试中同样操作。
示例：
- 传一个路径 或查询 参数，添加到URL上。
- 传一个JSON体，传一个Python对象(例如一个
dict)到参数json。 - 如果你需要发送 Form Data 而不是 JSON，使用
data参数。 - 要发送 headers，传
dict给headers参数。 - 对于 cookies，传
dict给cookies参数。
关于如何传数据给后端的更多信息 (使用httpx 或 TestClient)，请查阅 HTTPX 文档.
信息
注意 TestClient 接收可以被转化为JSON的数据，而不是Pydantic模型。
如果你在测试中有一个Pydantic模型，并且你想在测试时发送它的数据给应用，你可以使用在JSON Compatible Encoder介绍的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>