tessl/pypi-pytest-xdist

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

—

Pending

Overview

Eval results

Files

Worker Detection

Name: tessl/pypi-pytest-xdist
Author: tessl

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

Capabilities

Worker Process Detection

Determine if the current process is a worker process in a distributed test run.

def is_xdist_worker(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
    """
    Return True if this is an xdist worker, False otherwise.
    
    Args:
        request_or_session: pytest request fixture or session object
        
    Returns:
        bool: True if running in worker process
    """

Usage:

Check if test is running in a worker process
Conditional logic for worker-only or controller-only code
Resource management (e.g., shared resources only in controller)

Controller Process Detection

Determine if the current process is the controller (master) process coordinating workers.

def is_xdist_controller(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
    """
    Return True if this is the xdist controller, False otherwise.
    
    Note: this method also returns False when distribution has not been
    activated at all.
    
    Args:
        request_or_session: pytest request fixture or session object
        
    Returns:
        bool: True if running in controller process with distribution active
    """

Behavior:

Returns True only when distribution is active AND running in controller
Returns False for single-process runs (no distribution)
Returns False for worker processes

Legacy Master Detection (Deprecated)

Deprecated alias for controller detection. Use is_xdist_controller instead.

def is_xdist_master(request_or_session: pytest.FixtureRequest | pytest.Session) -> bool:
    """
    Deprecated alias for is_xdist_controller.
    
    Args:
        request_or_session: pytest request fixture or session object
        
    Returns:
        bool: True if running in controller process
    """

Worker ID Retrieval

Get the unique identifier for the current worker process.

def get_xdist_worker_id(request_or_session: pytest.FixtureRequest | pytest.Session) -> str:
    """
    Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
    if running on the controller node.
    
    If not distributing tests (for example passing -n0 or not passing -n at all)
    also return 'master'.
    
    Args:
        request_or_session: pytest request fixture or session object
        
    Returns:
        str: Worker ID ('gw0', 'gw1', etc.) or 'master' for controller/single-process
    """

Return Values:

'gw0', 'gw1', 'gw2', etc. for worker processes
'master' for controller process or single-process runs

Session-Scoped Fixtures

Convenient fixtures for accessing worker information in tests.

@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.
    
    Returns:
        str: Worker ID or 'master'
    """

@pytest.fixture(scope="session")  
def testrun_uid(request: pytest.FixtureRequest) -> str:
    """
    Return the unique id of the current test run.
    
    This ID is shared across all workers in the same test run,
    making it useful for coordinating between workers or creating
    unique resource names.
    
    Returns:
        str: Unique test run identifier (hex string)
    """

Usage Examples

Conditional Test Behavior

import pytest
from xdist import is_xdist_worker, get_xdist_worker_id

def test_with_worker_logic(request):
    if is_xdist_worker(request):
        # This code only runs in worker processes
        worker_id = get_xdist_worker_id(request)
        print(f"Running in worker {worker_id}")
        
        # Worker-specific setup
        setup_worker_resources()
    else:
        # This code runs in controller or single-process mode
        print("Running in controller process")
        
        # Controller-specific setup
        setup_shared_resources()

Resource Management

import pytest
from xdist import is_xdist_controller

@pytest.fixture(scope="session")
def database_connection(request):
    if is_xdist_controller(request):
        # Only controller sets up the database
        db = create_test_database()
        yield db
        cleanup_test_database(db)
    else:
        # Workers get connection info via other means
        yield get_worker_db_connection()

Using Fixtures

def test_worker_coordination(worker_id, testrun_uid):
    # worker_id is 'gw0', 'gw1', etc. for workers, 'master' for controller
    # testrun_uid is the same across all workers in this test run
    
    temp_file = f"/tmp/test_{testrun_uid}_{worker_id}.tmp"
    with open(temp_file, 'w') as f:
        f.write(f"Worker {worker_id} data")
    
    # Do test work
    assert os.path.exists(temp_file)
    
    # Cleanup
    os.unlink(temp_file)

Plugin Development

class MyPlugin:
    def pytest_runtest_setup(self, item):
        if is_xdist_worker(item.session):
            # Worker-specific test setup
            self.setup_worker_test_env()
        elif is_xdist_controller(item.session):
            # Controller-specific test setup  
            self.setup_controller_test_env()
        else:
            # Single-process test setup
            self.setup_single_process_test_env()

Shared State Management

import pytest
from xdist import is_xdist_worker, get_xdist_worker_id

# Global state that should only be initialized once per worker
_worker_state = None

def get_worker_state(request):
    global _worker_state
    if _worker_state is None:
        if is_xdist_worker(request):
            worker_id = get_xdist_worker_id(request)
            _worker_state = WorkerState(worker_id)
        else:
            _worker_state = ControllerState()
    return _worker_state

def test_using_worker_state(request):
    state = get_worker_state(request)
    # Use worker-specific or controller state
    state.do_something()

Install with Tessl CLI