pytest-xdistでPytestの並列実行をしてみる

kobayashi.m

2024.02.26

はじめに

データアナリティクス事業本部のkobayashiです。

Pytestでテストするsmallテストが多数ある場合はテストを並列実行してテスト結果を早く取得することが開発スピードを上げる一つの要素になるかと思います。今回はPytestでテストを並列実行してみたのでその内容をまとめます。

pytest-xdist · PyPI

Pytestを並列実行するプラグイン

Smallテストは単一プロセス内で動作するテストなので非常に高速に実行されかつスケールさせる事ができます。そのためpytestで並列実行させるには最適なテストとなります。

pytestで並列テストを行うにはpytestのプラグインを使いますが、調べてみると以下の2つが候補に上がりました。

pytest-xdist · PyPI
pytest-parallel · PyPI

それぞれのドキュメントやリリース履歴を見た所pytest-parallelに関してはここ数年リリースがなく開発が止まっていると思われるのでpytest-xdistを今後は使用すべきかと思い今回もpytest-xdistを使っていきたいと思います。

pytest-xdistを使ってみる

環境

Python: 3.11.4
pytest: 7.4.3

インストールはいつも通りpipでインストールします。

$ pip install pytest-xdist
$  pip list | grep xdist
pytest-xdist                     3.5.0

テスト対象とテストコード

テスト対象のコードは以下のモンテカルロ法で円周率を求める関数になります。

main.py

import numpy as np


def calc_pi(sim_num):
    count = 0

    for i in range(sim_num):

        x = np.random.rand()
        y = np.random.rand()

        d = x ** 2 + y ** 2

        if d <= 1:
            count += 1

    p = count / sim_num * 4

    return p

test_main.py

import pytest

from main import calc_pi


class TestPi_1:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_1(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_2(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14


class TestPi_2:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_2(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_4(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

test_main_2.py

import pytest

from main import calc_pi


class TestPi_3:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_5(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_6(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14


class TestPi_4:
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_7(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_8(self, data_in):
        ret = calc_pi(data_in)
        assert ret > 3.14

ではこれらのコードでテストを並列で行ってみます。

並列実行でテストする

python-xdistを使って並列でテストを行うには-n autoのオプションを付けてpytestを実行するだけです。

$ pytest -v -n auto  
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw3] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw4] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
[gw1] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
[gw7] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw5] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw6] [ 87%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw2] [100%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]

[gw0]等がpytest-xdistのワーカー名になります。

-n autoで実行するとpytest-xdistが実行環境マシーンのCPUのコアと同じ数のワーカー数でテストが並列実行されます。上記のコードだと8つのワーカーで8つのテストを行っているため大幅なテスト時間の短縮になります。

ワーカー数を指定する場合は-n {ワーカー数}でワーカー数を指定することで指定のワーカー数でテストを並列実行できます。

$ pytest -v -n 2 
=========================================================================== test session starts ============================================================================
2 workers [8 items]     
scheduling tests via LoadScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
[gw1] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw0] [ 25%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw1] [ 37%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw1] [ 62%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 87%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]

このような感じでワーカーが2つでテストが並列実行されています。

テスト配布方法（クラス単位）

pytest-xdistで特にオプションを指定しない場合はテストをどのワーカーで動かすかは自動的に利用可能なワーカーに割り振られ実行されます。したがってここまでの例ではクラスごと、ファイルごとにテストを記述してありますがテストの実行順序が担保されません。（--dist loadオプションがデフォルトなため）

そこでpytest-xdistでは配布オプションを使うことでクラス、ファイル、グループ単位で同一のワーカー上でテストを行うことができるので試してみます。

はじめにクラス単位でテストを並列実行してみます。この場合は--dist loadscopeでテストを実行します。

$ pytest -v -n auto --dist loadscope    
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadScopeScheduling

test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw3] [ 12%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw2] [ 50%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw3] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 87%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw2] [100%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]

ワーカー数はautoなので8つ立ち上がっていますがgw0のワーカーでTestPi_1クラス、gw1のワーカーでTestPi_3クラス、gw2のワーカーでTestPi_2クラス、gw3のワーカーでTestPi_4クラスとテストコードのクラスが4つなので、4つのワーカーでクラスが1つづつ実行されていることがわかります。

テスト配布方法（ファイル）

次にファイル単位位でテストを並列実行してみます。この場合は--dist loadfileでテストを実行します。

 pytest -v -n auto --dist loadfile
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadFileScheduling

test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main.py::TestPi_1::test_calc_pi_1[10000000] 
[gw0] [ 12%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000] 
test_main.py::TestPi_1::test_calc_pi_2[10000000] 
[gw1] [ 25%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000] 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000] 
test_main.py::TestPi_2::test_calc_pi_3[10000000] 
[gw1] [ 50%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
[gw0] [ 62%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000] 
test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000] 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000] 
[gw0] [ 87%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000] 
[gw1] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]

こちらもワーカーは8つ立ち上がっていますがgw0のワーカーでtest_main.pyファイル、gw1のワーカーでtest_main_2.pyファイルとテストコードのファイルが2つなので、2つのワーカーでファイル単位で1つづつ実行されていることがわかります。

テスト配布方法（指定グループ）

次に指定したグループ単位でテストを並列実行してみます。この場合は--dist loadgroupでテストを実行しますが、その前にグループを分ける必要があります。先ほど作成したtest_main.py、test_main_2.pyのクラスに@pytest.mark.xdist_groupデコレータを使ってグループ分けをしてみます。

test_main.py

import pytest

from main import calc_pi

@pytest.mark.xdist_group(name="GroupA")
class TestPi_1:
...


@pytest.mark.xdist_group(name="GroupB")
class TestPi_2:
...

test_main_2.py

import pytest

from main import calc_pi


@pytest.mark.xdist_group(name="GroupA")
class TestPi_3:
...


@pytest.mark.xdist_group(name="GroupB")
class TestPi_4:
...

$ pytest -v -n auto --dist loadgroup
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadGroupScheduling

test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupB 
test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
[gw0] [ 12%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
[gw1] [ 25%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupB 
test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
[gw0] [ 37%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
[gw1] [ 50%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
[gw0] [ 62%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw1] [ 75%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 87%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw1] [100%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB

こちらもワーカーは8つ立ち上がっていますがグループを2つ作成したのでgw0のワーカーでGroupAのグループを付けたTestPi_1・TestPi_3クラス、gw1のワーカーでGroupBのデコレータを付けたTestPi_2・TestPi_4クラスとテストコードのグループが2つなので、2つのワーカーで指定したグループ単位に実行されていることがわかります。

試しに以下のようにクラスではなくメソッドでグループを指定することもできます。

test_main.py

import pytest

from main import calc_pi

@pytest.mark.xdist_group(name="GroupA")
class TestPi_1:
...


class TestPi_2:
    @pytest.mark.xdist_group(name="GroupA")
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_3(self, data_in):
...

    @pytest.mark.xdist_group(name="GroupB")
    @pytest.mark.parametrize(
        ["data_in"],
        [
            pytest.param(10000000),
        ],
    )
    def test_calc_pi_4(self, data_in):
...

この場合は以下のような実行結果になります。

$ pytest -v -n auto --dist loadgroup
=========================================================================== test session starts ============================================================================
8 workers [8 items]     
scheduling tests via LoadGroupScheduling

test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
[gw1] [ 12%] PASSED test_main.py::TestPi_2::test_calc_pi_4[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
[gw0] [ 25%] PASSED test_main.py::TestPi_1::test_calc_pi_1[10000000]@GroupA 
test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
[gw1] [ 37%] PASSED test_main_2.py::TestPi_4::test_calc_pi_7[10000000]@GroupB 
test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 50%] PASSED test_main.py::TestPi_1::test_calc_pi_2[10000000]@GroupA 
test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupA 
[gw1] [ 62%] PASSED test_main_2.py::TestPi_4::test_calc_pi_8[10000000]@GroupB 
[gw0] [ 75%] PASSED test_main.py::TestPi_2::test_calc_pi_3[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
[gw0] [ 87%] PASSED test_main_2.py::TestPi_3::test_calc_pi_5[10000000]@GroupA 
test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA 
[gw0] [100%] PASSED test_main_2.py::TestPi_3::test_calc_pi_6[10000000]@GroupA

まとめ

Pytestでテストを並列実行するpytest-xdistプラグインを使ってみました。外部に依存関係が無く1プロセス上で動くようなテストでしたら-n autoで自動分散させてしまえば効率的にテストを行えますし、ある程度順序を制御したい場合は--distで適切な分散を指定すればいいので簡単に使えるかと思います。

最後まで読んで頂いてありがとうございました。

pytest-xdistでPytestの並列実行をしてみる

はじめに

Pytestを並列実行するプラグイン

pytest-xdistを使ってみる

環境

テスト対象とテストコード

並列実行でテストする

テスト配布方法（クラス単位）

テスト配布方法（ファイル）

テスト配布方法（指定グループ）

まとめ

イベント

EVENT【5/29（水）大阪】Alteryxの使い方から導入メリットまですべてがわかるセミナー＆ワークショップ

EVENT【5/15（水）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/8リモート】クラスメソッドのフリーランスエンジニア会社説明会〜フィンテック / リテール業界案件特集〜を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

pytest-xdistでPytestの並列実行をしてみる

はじめに

Pytestを並列実行するプラグイン

pytest-xdistを使ってみる

環境

テスト対象とテストコード

並列実行でテストする

テスト配布方法（クラス単位）

テスト配布方法（ファイル）

テスト配布方法（指定グループ）

まとめ

イベント

EVENT【5/29（水）大阪】Alteryxの使い方から導入メリットまですべてがわかるセミナー＆ワークショップ

EVENT【5/15（水）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/8リモート】クラスメソッドのフリーランスエンジニア会社説明会 〜フィンテック / リテール 業界案件特集〜 を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金） 名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

関連記事

愛知県の市区町村当てクイズをPythonライブラリを活用して作ってみた

PythonでCron式から実行日付を取得する

How to Fetch Latest Objects from AWS S3 and Store Locally using Python

WindowsマシンでPython環境構築をpoetry + pyenvで再度やってみた

EVENT【5/8リモート】クラスメソッドのフリーランスエンジニア会社説明会〜フィンテック / リテール業界案件特集〜を開催します

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！