FastAPI 測試與除錯：非同步測試（pytest‑asyncio）

簡介

在使用 FastAPI 建立非同步 API 時，許多開發者習慣只測試同步路徑，卻忽略了 async 端點的特殊行為。非同步程式碼若沒有正確測試，容易在高併發或 I/O 密集的情境下出現隱藏的 race condition、資源洩漏或回傳錯誤。

pytest-asyncio 是 pytest 官方支援的非同步測試外掛，讓我們可以在 pytest 中直接以 async def 撰寫測試函式，並在事件迴圈（event loop）內執行。結合 FastAPI 的 TestClient（同步）或 AsyncClient（非同步）後，開發者可以完整驗證路由、依賴、背景任務等所有非同步流程。

本篇文章將從核心概念說明、實作範例、常見陷阱與最佳實踐，最後帶出實務應用場景，協助你在 FastAPI 專案中建立可靠且可維護的非同步測試基礎。

---

核心概念

1. 為什麼需要 pytest-asyncio

• 事件迴圈管理：普通的 pytest 執行環境是同步的，直接呼叫 await 會拋出 RuntimeError: no running event loop。pytest-asyncio 會在每個測試函式前自動建立並關閉事件迴圈。
• 測試可讀性：使用 async def test_xxx()，測試程式碼與實際非同步實作保持同樣的語意，避免在測試中加入 loop.run_until_complete() 等雜湊程式。
• 與 FastAPI 完美結合：FastAPI 本身即支援 async 路由，搭配 httpx.AsyncClient（或 AsyncClient）即可在同一個事件迴圈內完成「發送請求 → 等待回應」的全流程。

2. 基本安裝與設定

pip install pytest pytest-asyncio httpx

在 pytest.ini（或 pyproject.toml）加入：

[pytest]
asyncio_mode = auto   # 讓 pytest-asyncio 自動偵測 async 測試

3. 測試非同步路由的最小範例

# app/main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/ping")
async def ping():
    return {"msg": "pong"}

# tests/test_ping.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_ping():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/ping")
    assert response.status_code == 200
    assert response.json() == {"msg": "pong"}

重點：AsyncClient 會自動在同一個事件迴圈內啟動 FastAPI 應用，省去啟動實體伺服器的時間與資源。

4. 使用 Fixtures 管理共用資源

# tests/conftest.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.fixture
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# tests/test_user.py
import pytest

@pytest.mark.asyncio
async def test_create_user(async_client):
    payload = {"username": "alice", "email": "alice@example.com"}
    resp = await async_client.post("/users/", json=payload)
    assert resp.status_code == 201
    data = resp.json()
    assert data["username"] == "alice"
    assert "id" in data

使用 fixture 可以避免在每個測試裡重複建立 AsyncClient，同時確保資源在測試結束後正確關閉。

5. 測試 Background Tasks

FastAPI 支援 BackgroundTasks，在非同步測試中，我們需要等待背景工作完成或利用 mock 來驗證呼叫次數。

# app/tasks.py
import asyncio

async def send_email(to: str, subject: str, body: str):
    await asyncio.sleep(0.1)   # 模擬 I/O
    return True

# app/main.py (續)
from fastapi import BackgroundTasks
from .tasks import send_email

@app.post("/notify/")
async def notify(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Hello", "Welcome!")
    return {"detail": "Notification scheduled"}

# tests/test_notify.py
import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_notify_background(async_client):
    mock_send = AsyncMock(return_value=True)
    with patch("app.tasks.send_email", mock_send):
        resp = await async_client.post("/notify/", json={"email": "bob@example.com"})
        assert resp.status_code == 200
        assert resp.json() == {"detail": "Notification scheduled"}
        # 確認背景任務已被加入
        mock_send.assert_awaited_once_with("bob@example.com", "Hello", "Welcome!")

技巧：使用 unittest.mock.patch 取代實際 I/O，讓測試保持快速且不依賴外部服務。

6. 測試依賴注入（Dependency Overrides）

FastAPI 允許在測試時覆寫依賴，配合 pytest-asyncio 可完成完整的非同步依賴測試。

# app/deps.py
from fastapi import Depends

async def get_db():
    # 假設這裡回傳 async DB session
    ...

# app/main.py (續)
@app.get("/items/")
async def read_items(db = Depends(get_db)):
    result = await db.fetch_all()
    return result

# tests/test_items.py
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from app.main import app
from app.deps import get_db

class FakeDB:
    async def fetch_all(self):
        return [{"id": 1, "name": "demo"}]

@pytest.fixture
def override_get_db():
    async def _override():
        return FakeDB()
    return _override

@pytest.mark.asyncio
async def test_read_items(async_client, override_get_db):
    app.dependency_overrides[get_db] = override_get_db
    resp = await async_client.get("/items/")
    assert resp.status_code == 200
    assert resp.json() == [{"id": 1, "name": "demo"}]
    # 清除覆寫，避免影響其他測試
    app.dependency_overrides.clear()

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記加 @pytest.mark.asyncio	pytest 會把測試當成同步函式，直接拋出 RuntimeError。	必須在所有 async def 測試前加上 @pytest.mark.asyncio（或使用 asyncio_mode = auto）。
在測試中使用 asyncio.run()	會產生「已存在事件迴圈」的衝突。	交由 pytest-asyncio 管理事件迴圈，直接 await。
BackgroundTasks 未被等待	測試結束前背景工作仍在執行，導致資料不一致。	使用 mock 或在測試中手動 await 背景函式（若可取得）。
依賴覆寫未清除	其他測試會意外使用測試專用的依賴，造成結果錯亂。	測試結束後 app.dependency_overrides.clear()，或使用 fixture 自動清理。
使用同步 TestClient 測試非同步路由	會產生阻塞，失去非同步效能測試的意義。	盡可能改用 httpx.AsyncClient；若必須使用 TestClient，確認只測試 同步 部分。

最佳實踐

1. 統一使用 async client：所有與 API 互動的測試都使用 AsyncClient，保持測試環境與實際運行時一致。
2. 以 fixture 管理 client、資料庫、環境變數：減少重複程式碼，提升可讀性與維護性。
3. Mock 外部服務（如郵件、第三方 API）：使用 AsyncMock，避免測試受網路或服務可用性影響。
4. 測試例外與錯誤路徑：除了成功案例，亦要寫測試驗證 4xx、5xx 回應與自訂例外處理。
5. 持續集成 (CI) 中加入 --cov：確保非同步程式碼的測試覆蓋率不低於 80%。

---

實際應用場景

1. 高併發的即時聊天服務

   • 每條訊息透過 async DB、Redis Pub/Sub、WebSocket 推送。使用 pytest-asyncio 同時測試 API、背景任務與 WebSocket 事件，確保訊息不會遺失或重複。

2. 資料匯入/匯出工作流

   • 大量 CSV 讀寫、S3 上傳皆為 I/O 密集型。測試時可 mock S3 客戶端，並以 async 測試驗證「上傳成功」與「失敗回滾」的邏輯。

3. OAuth2 / JWT 認證流程

   • 取得 token、驗證 token、刷新 token 均涉及非同步 HTTP 請求。使用 AsyncClient 搭配 pytest-asyncio，模擬整條認證鏈，確保 token 失效或過期時的回應正確。

4. 定時任務 (APScheduler) 與 BackgroundTasks

   • 例如每日報表產生、郵件提醒。透過 async 測試驗證排程觸發、任務執行與結果寫入資料庫的完整流程。

---

總結

非同步測試 是建構高效能 FastAPI 服務不可或缺的一環。pytest-asyncio 為我們提供了 自動事件迴圈管理、簡潔的 async 測試語法，以及與 FastAPI、httpx.AsyncClient 的天然結合。透過本文的概念說明、實作範例、常見陷阱與最佳實踐，你應該能夠：

1. 快速搭建 可重用的 async 測試環境。
2. 正確驗證 非同步路由、背景任務、依賴注入與外部服務的行為。
3. 避免 事件迴圈衝突、資源泄漏與測試間相互污染的問題。

在日常開發與 CI 流程中加入完整的非同步測試，不僅提升程式碼品質，也讓專案在面對高併發與 I/O 密集的真實場景時，更加穩定、可預測。祝你在 FastAPI 的測試之路上順利前行！