Python Pytest 非同步測試完全指南：pytest-asyncio 實戰 2026

编程语言

非同步測試的痛點

當你用 asyncio 寫了一個漂亮的非同步服務，信心滿滿地準備寫測試時，卻發現——普通的 pytest 測試函數根本跑不了 async def。你嘗試手動 asyncio.run()，fixture 裡的非同步初始化又成了問題；你想 Mock 一個非同步函數，unittest.mock.patch 卻告訴你協程物件不是可呼叫的；你寫了參數化測試，結果事件循環衝突導致用例全部報錯。非同步測試的複雜度遠超同步測試，配置不當甚至會讓測試套件完全無法執行。本文將系統講解 pytest-asyncio 的配置、模式和最佳實踐，幫你徹底解決 Python 非同步測試難題。

核心概念

概念 說明
pytest-asyncio pytest 的非同步測試外掛，自動為標記的測試函數建立事件循環
@pytest.mark.asyncio 標記測試函數為非同步，pytest-asyncio 會自動在事件循環中執行
async fixture 使用 async def 定義的 fixture，支援非同步資源初始化和清理
event_loop fixture pytest-asyncio 內建 fixture，提供測試用的事件循環實例
asyncio_mode 配置項，控制 pytest-asyncio 的行為模式（auto/strict）
AsyncMock unittest.mock 提供的非同步 Mock 物件，用於替換非同步函數
scope fixture 作用域，控制非同步 fixture 的生命週期（function/session）

問題分析：非同步測試的 5 大挑戰

  1. 事件循環管理混亂：每個測試需要獨立的事件循環，循環復用或衝突會導致狀態污染和不可預期的失敗
  2. 非同步 Fixture 難以編寫：資料庫連線、HTTP 客戶端等非同步資源的初始化和清理需要特殊處理
  3. 非同步 Mock 行為異常：普通 Mock 無法正確模擬協程函數，回傳值是協程物件而非實際結果
  4. 測試執行順序不確定：非同步測試並發執行時，共享狀態的測試可能因執行順序不同而失敗
  5. 參數化與非同步衝突：@pytest.mark.parametrize 與 @pytest.mark.asyncio 組合使用時標記順序和作用域容易出錯

模式一：pytest-asyncio 基礎配置

安裝與配置

pip install pytest pytest-asyncio

pyproject.toml 配置

[tool.pytest.ini_options]
asyncio_mode = "auto"

基礎非同步測試

# test_basic.py
import asyncio
import pytest

# 方式一：auto 模式，無需手動標記
async def test_auto_mode():
    """asyncio_mode = "auto" 時，所有 async def 測試自動識別"""
    await asyncio.sleep(0.1)
    result = await fetch_data()
    assert result == "expected"

# 方式二：strict 模式，需要顯式標記
@pytest.mark.asyncio
async def test_explicit_mark():
    """asyncio_mode = "strict" 時，必須使用 @pytest.mark.asyncio"""
    await asyncio.sleep(0.1)
    assert True

async def fetch_data():
    await asyncio.sleep(0.01)
    return "expected"

# 方式三：指定事件循環策略
@pytest.mark.asyncio(loop_scope="session")
async def test_session_loop():
    """使用 session 級別的事件循環，跨測試共享"""
    await asyncio.sleep(0.01)
    assert True

conftest.py 全域配置

# conftest.py
import asyncio
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    """自訂事件循環策略"""
    return asyncio.DefaultEventLoopPolicy()

@pytest.fixture(scope="session")
def loop_scope():
    """設定事件循環作用域為 session"""
    return "session"

模式二：Async Fixture 與依賴注入

基礎非同步 Fixture

# conftest.py
import asyncio
import pytest
from typing import AsyncGenerator

@pytest.fixture
async def db_connection() -> AsyncGenerator:
    """非同步資料庫連線 fixture"""
    conn = await create_connection("postgresql://localhost/test")
    yield conn
    await conn.close()

@pytest.fixture
async def http_client() -> AsyncGenerator:
    """非同步 HTTP 客戶端 fixture"""
    import httpx
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        yield client

async def create_connection(dsn: str):
    """模擬非同步資料庫連線"""
    await asyncio.sleep(0.01)
    return {"dsn": dsn, "status": "connected"}

# test_fixtures.py
@pytest.mark.asyncio
async def test_with_db(db_connection):
    """使用非同步 fixture 的測試"""
    assert db_connection["status"] == "connected"
    result = await db_query(db_connection, "SELECT 1")
    assert result is not None

@pytest.mark.asyncio
async def test_with_http_client(http_client):
    """使用非同步 HTTP 客戶端 fixture 的測試"""
    response = await http_client.get("/api/health")
    assert response.status_code == 200

async def db_query(conn, sql):
    await asyncio.sleep(0.01)
    return {"sql": sql, "result": True}

Fixture 依賴鏈

# conftest.py
@pytest.fixture
async def app():
    """應用實例 fixture"""
    app = await create_app()
    yield app
    await app.shutdown()

@pytest.fixture
async def db_connection(app):
    """依賴 app fixture 的資料庫連線"""
    conn = await app.get_db()
    yield conn
    await conn.close()

@pytest.fixture
async def user_service(db_connection):
    """依賴 db_connection 的使用者服務"""
    return UserService(db_connection)

# test_service.py
@pytest.mark.asyncio
async def test_create_user(user_service):
    user = await user_service.create("test@example.com", "password123")
    assert user.email == "test@example.com"

class UserService:
    def __init__(self, db):
        self.db = db

    async def create(self, email, password):
        await asyncio.sleep(0.01)
        return type("User", (), {"email": email})()

async def create_app():
    await asyncio.sleep(0.01)
    return type("App", (), {
        "get_db": lambda self: create_connection("postgresql://localhost/test"),
        "shutdown": lambda self: asyncio.sleep(0.01)
    })()

模式三：Mock 非同步函數與外部 API

AsyncMock 基礎

# test_mock.py
import asyncio
from unittest.mock import AsyncMock, patch, MagicMock
import pytest
import httpx

async def fetch_user(user_id: int):
    """被測函數：從外部 API 取得使用者"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

@pytest.mark.asyncio
async def test_fetch_user_mocked():
    """使用 AsyncMock 替換非同步函數"""
    mock_get = AsyncMock(return_value={"id": 1, "name": "張三"})

    with patch("httpx.AsyncClient.get", mock_get):
        result = await fetch_user(1)

    assert result == {"id": 1, "name": "張三"}
    mock_get.assert_called_once()

@pytest.mark.asyncio
async def test_fetch_user_side_effect():
    """使用 side_effect 模擬多次呼叫的不同回傳值"""
    mock_get = AsyncMock(side_effect=[
        {"id": 1, "name": "張三"},
        {"id": 2, "name": "李四"},
    ])

    with patch("httpx.AsyncClient.get", mock_get):
        result1 = await fetch_user(1)
        result2 = await fetch_user(2)

    assert result1["name"] == "張三"
    assert result2["name"] == "李四"

Mock 非同步上下文管理器

@pytest.mark.asyncio
async def test_mock_async_context_manager():
    """Mock 非同步上下文管理器"""
    mock_client = AsyncMock()
    mock_client.get.return_value = MagicMock(
        status_code=200,
        json=lambda: {"status": "ok"}
    )
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=False)

    with patch("httpx.AsyncClient", return_value=mock_client):
        async with httpx.AsyncClient() as client:
            response = await client.get("/health")
            assert response.status_code == 200

Mock 非同步異常

@pytest.mark.asyncio
async def test_fetch_user_timeout():
    """測試非同步超時異常處理"""
    mock_get = AsyncMock(side_effect=httpx.TimeoutException("Request timeout"))

    with patch("httpx.AsyncClient.get", mock_get):
        with pytest.raises(httpx.TimeoutException):
            await fetch_user(1)

@pytest.mark.asyncio
async def test_fetch_user_retry():
    """測試非同步重試邏輯"""
    call_count = 0

    async def failing_then_success(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise httpx.TimeoutException("timeout")
        return MagicMock(status_code=200, json=lambda: {"id": 1})

    mock_get = AsyncMock(side_effect=failing_then_success)

    with patch("httpx.AsyncClient.get", mock_get):
        result = await fetch_with_retry(1, max_retries=3)
        assert result["id"] == 1

async def fetch_with_retry(user_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(f"https://api.example.com/users/{user_id}")
                return response.json()
        except httpx.TimeoutException:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(0.1 * (attempt + 1))

模式四：測試非同步上下文管理器與生成器

非同步上下文管理器測試

# test_context_manager.py
import asyncio
import pytest

class AsyncDatabasePool:
    def __init__(self, dsn):
        self.dsn = dsn
        self._connections = []

    async def __aenter__(self):
        await asyncio.sleep(0.01)
        self._pool = {"status": "open", "size": 10}
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await asyncio.sleep(0.01)
        self._pool = None

    async def acquire(self):
        await asyncio.sleep(0.01)
        if self._pool is None:
            raise RuntimeError("Pool is closed")
        return {"id": len(self._connections), "pool": self.dsn}

@pytest.mark.asyncio
async def test_async_context_manager():
    """測試非同步上下文管理器的進入和退出"""
    pool = AsyncDatabasePool("postgresql://localhost/test")
    async with pool as p:
        assert p._pool["status"] == "open"
        conn = await p.acquire()
        assert conn["id"] == 0

    assert pool._pool is None

@pytest.mark.asyncio
async def test_async_context_manager_exception():
    """測試非同步上下文管理器異常時的清理"""
    pool = AsyncDatabasePool("postgresql://localhost/test")
    with pytest.raises(ValueError):
        async with pool as p:
            raise ValueError("test error")

    assert pool._pool is None

非同步生成器測試

# test_async_generator.py
async def async_stream_data(count: int):
    """非同步生成器：模擬串流資料"""
    for i in range(count):
        await asyncio.sleep(0.01)
        yield {"index": i, "data": f"item_{i}"}

@pytest.mark.asyncio
async def test_async_generator():
    """測試非同步生成器的輸出"""
    results = []
    async for item in async_stream_data(3):
        results.append(item)

    assert len(results) == 3
    assert results[0] == {"index": 0, "data": "item_0"}
    assert results[2] == {"index": 2, "data": "item_2"}

@pytest.mark.asyncio
async def test_async_generator_with_anext():
    """使用 anext 測試非同步生成器"""
    gen = async_stream_data(2)
    first = await anext(gen)
    assert first["index"] == 0
    second = await anext(gen)
    assert second["index"] == 1

    with pytest.raises(StopAsyncIteration):
        await anext(gen)

模式五：參數化非同步測試與組織

參數化非同步測試

# test_parametrize.py
import pytest

async def compute(x: int, y: int) -> int:
    await asyncio.sleep(0.01)
    return x + y

@pytest.mark.asyncio
@pytest.mark.parametrize("x, y, expected", [
    (1, 2, 3),
    (10, 20, 30),
    (-1, 1, 0),
    (100, 200, 300),
])
async def test_compute_parametrize(x, y, expected):
    """參數化非同步測試"""
    result = await compute(x, y)
    assert result == expected

@pytest.mark.asyncio
@pytest.mark.parametrize("input_data", [
    {"name": "張三", "age": 25},
    {"name": "李四", "age": 30},
    {"name": "王五", "age": 35},
], ids=["zhangsan", "lisi", "wangwu"])
async def test_create_user_parametrize(input_data):
    """參數化非同步測試：自訂 ID"""
    result = await create_user(input_data)
    assert result["name"] == input_data["name"]

async def create_user(data):
    await asyncio.sleep(0.01)
    return {**data, "id": 1}

測試類別組織

# test_user_api.py
import pytest

@pytest.mark.asyncio
class TestUserAPI:
    """非同步測試類別組織"""

    async def test_list_users(self):
        users = await list_users()
        assert len(users) > 0

    async def test_get_user(self):
        user = await get_user(1)
        assert user["id"] == 1

    async def test_create_user(self):
        user = await create_user({"name": "test"})
        assert user["name"] == "test"

    @pytest.mark.parametrize("user_id", [1, 2, 3])
    async def test_get_user_parametrize(self, user_id):
        user = await get_user(user_id)
        assert user["id"] == user_id

async def list_users():
    await asyncio.sleep(0.01)
    return [{"id": 1}, {"id": 2}]

async def get_user(user_id):
    await asyncio.sleep(0.01)
    return {"id": user_id}

async def create_user(data):
    await asyncio.sleep(0.01)
    return {**data, "id": 99}

陷阱指南

陷阱 1：忘記標記非同步測試

❌ 錯誤寫法：

# strict 模式下，忘記 @pytest.mark.asyncio
async def test_something():
    result = await fetch_data()
    assert result is not None
# 報錯：coroutine was never awaited

✅ 正確寫法：

@pytest.mark.asyncio
async def test_something():
    result = await fetch_data()
    assert result is not None

陷阱 2：在同步 Fixture 中回傳協程

❌ 錯誤寫法：

@pytest.fixture
def db_connection():
    return create_connection("postgresql://localhost/test")
    # 回傳的是協程物件，不是連線

✅ 正確寫法：

@pytest.fixture
async def db_connection():
    conn = await create_connection("postgresql://localhost/test")
    yield conn
    await conn.close()

陷阱 3：Mock 非同步函數使用普通 Mock

❌ 錯誤寫法：

from unittest.mock import MagicMock

mock_fetch = MagicMock(return_value={"id": 1})
# 呼叫 mock_fetch() 回傳 MagicMock，不是協程
result = await mock_fetch()  # TypeError

✅ 正確寫法：

from unittest.mock import AsyncMock

mock_fetch = AsyncMock(return_value={"id": 1})
result = await mock_fetch()
assert result == {"id": 1}

陷阱 4：在非同步測試中使用 time.sleep

❌ 錯誤寫法：

@pytest.mark.asyncio
async def test_with_delay():
    time.sleep(1)  # 阻塞事件循環！
    result = await fetch_data()
    assert result is not None

✅ 正確寫法：

@pytest.mark.asyncio
async def test_with_delay():
    await asyncio.sleep(1)  # 非阻塞等待
    result = await fetch_data()
    assert result is not None

陷阱 5：參數化標記順序錯誤

❌ 錯誤寫法：

@pytest.mark.asyncio
@pytest.mark.parametrize("x", [1, 2, 3])
# parametrize 在 asyncio 之後，可能導致標記不生效
async def test_value(x):
    assert x > 0

✅ 正確寫法：

@pytest.mark.parametrize("x", [1, 2, 3])
@pytest.mark.asyncio
# parametrize 在外層，asyncio 在內層
async def test_value(x):
    assert x > 0

錯誤排查表

序號 報錯資訊 原因 解決方法
1 coroutine was never awaited 非同步函數未在事件循環中執行 新增 @pytest.mark.asyncio 或設定 asyncio_mode = "auto"
2 RuntimeError: no running event loop 在同步上下文中呼叫了 await 確保測試函數是 async def 並正確標記
3 Fixture "xxx" is async but test is not 同步測試使用了非同步 fixture 將測試改為 async def 並新增 @pytest.mark.asyncio
4 TypeError: object MagicMock can't be used in await expression 使用普通 Mock 替代非同步函數 改用 AsyncMock 替換非同步函數
5 DeprecationWarning: The 'event_loop' fixture is deprecated 使用了舊版 event_loop fixture 升級 pytest-asyncio 到 0.23+，使用 loop_scope 參數
6 assert False on asyncio_mode="strict" strict 模式下未標記非同步測試 新增 @pytest.mark.asyncio 或切換為 auto 模式
7 RuntimeError: Event loop is closed 測試結束後事件循環被關閉 檢查 fixture scope，避免 session 級 fixture 使用 function 級循環
8 pluggy.PluggyTeardownError 非同步 fixture 清理失敗 確保 yield 後的清理程式碼也是非同步的，使用 async with
9 TimeoutError in async test 非同步測試超時 使用 @pytest.mark.timeout 或 asyncio.wait_for 控制超時
10 AssertionError: expected coroutine, got MagicMock Mock 回傳值型別不匹配 使用 AsyncMock(return_value=...) 確保回傳協程

進階最佳化

1. 使用 session 級事件循環提升效能

# conftest.py
@pytest.fixture(scope="session")
def loop_scope():
    return "session"

多個測試共享一個事件循環，避免重複建立和銷毀的開銷，特別適合有大量非同步測試的專案。

2. 使用 aioresponses Mock HTTP 請求

from aioresponses import aioresponses

@pytest.mark.asyncio
async def test_with_aioresponses():
    with aioresponses() as m:
        m.get("https://api.example.com/users/1", payload={"id": 1, "name": "張三"})
        result = await fetch_user(1)
        assert result["name"] == "張三"

3. 自訂非同步測試超時

@pytest.mark.asyncio
async def test_with_custom_timeout():
    async with asyncio.timeout(2.0):
        result = await slow_operation()
        assert result is not None

async def slow_operation():
    await asyncio.sleep(0.5)
    return "done"

4. 平行執行非同步測試

pip install pytest-xdist
pytest -n auto  # 自動平行執行

結合 pytest-xdist 可以讓非同步測試在多個 worker 中平行執行，大幅縮短測試時間。

5. 使用 hypothesis 進行非同步屬性測試

from hypothesis import given, strategies as st

@pytest.mark.asyncio
@given(st.integers(min_value=0, max_value=100))
async def test_async_property(x):
    result = await compute(x, x)
    assert result == x * 2

async def compute(a, b):
    await asyncio.sleep(0.001)
    return a + b

方案對比

特性 pytest-asyncio asyncio.run() unittest.IsolatedAsyncioTestCase 手動事件循環
配置複雜度 低（安裝即用） 無需配置 中（需繼承類別） 高（手動管理）
Fixture 支援 完整支援 不支援 有限支援 不支援
參數化測試 完整支援 需手動包裝 有限支援 需手動包裝
Mock 支援 AsyncMock 完美配合 需手動處理 需手動處理 需手動處理
測試隔離 自動隔離 手動隔離 自動隔離 手動隔離
平行執行 支援 xdist 不支援 不支援 不支援
社群生態 豐富 Python 內建 Python 內建 無
推薦場景 生產級非同步測試 簡單腳本驗證 unittest 專案遷移 不推薦

工具推薦

在 Python pytest 非同步測試實踐中，以下 工具庫 工具可以幫到你：

  • JSON 格式化 — 格式化非同步 API 回傳的 JSON 資料，快速排查回應結構問題
  • Hash 計算 — 為非同步快取產生鍵名，實現請求去重和快取驗證
  • cURL 轉程式碼 — 將 cURL 命令轉換為 Python 非同步 HTTP 請求程式碼

pytest-asyncio 是 Python 非同步測試的事實標準。掌握其配置模式、async fixture 編寫、AsyncMock 使用和參數化測試組織，你就能建構穩定、高效的生產級非同步測試套件。記住：非同步測試的核心是正確管理事件循環生命週期，避免同步阻塞，善用 AsyncMock。

本站提供瀏覽器本地工具，免註冊即可試用 →

#Python测试#pytest异步#asyncio测试#pytest-asyncio#2026#编程语言