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 の組み合わせで、マーカーの順序とスコープが間違いやすい

パターン1：pytest-asyncio の基本設定

インストールと設定

pip install pytest pytest-asyncio

pyproject.toml 設定

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

基本的な非同期テスト

# test_basic.py
import asyncio
import pytest

# 方法1：auto モード — 手動マーク不要
async def test_auto_mode():
    """asyncio_mode = "auto" の場合、すべての async def テストが自動認識される"""
    await asyncio.sleep(0.1)
    result = await fetch_data()
    assert result == "expected"

# 方法2：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"

# 方法3：イベントループスコープの指定
@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"

パターン2：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)
    })()

パターン3：非同期関数と外部 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"] == "鈴木"

非同期コンテキストマネージャーのモック

@pytest.mark.asyncio
async def test_mock_async_context_manager():
    """非同期コンテキストマネージャーのモック"""
    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

非同期例外のモック

@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))

パターン4：非同期コンテキストマネージャーとジェネレーターのテスト

非同期コンテキストマネージャーのテスト

# 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)

パターン5：パラメータ化非同期テストとテスト構成

パラメータ化非同期テスト

# 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=["tanaka", "suzuki", "sato"])
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 を使用

❌ 間違った書き方：

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 のスコープを確認。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"

複数のテストが1つのイベントループを共有し、繰り返しの作成と破棄のオーバーヘッドを回避。非同期テストが多数あるプロジェクトに特に有効。

2. aioresponses で 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 と組み合わせることで、非同期テストを複数のワーカーで並列実行し、テスト時間を大幅に短縮できる。

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 非同期テストの実践において、以下の ToolsKu ツールが役立ちます：

  • JSON フォーマッター — 非同期 API が返す JSON データをフォーマットし、レスポンス構造の問題を素早く特定
  • ハッシュ計算 — 非同期キャッシュのキー名を生成し、リクエストの重複排除とキャッシュ検証を実現
  • cURL to Code — cURL コマンドを Python 非同期 HTTP リクエストコードに変換

pytest-asyncio は Python 非同期テストの事実上の標準である。設定パターン、async fixture の作成、AsyncMock の使用、パラメータ化テストの構成をマスターすれば、安定した効率的な本番級非同期テストスイートを構築できる。非同期テストの核心は、イベントループのライフサイクルを正しく管理し、同期ブロッキングを避け、AsyncMock を活用することである。

ブラウザローカルツールを無料で試す →

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