or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

distribution-scheduling.mdhook-specifications.mdindex.mdloop-on-fail.mdplugin-configuration.mdsession-management.mdworker-detection.md
tile.json

tessl/pypi-pytest-xdist

pytest xdist plugin for distributed testing, most importantly across multiple CPUs

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pytest-xdist@3.8.x

To install, run

npx @tessl/cli install tessl/pypi-pytest-xdist@3.8.0

index.mddocs/

pytest-xdist

A pytest plugin that extends pytest with new test execution modes for distributed testing, most importantly across multiple CPUs. It enables parallel test execution through various distribution strategies while maintaining compatibility with pytest's existing ecosystem and other plugins.

Package Information

  • Package Name: pytest-xdist
  • Language: Python
  • Installation: pip install pytest-xdist
  • Requires: pytest >= 7.0.0, execnet >= 2.1
  • Optional Dependencies: psutil (for CPU detection), setproctitle (for process naming)

Core Imports

import xdist

For using the main API functions:

from xdist import get_xdist_worker_id, is_xdist_worker, is_xdist_controller

Basic Usage

# Command line usage - most common way to use pytest-xdist
# Run tests on auto-detected number of CPUs
pytest -n auto

# Run tests on specific number of processes
pytest -n 4

# Run tests with load balancing distribution
pytest --dist load

# Check if running in xdist worker in test code
from xdist import is_xdist_worker

def test_something(request):
    if is_xdist_worker(request):
        # Code that only runs in worker processes
        worker_id = get_xdist_worker_id(request)
        print(f"Running in worker {worker_id}")
    else:
        # Code that only runs in controller/master process
        print("Running in controller process")

Architecture

pytest-xdist implements a controller-worker architecture:

  • Controller Process: Manages test collection, coordinates workers, and aggregates results
  • Worker Processes: Execute individual tests and report results back to controller
  • Schedulers: Different algorithms for distributing tests across workers (load, loadscope, loadfile, etc.)
  • Communication: execnet-based communication between controller and workers
  • Test Distribution: Various modes for how tests are distributed (each, load, loadscope, loadfile, loadgroup, worksteal)

Capabilities

Plugin Installation and Configuration

Core plugin setup, command-line options, and pytest integration. This includes all the command-line flags and configuration options that control how pytest-xdist operates.

def pytest_addoption(parser: pytest.Parser) -> None: ...
def pytest_configure(config: pytest.Config) -> None: ...
def pytest_cmdline_main(config: pytest.Config) -> None: ...

Plugin Configuration

Worker Detection and Control

Functions to detect and control the execution environment, allowing tests and plugins to behave differently when running in distributed mode versus single-process mode.

def is_xdist_worker(request_or_session) -> bool: ...
def is_xdist_controller(request_or_session) -> bool: ...
def is_xdist_master(request_or_session) -> bool: ...  # deprecated alias
def get_xdist_worker_id(request_or_session) -> str: ...

Worker Detection

Test Distribution and Scheduling

Multiple scheduling algorithms for distributing tests across workers, each optimized for different test suite characteristics and performance requirements.

class Scheduling(Protocol):
    def add_node(self, node: WorkerController) -> None: ...
    def schedule(self) -> None: ...
    def mark_test_complete(self, node: WorkerController, item_index: int, duration: float = 0) -> None: ...

Distribution Scheduling

Distributed Session Management

Core session management for coordinating distributed test execution, including worker lifecycle management and result aggregation.

class DSession:
    def __init__(self, config: pytest.Config) -> None: ...
    def session_finished(self) -> bool: ...
    def pytest_sessionstart(self, session: pytest.Session) -> None: ...

Session Management

Hook Specifications

Custom pytest hooks specific to xdist for plugin authors to extend and customize distributed testing behavior.

def pytest_xdist_setupnodes(config: pytest.Config, specs: Sequence[execnet.XSpec]) -> None: ...
def pytest_configure_node(node: WorkerController) -> None: ...
def pytest_testnodeready(node: WorkerController) -> None: ...
def pytest_xdist_auto_num_workers(config: pytest.Config) -> int: ...

Hook Specifications

Loop-on-Fail (Deprecated)

Legacy functionality for automatically re-running failed tests when files change. This feature is deprecated and will be removed in pytest-xdist 4.0.

def pytest_cmdline_main(config: pytest.Config) -> int | None: ...
def looponfail_main(config: pytest.Config) -> None: ...

Loop-on-Fail

Fixtures

@pytest.fixture(scope="session")
def worker_id(request: pytest.FixtureRequest) -> str:
    """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
    if running on the master node."""

@pytest.fixture(scope="session") 
def testrun_uid(request: pytest.FixtureRequest) -> str:
    """Return the unique id of the current test run."""

Markers

# Mark tests to run in the same worker session
@pytest.mark.xdist_group(name="group_name")
def test_example():
    pass

Types

# Main protocol for scheduling implementations
class Scheduling(Protocol):
    @property
    def nodes(self) -> list[WorkerController]: ...
    @property 
    def collection_is_completed(self) -> bool: ...
    @property
    def tests_finished(self) -> bool: ...
    @property
    def has_pending(self) -> bool: ...
    
    def add_node(self, node: WorkerController) -> None: ...
    def add_node_collection(self, node: WorkerController, collection: Sequence[str]) -> None: ...
    def mark_test_complete(self, node: WorkerController, item_index: int, duration: float = 0) -> None: ...
    def mark_test_pending(self, item: str) -> None: ...
    def remove_pending_tests_from_node(self, node: WorkerController, indices: Sequence[int]) -> None: ...
    def remove_node(self, node: WorkerController) -> str | None: ...
    def schedule(self) -> None: ...

# Worker controller for managing individual worker processes
class WorkerController: ...

# Distributed session class for coordinating execution
class DSession: ...

# Node manager for setting up and managing workers
class NodeManager: ...

# Remote execution utilities
class Producer: ...
class TestQueue: ...