62 lines
2.0 KiB
Python
Executable File
62 lines
2.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# This script is a wrapper around pytest which starts a separate pytest process
|
|
# for each separate motor given in FOLDER. This allows to run tests on different
|
|
# motors in parallel, cutting down execution time and testing parallelism.
|
|
# The script accepts any arguments and forwards them to the pytest call.
|
|
# If a folder is given as an argument, all tests which are not within this
|
|
# folder are ignored:
|
|
# - `pytestpar` will run all tests parallelized per motor
|
|
# - `pytestpar tests/sinqMotor/turboPmac` will run all Turbo PMAC tests parallelized per motor
|
|
# - `pytestpar tests/sinqMotor/turboPmac/ax1` will only run the tests for motor `ax1`.
|
|
# These tests are run sequentially.
|
|
|
|
import sys
|
|
import subprocess
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Define the test folders you want to run in parallel
|
|
FOLDERS = ["tests/sinqMotor/turboPmac/ax1", "tests/sinqMotor/turboPmac/ax5"]
|
|
|
|
# Separate path-like arguments from other pytest args
|
|
path_args = []
|
|
extra_args = []
|
|
|
|
for arg in sys.argv[1:]:
|
|
if '/' in arg or os.path.exists(arg):
|
|
path_args.append(arg)
|
|
else:
|
|
extra_args.append(arg)
|
|
|
|
# Filter folders to run based on path_args
|
|
enabled_folders = []
|
|
if not path_args:
|
|
enabled_folders = FOLDERS
|
|
else:
|
|
for folder in FOLDERS:
|
|
if any(Path(arg).resolve().as_posix().startswith(Path(folder).resolve().as_posix()) for arg in path_args):
|
|
enabled_folders.append(folder)
|
|
|
|
# Run each enabled folder's relevant tests in parallel
|
|
processes = []
|
|
for folder in enabled_folders:
|
|
folder_path_args = (
|
|
[arg for arg in path_args if Path(arg).resolve().as_posix().startswith(Path(folder).resolve().as_posix())]
|
|
if path_args else [folder]
|
|
)
|
|
|
|
command = ["pytest"] + folder_path_args + extra_args
|
|
print(f"Running: {' '.join(command)}")
|
|
|
|
proc = subprocess.Popen(command)
|
|
processes.append(proc)
|
|
|
|
# Wait for all processes and collect exit codes
|
|
exit_code = 0
|
|
for proc in processes:
|
|
code = proc.wait()
|
|
if code != 0:
|
|
exit_code = code
|
|
|
|
sys.exit(exit_code) |