Merge pull request #122 from wronglink/custom_scheduler_notes

Added notes about new pytest_xdist_make_scheduler hook

.travis.yml

@@ -5,26 +5,28 @@ python:
 install: pip install -U tox setuptools_scm
 env:
   matrix:
-  - TESTENV=py26-pytest26
+  # note: please use "tox --listenvs" to populate the build matrix
-  - TESTENV=py26-pytest27
+  - TOXENV=py26-pytest27
-  - TESTENV=py26-pytest28
+  - TOXENV=py26-pytest28
-  - TESTENV=py26-pytest29
+  - TOXENV=py26-pytest29
-  - TESTENV=py27-pytest26
+  - TOXENV=py26-pytest30
-  - TESTENV=py27-pytest27
+  - TOXENV=py27-pytest27
-  - TESTENV=py27-pytest28
+  - TOXENV=py27-pytest28
-  - TESTENV=py27-pytest29
+  - TOXENV=py27-pytest29
-  - TESTENV=py34-pytest26
+  - TOXENV=py27-pytest30
-  - TESTENV=py34-pytest27
+  - TOXENV=py34-pytest27
-  - TESTENV=py34-pytest28
+  - TOXENV=py34-pytest28
-  - TESTENV=py34-pytest29
+  - TOXENV=py34-pytest29
-  - TESTENV=py35-pytest27
+  - TOXENV=py34-pytest30
-  - TESTENV=py35-pytest28
+  - TOXENV=py35-pytest27
-  - TESTENV=py35-pytest29
+  - TOXENV=py35-pytest28
-  - TESTENV=py27-pytest28-pexpect
+  - TOXENV=py35-pytest29
-  - TESTENV=py35-pytest28-pexpect
+  - TOXENV=py27-pytest28-pexpect
-  - TESTENV=flakes
+  - TOXENV=py35-pytest28-pexpect
-  - TESTENV=readme
+  - TOXENV=flakes
-script: tox --recreate -e $TESTENV
+  - TOXENV=readme
+
+script: tox
 notifications:
   irc:
     channels:

CHANGELOG

@@ -1,3 +1,15 @@
+1.16.0
+------
+
+- ``pytest-xdist`` now requires pytest 2.7 or later.
+
+- Add ``worker_id`` attribute in the TestReport
+
+- new hook: ``pytest_xdist_make_scheduler(config, log)``, can return custom tests items
+  distribution logic implementation. You can take a look at built-in ``LoadScheduling``
+  and ``EachScheduling`` implementations. Note that required scheduler class public
+  API may change in next ``pytest-xdist`` versions.
+
 1.15.0
 ------
 

OVERVIEW.md

@@ -35,6 +35,8 @@ The execution flow is:
    fashion. The rest of the tests will be distributed later as **workers**
    finish tests (see below).
 
+1. Note that `pytest_xdist_make_scheduler` hook can be used to implement custom tests distribution logic.
+   
 1. **workers** re-implement `pytest_runtestloop`: pytest's default implementation
    basically loops over all collected items in the `session` object and executes
    the `pytest_runtest_protocol` for each test item, but in xdist **workers** sit idly 

README.rst

@@ -46,7 +46,7 @@ Install the plugin with::
 
     pip install pytest-xdist
 
-or use the package in develope/in-place mode with
+or use the package in develop/in-place mode with
 a checkout of the `pytest-xdist repository`_ ::
 
     python setup.py develop
@@ -209,6 +209,10 @@ defined:
 * ``PYTEST_XDIST_WORKER_COUNT``: the total number of workers in this session,
   e.g., ``"4"`` when ``-n 4`` is given in the command-line.
 
+The information about the worker_id in a test is stored in the TestReport as
+well, under worker_id attribute.
+
+
 *New in version 1.15.*
 
 Specifying test exec environments in an ini file

appveyor.yml

@@ -1,3 +1,26 @@
+environment:
+  matrix:
+  # note: please use "tox --listenvs" to populate the build matrix
+  - TOXENV: "py26-pytest27"
+  - TOXENV: "py26-pytest28"
+  - TOXENV: "py26-pytest29"
+  - TOXENV: "py26-pytest30"
+  - TOXENV: "py27-pytest27"
+  - TOXENV: "py27-pytest28"
+  - TOXENV: "py27-pytest29"
+  - TOXENV: "py27-pytest30"
+  - TOXENV: "py34-pytest27"
+  - TOXENV: "py34-pytest28"
+  - TOXENV: "py34-pytest29"
+  - TOXENV: "py34-pytest30"
+  - TOXENV: "py35-pytest27"
+  - TOXENV: "py35-pytest28"
+  - TOXENV: "py35-pytest29"
+  - TOXENV: "py27-pytest28-pexpect"
+  - TOXENV: "py35-pytest28-pexpect"
+  - TOXENV: "flakes"
+  - TOXENV: "readme"
+
 install:
   - C:\Python35\python -m pip install tox setuptools_scm
 

setup.cfg

@@ -1,2 +1,5 @@
 [bdist_wheel]
 universal = 1
+
+[metadata]
+license_file = LICENSE

setup.py

@@ -20,7 +20,7 @@ setup(
         ],
     },
     zip_safe=False,
-    install_requires=['execnet>=1.1', 'pytest>=2.4.2', 'py>=1.4.22'],
+    install_requires=['execnet>=1.1', 'pytest>=2.7.0', 'py>=1.4.22'],
     setup_requires=['setuptools_scm'],
     classifiers=[
         'Development Status :: 5 - Production/Stable',

testing/acceptance_test.py

@@ -62,7 +62,7 @@ class TestDistribution:
                 pass
         """)
         result = testdir.runpytest(p1, '--tx=popen', '--tx=popen')
-        assert result.ret == 1
+        assert result.ret in (1, 2)
         result.stdout.fnmatch_lines([
             "E   ImportError: *__import_of_missing_module*",
         ])

testing/test_dsession.py

@@ -49,80 +49,84 @@ def dumpqueue(queue):
 
 
 class TestEachScheduling:
-    def test_schedule_load_simple(self):
+    def test_schedule_load_simple(self, testdir):
         node1 = MockNode()
         node2 = MockNode()
-        sched = EachScheduling(2)
+        config = testdir.parseconfig("--tx=2*popen")
-        sched.addnode(node1)
+        sched = EachScheduling(config)
-        sched.addnode(node2)
+        sched.add_node(node1)
+        sched.add_node(node2)
         collection = ["a.py::test_1", ]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert not sched.collection_is_completed
-        sched.addnode_collection(node2, collection)
+        sched.add_node_collection(node2, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
         assert sched.node2collection[node2] == collection
-        sched.init_distribute()
+        sched.schedule()
-        assert sched.tests_finished()
+        assert sched.tests_finished
         assert node1.sent == ['ALL']
         assert node2.sent == ['ALL']
-        sched.remove_item(node1, 0)
+        sched.mark_test_complete(node1, 0)
-        assert sched.tests_finished()
+        assert sched.tests_finished
-        sched.remove_item(node2, 0)
+        sched.mark_test_complete(node2, 0)
-        assert sched.tests_finished()
+        assert sched.tests_finished
 
-    def test_schedule_remove_node(self):
+    def test_schedule_remove_node(self, testdir):
         node1 = MockNode()
-        sched = EachScheduling(1)
+        config = testdir.parseconfig("--tx=popen")
-        sched.addnode(node1)
+        sched = EachScheduling(config)
+        sched.add_node(node1)
         collection = ["a.py::test_1", ]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
-        sched.init_distribute()
+        sched.schedule()
-        assert sched.tests_finished()
+        assert sched.tests_finished
         crashitem = sched.remove_node(node1)
         assert crashitem
-        assert sched.tests_finished()
+        assert sched.tests_finished
-        assert not sched.hasnodes()
+        assert not sched.nodes
 
 
 class TestLoadScheduling:
-    def test_schedule_load_simple(self):
+    def test_schedule_load_simple(self, testdir):
-        sched = LoadScheduling(2)
+        config = testdir.parseconfig("--tx=2*popen")
-        sched.addnode(MockNode())
+        sched = LoadScheduling(config)
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2 = sched.nodes
         collection = ["a.py::test_1", "a.py::test_2"]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert not sched.collection_is_completed
-        sched.addnode_collection(node2, collection)
+        sched.add_node_collection(node2, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
         assert sched.node2collection[node2] == collection
-        sched.init_distribute()
+        sched.schedule()
         assert not sched.pending
-        assert sched.tests_finished()
+        assert sched.tests_finished
         assert len(node1.sent) == 1
         assert len(node2.sent) == 1
         assert node1.sent == [0]
         assert node2.sent == [1]
-        sched.remove_item(node1, node1.sent[0])
+        sched.mark_test_complete(node1, node1.sent[0])
-        assert sched.tests_finished()
+        assert sched.tests_finished
 
-    def test_init_distribute_batch_size(self):
+    def test_schedule_batch_size(self, testdir):
-        sched = LoadScheduling(2)
+        config = testdir.parseconfig("--tx=2*popen")
-        sched.addnode(MockNode())
+        sched = LoadScheduling(config)
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2 = sched.nodes
         col = ["xyz"] * (6)
-        sched.addnode_collection(node1, col)
+        sched.add_node_collection(node1, col)
-        sched.addnode_collection(node2, col)
+        sched.add_node_collection(node2, col)
-        sched.init_distribute()
+        sched.schedule()
-        # assert not sched.tests_finished()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         assert sent1 == [0, 2]
@@ -131,25 +135,26 @@ class TestLoadScheduling:
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
         assert len(sched.pending) == 2
-        sched.remove_item(node1, 0)
+        sched.mark_test_complete(node1, 0)
         assert node1.sent == [0, 2, 4]
         assert sched.pending == [5]
         assert node2.sent == [1, 3]
-        sched.remove_item(node1, 2)
+        sched.mark_test_complete(node1, 2)
         assert node1.sent == [0, 2, 4, 5]
         assert not sched.pending
 
-    def test_init_distribute_fewer_tests_than_nodes(self):
+    def test_schedule_fewer_tests_than_nodes(self, testdir):
-        sched = LoadScheduling(2)
+        config = testdir.parseconfig("--tx=2*popen")
-        sched.addnode(MockNode())
+        sched = LoadScheduling(config)
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2, node3 = sched.nodes
         col = ["xyz"] * 2
-        sched.addnode_collection(node1, col)
+        sched.add_node_collection(node1, col)
-        sched.addnode_collection(node2, col)
+        sched.add_node_collection(node2, col)
-        sched.init_distribute()
+        sched.schedule()
-        # assert not sched.tests_finished()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         sent3 = node3.sent
@@ -158,17 +163,18 @@ class TestLoadScheduling:
         assert sent3 == []
         assert not sched.pending
 
-    def test_init_distribute_fewer_than_two_tests_per_node(self):
+    def test_schedule_fewer_than_two_tests_per_node(self, testdir):
-        sched = LoadScheduling(2)
+        config = testdir.parseconfig("--tx=2*popen")
-        sched.addnode(MockNode())
+        sched = LoadScheduling(config)
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
-        sched.addnode(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2, node3 = sched.nodes
         col = ["xyz"] * 5
-        sched.addnode_collection(node1, col)
+        sched.add_node_collection(node1, col)
-        sched.addnode_collection(node2, col)
+        sched.add_node_collection(node2, col)
-        sched.init_distribute()
+        sched.schedule()
-        # assert not sched.tests_finished()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         sent3 = node3.sent
@@ -177,14 +183,15 @@ class TestLoadScheduling:
         assert sent3 == [2]
         assert not sched.pending
 
-    def test_add_remove_node(self):
+    def test_add_remove_node(self, testdir):
         node = MockNode()
-        sched = LoadScheduling(1)
+        config = testdir.parseconfig("--tx=popen")
-        sched.addnode(node)
+        sched = LoadScheduling(config)
+        sched.add_node(node)
         collection = ["test_file.py::test_func"]
-        sched.addnode_collection(node, collection)
+        sched.add_node_collection(node, collection)
         assert sched.collection_is_completed
-        sched.init_distribute()
+        sched.schedule()
         assert not sched.pending
         crashitem = sched.remove_node(node)
         assert crashitem == collection[0]
@@ -207,16 +214,16 @@ class TestLoadScheduling:
                 self.reports.append(report)
 
         collect_hook = CollectHook()
-        config = testdir.parseconfig()
+        config = testdir.parseconfig("--tx=2*popen")
         config.pluginmanager.register(collect_hook, "collect_hook")
         node1 = MockNode()
         node2 = MockNode()
-        sched = LoadScheduling(2, config=config)
+        sched = LoadScheduling(config)
-        sched.addnode(node1)
+        sched.add_node(node1)
-        sched.addnode(node2)
+        sched.add_node(node2)
-        sched.addnode_collection(node1, ["a.py::test_1"])
+        sched.add_node_collection(node1, ["a.py::test_1"])
-        sched.addnode_collection(node2, ["a.py::test_2"])
+        sched.add_node_collection(node2, ["a.py::test_2"])
-        sched.init_distribute()
+        sched.schedule()
         assert len(collect_hook.reports) == 1
         rep = collect_hook.reports[0]
         assert 'Different tests were collected between' in rep.longrepr

testing/test_newhooks.py

@@ -14,14 +14,19 @@ class TestHooks:
 
     def test_runtest_logreport(self, testdir):
         """Test that log reports from pytest_runtest_logreport when running
-        with xdist contain a "node" attribute. (#8)
+        with xdist contain "node", "nodeid" and "worker_id" attributes. (#8)
         """
         testdir.makeconftest("""
             def pytest_runtest_logreport(report):
                 if hasattr(report, 'node'):
-                    slaveid = report.node.slaveinput['slaveid']
                     if report.when == "call":
-                        print("HOOK: %s %s" % (report.nodeid, slaveid))
+                        slaveid = report.node.slaveinput['slaveid']
+                        if slaveid != report.worker_id:
+                            print("HOOK: Worker id mismatch: %s %s"
+                                   % (slaveid, report.worker_id))
+                        else:
+                            print("HOOK: %s %s"
+                                   % (report.nodeid, report.worker_id))
         """)
         res = testdir.runpytest('-n1', '-s')
         res.stdout.fnmatch_lines([

tox.ini

@@ -1,8 +1,7 @@
 [tox]
 # if you change the envlist, please update .travis.yml file as well
 envlist=
-  py{26,27,34}-pytest2{6,7,8,9}
+  py{26,27,34,35}-pytest{27,28,29,30}
-  py35-pytest2{7,8,9}
   py{27,35}-pytest28-pexpect
   flakes
   readme  
@@ -15,10 +14,10 @@ deps =
   pycmd
   # to avoid .eggs
   setuptools_scm
-  pytest26: pytest~=2.6.1
   pytest27: pytest~=2.7.2
   pytest28: pytest~=2.8.7
-  pytest29: pytest~=2.9.1
+  pytest29: pytest~=2.9.2
+  pytest30: pytest~=3.0.5
   pexpect: pexpect
 platform=
   pexpect: linux|darwin    

xdist/dsession.py

@@ -4,7 +4,7 @@ from _pytest.runner import CollectReport
 
 import pytest
 import py
-from xdist.slavemanage import NodeManager
+from xdist.slavemanage import NodeManager, parse_spec_config
 
 
 queue = py.builtin._tryimport('queue', 'Queue')
@@ -24,8 +24,9 @@ class EachScheduling:
     assigned the remaining items from the removed node.
     """
 
-    def __init__(self, numnodes, log=None):
+    def __init__(self, config, log=None):
-        self.numnodes = numnodes
+        self.config = config
+        self.numnodes = len(parse_spec_config(config))
         self.node2collection = {}
         self.node2pending = {}
         self._started = []
@@ -41,10 +42,19 @@ class EachScheduling:
         """A list of all nodes in the scheduler."""
         return list(self.node2pending.keys())
 
-    def hasnodes(self):
+    @property
-        return bool(self.node2pending)
+    def tests_finished(self):
+        if not self.collection_is_completed:
+            return False
+        if self._removed2pending:
+            return False
+        for pending in self.node2pending.values():
+            if len(pending) >= 2:
+                return False
+        return True
 
-    def haspending(self):
+    @property
+    def has_pending(self):
         """Return True if there are pending test items
 
         This indicates that collection has finished and nodes are
@@ -56,21 +66,11 @@ class EachScheduling:
                 return True
         return False
 
-    def addnode(self, node):
+    def add_node(self, node):
         assert node not in self.node2pending
         self.node2pending[node] = []
 
-    def tests_finished(self):
+    def add_node_collection(self, node, collection):
-        if not self.collection_is_completed:
-            return False
-        if self._removed2pending:
-            return False
-        for pending in self.node2pending.values():
-            if len(pending) >= 2:
-                return False
-        return True
-
-    def addnode_collection(self, node, collection):
         """Add the collected test items from a node
 
         Collection is complete once all nodes have submitted their
@@ -78,7 +78,7 @@ class EachScheduling:
         list.  When the collection is already completed this
         submission is from a node which was restarted to replace a
         dead node.  In this case we already assign the pending items
-        here.  In either case ``.init_distribute()`` will instruct the
+        here.  In either case ``.schedule()`` will instruct the
         node to start running the required tests.
         """
         assert node in self.node2pending
@@ -102,11 +102,11 @@ class EachScheduling:
                     self.node2pending[node] = pending
                     break
 
-    def remove_item(self, node, item_index, duration=0):
+    def mark_test_complete(self, node, item_index, duration=0):
         self.node2pending[node].remove(item_index)
 
     def remove_node(self, node):
-        # KeyError if we didn't get an addnode() yet
+        # KeyError if we didn't get an add_node() yet
         pending = self.node2pending.pop(node)
         if not pending:
             return
@@ -115,12 +115,12 @@ class EachScheduling:
             self._removed2pending[node] = pending
         return crashitem
 
-    def init_distribute(self):
+    def schedule(self):
         """Schedule the test items on the nodes
 
         If the node's pending list is empty it is a new node which
         needs to run all the tests.  If the pending list is already
-        populated (by ``.addnode_collection()``) then it replaces a
+        populated (by ``.add_node_collection()``) then it replaces a
         dead node and we only need to run those tests.
         """
         assert self.collection_is_completed
@@ -143,7 +143,7 @@ class LoadScheduling:
     when all collections are received it is verified they are
     identical collections.  Then the collection gets divided up in
     chunks and chunks get submitted to nodes.  Whenever a node finishes
-    an item, it calls ``.remove_item()`` which will trigger the
+    an item, it calls ``.mark_test_complete()`` which will trigger the
     scheduler to assign more tests if the number of pending tests for
     the node falls below a low-watermark.
 
@@ -170,7 +170,7 @@ class LoadScheduling:
 
     :collection: The one collection once it is validated to be
        identical between all the nodes.  It is initialised to None
-       until ``.init_distribute()`` is called.
+       until ``.schedule()`` is called.
 
     :pending: List of indices of globally pending tests.  These are
        tests which have not yet been allocated to a chunk for a node
@@ -181,8 +181,8 @@ class LoadScheduling:
     :config: Config object, used for handling hooks.
     """
 
-    def __init__(self, numnodes, log=None, config=None):
+    def __init__(self, config, log=None):
-        self.numnodes = numnodes
+        self.numnodes = len(parse_spec_config(config))
         self.node2collection = {}
         self.node2pending = {}
         self.pending = []
@@ -208,7 +208,20 @@ class LoadScheduling:
         """
         return len(self.node2collection) >= self.numnodes
 
-    def haspending(self):
+    @property
+    def tests_finished(self):
+        """Return True if all tests have been executed by the nodes."""
+        if not self.collection_is_completed:
+            return False
+        if self.pending:
+            return False
+        for pending in self.node2pending.values():
+            if len(pending) >= 2:
+                return False
+        return True
+
+    @property
+    def has_pending(self):
         """Return True if there are pending test items
 
         This indicates that collection has finished and nodes are
@@ -222,11 +235,7 @@ class LoadScheduling:
                 return True
         return False
 
-    def hasnodes(self):
+    def add_node(self, node):
-        """Return True if nodes exist in the scheduler."""
-        return bool(self.node2pending)
-
-    def addnode(self, node):
         """Add a new node to the scheduler.
 
         From now on the node will be allocated chunks of tests to
@@ -238,18 +247,7 @@ class LoadScheduling:
         assert node not in self.node2pending
         self.node2pending[node] = []
 
-    def tests_finished(self):
+    def add_node_collection(self, node, collection):
-        """Return True if all tests have been executed by the nodes."""
-        if not self.collection_is_completed:
-            return False
-        if self.pending:
-            return False
-        for pending in self.node2pending.values():
-            if len(pending) >= 2:
-                return False
-        return True
-
-    def addnode_collection(self, node, collection):
         """Add the collected test items from a node
 
         The collection is stored in the ``.node2collection`` map.
@@ -258,7 +256,7 @@ class LoadScheduling:
         assert node in self.node2pending
         if self.collection_is_completed:
             # A new node has been added later, perhaps an original one died.
-            # .init_distribute() should have
+            # .schedule() should have
             # been called by now
             assert self.collection
             if collection != self.collection:
@@ -271,7 +269,7 @@ class LoadScheduling:
                 return
         self.node2collection[node] = list(collection)
 
-    def remove_item(self, node, item_index, duration=0):
+    def mark_test_complete(self, node, item_index, duration=0):
         """Mark test item as completed by node
 
         The duration it took to execute the item is used as a hint to
@@ -335,7 +333,7 @@ class LoadScheduling:
             self.check_schedule(node)
         return crashitem
 
-    def init_distribute(self):
+    def schedule(self):
         """Initiate distribution of the test collection
 
         Initiate scheduling of the items across the nodes.  If this
@@ -345,8 +343,6 @@ class LoadScheduling:
 
         This is called by the ``DSession.slave_collectionfinish`` hook
         if ``.collection_is_completed`` is True.
-
-        XXX Perhaps this method should have been called ".schedule()".
         """
         assert self.collection_is_completed
 
@@ -466,6 +462,8 @@ class DSession:
         self.log = py.log.Producer("dsession")
         if not config.option.debug:
             py.log.setconsumer(self.log._keywords, None)
+        self.nodemanager = None
+        self.sched = None
         self.shuttingdown = False
         self.countfailures = 0
         self.maxfail = config.getvalue("maxfail")
@@ -521,16 +519,21 @@ class DSession:
         # prohibit collection of test items in master process
         return True
 
-    def pytest_runtestloop(self):
+    @pytest.mark.trylast
-        numnodes = len(self.nodemanager.specs)
+    def pytest_xdist_make_scheduler(self, config, log):
-        dist = self.config.getvalue("dist")
+        dist = config.getvalue("dist")
         if dist == "load":
-            self.sched = LoadScheduling(numnodes, log=self.log,
+            return LoadScheduling(config, log)
-                                        config=self.config)
         elif dist == "each":
-            self.sched = EachScheduling(numnodes, log=self.log)
+            return EachScheduling(config, log)
-        else:
+
-            assert 0, dist
+    def pytest_runtestloop(self):
+        self.sched = self.config.hook.pytest_xdist_make_scheduler(
+            config=self.config,
+            log=self.log
+        )
+        assert self.sched is not None
+
         self.shouldstop = False
         while not self.session_finished:
             self.loop_once()
@@ -553,7 +556,7 @@ class DSession:
         call = getattr(self, method)
         self.log("calling method", method, kwargs)
         call(**kwargs)
-        if self.sched.tests_finished():
+        if self.sched.tests_finished:
             self.triggershutdown()
 
     #
@@ -573,7 +576,7 @@ class DSession:
         if self.shuttingdown:
             node.shutdown()
         else:
-            self.sched.addnode(node)
+            self.sched.add_node(node)
 
     def slave_slavefinished(self, node):
         """Emitted when node executes its pytest_sessionfinish hook.
@@ -635,16 +638,16 @@ class DSession:
         # tell session which items were effectively collected otherwise
         # the master node will finish the session with EXIT_NOTESTSCOLLECTED
         self._session.testscollected = len(ids)
-        self.sched.addnode_collection(node, ids)
+        self.sched.add_node_collection(node, ids)
         if self.terminal:
             self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
         if self.sched.collection_is_completed:
-            if self.terminal and not self.sched.haspending():
+            if self.terminal and not self.sched.has_pending:
                 self.trdist.ensure_show_status()
                 self.terminal.write_line("")
                 self.terminal.write_line("scheduling tests via %s" % (
                     self.sched.__class__.__name__))
-            self.sched.init_distribute()
+            self.sched.schedule()
 
     def slave_logstart(self, node, nodeid, location):
         """Emitted when a node calls the pytest_runtest_logstart hook."""
@@ -658,7 +661,7 @@ class DSession:
         the item from the pending list in the scheduler.
         """
         if rep.when == "call" or (rep.when == "setup" and not rep.passed):
-            self.sched.remove_item(node, rep.item_index, rep.duration)
+            self.sched.mark_test_complete(node, rep.item_index, rep.duration)
         # self.report_line("testreport %s: %s" %(rep.id, rep.status))
         rep.node = node
         self.config.hook.pytest_runtest_logreport(report=rep)

xdist/newhooks.py

@@ -11,6 +11,7 @@ must be taken in plugins in case ``xdist`` is not installed. Please see:
 
     http://pytest.org/latest/writing_plugins.html#optionally-using-hooks-from-3rd-party-plugins
 """
+import pytest
 
 
 def pytest_xdist_setupnodes(config, specs):
@@ -44,3 +45,8 @@ def pytest_testnodedown(node, error):
 def pytest_xdist_node_collection_finished(node, ids):
     """called by the master node when a node finishes collecting.
     """
+
+
+@pytest.mark.firstresult
+def pytest_xdist_make_scheduler(config, log):
+    """ return a node scheduler implementation """

xdist/plugin.py

@@ -32,7 +32,7 @@ def pytest_addoption(parser):
     group._addoption(
         '--dist', metavar="distmode",
         action="store", choices=['load', 'each', 'no'],
-        type="choice", dest="dist", default="no",
+        dest="dist", default="no",
         help=("set mode for distributing tests to exec environments.\n\n"
               "each: send each test to each available environment.\n\n"
               "load: send each test to available environment.\n\n"

xdist/remote.py

@@ -88,6 +88,7 @@ class SlaveInteractor:
     def pytest_runtest_logreport(self, report):
         data = serialize_report(report)
         data["item_index"] = self.item_index
+        data["worker_id"] = self.slaveid
         assert self.session.items[self.item_index].nodeid == report.nodeid
         self.sendevent("testreport", data=data)
 

xdist/slavemanage.py

@@ -10,6 +10,22 @@ import xdist.remote
 from _pytest import runner  # XXX load dynamically
 
 
+def parse_spec_config(config):
+    xspeclist = []
+    for xspec in config.getvalue("tx"):
+        i = xspec.find("*")
+        try:
+            num = int(xspec[:i])
+        except ValueError:
+            xspeclist.append(xspec)
+        else:
+            xspeclist.extend([xspec[i + 1:]] * num)
+    if not xspeclist:
+        raise pytest.UsageError(
+            "MISSING test execution (tx) nodes: please specify --tx")
+    return xspeclist
+
+
 class NodeManager(object):
     EXIT_TIMEOUT = 10
     DEFAULT_IGNORES = ['.*', '*.pyc', '*.pyo', '*~']
@@ -62,19 +78,7 @@ class NodeManager(object):
         self.group.terminate(self.EXIT_TIMEOUT)
 
     def _getxspecs(self):
-        xspeclist = []
+        return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
-        for xspec in self.config.getvalue("tx"):
-            i = xspec.find("*")
-            try:
-                num = int(xspec[:i])
-            except ValueError:
-                xspeclist.append(xspec)
-            else:
-                xspeclist.extend([xspec[i+1:]] * num)
-        if not xspeclist:
-            raise pytest.UsageError(
-                "MISSING test execution (tx) nodes: please specify --tx")
-        return [execnet.XSpec(x) for x in xspeclist]
 
     def _getrsyncdirs(self):
         for spec in self.specs:
@@ -292,7 +296,7 @@ class SlaveController(object):
                     self._down = True
                 return
             eventname, kwargs = eventcall
-            if eventname in ("collectionstart"):
+            if eventname in ("collectionstart",):
                 self.log("ignoring %s(%s)" % (eventname, kwargs))
             elif eventname == "slaveready":
                 self.notify_inproc(eventname, node=self, **kwargs)