Posted inPython

Pytest-xdist：Pythonでテストを並列実行し、テストスイートを4倍高速化する

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Pythonは日常のほぼすべての自動化タスクに使っていて、デプロイスクリプトから監視アラートまで幅広く活用しています。プロジェクトが大きくなるにつれ、テストスイートは50件から300件超に増え、git pushのたびにCI/CDが10分近く回り続けるのをただ眺めるはめになりました。コードにバグがあるわけではなく、純粋にテストが遅かったのです。

実際の問題：300件のテストスイートが毎回のpushで8分を消費

具体的なプロジェクトとして、ファイル処理・サードパーティAPI呼び出し・データベースクエリなど複数モジュールで構成されたPythonツール群があります。テストスイートは約320件で、そのうち60%はモックHTTPとファイルI/Oを使った複雑なロジックの結合テストです。

フルテスト実行の結果：

$ pytest --tb=no -q
320 passed in 487.23s (8 minutes 7 seconds)

8分です。1日10〜15回のpushで計算すると、テスト待機だけで1時間以上かかります。CI/CDではCIマシンのコア数が開発機より少ないことが多く、さらに悪化します。

原因の分析：Pytestは設計上シーケンシャルに実行される

Pytestはデフォルトでテストを1件ずつ順番に実行します。一貫性の確保とデバッグのしやすさを優先した意図的な設計ですが、マルチコアCPUを活かせません。

8コアマシンでテスト実行中のリソースモニターを見てみると：

  • CPU全体：12〜15%（1コアが稼働、7コアがアイドル）
  • ディスクI/O：ファイル読み書きのテストごとにスパイク発生
  • ネットワーク：モックHTTPレスポンスへの散発的なアクセス

問題は明確です。テストはCPUをほとんど使わず、主にI/O待ちが発生しています——モックレスポンス待ち、ファイル書き込み待ち、DBクエリの応答待ちです。その待機中、残り7コアのCPUは何もしていません。典型的なリソースの無駄遣いです。

解決方法

方法1：個別テストの最適化

xdistを使うかどうかに関わらず、まず取り組むべきステップです。よく使われるテクニックを紹介します：

import pytest

# 必要に応じてスキップできるよう遅いテストにマークを付ける
@pytest.mark.slow
def test_heavy_integration():
    result = call_real_api()  # 2〜3秒かかる
    assert result["status"] == "ok"
# ローカル開発：遅いテストをスキップ
pytest -m "not slow"

# CI全実行：すべてのテストを実行
pytest

最適化後は8分から約6分に短縮できます。それでもまだ遅く、十分ではありません。

方法2：変更に関連するテストの一部だけを実行

# 修正したモジュールのテストのみ実行
pytest tests/test_api_client.py

# キーワードで絞り込んで実行
pytest -k "upload or download"

# 前回失敗したテストを優先実行
pytest --lf  # 前回失敗したテスト
pytest --lf --nf  # 前回失敗 + 新しいファイル

ローカル開発では有効ですが、CI/CDではフルスイートを実行する必要があります。根本的な問題の解決にはなりません。

方法3：pytest-xdistによる並列実行

pytest-xdistは複数のワーカープロセスを起動し、各ワーカーがテストスイートのサブセットを受け取って並列に実行します。シーケンシャルな待機をなくし、すべてのCPUコアを同時に稼働させる根本的な解決策です。

ベストな方法：インストールから実践まで徹底解説

インストール

pip install pytest-xdist

# またはrequirements-dev.txtに追加
pytest-xdist>=3.0

基本的な使い方

# CPUコア数を自動検出 — 最もよく使うコマンド
pytest -n auto

# ワーカー数を固定で指定
pytest -n 4

# 詳細出力と組み合わせる
pytest -n auto -v --tb=short

有効化直後の結果：

320 passed in 127.41s (2 minutes 7 seconds)
# 8分 → 2分、約4倍高速化

–distによるテスト配分戦略

ここが一番調べるのに時間がかかった部分です。pytest-xdistには複数の配分戦略があります：

# load（デフォルト）：空きワーカーが次のテストを受け取る
pytest -n 4 --dist=load

# loadscope：クラスまたはモジュールでグループ化 — クラススコープfixture使用時に有効
pytest -n 4 --dist=loadscope

# loadfile：同一ファイルのテストを同じワーカーで実行
pytest -n 4 --dist=loadfile

複雑なクラスfixtureが多いプロジェクトでは--dist=loadfileを使っています。同一ファイルのテストが別々のワーカーに分散されないため、競合状態を大幅に回避できます。

FixtureとXdist：最も注意すべきポイント

xdistを使い始めたときに最もバグを起こしやすい箇所です。Fixture scope="session"は、各ワーカーで独立して実行され、セッション全体で共有されるわけではありません。

# 競合を避けるため各ワーカーに専用ポートを割り当てる
@pytest.fixture(scope="session")
def server_port(worker_id):
    # worker_id："gw0"、"gw1"、"gw2"...、xdist未使用時は"master"
    base_port = 8000
    if worker_id == "master":
        return base_port
    worker_num = int(worker_id.replace("gw", ""))
    return base_port + worker_num  # gw0=8000, gw1=8001, gw2=8002...

データベーステストでは、各テストが独自のDBを持てるようtmp_pathを使っています：

@pytest.fixture
def db(tmp_path):
    db_path = tmp_path / "test.db"
    conn = sqlite3.connect(str(db_path))
    yield conn
    conn.close()
    # tmp_pathはテスト終了後に自動削除される

pytest-covとの組み合わせ

pip install pytest-cov

.coveragercファイル：

[run]
concurrency = multiprocessing
parallel = true

[report]
omit =
    */tests/*
    */venv/*
# カバレッジを有効にしてテスト実行
pytest -n auto --cov=src --cov-report=html

# 複数ワーカーのデータを結合してレポートを出力
coverage combine
coverage report -m

実践的なtips

テストは完全にステートレスである必要があります — これは大前提です：

# NG：グローバル状態がワーカー間で共有される
counter = 0

def test_increment():
    global counter
    counter += 1
    assert counter == 1  # 並列実行時に失敗する！

# OK：各テストが独自の状態を管理する
def test_increment():
    counter = 0
    counter += 1
    assert counter == 1

ポートやファイルパスをハードコードしない：

import socket

def get_free_port() -> int:
    with socket.socket() as s:
        s.bind(('', 0))
        return s.getsockname()[1]

@pytest.fixture
def app_port():
    return get_free_port()  # 各ワーカーがランダムなポートを取得し競合しない

固定ファイルの代わりにtmp_pathを使う：

def test_file_processing(tmp_path):
    input_file = tmp_path / "input.csv"
    input_file.write_text("col1,col2\n1,2\n3,4")

    result = process_csv(input_file)
    assert len(result) == 2  # データ2行

最も遅いテストを特定するプロファイリング：

# 最も遅い20件のテストを表示
pytest --durations=20

# テストの実行順依存を検出するpytest-randomlyをインストール
pip install pytest-randomly
pytest -n auto  # テストがランダムにシャッフルされる

実際の結果とxdist有効化前のチェックリスト

pytest -n auto --dist=loadfileを適用した結果、8コアマシンでテストスイートが8分→約2分に短縮されました。CIパイプラインは30分から約12分に削減されています（残りの時間はDockerビルドとデプロイです）。

xdistを有効にする前に、以下の項目を素早く確認しましょう：

  • 各テストが共有可変状態を使用していない（グローバル変数、クラス属性）
  • テスト内でポートやファイルパスをハードコードしていない
  • Fixture scope="session"が他のテストに影響するサイドエフェクトを持っていない
  • データベーステストがトランザクションロールバックまたはin-memory DB / tmp_pathを個別に使用している
  • 他のテストの実行順序に依存するテストが存在しない

並列実行でテストが壊れる場合は、-n 1で再実行してシーケンシャル時にテストがパスすることを確認してください。-n 1でパスして-n autoで失敗するなら、共有状態またはポート/ファイルの競合が原因と断定できます。

Share:
Tags: