pytest_test_categories.xdist_compat

pytest-xdist compatibility module.

This module provides utilities and hooks for pytest-xdist compatibility. When tests are run in parallel with xdist, each worker process maintains its own plugin state. This module handles aggregating results from workers back to the controller for accurate reporting.

Key concepts: - Controller: The main pytest process that coordinates workers - Workers: Subprocesses (gw0, gw1, etc.) that execute tests - workeroutput: Dict on worker config for passing data to controller

Hooks used: - pytest_sessionfinish (worker): Send stats via workeroutput - pytest_testnodedown (controller): Aggregate stats from workers

Attributes

WORKEROUTPUT_DISTRIBUTION_KEY
WORKEROUTPUT_REPORT_KEY
XDIST_WORKER_ENV

Functions

deserialize_distribution_counts(data)	Deserialize distribution counts from worker output.
is_xdist_controller(config)	Check if current process is the xdist controller.
is_xdist_worker()	Check if current process is an xdist worker.
merge_report_data(target, worker_data)	Merge worker report data into target report.
serialize_distribution_counts(counts)	Serialize TestCounts to a dict for worker output.
serialize_report_data(report)	Serialize TestSizeReport data for worker output.

Module Contents

pytest_test_categories.xdist_compat.deserialize_distribution_counts(data)

Deserialize distribution counts from worker output.

Parameters:

data (dict[str, int]) – The serialized counts dict.

Returns:

A dict suitable for creating TestCounts.

Return type:

dict[str, int]

pytest_test_categories.xdist_compat.is_xdist_controller(config)

Check if current process is the xdist controller.

The controller is the main pytest process when xdist is active. It coordinates workers but does not execute tests itself.

Parameters:

config (pytest.Config) – The pytest configuration object.

Returns:

True if this is the xdist controller, False otherwise.

Return type:

bool

pytest_test_categories.xdist_compat.is_xdist_worker()

Check if current process is an xdist worker.

Returns:

True if running as an xdist worker, False otherwise.

Return type:

bool

pytest_test_categories.xdist_compat.merge_report_data(target, worker_data)

Merge worker report data into target report.

This modifies the target report in place, adding all tests from the worker.

Parameters:

• target (pytest_test_categories.reporting.TestSizeReport) – The target report to merge into.

• worker_data (dict[str, object]) – Serialized report data from a worker.

Return type:

None

pytest_test_categories.xdist_compat.serialize_distribution_counts(counts)

Serialize TestCounts to a dict for worker output.

Parameters:

counts (pytest_test_categories.distribution.stats.TestCounts) – The test counts to serialize.

Returns:

A dict that can be safely passed through workeroutput.

Return type:

dict[str, int]

pytest_test_categories.xdist_compat.serialize_report_data(report)

Serialize TestSizeReport data for worker output.

Parameters:

report (pytest_test_categories.reporting.TestSizeReport) – The test size report to serialize.

Returns:

A dict that can be safely passed through workeroutput.

Return type:

dict[str, object]

pytest_test_categories.xdist_compat.WORKEROUTPUT_DISTRIBUTION_KEY = 'test_categories_distribution'

pytest_test_categories.xdist_compat.WORKEROUTPUT_REPORT_KEY = 'test_categories_report'

pytest_test_categories.xdist_compat.XDIST_WORKER_ENV = 'PYTEST_XDIST_WORKER'