pytestのプラグインであるpytest-xdistを利用すると、テストを並列実行できます。これにより、テスト時間の短縮が見込めます。
しかし、並列実行するため、sessionスコープ(1度だけ実行される)のfixtureも並列数だけ実行されます。 そのため、次のような場合で課題となる可能性もあります。
- テスト用データの用意などで、時間がかかる
- 外部APIを利用するため、rate limitの上限が気になる
- など
たとえば、APIの認可にAuth0を利用している場合、E2Eテスト用のアクセストークンの取得(M2M認証)は、月1000回の制限があります(プランにもよる)。
テストの実行回数や並列数にもよりますが、少し心配になります。(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
一時フォルダを確認すると次のようになっていました。
- テスト実行のたびに、新しいフォルダが作成される
- テスト実行のたびに、古いフォルダが削除される