pytest-xdistでテストを並列実行するとき、sessionのfixtureを1度だけ実行させる

pytestでテストを並列実行するときの参考にどうぞ。
2024.03.27

pytestのプラグインであるpytest-xdistを利用すると、テストを並列実行できます。これにより、テスト時間の短縮が見込めます。

しかし、並列実行するため、sessionスコープ(1度だけ実行される)のfixtureも並列数だけ実行されます。 そのため、次のような場合で課題となる可能性もあります。

  • テスト用データの用意などで、時間がかかる
  • 外部APIを利用するため、rate limitの上限が気になる
  • など

たとえば、APIの認可にAuth0を利用している場合、E2Eテスト用のアクセストークンの取得(M2M認証)は、月1000回の制限があります(プランにもよる)。

01_auth0_price

テストの実行回数や並列数にもよりますが、少し心配になります。(1日あたり50回&10並列としたとき、1日あたり5回まで実行できる)

そこで、並列実行時でもsessionスコープのfixtureを1度だけ実行する方法を試してみました。

おすすめの方

  • pytestで並列実行したい方
  • pytestの並列実行でsessionスコープのfixtureを1度だけ実行したい方

本記事の補足

公式ドキュメントの紹介を参考にしています。

1度だけ実行するsessionスコープのfixtureは、Auth0のアクセストークン取得を想定します。 そのため、各プロセスで同じ値を参照させたいですが、環境変数にアクセストークンを保存するのはセキュリティ的に怖いため、ローカルのtempフォルダに保存します。 セキュリティをさらに考慮する場合は、一時保存場所をAWS Systems Managerのパラメータストアなどにすると良さそうです。

実験環境を構築する

python -m venv .venv
source .venv/bin/activate

pip install pytest pytest-xdist filelock

並列実行したとき、sessionのfixtureが複数実行することを確認する(アクセストークン取得)

pytest-xdistを利用すると通常のprintが表示されないため、実験用のログ確認のためstderrに出力しています。

conftest.py

1度だけ実行したいsessionの関数として、「token()」を用意しました。 Auth0からE2Eテスト用のアクセストークンを取得する想定です。

また、sessionとfunctionのスコープを毎回実行させています。

conftest.py

import random
import string
import sys

import pytest

@pytest.fixture(scope="session")
def token():
    # tokenを取得したことにする
    return "".join(random.choices(string.ascii_letters + string.digits, k=5))

@pytest.fixture(scope="session", autouse=True)
def my_setup_session():
    print("my_setup_session", file=sys.stderr)

@pytest.fixture(scope="function", autouse=True)
def my_setup_function():
    print("my_setup_function", file=sys.stderr)

test_e2e.py

E2Eテスト用のコードです。受け取ったtokenを出力するだけです。 本来はこの部分で、任意のAPIに対してリクエストする想定です。

test_e2e.py

import sys

import pytest

def test_1(token):
    print(token, file=sys.stderr)
    assert True

def test_2(token):
    print(token, file=sys.stderr)
    assert True

def test_3(token):
    print(token, file=sys.stderr)
    assert True

def test_4(token):
    print(token, file=sys.stderr)
    assert True

def test_5(token):
    print(token, file=sys.stderr)
    assert True

テストを並列実行する

worker数は3つにしてみました。

pytest -v --capture=no -n 3

結果は下記です。改行的な問題で見づらいですが、sessionの関数が3回実行されており、tokenの値も3種類あります。

plugins: xdist-3.4.0
3 workers [5 items]     
scheduling tests via LoadScheduling
my_setup_session

test_e2e.py::test_3 my_setup_session
my_setup_session
my_setup_function
my_setup_function
my_setup_function

test_e2e.py::test_2 
test_e2e.py::test_1 mmVGf
mvwyD
KNe0n
my_setup_function
my_setup_function
KNe0n
mvwyD

[gw2] PASSED test_e2e.py::test_3 
[gw1] PASSED test_e2e.py::test_2 
[gw0] PASSED test_e2e.py::test_1 
test_e2e.py::test_4 
[gw0] PASSED test_e2e.py::test_4 
test_e2e.py::test_5 
[gw1] PASSED test_e2e.py::test_5

sessionスコープのfixtureを1度だけ実行させる

conftest.py

公式ドキュメントを参考にして、token()を更新します。

「tmp_path_factory」は、pytestが用意している一時作業用のディレクトリを作成するfixtureです。

conftest.py

import random
import string
import sys

import pytest

from filelock import FileLock
from time import sleep

@pytest.fixture(scope="session")
def token(tmp_path_factory, worker_id):
    def get_token():
        # tokenを取得したことにする(取得まで2秒かかる想定とする)
        sleep(2)
        return "".join(random.choices(string.ascii_letters + string.digits, k=5))

    if worker_id == "master":
        # 並列実行でない場合は、tokenを取得して返す
        return get_token()

    root_tmp_dir = tmp_path_factory.getbasetemp().parent

    fn = root_tmp_dir / "token.txt"

    with FileLock(str(fn) + ".lock"):
        if fn.is_file():
            return fn.read_text()

        token = get_token()
        fn.write_text(token)
        return token

@pytest.fixture(scope="session", autouse=True)
def my_setup_session():
    print("my_setup_session", file=sys.stderr)

@pytest.fixture(scope="function", autouse=True)
def my_setup_function():
    print("my_setup_function", file=sys.stderr)

テストを並列実行すると、同じtokenの値になっている

並列実行したすべてのテストで「同じtoken」を取得できました。 動作的には、token取得時にログの表示が約2秒ほど停止していました(get_token()が終了するまで待ってる)。

plugins: xdist-3.4.0
3 workers [5 items]     
scheduling tests via LoadScheduling

my_setup_session
test_e2e.py::test_3 
test_e2e.py::test_2 my_setup_session

test_e2e.py::test_1 my_setup_session
my_setup_function
fYXrk

[gw2] PASSED test_e2e.py::test_3 my_setup_function
fYXrk

[gw0] PASSED test_e2e.py::test_1 
test_e2e.py::test_4 my_setup_function
fYXrk

[gw0] PASSED test_e2e.py::test_4 my_setup_function
fYXrk

[gw1] PASSED test_e2e.py::test_2 
test_e2e.py::test_5 my_setup_function
fYXrk

[gw1] PASSED test_e2e.py::test_5

一時フォルダを確認すると次のようになっていました。

  • テスト実行のたびに、新しいフォルダが作成される
  • テスト実行のたびに、古いフォルダが削除される

11_pytest_temp_directory

参考