New methods is_xdist_worker, is_xdist_master, get_xdist_worker_id. (#505)

Co-authored-by: Sylvain MARIE <[email protected]>
Co-authored-by: Zac Hatfield-Dodds <[email protected]>
Co-authored-by: Bruno Oliveira <[email protected]>

Author: 4 people

README.rst

@@ -268,7 +268,7 @@ a test or fixture, you may use the ``worker_id`` fixture to do so:
 When ``xdist`` is disabled (running with ``-n0`` for example), then
 ``worker_id`` will return ``"master"``.
 
-Additionally, worker processes have the following environment variables
+Worker processes also have the following environment variables
 defined:
 
 * ``PYTEST_XDIST_WORKER``: the name of the worker, e.g., ``"gw2"``.
@@ -278,6 +278,34 @@ defined:
 The information about the worker_id in a test is stored in the ``TestReport`` as
 well, under the ``worker_id`` attribute.
 
+Since version 2.0, the following functions are also available in the ``xdist`` module:
+
+.. code-block:: python
+
+    def is_xdist_worker(request_or_session) -> bool:
+        """Return `True` if this is an xdist worker, `False` otherwise
+
+        :param request_or_session: the `pytest` `request` or `session` object
+        """
+
+    def is_xdist_master(request_or_session) -> bool:
+        """Return `True` if this is the xdist master, `False` otherwise
+
+        Note: this method also returns `False` when distribution has not been
+        activated at all.
+
+        :param request_or_session: the `pytest` `request` or `session` object
+        """
+
+    def get_xdist_worker_id(request_or_session) -> str:
+        """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
+        if running on the 'master' node.
+
+        If not distributing tests (for example passing `-n0` or not passing `-n` at all) also return 'master'.
+
+        :param request_or_session: the `pytest` `request` or `session` object
+        """
+
 
 Uniquely identifying the current test run
 -----------------------------------------

changelog/504.feature.rst

@@ -0,0 +1 @@
+New functions ``xdist.is_xdist_worker``, ``xdist.is_xdist_master``, ``xdist.get_xdist_worker_id``, to easily identify the current node.

src/xdist/__init__.py

@@ -1,3 +1,4 @@
+from xdist.plugin import is_xdist_worker, is_xdist_master, get_xdist_worker_id
 from xdist._version import version as __version__
 
-__all__ = ["__version__"]
+__all__ = ["__version__", "is_xdist_worker", "is_xdist_master", "get_xdist_worker_id"]

src/xdist/plugin.py

@@ -192,19 +192,53 @@ def pytest_cmdline_main(config):
 
 
 # -------------------------------------------------------------------------
-# fixtures
+# fixtures and API to easily know the role of current node
 # -------------------------------------------------------------------------
 
 
+def is_xdist_worker(request_or_session) -> bool:
+    """Return `True` if this is an xdist worker, `False` otherwise
+
+    :param request_or_session: the `pytest` `request` or `session` object
+    """
+    return hasattr(request_or_session.config, "workerinput")
+
+
+def is_xdist_master(request_or_session) -> bool:
+    """Return `True` if this is the xdist master, `False` otherwise
+
+    Note: this method also returns `False` when distribution has not been
+    activated at all.
+
+    :param request_or_session: the `pytest` `request` or `session` object
+    """
+    return (
+        not is_xdist_worker(request_or_session)
+        and request_or_session.config.option.dist != "no"
+    )
+
+
+def get_xdist_worker_id(request_or_session) -> str:
+    """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
+    if running on the 'master' node.
+
+    If not distributing tests (for example passing `-n0` or not passing `-n` at all)
+    also return 'master'.
+
+    :param request_or_session: the `pytest` `request` or `session` object
+    """
+    if hasattr(request_or_session.config, "workerinput"):
+        return request_or_session.config.workerinput["workerid"]
+    else:
+        return "master"
+
+
 @pytest.fixture(scope="session")
 def worker_id(request):
     """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
     if running on the master node.
     """
-    if hasattr(request.config, "workerinput"):
-        return request.config.workerinput["workerid"]
-    else:
-        return "master"
+    return get_xdist_worker_id(request)
 
 
 @pytest.fixture(scope="session")

testing/acceptance_test.py

@@ -3,6 +3,7 @@
 
 import py
 import pytest
+import xdist
 
 
 class TestDistribution:
@@ -1366,3 +1367,39 @@ def get_workers_and_test_count_by_prefix(prefix, lines, expected_status="PASSED"
         if expected_status == status and nodeid.startswith(prefix):
             result[worker] = result.get(worker, 0) + 1
     return result
+
+
+class TestAPI:
+    @pytest.fixture
+    def fake_request(self):
+        class FakeOption:
+            def __init__(self):
+                self.dist = "load"
+
+        class FakeConfig:
+            def __init__(self):
+                self.workerinput = {"workerid": "gw5"}
+                self.option = FakeOption()
+
+        class FakeRequest:
+            def __init__(self):
+                self.config = FakeConfig()
+
+        return FakeRequest()
+
+    def test_is_xdist_worker(self, fake_request):
+        assert xdist.is_xdist_worker(fake_request)
+        del fake_request.config.workerinput
+        assert not xdist.is_xdist_worker(fake_request)
+
+    def test_is_xdist_master(self, fake_request):
+        assert not xdist.is_xdist_master(fake_request)
+        del fake_request.config.workerinput
+        assert xdist.is_xdist_master(fake_request)
+        fake_request.config.option.dist = "no"
+        assert not xdist.is_xdist_master(fake_request)
+
+    def test_get_xdist_worker_id(self, fake_request):
+        assert xdist.get_xdist_worker_id(fake_request) == "gw5"
+        del fake_request.config.workerinput
+        assert xdist.get_xdist_worker_id(fake_request) == "master"