測試¶
感謝 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"}
提示
請注意,測試函式是普通的 def
,而不是 async def
。
而且對 client 的呼叫也是普通的呼叫,而不是使用 await
。
這讓您可以直接使用 pytest
而不產生複雜性。
「技術細節」
您也可以使用 from starlette.testclient import TestClient
。
FastAPI 提供與 starlette.testclient
相同的 fastapi.testclient
,只是為了方便您,開發人員。但它直接來自 Starlette。
提示
如果您想在測試中呼叫 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"}
測試檔案¶
然後您可以有一個包含測試的 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
標頭。
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>