在現代軟體開發中,測試是確保程式碼品質和穩健性的關鍵環節。本文將示範如何使用 pytest 和 httpx 測試 FastAPI 應用程式,包含設定測試環境、使用 fixtures 管理測試資料,以及編寫涵蓋 CRUD 操作的測試案例。我們將使用 pytest-asyncio 執行非同步測試,並透過模擬請求和驗證回應,確保 API 的正確性和可靠性。此外,文章也將探討如何使用 fixtures 有效地管理測試資料,減少程式碼重複,並提高測試效率。最後,我們將示範如何測試受保護的路由,以及如何清理測試資料,確保測試環境的乾淨和一致性。
def add(a: int, b: int) -> int:
return a + b
def subtract(a: int, b: int) -> int:
return b - a
def multiply(a: int, b: int) -> int:
return a * b
def divide(a: int, b: int) -> int:
return b // a
測試 FastAPI 應用程式
使用 pytest 進行單元測試
單元測試是一種測試程式,用於測試應用程式的個別元件,以驗證其功能是否正確。例如,單元測試可用於測試應用程式中的個別路由,以確保傳回正確的回應。
本章將使用 pytest,一個 Python 測試函式庫,來進行單元測試操作。雖然 Python 內建了一個名為 unittest 的單元測試函式庫,但 pytest 的語法更簡潔,因此更受歡迎。
安裝 pytest
首先,安裝 pytest 函式庫:
(venv)$ pip install pytest
接下來,建立一個名為 tests 的資料夾,用於存放測試檔案:
(venv)$ mkdir tests && cd tests
(venv)$ touch __init__.py
個別測試檔名的建立將以 test_ 為字首,以便 pytest 函式庫能夠識別並執行測試檔。
編寫第一個測試
首先,定義一個執行算術運算的函式:
def add(a: int, b: int) -> int:
return a + b
def subtract(a: int, b: int) -> int:
return b - a
def multiply(a: int, b: int) -> int:
return a * b
def divide(a: int, b: int) -> int:
return b // a
接下來,編寫測試函式,使用 assert 關鍵字驗證輸出是否正確:
def test_add() -> None:
assert add(1, 1) == 2
def test_subtract() -> None:
assert subtract(2, 5) == 3
def test_multiply() -> None:
assert multiply(10, 10) == 100
def test_divide() -> None:
assert divide(25, 100) == 4
執行測試
使用 pytest 命令執行測試檔:
(venv)$ pytest test_arithmetic_operations.py
如果所有測試透過,將顯示綠色的回應。
使用 pytest fixtures 消除重複程式碼
Fixtures 是可重複使用的函式,用於傳回測試函式所需的資料。Fixtures 使用 @pytest.fixture 修飾符進行裝飾。
例如,定義一個傳回 EventUpdate pydantic 模型例項的 fixture:
import pytest
from models.events import EventUpdate
@pytest.fixture
def event() -> EventUpdate:
return EventUpdate(
title="FastAPI Book Launch",
description="We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!",
tags=["python", "fastapi", "book", "launch"],
location="Google Meet"
)
def test_event_name(event: EventUpdate) -> None:
assert event.title == "FastAPI Book Launch"
在上述程式碼中,fixture 傳回一個 EventUpdate pydantic 模型的例項,並將其作為引數傳遞給 test_event_name 函式。
設定測試環境
在本文中,我們將學習如何測試 CRUD 操作的端點以及使用者身份驗證。為了測試非同步 API,我們將使用 httpx 和 pytest-asyncio 函式庫。
安裝所需的函式庫:
(venv)$ pip install httpx pytest-asyncio
內容解密:
- pytest 的基本使用:本文介紹瞭如何使用 pytest 進行單元測試,包括安裝 pytest、編寫測試函式和執行測試。
- pytest fixtures 的使用:本文介紹瞭如何使用 pytest fixtures 消除重複程式碼,包括定義 fixtures 和使用 fixtures。
- 設定測試環境:本文介紹瞭如何設定測試環境,包括安裝所需的函式庫。
pytest 的工作流程
此圖示展示了使用 pytest 進行單元測試的工作流程。
測試 FastAPI 應用程式
設定測試環境
首先,我們需要建立一個名為 pytest.ini 的設定檔,並在其中加入以下程式碼:
[pytest]
asyncio_mode = True
這個設定檔會在執行 pytest 時被讀取,並且自動讓 pytest 以非同步模式執行所有測試。
接下來,我們需要建立一個名為 conftest.py 的測試檔案,這個檔案將負責建立我們的應用程式例項。請在 tests 資料夾中建立這個檔案:
(venv)$ touch tests/conftest.py
然後,在 conftest.py 中匯入所需的相依性:
import asyncio
import httpx
import pytest
from main import app
from database.connection import Settings
from models.events import Event
from models.users import User
定義迴圈階段固定裝置
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.get_event_loop()
yield loop
loop.close()
建立新的資料函式庫例項
async def init_db():
test_settings = Settings()
test_settings.DATABASE_URL = "mongodb://localhost:27017/testdb"
await test_settings.initialize_database()
在上述程式碼中,我們定義了一個新的 DATABASE_URL,並呼叫了第 6 章中定義的初始化函式。
定義預設客戶端固定裝置
@pytest.fixture(scope="session")
async def default_client():
await init_db()
async with httpx.AsyncClient(app=app, base_url="http://app") as client:
yield client
# 清理資源
await Event.find_all().delete()
await User.find_all().delete()
在上述程式碼中,首先初始化資料函式庫,然後以 AsyncClient 的形式啟動應用程式,並保持其存活直到測試階段結束。測試階段結束後,將清除事件和使用者集合,以確保每次測試執行前資料函式庫都是空的。
為 REST API 端點撰寫測試
測試註冊路由
首先,建立一個名為 test_login.py 的測試檔案:
(venv)$ touch tests/test_login.py
然後,在測試檔案中匯入所需的相依性:
import httpx
import pytest
接下來,定義測試函式和請求負載:
@pytest.mark.asyncio
async def test_sign_new_user(default_client: httpx.AsyncClient) -> None:
payload = {
"email": "[email protected]",
"password": "testpassword",
}
定義請求標頭和預期回應
headers = {
"accept": "application/json",
"Content-Type": "application/json"
}
test_response = {
"message": "User created successfully"
}
發起請求並測試回應
response = await default_client.post("/user/signup", json=payload, headers=headers)
assert response.status_code == 200
assert response.json() == test_response
測試登入路由
在註冊路由測試下方,定義登入路由的測試:
@pytest.mark.asyncio
async def test_sign_user_in(default_client: httpx.AsyncClient) -> None:
payload = {
"username": "[email protected]",
"password": "testpassword"
}
headers = {
"accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded"
}
發起請求並測試回應
response = await default_client.post("/user/signin", data=payload, headers=headers)
assert response.status_code == 200
assert response.json()["token_type"] == "Bearer"
測試 CRUD 端點
首先,建立一個名為 test_routes.py 的新檔案:
(venv)$ touch test_routes.py
然後,在新檔案中加入以下程式碼:
import httpx
import pytest
from auth.jwt_handler import create_access_token
from models.events import Event
詳細內容解說:
pytest.ini設定檔:此設定檔讓pytest以非同步模式執行所有測試。conftest.py測試檔案:負責建立應用程式例項,定義迴圈階段固定裝置、新的資料函式庫例項和預設客戶端固定裝置。test_login.py測試檔案:包含註冊路由和登入路由的測試,使用httpx.AsyncClient發起請求並驗證回應。test_routes.py測試檔案:用於測試 CRUD 端點,匯入必要的相依性。
這些步驟和程式碼片段共同構成了對 FastAPI 應用程式進行全面測試的基礎。透過這些測試,可以確保應用程式的各個端點功能正常且穩定。
圖表說明:測試流程圖示
@startuml
skinparam backgroundColor #FEFEFE
skinparam sequenceArrowThickness 2
title FastAPI 應用程式單元測試實踐
actor "客戶端" as client
participant "API Gateway" as gateway
participant "認證服務" as auth
participant "業務服務" as service
database "資料庫" as db
queue "訊息佇列" as mq
client -> gateway : HTTP 請求
gateway -> auth : 驗證 Token
auth --> gateway : 認證結果
alt 認證成功
gateway -> service : 轉發請求
service -> db : 查詢/更新資料
db --> service : 回傳結果
service -> mq : 發送事件
service --> gateway : 回應資料
gateway --> client : HTTP 200 OK
else 認證失敗
gateway --> client : HTTP 401 Unauthorized
end
@enduml
此圖示展示了從設定測試環境到執行各項測試的流程。透過這個流程,可以確保 FastAPI 應用程式的各個元件正確無誤地運作。
重點回顧:
- 正確設定
pytest.ini以啟用非同步模式。 - 使用
conftest.py建立應用程式例項和固定裝置。 - 為註冊和登入路由撰寫全面的非同步測試。
- 利用
httpx.AsyncClient發起請求並驗證回應。
這些技術和最佳實踐將有助於確保 FastAPI 應用程式的長期穩定性和可維護性。
為REST API端點撰寫測試
在前面的程式碼區塊中,我們已經匯入了常規的相依性。同時也匯入了create_access_token函式和Event模型。由於某些路由受到保護,我們將自行產生存取權杖。讓我們建立一個新的fixture,它在被呼叫時傳回一個存取權杖。這個fixture的作用域是模組層級,這意味著它只會在測試檔案被執行時執行一次,而不是在每次函式呼叫時都被呼叫。新增以下程式碼:
@pytest.fixture(scope="module")
async def access_token() -> str:
return create_access_token("[email protected]")
接下來,讓我們建立一個新的fixture,它會在資料函式庫中新增一個事件。在測試CRUD端點之前執行這個動作,以進行初步測試。新增以下程式碼:
@pytest.fixture(scope="module")
async def mock_event() -> Event:
new_event = Event(
creator="[email protected]",
title="FastAPI Book Launch",
image="https://linktomyimage.com/image.png",
description="We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
tags=["python", "fastapi", "book", "launch"],
location="Google Meet"
)
await Event.insert_one(new_event)
yield new_event
測試讀取端點
接下來,讓我們撰寫一個測試函式,用於測試/event路由上的GET HTTP方法:
@pytest.mark.asyncio
async def test_get_events(default_client: httpx.AsyncClient, mock_event: Event) -> None:
response = await default_client.get("/event/")
assert response.status_code == 200
assert response.json()[0]["_id"] == str(mock_event.id)
內容解密:
- 這個測試函式驗證了
/event端點是否能夠正確傳回事件列表。 default_client是用於傳送HTTP請求的客戶端。mock_event是預先在資料函式庫中建立的事件,用於驗證傳回結果。
測試建立端點
讓我們撰寫一個測試函式,用於測試建立新事件的端點:
@pytest.mark.asyncio
async def test_post_event(default_client: httpx.AsyncClient, access_token: str) -> None:
payload = {
"title": "FastAPI Book Launch",
"image": "https://linktomyimage.com/image.png",
"description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
"tags": ["python", "fastapi", "book", "launch"],
"location": "Google Meet",
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
test_response = {
"message": "Event created successfully"
}
response = await default_client.post("/event/new", json=payload, headers=headers)
assert response.status_code == 200
assert response.json() == test_response
內容解密:
- 這個測試函式驗證了
/event/new端點是否能夠正確建立新事件。 access_token是用於身份驗證的存取權杖。payload是傳送到伺服器的請求主體,包含了新事件的詳細資訊。
測試更新和刪除端點
讓我們繼續撰寫測試函式,用於測試更新和刪除事件的端點:
@pytest.mark.asyncio
async def test_update_event(default_client: httpx.AsyncClient, mock_event: Event, access_token: str) -> None:
test_payload = {
"title": "Updated FastAPI event"
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
url = f"/event/{str(mock_event.id)}"
response = await default_client.put(url, json=test_payload, headers=headers)
assert response.status_code == 200
assert response.json()["title"] == test_payload["title"]
內容解密:
- 這個測試函式驗證了
/event/{id}端點是否能夠正確更新事件。 mock_event是用於取得事件ID的fixture。
@pytest.mark.asyncio
async def test_delete_event(default_client: httpx.AsyncClient, mock_event: Event, access_token: str) -> None:
test_response = {
"message": "Event deleted successfully."
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
url = f"/event/{mock_event.id}"
response = await default_client.delete(url, headers=headers)
assert response.status_code == 200
assert response.json() == test_response
內容解密:
- 這個測試函式驗證了
/event/{id}端點是否能夠正確刪除事件。