Fix CHANGELOG for 1.14 release

Add new pytest_xdist_node_collection_finished hook

.gitignore

@@ -22,10 +22,11 @@ dist/
 include/
 lib/
 bin/
-xdist/_version.py
+xdist/_version.py*
 pytest_xdist.egg-info
 issue/
 3rdparty/
 pytestdebug.log
-.tox
+.tox/
-.cache
+.cache/
+.eggs/

.hgtags

@@ -18,3 +18,4 @@ cd44a941c833c098e4899fe3d42a96703754d0d5 1.5
 220f6e46eb71a6212ccbe6b67b9e6edcf8ee4fa5 1.11
 39ef85dbc893cc63dede11601208098a667b58e9 1.12
 4e25f4c568be2d7cb4d1739638a9e66bbf28f588 v1.13
+67ff3aa4d294f75ade0ebf03267e5739e1bd9473 v1.13.1

.travis.yml

@@ -0,0 +1,46 @@
+sudo: false
+language: python
+python:
+    - '3.5'
+# command to install dependencies
+install: "pip install -U tox setuptools_scm"
+# # command to run tests
+env:
+  matrix:
+  - TESTENV=flakes
+  - TESTENV=readme
+  # matrix was trimmed to skip
+  # some builds that are unnecessary/perceived redundant:
+  # py27 and py35 are tested with all pytest versions we declare to support;
+  # py26 and py34 only with latest pytest versions.
+  # pytest24,25,26 don't work on py35 due to changes in ast module
+  - TESTENV=py26-pytest27
+  - TESTENV=py26-pytest28
+
+  - TESTENV=py27-pytest24
+  - TESTENV=py27-pytest25
+  - TESTENV=py27-pytest26
+  - TESTENV=py27-pytest27
+  - TESTENV=py27-pytest28
+
+  - TESTENV=py34-pytest27
+  - TESTENV=py34-pytest28
+
+  - TESTENV=py35-pytest27
+  - TESTENV=py35-pytest28
+
+  - TESTENV=py27-pytest28-pexpect
+  - TESTENV=py34-pytest28-pexpect
+  - TESTENV=pypy-pytest28
+
+script: tox --recreate -e $TESTENV
+
+notifications:
+  irc:
+    channels:
+      - "chat.freenode.net#pytest"
+    on_success: change
+    on_failure: change
+    skip_join: true
+  email:
+    - pytest-commit@python.org

CHANGELOG

@@ -1,3 +1,19 @@
+1.14
+----
+
+- new hook: ``pytest_xdist_node_collection_finished(node, ids)``, called when
+  a worker has finished collection. Thanks Omer Katz for the request and
+  Bruno Oliveira for the PR.
+
+- fix README display on pypi
+
+- fix #22: xdist now works if the internal tmpdir plugin is disabled.
+  Thanks Bruno Oliveira for the PR.
+
+- fix #32: xdist now works if looponfail or boxed are disabled.
+  Thanks Bruno Oliveira for the PR.
+
+
 1.13.1
 -------
 

OVERVIEW.md

@@ -0,0 +1,76 @@
+# Overview #
+
+`xdist` works by spawning one or more **workers**, which are controlled
+by the **master**. Each **worker** is responsible for performing 
+a full test collection and afterwards running tests as dictated by the **master**.
+
+The execution flow is:
+
+1. **master** spawns one or more **workers** at the beginning of
+   the test session. The communication between **master** and **worker** nodes makes use of 
+   [execnet](http://codespeak.net/execnet/) and its [gateways](http://codespeak.net/execnet/basics.html#gateways-bootstrapping-python-interpreters).
+   The actual interpreters executing the code for the **workers** might
+   be remote or local. 
+  
+1. Each **worker** itself is a mini pytest runner. **workers** at this
+   point perform a full test collection, sending back the collected 
+   test-ids back to the **master** which does not
+   perform any collection itself.
+     
+1. The **master** receives the result of the collection from all nodes.
+   At this point the **master** performs some sanity check to ensure that
+   all **workers** collected the same tests (including order), bailing out otherwise.
+   If all is well, it converts the list of test-ids into a list of simple
+   indexes, where each index corresponds to the position of that test in the
+   original collection list. This works because all nodes have the same 
+   collection list, and saves bandwidth because the **master** can now tell
+   one of the workers to just *execute test index 3* index of passing the
+   full test id.
+   
+1. If **dist-mode** is **each**: the **master** just sends the full list
+   of test indexes to each node at this moment.
+   
+1. If **dist-mode** is **load**: the **master** takes around 25% of the
+   tests and sends them one by one to each **worker** in a round robin
+   fashion. The rest of the tests will be distributed later as **workers**
+   finish tests (see below).
+   
+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 
+   waiting for **master** to send tests for execution. As tests are
+   received by **workers**, `pytest_runtest_protocol` is executed for each test. 
+   Here it worth noting an implementation detail: **workers** always must keep at 
+   least one test item on their queue due to how the `pytest_runtest_protocol(item, nextitem)` 
+   hook is defined: in order to pass the `nextitem` to the hook, the worker must wait for more 
+   instructions from master before executing that remaining test. If it receives more tests, 
+   then it can safely call `pytest_runtest_protocol` because it knows what the `nextitem` parameter will be. 
+   If it receives a "shutdown" signal, then it can execute the hook passing `nextitem` as `None`. 
+   
+1. As tests are started and completed at the **workers**, the results are sent
+   back to the **master**, which then just forwards the results to 
+   the appropriate pytest hooks: `pytest_runtest_logstart` and 
+   `pytest_runtest_logreport`. This way other plugins (for example `junitxml`)
+   can work normally. The **master** (when in dist-mode **load**) 
+   decides to send more tests to a node when a test completes, using
+   some heuristics such as test durations and how many tests each **worker**
+   still has to run.
+   
+1. When the **master** has no more pending tests it will
+   send a "shutdown" signal to all **workers**, which will then run their 
+   remaining tests to completion and shut down. At this point the 
+   **master** will sit waiting for **workers** to shut down, still
+   processing events such as `pytest_runtest_logreport`.
+ 
+## FAQ ##
+
+> Why does each worker do its own collection, as opposed to having 
+the master collect once and distribute from that collection to the workers?
+
+If collection was performed by master then it would have to 
+serialize collected items to send them through the wire, as workers live in another process. 
+The problem is that test items are not easily (impossible?) to serialize, as they contain references to 
+the test functions, fixture managers, config objects, etc. Even if one manages to serialize it, 
+it seems it would be very hard to get it right and easy to break by any small change in pytest. 
+  
+

README.rst

@@ -1,10 +1,15 @@
-.. image:: https://drone.io/bitbucket.org/pytest-dev/pytest-xdist/status.png
+
-   :target: https://drone.io/bitbucket.org/pytest-dev/pytest-xdist/latest
+.. image:: https://travis-ci.org/pytest-dev/pytest-xdist.svg?branch=master
-.. image:: https://pypip.in/v/pytest-xdist/badge.png
+    :target: https://travis-ci.org/pytest-dev/pytest-xdist
+
+.. image:: http://img.shields.io/pypi/v/pytest-xdist.svg
    :target: https://pypi.python.org/pypi/pytest-xdist
 
+.. image:: https://ci.appveyor.com/api/projects/status/56eq1a1avd4sdd7e/branch/master?svg=true
+    :target: https://ci.appveyor.com/project/pytestbot/pytest-xdist
+
 xdist: pytest distributed testing plugin
-============================
+=========================================
 
 The `pytest-xdist`_ plugin extends py.test with some unique
 test execution modes:
@@ -118,24 +123,27 @@ you can ad-hoc distribute your tests by typing::
 
     py.test -d --tx ssh=myhostpopen --rsyncdir mypkg mypkg
 
-This will synchronize your ``mypkg`` package directory
+This will synchronize your :code:`mypkg` package directory
 to an remote ssh account and then locally collect tests
 and send them to remote places for execution.
 
-You can specify multiple ``--rsyncdir`` directories
+You can specify multiple :code:`--rsyncdir` directories
 to be sent to the remote side.
 
-**NOTE:** For py.test to collect and send tests correctly
+.. note::
+
+  For py.test to collect and send tests correctly
   you not only need to make sure all code and tests
   directories are rsynced, but that any test (sub) directory
-also has an ``__init__.py`` file because internally
+  also has an :code:`__init__.py` file because internally
   py.test references tests as a fully qualified python
   module path.  **You will otherwise get strange errors**
   during setup of the remote side.
 
-You can specify multiple ``--rsyncignore`` glob-patterns
+
+You can specify multiple :code:`--rsyncignore` glob patterns
 to be ignored when file are sent to the remote side.
-There are also internal ignores: .*, *.pyc, *.pyo, *~
+There are also internal ignores: :code:`.*, *.pyc, *.pyo, *~`
 Those you cannot override using rsyncignore command-line or
 ini-file option(s).
 
@@ -171,7 +179,7 @@ environment this command will send each tests to all
 platforms - and report back failures from all platforms
 at once.   The specifications strings use the `xspec syntax`_.
 
-.. _`xspec syntax`: http://codespeak.net/execnet/trunk/basics.html#xspec
+.. _`xspec syntax`: http://codespeak.net/execnet/basics.html#xspec
 
 .. _`socketserver.py`: http://bitbucket.org/hpk42/execnet/raw/2af991418160/execnet/script/socketserver.py
 
@@ -212,10 +220,11 @@ These directory specifications are relative to the directory
 where the configuration file was found.
 
 .. _`pytest-xdist`: http://pypi.python.org/pypi/pytest-xdist
-.. _`pytest-xdist repository`: http://bitbucket.org/pytest-dev/pytest-xdist
+.. _`pytest-xdist repository`: https://github.com/pytest-dev/pytest-xdist
 .. _`pytest`: http://pytest.org
 
 Issue and Bug Tracker
 ------------------------
 
-Please use the pytest issue tracker for bugs in this plugin, see https://bitbucket.org/hpk42/pytest/issues .
+Please use the `pytest issue tracker <https://github.com/pytest-dev/pytest/issues>`_
+for reporting bugs in this plugin.

appveyor.yml

@@ -0,0 +1,17 @@
+environment:
+  matrix:
+    - PYTHON: "C:\\Python35"
+      TESTENV: "py35"
+
+    - PYTHON: "C:\\Python27"
+      TESTENV: "py27"
+
+install:
+  - "SET PATH=%PYTHON%;%PYTHON%\\Scripts;%PATH%"
+  - pip install tox setuptools_scm"
+
+build: false  # Not a C# project, build stuff at the test step instead.
+
+test_script:
+  # Build the compiled extension and run the project tests
+  - tox -e %TESTENV%-pytest27,%TESTENV%-pytest28,readme,flakes

example/boxed.txt

@@ -2,7 +2,7 @@
 
 If your testing involves C or C++ libraries you might have to deal
 with crashing processes.  The xdist-plugin provides the ``--boxed`` option
-to run each test in a controled subprocess.  Here is a basic example::
+to run each test in a controlled subprocess.  Here is a basic example::
 
     # content of test_module.py
 

setup.py

@@ -3,12 +3,13 @@ from setuptools import setup
 setup(
     name="pytest-xdist",
     use_scm_version={'write_to': 'xdist/_version.py'},
-    description='py.test xdist plugin for distributed testing and loop-on-failing modes',
+    description='py.test xdist plugin for distributed testing'
-    long_description=open('README.txt').read(),
+                ' and loop-on-failing modes',
+    long_description=open('README.rst').read(),
     license='MIT',
     author='holger krekel and contributors',
     author_email='pytest-dev@python.org,holger@merlinux.eu',
-    url='http://bitbucket.org/hpk42/pytest-xdist',
+    url='https://github.com/pytest-dev/pytest-xdist',
     platforms=['linux', 'osx', 'win32'],
     packages=['xdist'],
     entry_points={

testing/acceptance_test.py

@@ -10,9 +10,7 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines([
+        result.stdout.fnmatch_lines(["*1 passed*", ])
-            "*1 passed*",
-        ])
 
     def test_n1_fail(self, testdir):
         p1 = testdir.makepyfile("""
@@ -21,9 +19,7 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 1
-        result.stdout.fnmatch_lines([
+        result.stdout.fnmatch_lines(["*1 failed*", ])
-            "*1 failed*",
-        ])
 
     def test_n1_import_error(self, testdir):
         p1 = testdir.makepyfile("""
@@ -57,9 +53,7 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines([
+        result.stdout.fnmatch_lines(["*1 skipped*", ])
-            "*1 skipped*",
-        ])
 
     def test_manytests_to_one_import_error(self, testdir):
         p1 = testdir.makepyfile("""
@@ -84,8 +78,7 @@ class TestDistribution:
                     pass
                 def test_skip():
                     py.test.skip("hello")
-            """,
+            """, )
-        )
         result = testdir.runpytest(p1, "-v", '-d', '--tx=popen', '--tx=popen')
         result.stdout.fnmatch_lines([
             "*1*Python*",
@@ -115,9 +108,7 @@ class TestDistribution:
         """ % str(testdir.tmpdir))
         result = testdir.runpytest_subprocess(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines([
+        result.stdout.fnmatch_lines(["*1 passed*", ])
-            "*1 passed*",
-        ])
 
     def test_dist_ini_specified(self, testdir):
         p1 = testdir.makepyfile("""
@@ -130,8 +121,7 @@ class TestDistribution:
                     pass
                 def test_skip():
                     py.test.skip("hello")
-            """,
+            """, )
-        )
         testdir.makeini("""
             [pytest]
             addopts = --tx=3*popen
@@ -163,13 +153,10 @@ class TestDistribution:
                     import os
                     time.sleep(0.5)
                     os.kill(os.getpid(), 15)
-            """
+            """)
-        )
         result = testdir.runpytest(p1, "-v", '-d', '-n1')
         result.stdout.fnmatch_lines([
-            "*Python*",
+            "*Python*", "*PASS**test_ok*", "*node*down*",
-            "*PASS**test_ok*",
-            "*node*down*",
             "*3 failed, 1 passed, 1 skipped*"
         ])
         assert result.ret == 1
@@ -221,14 +208,13 @@ class TestDistribution:
         p1 = testdir.makepyfile("def test_func(): pass")
         result = testdir.runpytest("-v", p1, '-d', '--tx=popen')
         result.stdout.fnmatch_lines([
-            "*0*Python*",
+            "*0*Python*", "*calculated result is 49*", "*1 passed*"
-            "*calculated result is 49*",
-            "*1 passed*"
         ])
         assert result.ret == 0
 
     def test_keyboardinterrupt_hooks_issue79(self, testdir):
-        testdir.makepyfile(__init__="", test_one="""
+        testdir.makepyfile(__init__="",
+                           test_one="""
             def test_hello():
                 raise KeyboardInterrupt()
         """)
@@ -264,17 +250,20 @@ class TestDistribution:
         child.close()
         # assert ret == 2
 
+
 class TestDistEach:
     def test_simple(self, testdir):
         testdir.makepyfile("""
             def test_hello():
                 pass
         """)
-        result = testdir.runpytest_subprocess("--debug", "--dist=each", "--tx=2*popen")
+        result = testdir.runpytest_subprocess("--debug", "--dist=each",
+                                              "--tx=2*popen")
         assert not result.ret
         result.stdout.fnmatch_lines(["*2 pass*"])
 
-    @py.test.mark.xfail(run=False,
+    @py.test.mark.xfail(
+        run=False,
         reason="other python versions might not have py.test installed")
     def test_simple_diffoutput(self, testdir):
         interpreters = []
@@ -284,7 +273,8 @@ class TestDistEach:
                 py.test.skip("%s not found" % name)
             interpreters.append(interp)
 
-        testdir.makepyfile(__init__="", test_one="""
+        testdir.makepyfile(__init__="",
+                           test_one="""
             import sys
             def test_hello():
                 print("%s...%s" % sys.version_info[:2])
@@ -298,6 +288,7 @@ class TestDistEach:
         assert "2...5" in s
         assert "2...6" in s
 
+
 class TestTerminalReporting:
     def test_pass_skip_fail(self, testdir):
         testdir.makepyfile("""
@@ -335,6 +326,7 @@ class TestTerminalReporting:
             "E       assert 0",
         ])
 
+
 def test_teardownfails_one_function(testdir):
     p = testdir.makepyfile("""
         def test_func():
@@ -344,10 +336,10 @@ def test_teardownfails_one_function(testdir):
     """)
     result = testdir.runpytest(p, '-n1', '--tx=popen')
     result.stdout.fnmatch_lines([
-        "*def teardown_function(function):*",
+        "*def teardown_function(function):*", "*1 passed*1 error*"
-        "*1 passed*1 error*"
     ])
 
+
 @py.test.mark.xfail
 def test_terminate_on_hangingnode(testdir):
     p = testdir.makeconftest("""
@@ -358,22 +350,7 @@ def test_terminate_on_hangingnode(testdir):
     """)
     result = testdir.runpytest(p, '--dist=each', '--tx=popen//id=my')
     assert result.duration < 2.0
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*killed*my*", ])
-        "*killed*my*",
-    ])
-
-
-def test_auto_detect_cpus(testdir, monkeypatch):
-    import multiprocessing
-    monkeypatch.setattr(multiprocessing, 'cpu_count', lambda: 3)
-    testdir.makeconftest("""
-        def pytest_unconfigure(config):
-            with open('cpus', 'w') as f:
-                f.write('cpus = %s' % config.option.numprocesses)
-    """)
-    testdir.inline_run('-n=auto')
-    cpus_file = testdir.tmpdir.join('cpus')
-    assert cpus_file.read() == 'cpus = 3'
 
 
 @pytest.mark.xfail(reason="works if run outside test suite", run=False)
@@ -400,10 +377,7 @@ def test_session_hooks(testdir):
             assert hasattr(sys, 'pytestsessionhooks')
     """)
     result = testdir.runpytest(p, "--dist=each", "--tx=popen")
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*ValueError*", "*1 passed*", ])
-        "*ValueError*",
-        "*1 passed*",
-    ])
     assert not result.ret
     d = result.parseoutcomes()
     assert d['passed'] == 1
@@ -446,12 +420,10 @@ def test_funcarg_teardown_failure(testdir):
             pass
     """)
     result = testdir.runpytest_subprocess("--debug", p)  # , "-n1")
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*ValueError*42*", "*1 passed*1 error*", ])
-        "*ValueError*42*",
-        "*1 passed*1 error*",
-    ])
     assert result.ret
 
+
 def test_crashing_item(testdir):
     p = testdir.makepyfile("""
         import py
@@ -463,12 +435,10 @@ def test_crashing_item(testdir):
     """)
     result = testdir.runpytest("-n2", p)
     result.stdout.fnmatch_lines([
-        "*crashed*test_crash*",
+        "*crashed*test_crash*", "*1 failed*1 passed*"
-        "*1 failed*1 passed*"
     ])
 
 
-
 def test_skipping(testdir):
     p = testdir.makepyfile("""
         import pytest
@@ -477,10 +447,8 @@ def test_skipping(testdir):
     """)
     result = testdir.runpytest("-n1", '-rs', p)
     assert result.ret == 0
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*hello*", "*1 skipped*"])
-        "*hello*",
+
-        "*1 skipped*"
-    ])
 
 def test_issue34_pluginloading_in_subprocess(testdir):
     testdir.tmpdir.join("plugin123.py").write(py.code.Source("""
@@ -494,9 +462,7 @@ def test_issue34_pluginloading_in_subprocess(testdir):
     """)
     result = testdir.runpytest_subprocess("-n1", "-p", "plugin123")
     assert result.ret == 0
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*1 passed*", ])
-        "*1 passed*",
-    ])
 
 
 def test_fixture_scope_caching_issue503(testdir):
@@ -505,7 +471,8 @@ def test_fixture_scope_caching_issue503(testdir):
 
             @pytest.fixture(scope='session')
             def fix():
-                assert fix.counter == 0, 'session fixture was invoked multiple times'
+                assert fix.counter == 0, \
+                    'session fixture was invoked multiple times'
                 fix.counter += 1
             fix.counter = 0
 
@@ -517,9 +484,7 @@ def test_fixture_scope_caching_issue503(testdir):
     """)
     result = testdir.runpytest(p1, '-v', '-n1')
     assert result.ret == 0
-    result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines(["*2 passed*", ])
-    "*2 passed*",
-    ])
 
 
 def test_issue_594_random_parametrize(testdir):
@@ -544,8 +509,32 @@ def test_issue_594_random_parametrize(testdir):
     ])
 
 
-class TestNodeFailure:
+def test_tmpdir_disabled(testdir):
+    """Test xdist doesn't break if internal tmpdir plugin is disabled (#22).
+    """
+    p1 = testdir.makepyfile("""
+        def test_ok():
+            pass
+    """)
+    result = testdir.runpytest(p1, "-n1", '-p', 'no:tmpdir')
+    assert result.ret == 0
+    result.stdout.fnmatch_lines("*1 passed*")
 
+
+@pytest.mark.parametrize('plugin', ['xdist.looponfail', 'xdist.boxed'])
+def test_sub_plugins_disabled(testdir, plugin):
+    """Test that xdist doesn't break if we disable any of its sub-plugins. (#32)
+    """
+    p1 = testdir.makepyfile("""
+        def test_ok():
+            pass
+    """)
+    result = testdir.runpytest(p1, "-n1", '-p', 'no:%s' % plugin)
+    assert result.ret == 0
+    result.stdout.fnmatch_lines("*1 passed*")
+
+
+class TestNodeFailure:
     def test_load_single(self, testdir):
         f = testdir.makepyfile("""
             import os
@@ -587,6 +576,7 @@ class TestNodeFailure:
             "*1 failed*1 passed*",
         ])
 
+    @pytest.mark.xfail(reason='#20: xdist race condition on node restart')
     def test_each_multiple(self, testdir):
         f = testdir.makepyfile("""
             import os
@@ -617,7 +607,6 @@ class TestNodeFailure:
             "*2 failed*2 passed*",
         ])
 
-
     def test_disable_restart(self, testdir):
         f = testdir.makepyfile("""
             import os

testing/conftest.py

@@ -2,6 +2,7 @@ import py
 import pytest
 import execnet
 
+
 @pytest.fixture(scope="session", autouse=True)
 def _ensure_imports():
     # we import some modules because pytest-2.8's testdir fixture
@@ -10,28 +11,36 @@ def _ensure_imports():
     execnet.Group
     execnet.makegateway
 
+
 pytest_plugins = "pytester"
 
 # rsyncdirs = ['.', '../xdist', py.path.local(execnet.__file__).dirpath()]
 
+
 @pytest.fixture(autouse=True)
 def _divert_atexit(request, monkeypatch):
     import atexit
     l = []
+
     def finish():
         while l:
             l.pop()()
+
     monkeypatch.setattr(atexit, "register", l.append)
     request.addfinalizer(finish)
 
+
 def pytest_addoption(parser):
     parser.addoption('--gx',
-       action="append", dest="gspecs",
+                     action="append",
+                     dest="gspecs",
                      help=("add a global test environment, XSpec-syntax. "))
 
+
 def pytest_funcarg__specssh(request):
     return getspecssh(request.config)
 
+
 @pytest.fixture
 def testdir(testdir):
     # pytest before 2.8 did not have a runpytest_subprocess
@@ -39,10 +48,11 @@ def testdir(testdir):
         testdir.runpytest_subprocess = testdir.runpytest
     return testdir
 
+
 # configuration information for tests
 def getgspecs(config):
-    return [execnet.XSpec(spec)
+    return [execnet.XSpec(spec) for spec in config.getvalueorskip("gspecs")]
-                for spec in config.getvalueorskip("gspecs")]
+
 
 def getspecssh(config):
     xspecs = getgspecs(config)
@@ -53,10 +63,10 @@ def getspecssh(config):
             return str(spec)
     py.test.skip("need '--gx ssh=...'")
 
+
 def getsocketspec(config):
     xspecs = getgspecs(config)
     for spec in xspecs:
         if spec.socket:
             return spec
     py.test.skip("need '--gx socket=...'")
-

testing/test_boxed.py

@@ -4,6 +4,7 @@ import os
 needsfork = pytest.mark.skipif(not hasattr(os, "fork"),
                                reason="os.fork required")
 
+
 @needsfork
 def test_functional_boxed(testdir):
     p1 = testdir.makepyfile("""
@@ -17,6 +18,7 @@ def test_functional_boxed(testdir):
         "*1 failed*"
     ])
 
+
 @needsfork
 @pytest.mark.parametrize("capmode", [
     "no",
@@ -41,6 +43,7 @@ def test_functional_boxed_capturing(testdir, capmode):
         *1 failed*
     """)
 
+
 class TestOptionEffects:
     def test_boxed_option_default(self, testdir):
         tmpdir = testdir.tmpdir.ensure("subdir", dir=1)
@@ -53,4 +56,3 @@ class TestOptionEffects:
     def test_is_not_boxed_by_default(self, testdir):
         config = testdir.parseconfig(testdir.tmpdir)
         assert not config.option.boxed
-

testing/test_dsession.py

@@ -1,8 +1,5 @@
 from xdist.dsession import (
-    DSession,
+    DSession, LoadScheduling, EachScheduling, report_collection_diff,
-    LoadScheduling,
-    EachScheduling,
-    report_collection_diff,
 )
 import py
 import pytest
@@ -10,23 +7,27 @@ import execnet
 
 XSpec = execnet.XSpec
 
+
 def run(item, node, excinfo=None):
     runner = item.config.pluginmanager.getplugin("runner")
-    rep = runner.ItemTestReport(item=item,
+    rep = runner.ItemTestReport(item=item, excinfo=excinfo, when="call")
-        excinfo=excinfo, when="call")
     rep.node = node
     return rep
 
+
 class MockGateway:
     _count = 0
+
     def __init__(self):
         self.id = str(self._count)
         self._count += 1
 
+
 class MockNode:
     def __init__(self):
         self.sent = []
         self.gateway = MockGateway()
+        self._shutdown = False
 
     def send_runtest_some(self, indices):
         self.sent.extend(indices)
@@ -37,10 +38,16 @@ class MockNode:
     def shutdown(self):
         self._shutdown = True
 
+    @property
+    def shutting_down(self):
+        return self._shutdown
+
+
 def dumpqueue(queue):
     while queue.qsize():
         print(queue.get())
 
+
 class TestEachScheduling:
     def test_schedule_load_simple(self):
         node1 = MockNode()
@@ -81,6 +88,7 @@ class TestEachScheduling:
         assert sched.tests_finished()
         assert not sched.hasnodes()
 
+
 class TestLoadScheduling:
     def test_schedule_load_simple(self):
         sched = LoadScheduling(2)
@@ -97,16 +105,15 @@ class TestLoadScheduling:
         assert sched.node2collection[node2] == collection
         sched.init_distribute()
         assert not sched.pending
-        assert not sched.tests_finished()
+        assert sched.tests_finished()
-        assert len(node1.sent) == 2
+        assert len(node1.sent) == 1
-        assert len(node2.sent) == 0
+        assert len(node2.sent) == 1
-        assert node1.sent == [0, 1]
+        assert node1.sent == [0]
+        assert node2.sent == [1]
         sched.remove_item(node1, node1.sent[0])
         assert sched.tests_finished()
-        sched.remove_item(node1, node1.sent[1])
-        assert sched.tests_finished()
 
-    def test_init_distribute_chunksize(self):
+    def test_init_distribute_batch_size(self):
         sched = LoadScheduling(2)
         sched.addnode(MockNode())
         sched.addnode(MockNode())
@@ -118,18 +125,56 @@ class TestLoadScheduling:
         # assert not sched.tests_finished()
         sent1 = node1.sent
         sent2 = node2.sent
-        assert sent1 == [0, 1]
+        assert sent1 == [0, 2]
-        assert sent2 == [2, 3]
+        assert sent2 == [1, 3]
         assert sched.pending == [4, 5]
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
         assert len(sched.pending) == 2
         sched.remove_item(node1, 0)
-        assert node1.sent == [0, 1, 4]
+        assert node1.sent == [0, 2, 4]
         assert sched.pending == [5]
-        assert node2.sent == [2, 3]
+        assert node2.sent == [1, 3]
-        sched.remove_item(node1, 1)
+        sched.remove_item(node1, 2)
-        assert node1.sent == [0, 1, 4, 5]
+        assert node1.sent == [0, 2, 4, 5]
+        assert not sched.pending
+
+    def test_init_distribute_fewer_tests_than_nodes(self):
+        sched = LoadScheduling(2)
+        sched.addnode(MockNode())
+        sched.addnode(MockNode())
+        sched.addnode(MockNode())
+        node1, node2, node3 = sched.nodes
+        col = ["xyz"] * 2
+        sched.addnode_collection(node1, col)
+        sched.addnode_collection(node2, col)
+        sched.init_distribute()
+        # assert not sched.tests_finished()
+        sent1 = node1.sent
+        sent2 = node2.sent
+        sent3 = node3.sent
+        assert sent1 == [0]
+        assert sent2 == [1]
+        assert sent3 == []
+        assert not sched.pending
+
+    def test_init_distribute_fewer_than_two_tests_per_node(self):
+        sched = LoadScheduling(2)
+        sched.addnode(MockNode())
+        sched.addnode(MockNode())
+        sched.addnode(MockNode())
+        node1, node2, node3 = sched.nodes
+        col = ["xyz"] * 5
+        sched.addnode_collection(node1, col)
+        sched.addnode_collection(node2, col)
+        sched.init_distribute()
+        # assert not sched.tests_finished()
+        sent1 = node1.sent
+        sent2 = node2.sent
+        sent3 = node3.sent
+        assert sent1 == [0, 3]
+        assert sent2 == [1, 4]
+        assert sent3 == [2]
         assert not sched.pending
 
     def test_add_remove_node(self):
@@ -149,6 +194,7 @@ class TestLoadScheduling:
         Test that LoadScheduling is reporting collection errors when
         different test ids are collected by slaves.
         """
+
         class CollectHook(object):
             """
             Dummy hook that stores collection reports.
@@ -177,7 +223,6 @@ class TestLoadScheduling:
 
 
 class TestDistReporter:
-
     @py.test.mark.xfail
     def test_rsync_printing(self, testdir, linecomp):
         config = testdir.parseconfig()
@@ -185,9 +230,11 @@ class TestDistReporter:
         rep = TerminalReporter(config, file=linecomp.stringio)
         config.pluginmanager.register(rep, "terminalreporter")
         dsession = DSession(config)
+
         class gw1:
             id = "X1"
             spec = execnet.XSpec("popen")
+
         class gw2:
             id = "X2"
             spec = execnet.XSpec("popen")
@@ -202,9 +249,7 @@ class TestDistReporter:
         #     "*X1*popen*xyz*2.5*"
         # ])
         dsession.pytest_xdist_rsyncstart(source="hello", gateways=[gw1, gw2])
-        linecomp.assert_contains_lines([
+        linecomp.assert_contains_lines(["[X1,X2] rsyncing: hello", ])
-            "[X1,X2] rsyncing: hello",
-        ])
 
 
 def test_report_collection_diff_equal():
@@ -230,12 +275,12 @@ def test_report_collection_diff_different():
         ' bbb\n'
         '+XXX\n'
         ' ccc\n'
-        '-YYY'
+        '-YYY')
-    )
 
-    msg = report_collection_diff(from_collection, to_collection, 1, 2)
+    msg = report_collection_diff(from_collection, to_collection, '1', '2')
     assert msg == error_message
 
+
 @pytest.mark.xfail(reason="duplicate test ids not supported yet")
 def test_pytest_issue419(testdir):
     testdir.makepyfile("""

testing/test_looponfail.py

@@ -2,6 +2,7 @@ import py
 from xdist.looponfail import RemoteControl
 from xdist.looponfail import StatRecorder
 
+
 class TestStatRecorder:
     def test_filechange(self, tmpdir):
         tmp = tmpdir
@@ -87,6 +88,7 @@ class TestStatRecorder:
         sd.waitonchange(checkinterval=0.2)
         assert not l
 
+
 class TestRemoteControl:
     def test_nofailures(self, testdir):
         item = testdir.getitem("def test_func(): pass\n")
@@ -143,6 +145,7 @@ class TestRemoteControl:
         control.loop_once()
         assert control.failures
 
+
 class TestLooponFailing:
     def test_looponfail_from_fail_to_ok(self, testdir):
         modcol = testdir.getmodulecol("""
@@ -269,6 +272,7 @@ class TestFunctional:
         child.expect("waiting for changes")
         child.kill(15)
 
+
 def removepyc(path):
     # XXX damn those pyc files
     pyc = path + "c"
@@ -277,4 +281,3 @@ def removepyc(path):
     c = path.dirpath("__pycache__")
     if c.check():
         c.remove()
-

testing/test_newhooks.py

@@ -0,0 +1,50 @@
+import pytest
+
+
+class TestHooks:
+
+    @pytest.fixture(autouse=True)
+    def create_test_file(self, testdir):
+        testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): pass
+            def test_c(): pass
+        """)
+
+    def test_runtest_logreport(self, testdir):
+        """Test that log reports from pytest_runtest_logreport when running
+        with xdist contain a "node" attribute. (#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))
+        """)
+        res = testdir.runpytest('-n1', '-s')
+        res.stdout.fnmatch_lines([
+            '*HOOK: test_runtest_logreport.py::test_a gw0*',
+            '*HOOK: test_runtest_logreport.py::test_b gw0*',
+            '*HOOK: test_runtest_logreport.py::test_c gw0*',
+            '*3 passed*',
+        ])
+
+    def test_node_collection_finished(self, testdir):
+        """Test pytest_xdist_node_collection_finished hook (#8).
+        """
+        testdir.makeconftest("""
+            def pytest_xdist_node_collection_finished(node, ids):
+                slaveid = node.slaveinput['slaveid']
+                stripped_ids = [x.split('::')[1] for x in ids]
+                print("HOOK: %s %s" % (slaveid, ', '.join(stripped_ids)))
+        """)
+        res = testdir.runpytest('-n2', '-s')
+        res.stdout.fnmatch_lines_random([
+            '*HOOK: gw0 test_a, test_b, test_c',
+            '*HOOK: gw1 test_a, test_b, test_c',
+        ])
+        res.stdout.fnmatch_lines([
+            '*3 passed*',
+        ])

testing/test_plugin.py

@@ -2,6 +2,7 @@ import py
 import execnet
 from xdist.slavemanage import NodeManager
 
+
 def test_dist_incompatibility_messages(testdir):
     result = testdir.runpytest("--pdb", "--looponfail")
     assert result.ret != 0
@@ -12,6 +13,7 @@ def test_dist_incompatibility_messages(testdir):
     assert result.ret != 0
     assert "incompatible" in result.stderr.str()
 
+
 def test_dist_options(testdir):
     from xdist.plugin import pytest_cmdline_main as check_options
     config = testdir.parseconfigure("-n 2")
@@ -22,6 +24,18 @@ def test_dist_options(testdir):
     check_options(config)
     assert config.option.dist == "load"
 
+
+def test_auto_detect_cpus(testdir, monkeypatch):
+    import multiprocessing
+    monkeypatch.setattr(multiprocessing, 'cpu_count', lambda: 99)
+
+    config = testdir.parseconfigure("-n2")
+    assert config.getoption('numprocesses') == 2
+
+    config = testdir.parseconfigure("-nauto")
+    assert config.getoption('numprocesses') == 99
+
+
 class TestDistOptions:
     def test_getxspecs(self, testdir):
         config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz")

testing/test_remote.py

@@ -7,6 +7,7 @@ import marshal
 
 WAIT_TIMEOUT = 10.0
 
+
 def check_marshallable(d):
     try:
         marshal.dumps(d)
@@ -14,6 +15,7 @@ def check_marshallable(d):
         py.std.pprint.pprint(d)
         raise ValueError("not marshallable")
 
+
 class EventCall:
     def __init__(self, eventcall):
         self.name, self.kwargs = eventcall
@@ -21,6 +23,7 @@ class EventCall:
     def __str__(self):
         return "<EventCall %s(**%s)>" % (self.name, self.kwargs)
 
+
 class SlaveSetup:
     use_callback = False
 
@@ -53,9 +56,11 @@ class SlaveSetup:
     def sendcommand(self, name, **kwargs):
         self.slp.sendcommand(name, **kwargs)
 
+
 def pytest_funcarg__slave(request):
     return SlaveSetup(request)
 
+
 def test_remoteinitconfig(testdir):
     from xdist.remote import remote_initconfig
     config1 = testdir.parseconfig()
@@ -63,6 +68,7 @@ def test_remoteinitconfig(testdir):
     assert config2.option.__dict__ == config1.option.__dict__
     assert config2.pluginmanager.getplugin("terminal") in (-1, None)
 
+
 class TestReportSerialization:
     def test_itemreport_outcomes(self, testdir):
         reprec = testdir.inline_runsource("""
@@ -245,4 +251,3 @@ class TestSlaveInteractor:
             ("pytest_pycollect_makeitem", "name == 'test_func'"),
             ("pytest_collectreport", "report.collector.fspath == bbb"),
         ])
-

testing/test_slavemanage.py

@@ -7,6 +7,7 @@ from xdist.slavemanage import HostRSync, NodeManager
 
 pytest_plugins = "pytester"
 
+
 def pytest_funcarg__hookrecorder(request, config):
     hookrecorder = HookRecorder(config.pluginmanager)
     if hasattr(hookrecorder, "start_recording"):
@@ -14,23 +15,32 @@ def pytest_funcarg__hookrecorder(request, config):
     request.addfinalizer(hookrecorder.finish_recording)
     return hookrecorder
 
+
 def pytest_funcarg__config(testdir):
     return testdir.parseconfig()
 
+
 def pytest_funcarg__mysetup(tmpdir):
     class mysetup:
         source = tmpdir.mkdir("source")
         dest = tmpdir.mkdir("dest")
+
     return mysetup()
 
+
 @pytest.fixture
 def slavecontroller(monkeypatch):
     class MockController(object):
-        def __init__(self, *args): pass
+        def __init__(self, *args):
-        def setup(self): pass
+            pass
+
+        def setup(self):
+            pass
+
     monkeypatch.setattr(slavemanage, 'SlaveController', MockController)
     return MockController
 
+
 class TestNodeManagerPopen:
     def test_popen_no_default_chdir(self, config):
         gm = NodeManager(config, ["popen"])
@@ -43,7 +53,8 @@ class TestNodeManagerPopen:
         for spec in NodeManager(config, l, defaultchdir="abc").specs:
             assert spec.chdir == "abc"
 
-    def test_popen_makegateway_events(self, config, hookrecorder, slavecontroller):
+    def test_popen_makegateway_events(self, config, hookrecorder,
+                                      slavecontroller):
         hm = NodeManager(config, ["popen"] * 2)
         hm.setup_nodes(None)
         call = hookrecorder.popcall("pytest_xdist_setupnodes")
@@ -64,12 +75,16 @@ class TestNodeManagerPopen:
         hm.setup_nodes(None)
         assert len(hm.group) == 2
         for gw in hm.group:
+
             class pseudoexec:
                 args = []
+
                 def __init__(self, *args):
                     self.args.extend(args)
+
                 def waitclose(self):
                     pass
+
             gw.remote_exec = pseudoexec
         l = []
         for gw in hm.group:
@@ -95,8 +110,8 @@ class TestNodeManagerPopen:
         assert dest.join("dir1", "dir2").check()
         assert dest.join("dir1", "dir2", 'hello').check()
 
-    def test_rsync_same_popen_twice(self, config, mysetup,
+    def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
-                                    hookrecorder, slavecontroller):
+                                    slavecontroller):
         source, dest = mysetup.source, mysetup.dest
         hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
         hm.roots = []
@@ -110,6 +125,7 @@ class TestNodeManagerPopen:
         assert call.gateways[0] in hm.group
         call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
 
+
 class TestHRSync:
     def test_hrsync_filter(self, mysetup):
         source, _ = mysetup.source, mysetup.dest  # noqa
@@ -118,8 +134,7 @@ class TestHRSync:
         source.ensure(".somedotfile", "moreentries")
         source.ensure("somedir", "editfile~")
         syncer = HostRSync(source, ignores=NodeManager.DEFAULT_IGNORES)
-        l = list(source.visit(rec=syncer.filter,
+        l = list(source.visit(rec=syncer.filter, fil=syncer.filter))
-                                   fil=syncer.filter))
         assert len(l) == 3
         basenames = [x.basename for x in l]
         assert 'dir' in basenames
@@ -164,10 +179,8 @@ class TestNodeManager:
         for rsyncroot in (dir1, source):
             dest.remove()
             nodemanager = NodeManager(testdir.parseconfig(
-                "--tx", "popen//chdir=%s" % dest,
+                "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot,
-                "--rsyncdir", rsyncroot,
+                source, ))
-                source,
-            ))
             nodemanager.setup_nodes(None)  # calls .rsync_roots()
             if rsyncroot == source:
                 dest = dest.join("source")
@@ -230,7 +243,8 @@ class TestNodeManager:
             assert not gwspec.chdir
 
     def test_ssh_setup_nodes(self, specssh, testdir):
-        testdir.makepyfile(__init__="", test_x="""
+        testdir.makepyfile(__init__="",
+                           test_x="""
             def test_one():
                 pass
         """)

tox.ini

@@ -1,17 +1,19 @@
 [tox]
 envlist=
-  py{26,33,34,27}-pytest2{4,5,6,7},py{27,34}-pytest27-pexpect,flakes
+  py{26,27,34,35}-pytest2{4,5,6,7,8},py{27,34}-pytest28-pexpect,flakes,readme
 
 
 [testenv]
 changedir=testing
+passenv = USER USERNAME
 deps =
     pycmd
+    setuptools_scm # to avoid .eggs
     pytest24: pytest~=2.4.0
     pytest25: pytest~=2.5.0
-
     pytest26: pytest~=2.6.1
     pytest27: pytest~=2.7.2
+    pytest28: pytest~=2.8.3
     pexpect: pexpect
 commands=
   # always clean to avoid code unmarshal mismatch on old python/pytest
@@ -20,8 +22,15 @@ commands=
 
 [testenv:flakes]
 changedir=
-deps = pytest-flakes>=0.2
+deps = flake8
-commands = py.test --flakes -m flakes testing xdist
+commands = flake8 setup.py testing xdist
+
+[testenv:readme]
+changedir =
+deps = readme
+skip_install = true
+commands = python setup.py check -r -s
+
 
 [pytest]
 addopts = -rsfxX

xdist/boxed.py

@@ -4,7 +4,8 @@ import py
 
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
-    group.addoption('--boxed',
+    group.addoption(
+        '--boxed',
         action="store_true", dest="boxed", default=False,
         help="box each test run in a separate process (unix)")
 
@@ -16,6 +17,7 @@ def pytest_runtest_protocol(item):
             item.ihook.pytest_runtest_logreport(report=rep)
         return True
 
+
 def forked_run_report(item):
     # for now, we run setup/teardown in the subprocess
     # XXX optionally allow sharing of setup/teardown
@@ -24,6 +26,7 @@ def forked_run_report(item):
     import marshal
     from xdist.remote import serialize_report
     from xdist.slavemanage import unserialize_report
+
     def runforked():
         try:
             reports = runtestprotocol(item, log=False)
@@ -41,6 +44,7 @@ def forked_run_report(item):
             py.test.exit("forked test item %s raised Exit" % (item,))
         return [report_process_crash(item, result)]
 
+
 def report_process_crash(item, result):
     path, lineno = item._getfslineno()
     info = ("%s:%s: running the test CRASHED with signal %d" %

xdist/dsession.py

@@ -1,4 +1,5 @@
 import difflib
+import itertools
 from _pytest.runner import CollectReport
 
 import pytest
@@ -14,12 +15,12 @@ class EachScheduling:
 
     If a node gets added after the test run is started then it is
     assumed to replace a node which got removed before it finished
-    it's collection.  In this case it will only be used if a a node
+    its collection.  In this case it will only be used if a node
     with the same spec got removed earlier.
 
     Any nodes added after the run is started will only get items
-    assigned if a node with matching spec was removed before it
+    assigned if a node with a matching spec was removed before it
-    finished all it's pending items.  The new node will then be
+    finished all its pending items.  The new node will then be
     assigned the remaining items from the removed node.
     """
 
@@ -47,8 +48,8 @@ class EachScheduling:
         """Return True if there are pending test items
 
         This indicates that collection has finished and nodes are
-        still processing test items, so can be thought of as "the
+        still processing test items, so this can be thought of as
-        scheduler is active".
+        "the scheduler is active".
         """
         for pending in self.node2pending.values():
             if pending:
@@ -73,10 +74,10 @@ class EachScheduling:
         """Add the collected test items from a node
 
         Collection is complete once all nodes have submitted their
-        collection.  In this case it's peding list is set to an empty
+        collection.  In this case its pending list is set to an empty
         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 assing the pending items
+        dead node.  In this case we already assign the pending items
         here.  In either case ``.init_distribute()`` will instruct the
         node to start running the required tests.
         """
@@ -120,7 +121,7 @@ class EachScheduling:
         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
-        died node and we only need to run those tests.
+        dead node and we only need to run those tests.
         """
         assert self.collection_is_completed
         for node, pending in self.node2pending.items():
@@ -135,19 +136,19 @@ class EachScheduling:
 
 
 class LoadScheduling:
-    """Implement load scheduling accross nodes.
+    """Implement load scheduling across nodes.
 
     This distributes the tests collected across all nodes so each test
-    is run just once.  All nodes collect and submit the test suit and
+    is run just once.  All nodes collect and submit the test suite and
     when all collections are received it is verified they are
-    identical collections.  Then the collection gets devided up in
+    identical collections.  Then the collection gets divided up in
-    chunks and chunks get submitted to nodes.  Whenver a node finishes
+    chunks and chunks get submitted to nodes.  Whenever a node finishes
-    an item they call ``.remove_item()`` which will trigger the
+    an item, it calls ``.remove_item()`` which will trigger the
     scheduler to assign more tests if the number of pending tests for
     the node falls below a low-watermark.
 
-    When created ``numnodes`` defines how many nodes are expected to
+    When created, ``numnodes`` defines how many nodes are expected to
-    submit a collection, this is used to know when all nodes have
+    submit a collection. This is used to know when all nodes have
     finished collection or how large the chunks need to be created.
 
     Attributes:
@@ -155,7 +156,7 @@ class LoadScheduling:
     :numnodes: The expected number of nodes taking part.  The actual
        number of nodes will vary during the scheduler's lifetime as
        nodes are added by the DSession as they are brought up and
-       removed either because of a died node or normal shutdown.  This
+       removed either because of a dead node or normal shutdown.  This
        number is primarily used to know when the initial collection is
        completed.
 
@@ -211,8 +212,8 @@ class LoadScheduling:
         """Return True if there are pending test items
 
         This indicates that collection has finished and nodes are
-        still processing test items, so can be thought of as "the
+        still processing test items, so this can be thought of as
-        scheduler is active".
+        "the scheduler is active".
         """
         if self.pending:
             return True
@@ -226,13 +227,13 @@ class LoadScheduling:
         return bool(self.node2pending)
 
     def addnode(self, node):
-        """Add a new node in the scheduler.
+        """Add a new node to the scheduler.
 
         From now on the node will be allocated chunks of tests to
         execute.
 
         Called by the ``DSession.slave_slaveready`` hook when it
-        sucessfully bootstrapped a new node.
+        successfully bootstraps a new node.
         """
         assert node not in self.node2pending
         self.node2pending[node] = []
@@ -257,8 +258,9 @@ class LoadScheduling:
         assert node in self.node2pending
         if self.collection_is_completed:
             # A new node has been added later, perhaps an original one died.
-            assert self.collection  # .init_distribute() should have
+            # .init_distribute() should have
             # been called by now
+            assert self.collection
             if collection != self.collection:
                 other_node = next(iter(self.node2collection.keys()))
                 msg = report_collection_diff(self.collection,
@@ -288,6 +290,9 @@ class LoadScheduling:
         ``duration`` of the last test is optionally used as a
         heuristic to influence how many tests the node is assigned.
         """
+        if node.shutting_down:
+            return
+
         if self.pending:
             # how many nodes do we have?
             num_nodes = len(self.node2pending)
@@ -307,7 +312,7 @@ class LoadScheduling:
         self.log("num items waiting for node:", len(self.pending))
 
     def remove_node(self, node):
-        """Remove an node from the scheduler
+        """Remove a node from the scheduler
 
         This should be called either when the node crashed or at
         shutdown time.  In the former case any pending items assigned
@@ -345,7 +350,7 @@ class LoadScheduling:
         """
         assert self.collection_is_completed
 
-        # Initial distribution already happend, reschedule on all nodes
+        # Initial distribution already happened, reschedule on all nodes
         if self.collection is not None:
             for node in self.nodes:
                 self.check_schedule(node)
@@ -362,13 +367,22 @@ class LoadScheduling:
         if not self.collection:
             return
 
-        # how many items per node do we have about?
+        # Send a batch of tests to run. If we don't have at least two
-        items_per_node = len(self.collection) // len(self.node2pending)
+        # tests per node, we have to send them all so that we can send
-        # take a fraction of tests for initial distribution
+        # shutdown signals and get all nodes working.
-        node_chunksize = max(items_per_node // 4, 2)
+        initial_batch = max(len(self.pending) // 4,
-        # and initialize each node with a chunk of tests
+                            2 * len(self.nodes))
+
+        # distribute tests round-robin up to the batch size
+        # (or until we run out)
+        nodes = itertools.cycle(self.nodes)
+        for i in range(initial_batch):
+            self._send_tests(next(nodes), 1)
+
+        if not self.pending:
+            # initial distribution sent all tests, start node shutdown
             for node in self.nodes:
-            self._send_tests(node, node_chunksize)
+                node.shutdown()
 
     def _send_tests(self, node, num):
         tests_per_node = self.pending[:num]
@@ -398,8 +412,9 @@ class LoadScheduling:
                 same_collection = False
                 self.log(msg)
                 if self.config is not None:
-                    rep = CollectReport(node.gateway.id, 'failed', longrepr=msg,
+                    rep = CollectReport(
-                                        result=[])
+                        node.gateway.id, 'failed',
+                        longrepr=msg, result=[])
                     self.config.hook.pytest_collectreport(report=rep)
 
         return same_collection
@@ -475,7 +490,7 @@ class DSession:
         """Return True if the distributed session has finished
 
         This means all nodes have executed all test items.  This is
-        used to by pytest_runtestloop to break out of it's loop.
+        used by pytest_runtestloop to break out of its loop.
         """
         return bool(self.shuttingdown and not self._active_nodes)
 
@@ -564,7 +579,7 @@ class DSession:
 
         Removes the node from the scheduler.
 
-        The node might not be the scheduler if it had not emitted
+        The node might not be in the scheduler if it had not emitted
         slaveready before shutdown was triggered.
         """
         self.config.hook.pytest_testnodedown(node=node, error=None)
@@ -608,12 +623,14 @@ class DSession:
 
         This adds the collection for this node to the scheduler.  If
         the scheduler indicates collection is finished (i.e. all
-        initial nodes have submitted their collection), then tells the
+        initial nodes have submitted their collections), then tells the
         scheduler to schedule the collected items.  When initiating
         scheduling the first time it logs which scheduler is in use.
         """
         if self.shuttingdown:
             return
+        self.config.hook.pytest_xdist_node_collection_finished(node=node,
+                                                               ids=ids)
         # tell session which items were effectively collected otherwise
         # the master node will finish the session with EXIT_NOTESTSCOLLECTED
         self._session.testscollected = len(ids)
@@ -636,7 +653,7 @@ class DSession:
     def slave_testreport(self, node, rep):
         """Emitted when a node calls the pytest_runtest_logreport hook.
 
-        If the node indicates it is finished with a test item remove
+        If the node indicates it is finished with a test item, remove
         the item from the pending list in the scheduler.
         """
         if rep.when == "call" or (rep.when == "setup" and not rep.passed):
@@ -654,9 +671,9 @@ class DSession:
     def _clone_node(self, node):
         """Return new node based on an existing one.
 
-        This is normally for when a node died, this will copy the spec
+        This is normally for when a node dies, this will copy the spec
         of the existing node and create a new one with a new id.  The
-        new node will have been setup so will start calling the
+        new node will have been setup so it will start calling the
         "slave_*" hooks and do work soon.
         """
         spec = node.gateway.spec
@@ -769,4 +786,3 @@ class TerminalDistReporter:
     # def pytest_xdist_rsyncfinish(self, source, gateways):
     #    targets = ", ".join(["[%s]" % gw.id for gw in gateways])
     #    self.write_line("rsyncfinish: %s -> %s" %(source, targets))
-

xdist/looponfail.py

@@ -7,25 +7,32 @@
     the controlling process which should best never happen.
 """
 
-import py, pytest
+import py
+import pytest
 import sys
 import execnet
 
+
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
-    group._addoption('-f', '--looponfail',
+    group._addoption(
+        '-f', '--looponfail',
         action="store_true", dest="looponfail", default=False,
         help="run tests in subprocess, wait for modified files "
              "and re-run failing test set until all pass.")
 
+
 def pytest_cmdline_main(config):
 
     if config.getoption("looponfail"):
+        usepdb = config.getoption('usepdb')  # a core option
+        if usepdb:
+            raise pytest.UsageError(
+                "--pdb incompatible with --looponfail.")
         looponfail_main(config)
         return 2  # looponfail only can get stop with ctrl-C anyway
 
 
-
 def looponfail_main(config):
     remotecontrol = RemoteControl(config)
     rootdirs = config.getini("looponfailroots")
@@ -34,7 +41,8 @@ def looponfail_main(config):
         while 1:
             remotecontrol.loop_once()
             if not remotecontrol.failures and remotecontrol.wasfailing:
-                continue # the last failures passed, let's immediately rerun all
+                # the last failures passed, let's immediately rerun all
+                continue
             repr_pytest_looponfailinfo(
                 failreports=remotecontrol.failures,
                 rootdirs=rootdirs)
@@ -42,6 +50,7 @@ def looponfail_main(config):
     except KeyboardInterrupt:
         print()
 
+
 class RemoteControl(object):
     def __init__(self, config):
         self.config = config
@@ -62,11 +71,13 @@ class RemoteControl(object):
             raise ValueError("already have gateway %r" % self.gateway)
         self.trace("setting up slave session")
         self.gateway = self.initgateway()
-        self.channel = channel = self.gateway.remote_exec(init_slave_session,
+        self.channel = channel = self.gateway.remote_exec(
+            init_slave_session,
             args=self.config.args,
             option_dict=vars(self.config.option),
         )
         remote_outchannel = channel.receive()
+
         def write(s):
             out._file.write(s)
             out._file.flush()
@@ -110,6 +121,7 @@ class RemoteControl(object):
                     uniq_failures.append(failure)
             self.failures = uniq_failures
 
+
 def repr_pytest_looponfailinfo(failreports, rootdirs):
     tr = py.io.TerminalWriter()
     if failreports:
@@ -123,7 +135,8 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
 
 
 def init_slave_session(channel, args, option_dict):
-    import os, sys
+    import os
+    import sys
     outchannel = channel.gateway.newchannel()
     sys.stdout = sys.stderr = outchannel.makefile('w')
     channel.send(outchannel)
@@ -143,6 +156,7 @@ def init_slave_session(channel, args, option_dict):
     from xdist.looponfail import SlaveFailSession
     SlaveFailSession(config, channel).main()
 
+
 class SlaveFailSession:
     def __init__(self, config, channel):
         self.config = config
@@ -165,7 +179,8 @@ class SlaveFailSession:
             items = session.perform_collect(self.trails or None)
         except pytest.UsageError:
             items = session.perform_collect(None)
-        hook.pytest_collection_modifyitems(session=session, config=session.config, items=items)
+        hook.pytest_collection_modifyitems(
+            session=session, config=session.config, items=items)
         hook.pytest_collection_finish(session=session)
         return True
 
@@ -195,6 +210,7 @@ class SlaveFailSession:
             failreports.append(loc)
         self.channel.send((trails, failreports, self.collection_failed))
 
+
 class StatRecorder:
     def __init__(self, rootdirlist):
         self.rootdirlist = rootdirlist
@@ -203,6 +219,7 @@ class StatRecorder:
 
     def fil(self, p):
         return p.check(file=1, dotfile=0) and p.ext != ".pyc"
+
     def rec(self, p):
         return p.check(dotfile=0)
 
@@ -213,7 +230,7 @@ class StatRecorder:
                 return
             py.std.time.sleep(checkinterval)
 
-    def check(self, removepycfiles=True):
+    def check(self, removepycfiles=True):  # noqa, too complex
         changed = False
         statcache = self.statcache
         newstat = {}

xdist/newhooks.py

@@ -1,21 +1,46 @@
+"""
+xdist hooks.
+
+Additionally, pytest-xdist will also decorate a few other hooks
+with the worker instance that executed the hook originally:
+
+``pytest_runtest_logreport``: ``rep`` parameter has a ``node`` attribute.
+
+You can use this hooks just as you would use normal pytest hooks, but some care
+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
+"""
+
 
 def pytest_xdist_setupnodes(config, specs):
     """ called before any remote node is set up. """
 
+
 def pytest_xdist_newgateway(gateway):
     """ called on new raw gateway creation. """
 
+
 def pytest_xdist_rsyncstart(source, gateways):
     """ called before rsyncing a directory to remote gateways takes place. """
 
+
 def pytest_xdist_rsyncfinish(source, gateways):
     """ called after rsyncing a directory to remote gateways takes place. """
 
+
 def pytest_configure_node(node):
     """ configure node information before it gets instantiated. """
 
+
 def pytest_testnodeready(node):
     """ Test Node is ready to operate. """
 
+
 def pytest_testnodedown(node, error):
     """ Test Node is down. """
+
+
+def pytest_xdist_node_collection_finished(node, ids):
+    """called by the master node when a node finishes collecting.
+    """

xdist/plugin.py

@@ -12,7 +12,8 @@ def parse_numprocesses(s):
 
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
-    group._addoption('-n', dest="numprocesses", metavar="numprocesses",
+    group._addoption(
+        '-n', dest="numprocesses", metavar="numprocesses",
         action="store",
         type=parse_numprocesses,
         help="shortcut for '--dist=load --tx=NUM*popen', "
@@ -21,36 +22,46 @@ def pytest_addoption(parser):
     group._addoption('--max-slave-restart', action="store", default=None,
                      help="maximum number of slaves that can be restarted "
                           "when crashed (set to zero to disable this feature)")
-    group._addoption('--dist', metavar="distmode",
+    group._addoption(
+        '--dist', metavar="distmode",
         action="store", choices=['load', 'each', 'no'],
         type="choice", 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"
               "(default) no: run tests inprocess, don't distribute."))
-    group._addoption('--tx', dest="tx", action="append", default=[],
+    group._addoption(
+        '--tx', dest="tx", action="append", default=[],
         metavar="xspec",
         help=("add a test execution environment. some examples: "
               "--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
               "--tx ssh=user@codespeak.net//chdir=testcache"))
-    group._addoption('-d',
+    group._addoption(
+        '-d',
         action="store_true", dest="distload", default=False,
         help="load-balance tests.  shortcut for '--dist=load'")
-    group.addoption('--rsyncdir', action="append", default=[], metavar="DIR",
+    group.addoption(
+        '--rsyncdir', action="append", default=[], metavar="DIR",
         help="add directory for rsyncing to remote tx nodes.")
-    group.addoption('--rsyncignore', action="append", default=[], metavar="GLOB",
+    group.addoption(
+        '--rsyncignore', action="append", default=[], metavar="GLOB",
         help="add expression for ignores when rsyncing to remote tx nodes.")
 
-    parser.addini('rsyncdirs', 'list of (relative) paths to be rsynced for'
+    parser.addini(
+        'rsyncdirs', 'list of (relative) paths to be rsynced for'
         ' remote distributed testing.', type="pathlist")
-    parser.addini('rsyncignore', 'list of (relative) glob-style paths to be ignored '
+    parser.addini(
+        'rsyncignore', 'list of (relative) glob-style paths to be ignored '
         'for rsyncing.', type="pathlist")
-    parser.addini("looponfailroots", type="pathlist",
+    parser.addini(
+        "looponfailroots", type="pathlist",
         help="directories to check for changes", default=[py.path.local()])
 
 # -------------------------------------------------------------------------
 # distributed testing hooks
 # -------------------------------------------------------------------------
+
+
 def pytest_addhooks(pluginmanager):
     from xdist import newhooks
     # avoid warnings with pytest-2.8
@@ -73,6 +84,7 @@ def pytest_configure(config):
         tr = config.pluginmanager.getplugin("terminalreporter")
         tr.showfspath = False
 
+
 @pytest.mark.tryfirst
 def pytest_cmdline_main(config):
     if config.option.numprocesses:
@@ -82,10 +94,8 @@ def pytest_cmdline_main(config):
         config.option.dist = "load"
     val = config.getvalue
     if not val("collectonly"):
-        usepdb = config.option.usepdb  # a core option
+        usepdb = config.getoption('usepdb')  # a core option
-        if val("looponfail"):
+        if val("dist") != "no":
             if usepdb:
-                raise pytest.UsageError("--pdb incompatible with --looponfail.")
+                raise pytest.UsageError(
-        elif val("dist") != "no":
+                    "--pdb incompatible with distributing tests.")
-            if usepdb:
-                raise pytest.UsageError("--pdb incompatible with distributing tests.")

xdist/remote.py

@@ -6,7 +6,9 @@
     needs not to be installed in remote environments.
 """
 
-import sys, os
+import sys
+import os
+
 
 class SlaveInteractor:
     def __init__(self, config, channel):
@@ -72,7 +74,8 @@ class SlaveInteractor:
             nextitem=nextitem)
 
     def pytest_collection_finish(self, session):
-        self.sendevent("collectionfinish",
+        self.sendevent(
+            "collectionfinish",
             topdir=str(session.fspath),
             ids=[item.nodeid for item in session.items])
 
@@ -89,6 +92,7 @@ class SlaveInteractor:
         data = serialize_report(report)
         self.sendevent("collectreport", data=data)
 
+
 def serialize_report(rep):
     import py
     d = rep.__dict__.copy()
@@ -103,6 +107,7 @@ def serialize_report(rep):
             d[name] = None  # for now
     return d
 
+
 def getinfodict():
     import platform
     return dict(
@@ -114,6 +119,7 @@ def getinfodict():
         cwd=os.getcwd(),
     )
 
+
 def remote_initconfig(option_dict, args):
     from _pytest.config import Config
     option_dict['plugins'].append("no:terminal")
@@ -136,7 +142,8 @@ if __name__ == '__channelexec__':
     slaveinput, args, option_dict = channel.receive()
     importpath = os.getcwd()
     sys.path.insert(0, importpath)  # XXX only for remote situations
-    os.environ['PYTHONPATH'] = (importpath + os.pathsep +
+    os.environ['PYTHONPATH'] = (
+        importpath + os.pathsep +
         os.environ.get('PYTHONPATH', ''))
     # os.environ['PYTHONPATH'] = importpath
     import py

xdist/slavemanage.py

@@ -1,5 +1,6 @@
 import fnmatch
 import os
+import re
 
 import py
 import pytest
@@ -8,9 +9,11 @@ import xdist.remote
 
 from _pytest import runner  # XXX load dynamically
 
+
 class NodeManager(object):
     EXIT_TIMEOUT = 10
     DEFAULT_IGNORES = ['.*', '*.pyc', '*.pyo', '*~']
+
     def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
         self.config = config
         self._nodesready = py.std.threading.Event()
@@ -79,7 +82,8 @@ class NodeManager(object):
                 break
         else:
             return []
-        import pytest, _pytest
+        import pytest
+        import _pytest
         pytestpath = pytest.__file__.rstrip("co")
         pytestdir = py.path.local(_pytest.__file__).dirpath()
         config = self.config
@@ -124,6 +128,7 @@ class NodeManager(object):
             return
         if (spec, source) in self._rsynced_specs:
             return
+
         def finished():
             if notify:
                 notify("rsyncrootready", spec, source)
@@ -139,19 +144,23 @@ class NodeManager(object):
             gateways=[gateway],
         )
 
+
 class HostRSync(execnet.RSync):
     """ RSyncer that filters out common files
     """
     def __init__(self, sourcedir, *args, **kwargs):
         self._synced = {}
-        self._ignores = kwargs.pop('ignores', None) or []
+        self._ignores = []
+        ignores = kwargs.pop('ignores', None) or []
+        for x in ignores:
+            x = getattr(x, 'strpath', x)
+            self._ignores.append(re.compile(fnmatch.translate(x)))
         super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
 
     def filter(self, path):
         path = py.path.local(path)
-        for x in self._ignores:
+        for cre in self._ignores:
-            x = getattr(x, 'strpath', x)
+            if cre.match(path.basename) or cre.match(path.strpath):
-            if fnmatch.fnmatch(path.basename, x) or fnmatch.fnmatch(path.strpath, x):
                 return False
         else:
             return True
@@ -187,6 +196,7 @@ def make_reltoroot(roots, args):
         l.append(splitcode.join(parts))
     return l
 
+
 class SlaveController(object):
     ENDMARK = -1
 
@@ -197,6 +207,7 @@ class SlaveController(object):
         self.config = config
         self.slaveinput = {'slaveid': gateway.id}
         self._down = False
+        self._shutdown_sent = False
         self.log = py.log.Producer("slavectl-%s" % gateway.id)
         if not self.config.option.debug:
             py.log.setconsumer(self.log._keywords, None)
@@ -204,6 +215,10 @@ class SlaveController(object):
     def __repr__(self):
         return "<%s %s>" % (self.__class__.__name__, self.gateway.id,)
 
+    @property
+    def shutting_down(self):
+        return self._down or self._shutdown_sent
+
     def setup(self):
         self.log("setting up slave session")
         spec = self.gateway.spec
@@ -213,13 +228,15 @@ class SlaveController(object):
         option_dict = vars(self.config.option)
         if spec.popen:
             name = "popen-%s" % self.gateway.id
+            if hasattr(self.config, '_tmpdirhandler'):
                 basetemp = self.config._tmpdirhandler.getbasetemp()
                 option_dict['basetemp'] = str(basetemp.join(name))
         self.config.hook.pytest_configure_node(node=self)
         self.channel = self.gateway.remote_exec(xdist.remote)
         self.channel.send((self.slaveinput, args, option_dict))
         if self.putevent:
-            self.channel.setcallback(self.process_from_remote,
+            self.channel.setcallback(
+                self.process_from_remote,
                 endmarker=self.ENDMARK)
 
     def ensure_teardown(self):
@@ -245,6 +262,7 @@ class SlaveController(object):
                 self.sendcommand("shutdown")
             except IOError:
                 pass
+            self._shutdown_sent = True
 
     def sendcommand(self, name, **kwargs):
         """ send a named parametrized command to the other side. """
@@ -255,7 +273,7 @@ class SlaveController(object):
         self.log("queuing %s(**%s)" % (eventname, kwargs))
         self.putevent((eventname, kwargs))
 
-    def process_from_remote(self, eventcall):
+    def process_from_remote(self, eventcall):  # noqa too complex
         """ this gets called for each object we receive from
             the other side and if the channel closes.
 
@@ -283,7 +301,8 @@ class SlaveController(object):
                 self.notify_inproc("slavefinished", node=self)
             elif eventname == "logstart":
                 self.notify_inproc(eventname, node=self, **kwargs)
-            elif eventname in ("testreport", "collectreport", "teardownreport"):
+            elif eventname in (
+                    "testreport", "collectreport", "teardownreport"):
                 item_index = kwargs.pop("item_index", None)
                 rep = unserialize_report(eventname, kwargs['data'])
                 if item_index is not None:
@@ -301,6 +320,7 @@ class SlaveController(object):
             py.builtin.print_("!" * 20, excinfo)
             self.config.pluginmanager.notify_exception(excinfo)
 
+
 def unserialize_report(name, reportdict):
     if name == "testreport":
         return runner.TestReport(**reportdict)