pytest xdist plugin for distributed testing, most importantly across multiple CPUs
—
Core plugin setup, command-line options, and pytest integration for pytest-xdist. This module handles all the configuration and initialization needed to enable distributed testing.
pytest-xdist adds numerous command-line options to control distributed testing behavior.
def pytest_addoption(parser: pytest.Parser) -> None:
"""Add xdist-specific command line options to pytest."""Key Options:
-n/--numprocesses: Number of processes ('auto', 'logical', or integer)--maxprocesses: Limit maximum workers with auto detection--max-worker-restart: Maximum worker restart count--dist: Distribution mode (each, load, loadscope, loadfile, loadgroup, worksteal, no)--loadscope-reorder/--no-loadscope-reorder: Control test reordering in loadscope mode--tx: Test execution environments--px: Proxy gateways-d/--distload: Shortcut for '--dist=load'--rsyncdir: Directories for rsyncing (deprecated)--rsyncignore: Ignore patterns for rsyncing (deprecated)--testrunuid: Shared identifier across workers--maxschedchunk: Maximum tests scheduled per stepHandles main command-line processing and configuration validation.
def pytest_cmdline_main(config: pytest.Config) -> None:
"""Main command line processing for xdist options."""Functionality:
-d/--distload to --dist=load--numprocesses with 'auto' and 'logical' values--pdb is usedConfigures the xdist plugin and sets up distributed session if needed.
def pytest_configure(config: pytest.Config) -> None:
"""Configure xdist plugin and create distributed session if needed."""Functionality:
Registers xdist-specific hooks with pytest's plugin manager.
def pytest_addhooks(pluginmanager: pytest.PytestPluginManager) -> None:
"""Add xdist hook specifications to pytest."""Automatically detects the optimal number of workers based on system resources.
def pytest_xdist_auto_num_workers(config: pytest.Config) -> int:
"""
Hook implementation for determining auto worker count.
Returns:
int: Number of workers to spawn for 'auto' or 'logical' modes
"""Detection Logic:
PYTEST_XDIST_AUTO_NUM_WORKERS environment variableos.sched_getaffinity() on Unix systemsos.cpu_count() or multiprocessing.cpu_count() as final fallbackHelper functions for parsing and validating configuration options.
def parse_numprocesses(s: str) -> int | Literal["auto", "logical"]:
"""
Parse numprocesses argument value.
Args:
s: String value ('auto', 'logical', or numeric)
Returns:
Parsed value as int or literal string
"""
def _is_distribution_mode(config: pytest.Config) -> bool:
"""
Check if distribution mode is enabled.
Args:
config: pytest configuration object
Returns:
bool: True if distribution is enabled
"""# In conftest.py - check if xdist is being used
def pytest_configure(config):
if hasattr(config, 'workerinput'):
# Running as xdist worker
setup_worker_specific_config()
else:
# Running as controller or single process
setup_controller_config()# In conftest.py - customize auto worker detection
def pytest_xdist_auto_num_workers(config):
# Custom logic for determining worker count
if config.getoption("--slow-tests"):
return 2 # Use fewer workers for slow tests
return None # Use default detection# Set custom worker count via environment
export PYTEST_XDIST_AUTO_NUM_WORKERS=8
pytest -n auto
# Use logical CPU count (includes hyperthreading)
pytest -n logical
# Use physical CPU count only
pytest -n autoInstall with Tessl CLI
npx tessl i tessl/pypi-pytest-xdist