Load-Distribute test cases by filename

README.rst

@@ -321,6 +321,22 @@ the same worker ``gw0``, while the test methods from classes ``TestHDF`` and
 Currently the groupings can't be customized, with grouping by class takes
 priority over grouping by module.
 
+Sending tests to the same worker based on their file
+++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+*New in version 1.21.*
+
+.. note::
+    This is an **experimental** feature: the actual functionality will
+    likely stay the same, but the CLI might change slightly in future versions.
+
+You can send tests to the same worker grouped by their filename by using the
+``--dist=loadfile`` option, so tests of the same file are guaranteed to run
+in the same worker.
+
+Using the example in the previous section, all tests from ``test_container.py`` will
+run in the same worker, as well as the tests in ``test_io.py``.
+
 
 Specifying "rsync" dirs in an ini-file
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

changelog/242.feature

@@ -0,0 +1 @@
+New ``--dist=loadfile`` option which load-distributes test to workers grouped by the file the tests live in.

testing/acceptance_test.py

@@ -856,6 +856,61 @@ def test(self, i):
             'test_a.py::TestB', result.outlines) in ({'gw0': 10}, {'gw1': 10})
 
 
+class TestFileScope:
+
+    def test_by_module(self, testdir):
+        test_file = """
+            import pytest
+            class TestA:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+
+            class TestB:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+        """
+        testdir.makepyfile(
+            test_a=test_file,
+            test_b=test_file,
+        )
+        result = testdir.runpytest('-n2', '--dist=loadfile', '-v')
+        test_a_workers_and_test_count = get_workers_and_test_count_by_prefix(
+            'test_a.py::TestA', result.outlines)
+        test_b_workers_and_test_count = get_workers_and_test_count_by_prefix(
+            'test_b.py::TestB', result.outlines)
+
+        assert test_a_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \
+            test_a_workers_and_test_count in ({'gw0': 0}, {'gw1': 10})
+        assert test_b_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \
+            test_b_workers_and_test_count in ({'gw0': 0}, {'gw1': 10})
+
+    def test_by_class(self, testdir):
+        testdir.makepyfile(test_a="""
+            import pytest
+            class TestA:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+
+            class TestB:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+        """)
+        result = testdir.runpytest('-n2', '--dist=loadfile', '-v')
+        test_a_workers_and_test_count = get_workers_and_test_count_by_prefix(
+            'test_a.py::TestA', result.outlines)
+        test_b_workers_and_test_count = get_workers_and_test_count_by_prefix(
+            'test_a.py::TestB', result.outlines)
+
+        assert test_a_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \
+            test_a_workers_and_test_count in ({'gw0': 0}, {'gw1': 10})
+        assert test_b_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \
+            test_b_workers_and_test_count in ({'gw0': 0}, {'gw1': 10})
+
+
 def parse_tests_and_workers_from_output(lines):
     result = []
     for line in lines:

xdist/dsession.py

@@ -6,6 +6,7 @@
     EachScheduling,
     LoadScheduling,
     LoadScopeScheduling,
+    LoadFileScheduling,
 )
 
 
@@ -99,6 +100,7 @@ def pytest_xdist_make_scheduler(self, config, log):
             'each': EachScheduling,
             'load': LoadScheduling,
             'loadscope': LoadScopeScheduling,
+            'loadfile': LoadFileScheduling,
         }
         return schedulers[dist](config, log)
 

xdist/plugin.py

@@ -31,14 +31,16 @@ def pytest_addoption(parser):
                          "when crashed (set to zero to disable this feature)")
     group.addoption(
         '--dist', metavar="distmode",
-        action="store", choices=['each', 'load', 'loadscope', 'no'],
+        action="store", choices=['each', 'load', 'loadscope', 'loadfile', 'no'],
         dest="dist", default="no",
         help=("set mode for distributing tests to exec environments.\n\n"
               "each: send each test to all available environments.\n\n"
               "load: load balance by sending any pending test to any"
               " available environment.\n\n"
               "loadscope: load balance by sending pending groups of tests in"
               " the same scope to any available environment.\n\n"
+              "loadfile: load balance by sending test grouped by file"
+              " to any available environment.\n\n"
               "(default) no: run tests inprocess, don't distribute."))
     group.addoption(
         '--tx', dest="tx", action="append", default=[],

xdist/scheduler/__init__.py

@@ -1,3 +1,4 @@
 from xdist.scheduler.each import EachScheduling  # noqa
 from xdist.scheduler.load import LoadScheduling  # noqa
 from xdist.scheduler.loadscope import LoadScopeScheduling  # noqa
+from xdist.scheduler.filescope import LoadFileScheduling  # noqa

xdist/scheduler/filescope.py

@@ -0,0 +1,52 @@
+from . import LoadScopeScheduling
+from py.log import Producer
+
+
+class LoadFileScheduling(LoadScopeScheduling):
+    """Implement load scheduling across nodes, but grouping test test file.
+
+    This distributes the tests collected across all nodes so each test is run
+    just once.  All nodes collect and submit the list of tests and when all
+    collections are received it is verified they are identical collections.
+    Then the collection gets divided up in work units, grouped by test file,
+    and those work units get submitted to nodes.  Whenever a node finishes an
+    item, it calls ``.mark_test_complete()`` which will trigger the scheduler
+    to assign more work units if the number of pending tests for the node falls
+    below a low-watermark.
+
+    When created, ``numnodes`` defines how many nodes are expected to submit a
+    collection. This is used to know when all nodes have finished collection.
+
+    This class behaves very much like LoadScopeScheduling, but with a file-level scope.
+    """
+
+    def __init(self, config, log=None):
+        super(LoadFileScheduling, self).__init__(config, log)
+        if log is None:
+            self.log = Producer('loadfilesched')
+        else:
+            self.log = log.loadfilesched
+
+    def _split_scope(self, nodeid):
+        """Determine the scope (grouping) of a nodeid.
+
+        There are usually 3 cases for a nodeid::
+
+            example/loadsuite/test/test_beta.py::test_beta0
+            example/loadsuite/test/test_delta.py::Delta1::test_delta0
+            example/loadsuite/epsilon/__init__.py::epsilon.epsilon
+
+        #. Function in a test module.
+        #. Method of a class in a test module.
+        #. Doctest in a function in a package.
+
+        This function will group tests with the scope determined by splitting
+        the first ``::`` from the left. That is, test will be grouped in a
+        single work unit when they reside in the same file.
+         In the above example, scopes will be::
+
+            example/loadsuite/test/test_beta.py
+            example/loadsuite/test/test_delta.py
+            example/loadsuite/epsilon/__init__.py
+        """
+        return nodeid.split('::', 1)[0]

xdist/scheduler/loadscope.py

@@ -368,12 +368,12 @@ def schedule(self):
         extra_nodes = len(self.nodes) - len(self.workqueue)
 
         if extra_nodes > 0:
-            self.log('Shuting down {} nodes'.format(extra_nodes))
+            self.log('Shuting down {0} nodes'.format(extra_nodes))
 
             for _ in range(extra_nodes):
                 unused_node, assigned = self.assigned_work.popitem(last=True)
 
-                self.log('Shuting down unused node {}'.format(unused_node))
+                self.log('Shuting down unused node {0}'.format(unused_node))
                 unused_node.shutdown()
 
         # Assign initial workload