refactor cpu number autodetection to avoid the regression

Add --max-slave-restart option

.hgignore

@@ -14,9 +14,18 @@ syntax:glob
 *.class
 *.orig
 
+*.sublime-*
+.Python
+
 build/
 dist/
+include/
+lib/
+bin/
+xdist/_version.py
 pytest_xdist.egg-info
 issue/
 3rdparty/
+pytestdebug.log
 .tox
+.cache

.hgtags

@@ -11,3 +11,10 @@ cd44a941c833c098e4899fe3d42a96703754d0d5 1.5
 20875fed94e7f3dff50bdf762df91153b15ceca6 1.7
 29c38e195526f5f0fdd651fb51f59d6efaaafbb0 1.7
 0d1c00018008433956aa7d93007bab6ea7de96e4 1.8
+0d1c00018008433956aa7d93007bab6ea7de96e4 1.8
+1d27987c267577899350a25ba5828d55d87083ad 1.8
+5c5cb6d59e12e566fbb0217aea718dc31578bee1 1.9
+4406fc2a6427fadc021ed7e43e7aa5032b1ea91f 1.10
+220f6e46eb71a6212ccbe6b67b9e6edcf8ee4fa5 1.11
+39ef85dbc893cc63dede11601208098a667b58e9 1.12
+4e25f4c568be2d7cb4d1739638a9e66bbf28f588 v1.13

CHANGELOG

@@ -1,3 +1,94 @@
+1.13.1
+-------
+
+- fix a regression -n 0 now disables xdist again
+
+
+1.13
+-------------------------
+
+- extended the tox matrix with the supported py.test versions
+
+- split up the plugin into 3 plugin's
+  to prepare the departure of boxed and looponfail.
+
+  looponfail will be a part of core
+  and forked boxed will be replaced
+  with a more reliable primitive based on xdist
+
+- conforming with new pytest-2.8 behavior of returning non-zero when all
+  tests were skipped or deselected.
+
+- new "--max-slave-restart" option that can be used to control maximum
+  number of times pytest-xdist can restart slaves due to crashes. Thanks to
+  Anatoly Bubenkov for the report and Bruno Oliveira for the PR.
+
+- release as wheel
+
+- "-n" option now can be set to "auto" for automatic detection of number
+  of cpus in the host system. Thanks Suloev Dmitry for the PR.
+
+1.12
+-------------------------
+
+- fix issue594: properly report errors when the test collection
+  is random.  Thanks Bruno Oliveira.
+
+- some internal test suite adaptation (to become forward
+  compatible with the upcoming pytest-2.8)
+
+
+1.11
+-------------------------
+
+- fix pytest/xdist issue485 (also depends on py-1.4.22):
+  attach stdout/stderr on --boxed processes that die.
+
+- fix pytest/xdist issue503: make sure that a node has usually
+  two items to execute to avoid scoped fixtures to be torn down
+  pre-maturely (fixture teardown/setup is "nextitem" sensitive).
+  Thanks to Andreas Pelme for bug analysis and failing test.
+
+- restart crashed nodes by internally refactoring setup handling
+  of nodes.  Also includes better code documentation.
+  Many thanks to Floris Bruynooghe for the complete PR.
+
+
+1.10
+-------------------------
+
+- add glob support for rsyncignores, add command line option to pass
+  additional rsyncignores. Thanks Anatoly Bubenkov.
+
+- fix pytest issue382 - produce "pytest_runtest_logstart" event again
+  in master. Thanks Aron Curzon.
+
+- fix pytest issue419 by sending/receiving indices into the test
+  collection instead of node ids (which are not neccessarily unique
+  for functions parametrized with duplicate values)
+
+- send multiple "to test" indices in one network message to a slave
+  and improve heuristics for sending chunks where the chunksize
+  depends on the number of remaining tests rather than fixed numbers.
+  This reduces the number of master -> node messages (but not the
+  reverse direction)
+
+
+1.9
+-------------------------
+
+- changed LICENSE to MIT
+
+- fix duplicate reported test ids with --looponfailing
+  (thanks Jeremy Thurgood)
+
+- fix pytest issue41: re-run tests on all file changes, not just
+  randomly select ones like .py/.c.
+
+- fix pytest issue347: slaves running on top of Python3.2
+  will set PYTHONDONTWRITEYBTECODE to 1 to avoid import concurrency
+  bugs.
+
 1.8
 -------------------------
 
@@ -85,4 +176,3 @@
 - cleaned up termination handling
 - make -x cause hard killing of test nodes to decrease wait time
   until the traceback shows up on first failure
-

LICENSE

@@ -1,13 +1,19 @@
-The execnet package is released under the provisions of the Gnu Public
-License (GPL), version 2 or later.  
 
-See http://www.fsf.org/licensing/licenses/ for more information.
+  Permission is hereby granted, free of charge, to any person obtaining a copy
+  of this software and associated documentation files (the "Software"), to deal
+  in the Software without restriction, including without limitation the rights
+  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+  copies of the Software, and to permit persons to whom the Software is
+  furnished to do so, subject to the following conditions:
      
-This package also contains some minor parts which 
+  The above copyright notice and this permission notice shall be included in all
-which are useable under the MIT license. 
+  copies or substantial portions of the Software.
  
-If you have questions and/or want to use parts of 
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-the code under a different license than the GPL 
+  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-please contact me. 
+  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+  SOFTWARE.
 
-holger krekel, January 2010, holger at merlinux eu 

MANIFEST.in

@@ -2,5 +2,6 @@ include CHANGELOG
 include LICENSE 
 include README.txt
 include setup.py
+include tox.ini
 graft testing
 prune .hg 

README.txt

@@ -1,19 +1,27 @@
+.. 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://pypip.in/v/pytest-xdist/badge.png
+   :target: https://pypi.python.org/pypi/pytest-xdist
+
 xdist: pytest distributed testing plugin
-===============================================================
+============================
 
 The `pytest-xdist`_ plugin extends py.test with some unique
 test execution modes:
 
-* Looponfail: run your tests repeatedly in a subprocess.  After each run py.test
+* test run parallelization_: if you have multiple CPUs or hosts you can use
-  waits until a file in your project changes and then re-runs the previously
-  failing tests.  This is repeated until all tests pass after which again
-  a full run is performed.
-
-* multiprocess Load-balancing: if you have multiple CPUs or hosts you can use
   those for a combined test run.  This allows to speed up
-  development or to use special resources of remote machines.
+  development or to use special resources of `remote machines`_.
 
-* Multi-Platform coverage: you can specify different Python interpreters
+* ``--boxed``: (not available on Windows) run each test in a boxed_
+  subprocess to survive ``SEGFAULTS`` or otherwise dying processes
+
+* ``--looponfail``: run your tests repeatedly in a subprocess.  After each run
+  py.test waits until a file in your project changes and then re-runs
+  the previously failing tests.  This is repeated until all tests pass
+  after which again a full run is performed.
+
+* `Multi-Platform`_ coverage: you can specify different Python interpreters
   or different platforms and run tests in parallel on all of them.
 
 Before running tests remotely, ``py.test`` efficiently "rsyncs" your
@@ -41,6 +49,8 @@ a checkout of the `pytest-xdist repository`_ ::
 Usage examples
 ---------------------
 
+.. _parallelization:
+
 Speed up test runs by sending tests to multiple CPUs
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
@@ -49,26 +59,54 @@ To send tests to multiple CPUs, type::
     py.test -n NUM
 
 Especially for longer running tests or tests requiring
-a lot of IO this can lead to considerable speed ups.
+a lot of IO this can lead to considerable speed ups. This option can
+also be set to ``auto`` for automatic detection of the number of CPUs.
+
+If a test crashes the interpreter, pytest-xdist will automatically restart
+that slave and report the failure as usual. You can use the
+``--max-slave-restart`` option to limit the number of slaves that can
+be restarted, or disable restarting altogether using ``--max-slave-restart=0``.
 
 
 Running tests in a Python subprocess
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
-To instantiate a python2.4 sub process and send tests to it, you may type::
+To instantiate a python2.5 sub process and send tests to it, you may type::
 
-    py.test -d --tx popen//python=python2.4
+    py.test -d --tx popen//python=python2.5
 
-This will start a subprocess which is run with the "python2.4"
+This will start a subprocess which is run with the "python2.5"
 Python interpreter, found in your system binary lookup path.
 
 If you prefix the --tx option value like this::
 
-    --tx 3*popen//python=python2.4
+    --tx 3*popen//python=python2.5
 
 then three subprocesses would be created and tests
 will be load-balanced across these three processes.
 
+.. _boxed:
+
+Running tests in a boxed subprocess
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+If you have tests involving C or C++ libraries you might have to deal
+with tests crashing the process.  For this case you may use the boxing
+options::
+
+    py.test --boxed
+
+which will run each test in a subprocess and will report if a test
+crashed the process.  You can also combine this option with
+running multiple processes to speed up the test run and use your CPU cores::
+
+    py.test -n3 --boxed
+
+this would run 3 testing subprocesses in parallel which each
+create new boxed subprocesses for each test.
+
+
+.. _`remote machines`:
 
 Sending tests to remote SSH accounts
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -95,6 +133,13 @@ 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
+to be ignored when file are sent to the remote side.
+There are also internal ignores: .*, *.pyc, *.pyo, *~
+Those you cannot override using rsyncignore command-line or
+ini-file option(s).
+
+
 Sending tests to remote Socket Servers
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
@@ -111,6 +156,8 @@ new socket host with something like this::
 
 
 .. _`atonce`:
+.. _`Multi-Platform`:
+
 
 Running tests on many platforms at once
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -165,6 +212,10 @@ 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/hpk42/pytest-xdist
+.. _`pytest-xdist repository`: http://bitbucket.org/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 .

example/boxed.txt

@@ -0,0 +1,62 @@
+
+
+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::
+
+    # content of test_module.py
+
+    import pytest
+    import os
+    import time
+
+    # run test function 50 times with different argument
+    @pytest.mark.parametrize("arg", range(50))
+    def test_func(arg):
+        time.sleep(0.05) # each tests takes a while
+        if arg % 19 == 0: 
+            os.kill(os.getpid(), 15)
+
+If you run this with::
+
+    $ py.test --boxed
+    =========================== test session starts ============================
+    platform linux2 -- Python 2.7.3 -- pytest-2.3.0.dev8
+    plugins: xdist, bugzilla, cache, oejskit, cli, pep8, cov
+    collecting ... collected 50 items
+    
+    test_module.py f..................f..................f...........
+    
+    ================================= FAILURES =================================
+    _______________________________ test_func[0] _______________________________
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    ______________________________ test_func[19] _______________________________
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    ______________________________ test_func[38] _______________________________
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    =================== 3 failed, 47 passed in 3.41 seconds ====================
+
+You'll see that a couple of tests are reported as crashing, indicated
+by lower-case ``f`` and the respective failure summary. You can also use
+the xdist-provided parallelization feature to speed up your testing::
+
+    $ py.test --boxed -n3
+    =========================== test session starts ============================
+    platform linux2 -- Python 2.7.3 -- pytest-2.3.0.dev8
+    plugins: xdist, bugzilla, cache, oejskit, cli, pep8, cov
+    gw0 I / gw1 I / gw2 I
+    gw0 [50] / gw1 [50] / gw2 [50]
+    
+    scheduling tests via LoadScheduling
+    ..f...............f..................f............
+    ================================= FAILURES =================================
+    _______________________________ test_func[0] _______________________________
+    [gw0] linux2 -- Python 2.7.3 /home/hpk/venv/1/bin/python
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    ______________________________ test_func[19] _______________________________
+    [gw2] linux2 -- Python 2.7.3 /home/hpk/venv/1/bin/python
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    ______________________________ test_func[38] _______________________________
+    [gw2] linux2 -- Python 2.7.3 /home/hpk/venv/1/bin/python
+    /home/hpk/tmp/doc-exec-420/test_module.py:6: running the test CRASHED with signal 15
+    =================== 3 failed, 47 passed in 2.03 seconds ====================

setup.cfg

@@ -0,0 +1,2 @@
+[bdist_wheel]
+universal = 1

setup.py

@@ -2,29 +2,36 @@ from setuptools import setup
 
 setup(
     name="pytest-xdist",
-    version='1.8',
+    use_scm_version={'write_to': 'xdist/_version.py'},
     description='py.test xdist plugin for distributed testing and loop-on-failing modes',
     long_description=open('README.txt').read(),
-    license='GPLv2 or later',
+    license='MIT',
     author='holger krekel and contributors',
-    author_email='py-dev@codespeak.net,holger@merlinux.eu',
+    author_email='pytest-dev@python.org,holger@merlinux.eu',
     url='http://bitbucket.org/hpk42/pytest-xdist',
     platforms=['linux', 'osx', 'win32'],
     packages = ['xdist'],
-    entry_points = {'pytest11': ['xdist = xdist.plugin'],},
+    entry_points = {
+        'pytest11': [
+            'xdist = xdist.plugin',
+            'xdist.looponfail = xdist.looponfail',
+            'xdist.boxed = xdist.boxed',
+        ],
+    },
     zip_safe=False,
-    install_requires = ['execnet>=1.0.8', 'pytest>=2.2.1'],
+    install_requires=['execnet>=1.1', 'pytest>=2.4.2', 'py>=1.4.22'],
+    setup_requires=['setuptools_scm'],
     classifiers=[
-    'Development Status :: 5 - Production/Stable',
+        'Development Status :: 5 - Production/Stable',
-    'Intended Audience :: Developers',
+        'Intended Audience :: Developers',
-    'License :: OSI Approved :: GNU General Public License (GPL)',
+        'License :: OSI Approved :: MIT License',
-    'Operating System :: POSIX',
+        'Operating System :: POSIX',
-    'Operating System :: Microsoft :: Windows',
+        'Operating System :: Microsoft :: Windows',
-    'Operating System :: MacOS :: MacOS X',
+        'Operating System :: MacOS :: MacOS X',
-    'Topic :: Software Development :: Testing',
+        'Topic :: Software Development :: Testing',
-    'Topic :: Software Development :: Quality Assurance',
+        'Topic :: Software Development :: Quality Assurance',
-    'Topic :: Utilities',
+        'Topic :: Utilities',
-    'Programming Language :: Python',
+        'Programming Language :: Python',
-    'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3',
     ],
 )

testing/acceptance_test.py

@@ -1,5 +1,6 @@
 import py
-import sys
+import pytest
+
 
 class TestDistribution:
     def test_n1_pass(self, testdir):
@@ -24,6 +25,30 @@ class TestDistribution:
             "*1 failed*",
         ])
 
+    def test_n1_import_error(self, testdir):
+        p1 = testdir.makepyfile("""
+            import __import_of_missing_module
+            def test_import():
+                pass
+        """)
+        result = testdir.runpytest(p1, "-n1")
+        assert result.ret == 1
+        result.stdout.fnmatch_lines([
+            "E   ImportError: *__import_of_missing_module*",
+        ])
+
+    def test_n2_import_error(self, testdir):
+        """Check that we don't report the same import error multiple times
+        in distributed mode."""
+        p1 = testdir.makepyfile("""
+            import __import_of_missing_module
+            def test_import():
+                pass
+        """)
+        result1 = testdir.runpytest(p1, "-n2")
+        result2 = testdir.runpytest(p1, "-n1")
+        assert len(result1.stdout.lines) == len(result2.stdout.lines)
+
     def test_n1_skip(self, testdir):
         p1 = testdir.makepyfile("""
             def test_skip():
@@ -36,6 +61,18 @@ class TestDistribution:
             "*1 skipped*",
         ])
 
+    def test_manytests_to_one_import_error(self, testdir):
+        p1 = testdir.makepyfile("""
+            import __import_of_missing_module
+            def test_import():
+                pass
+        """)
+        result = testdir.runpytest(p1, '--tx=popen', '--tx=popen')
+        assert result.ret == 1
+        result.stdout.fnmatch_lines([
+            "E   ImportError: *__import_of_missing_module*",
+        ])
+
     def test_manytests_to_one_popen(self, testdir):
         p1 = testdir.makepyfile("""
                 import py
@@ -76,7 +113,7 @@ class TestDistribution:
                 import py
                 assert tmpdir.relto(py.path.local(%r)), tmpdir
         """ % str(testdir.tmpdir))
-        result = testdir.runpytest(p1, "-n1")
+        result = testdir.runpytest_subprocess(p1, "-n1")
         assert result.ret == 0
         result.stdout.fnmatch_lines([
             "*1 passed*",
@@ -157,7 +194,7 @@ class TestDistribution:
         assert dest.join(subdir.basename).check(dir=1)
 
     def test_data_exchange(self, testdir):
-        c1 = testdir.makeconftest("""
+        testdir.makeconftest("""
             # This hook only called on master.
             def pytest_configure_node(node):
                 node.slaveinput['a'] = 42
@@ -206,7 +243,7 @@ class TestDistribution:
                 print ("s2call-finished")
         """)
         args = ["-n1", "--debug"]
-        result = testdir.runpytest(*args)
+        result = testdir.runpytest_subprocess(*args)
         s = result.stdout.str()
         assert result.ret == 2
         assert 's2call' in s
@@ -214,13 +251,13 @@ class TestDistribution:
 
     def test_keyboard_interrupt_dist(self, testdir):
         # xxx could be refined to check for return code
-        p = testdir.makepyfile("""
+        testdir.makepyfile("""
             def test_sleep():
                 import time
                 time.sleep(10)
         """)
-        child = testdir.spawn_pytest("-n1")
+        child = testdir.spawn_pytest("-n1 -v")
-        child.expect(".*test session starts.*")
+        child.expect(".*test_sleep.*")
         child.kill(2) # keyboard interrupt
         child.expect(".*KeyboardInterrupt.*")
         #child.expect(".*seconds.*")
@@ -233,7 +270,7 @@ class TestDistEach:
             def test_hello():
                 pass
         """)
-        result = testdir.runpytest("--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*"])
 
@@ -263,7 +300,7 @@ class TestDistEach:
 
 class TestTerminalReporting:
     def test_pass_skip_fail(self, testdir):
-        p = testdir.makepyfile("""
+        testdir.makepyfile("""
             import py
             def test_ok():
                 pass
@@ -274,9 +311,9 @@ class TestTerminalReporting:
         """)
         result = testdir.runpytest("-n1", "-v")
         result.stdout.fnmatch_lines_random([
-            "*PASS*test_pass_skip_fail.py:2: *test_ok*",
+            "*PASS*test_pass_skip_fail.py*test_ok*",
-            "*SKIP*test_pass_skip_fail.py:4: *test_skip*",
+            "*SKIP*test_pass_skip_fail.py*test_skip*",
-            "*FAIL*test_pass_skip_fail.py:6: *test_func*",
+            "*FAIL*test_pass_skip_fail.py*test_func*",
         ])
         result.stdout.fnmatch_lines([
             "*def test_func():",
@@ -285,13 +322,13 @@ class TestTerminalReporting:
         ])
 
     def test_fail_platinfo(self, testdir):
-        p = testdir.makepyfile("""
+        testdir.makepyfile("""
             def test_func():
                 assert 0
         """)
         result = testdir.runpytest("-n1", "-v")
         result.stdout.fnmatch_lines([
-            "*FAIL*test_fail_platinfo.py:1: *test_func*",
+            "*FAIL*test_fail_platinfo.py*test_func*",
             "*0*Python*",
             "*def test_func():",
             ">       assert 0",
@@ -314,7 +351,7 @@ def test_teardownfails_one_function(testdir):
 @py.test.mark.xfail
 def test_terminate_on_hangingnode(testdir):
     p = testdir.makeconftest("""
-        def pytest_sessionfinishes(session):
+        def pytest_sessionfinish(session):
             if session.nodeid == "my": # running on slave
                 import time
                 time.sleep(3)
@@ -326,7 +363,20 @@ def test_terminate_on_hangingnode(testdir):
     ])
 
 
+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)
 def test_session_hooks(testdir):
     testdir.makeconftest("""
         import sys
@@ -360,6 +410,31 @@ def test_session_hooks(testdir):
     assert testdir.tmpdir.join("slave").check()
     assert testdir.tmpdir.join("master").check()
 
+
+def test_session_testscollected(testdir):
+    """
+    Make sure master node is updating the session object with the number
+    of tests collected from the slaves.
+    """
+    testdir.makepyfile(test_foo="""
+        import pytest
+        @pytest.mark.parametrize('i', range(3))
+        def test_ok(i):
+            pass
+    """)
+    testdir.makeconftest("""
+        def pytest_sessionfinish(session):
+            collected = getattr(session, 'testscollected', None)
+            with open('testscollected', 'w') as f:
+                f.write('collected = %s' % collected)
+    """)
+    result = testdir.inline_run("-n1")
+    result.assertoutcome(passed=3)
+    collected_file = testdir.tmpdir.join('testscollected')
+    assert collected_file.isfile()
+    assert collected_file.read() == 'collected = 3'
+
+
 def test_funcarg_teardown_failure(testdir):
     p = testdir.makepyfile("""
         def pytest_funcarg__myarg(request):
@@ -370,7 +445,7 @@ def test_funcarg_teardown_failure(testdir):
         def test_hello(myarg):
             pass
     """)
-    result = testdir.runpytest("--debug", p) # , "-n1")
+    result = testdir.runpytest_subprocess("--debug", p) # , "-n1")
     result.stdout.fnmatch_lines([
         "*ValueError*42*",
         "*1 passed*1 error*",
@@ -417,8 +492,142 @@ def test_issue34_pluginloading_in_subprocess(testdir):
         def test_hello():
             assert pytest.sample_variable == "testing"
     """)
-    result = testdir.runpytest("-n1", "-p", "plugin123")
+    result = testdir.runpytest_subprocess("-n1", "-p", "plugin123")
     assert result.ret == 0
     result.stdout.fnmatch_lines([
         "*1 passed*",
     ])
+
+
+def test_fixture_scope_caching_issue503(testdir):
+    p1 = testdir.makepyfile("""
+            import pytest
+
+            @pytest.fixture(scope='session')
+            def fix():
+                assert fix.counter == 0, 'session fixture was invoked multiple times'
+                fix.counter += 1
+            fix.counter = 0
+
+            def test_a(fix):
+                pass
+
+            def test_b(fix):
+                pass
+    """)
+    result = testdir.runpytest(p1, '-v', '-n1')
+    assert result.ret == 0
+    result.stdout.fnmatch_lines([
+    "*2 passed*",
+    ])
+
+
+def test_issue_594_random_parametrize(testdir):
+    """
+    Make sure that tests that are randomly parametrized display an appropriate
+    error message, instead of silently skipping the entire test run.
+    """
+    p1 = testdir.makepyfile("""
+        import pytest
+        import random
+
+        xs = list(range(10))
+        random.shuffle(xs)
+        @pytest.mark.parametrize('x', xs)
+        def test_foo(x):
+            assert 1
+    """)
+    result = testdir.runpytest(p1, '-v', '-n4')
+    assert result.ret == 1
+    result.stdout.fnmatch_lines([
+        "Different tests were collected between gw* and gw*",
+    ])
+
+
+class TestNodeFailure:
+
+    def test_load_single(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): os._exit(1)
+            def test_b(): pass
+        """)
+        res = testdir.runpytest(f, '-n1')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Slave*crashed while running*",
+            "*1 failed*1 passed*",
+        ])
+
+    def test_load_multiple(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): pass
+            def test_d(): pass
+        """)
+        res = testdir.runpytest(f, '-n2')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Slave*crashed while running*",
+            "*1 failed*3 passed*",
+        ])
+
+    def test_each_single(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): os._exit(1)
+            def test_b(): pass
+        """)
+        res = testdir.runpytest(f, '--dist=each', '--tx=popen')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Slave*crashed while running*",
+            "*1 failed*1 passed*",
+        ])
+
+    def test_each_multiple(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): os._exit(1)
+            def test_b(): pass
+        """)
+        res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Slave*crashed while running*",
+            "*2 failed*2 passed*",
+        ])
+
+    def test_max_slave_restart(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): os._exit(1)
+            def test_d(): pass
+        """)
+        res = testdir.runpytest(f, '-n4', '--max-slave-restart=1')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Maximum crashed slaves reached: 1*",
+            "*Slave*crashed while running*",
+            "*Slave*crashed while running*",
+            "*2 failed*2 passed*",
+        ])
+
+
+    def test_disable_restart(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): pass
+        """)
+        res = testdir.runpytest(f, '-n4', '--max-slave-restart=0')
+        res.stdout.fnmatch_lines([
+            "*Slave restarting disabled*",
+            "*Slave*crashed while running*",
+            "*1 failed*2 passed*",
+        ])

testing/conftest.py

@@ -1,20 +1,43 @@
 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
+    # will unload all modules after each test and this cause
+    # (unknown) problems with execnet.Group()
+    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", default=None,
+       action="append", dest="gspecs",
        help=("add a global test environment, XSpec-syntax. "))
 
 def pytest_funcarg__specssh(request):
     return getspecssh(request.config)
-def getgspecs(config):
+
-    return [execnet.XSpec(spec)
+@pytest.fixture
-                for spec in config.getvalueorskip("gspecs")]
+def testdir(testdir):
+    # pytest before 2.8 did not have a runpytest_subprocess
+    if not hasattr(testdir, "runpytest_subprocess"):
+        testdir.runpytest_subprocess = testdir.runpytest
+    return testdir
 
 # configuration information for tests
 def getgspecs(config):

testing/test_boxed.py

@@ -1,6 +1,10 @@
-import py
+import pytest
+import os
 
-@py.test.mark.skipif("not hasattr(os, 'fork')")
+needsfork = pytest.mark.skipif(not hasattr(os, "fork"),
+                               reason="os.fork required")
+
+@needsfork
 def test_functional_boxed(testdir):
     p1 = testdir.makepyfile("""
         import os
@@ -13,12 +17,36 @@ def test_functional_boxed(testdir):
         "*1 failed*"
     ])
 
+@needsfork
+@pytest.mark.parametrize("capmode", [
+    "no",
+    pytest.mark.xfail("sys", reason="capture cleanup needed"),
+    pytest.mark.xfail("fd", reason="capture cleanup needed")])
+def test_functional_boxed_capturing(testdir, capmode):
+    p1 = testdir.makepyfile("""
+        import os
+        import sys
+        def test_function():
+            sys.stdout.write("hello\\n")
+            sys.stderr.write("world\\n")
+            os.kill(os.getpid(), 15)
+    """)
+    result = testdir.runpytest(p1, "--boxed", "--capture=%s" % capmode)
+    result.stdout.fnmatch_lines("""
+        *CRASHED*
+        *stdout*
+        hello
+        *stderr*
+        world
+        *1 failed*
+    """)
+
 class TestOptionEffects:
     def test_boxed_option_default(self, testdir):
         tmpdir = testdir.tmpdir.ensure("subdir", dir=1)
         config = testdir.parseconfig()
         assert not config.option.boxed
-        py.test.importorskip("execnet")
+        pytest.importorskip("execnet")
         config = testdir.parseconfig('-d', tmpdir)
         assert not config.option.boxed
 

testing/test_dsession.py

@@ -1,6 +1,11 @@
-from xdist.dsession import DSession, LoadScheduling, EachScheduling
+from xdist.dsession import (
-from _pytest import main as outcome
+    DSession,
+    LoadScheduling,
+    EachScheduling,
+    report_collection_diff,
+)
 import py
+import pytest
 import execnet
 
 XSpec = execnet.XSpec
@@ -12,19 +17,23 @@ def run(item, node, excinfo=None):
     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()
 
-    def send_runtest(self, nodeid):
+    def send_runtest_some(self, indices):
-        self.sent.append(nodeid)
+        self.sent.extend(indices)
 
     def send_runtest_all(self):
         self.sent.append("ALL")
 
-    def sendlist(self, items):
-        self.sent.extend(items)
-
     def shutdown(self):
         self._shutdown=True
 
@@ -51,9 +60,9 @@ class TestEachScheduling:
         assert sched.tests_finished()
         assert node1.sent == ['ALL']
         assert node2.sent == ['ALL']
-        sched.remove_item(node1, collection[0])
+        sched.remove_item(node1, 0)
         assert sched.tests_finished()
-        sched.remove_item(node2, collection[0])
+        sched.remove_item(node2, 0)
         assert sched.tests_finished()
 
     def test_schedule_remove_node(self):
@@ -74,11 +83,10 @@ class TestEachScheduling:
 
 class TestLoadScheduling:
     def test_schedule_load_simple(self):
-        node1 = MockNode()
-        node2 = MockNode()
         sched = LoadScheduling(2)
-        sched.addnode(node1)
+        sched.addnode(MockNode())
-        sched.addnode(node2)
+        sched.addnode(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)
@@ -88,39 +96,40 @@ class TestLoadScheduling:
         assert sched.node2collection[node1] == collection
         assert sched.node2collection[node2] == collection
         sched.init_distribute()
-        assert sched.tests_finished()
-        assert len(node1.sent) == 1
-        assert len(node2.sent) == 1
-        x = sorted(node1.sent + node2.sent)
-        assert x == collection
-        sched.remove_item(node1, node1.sent[0])
-        sched.remove_item(node2, node2.sent[0])
-        assert sched.tests_finished()
         assert not sched.pending
+        assert not sched.tests_finished()
+        assert len(node1.sent) == 2
+        assert len(node2.sent) == 0
+        assert node1.sent == [0, 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):
         sched = LoadScheduling(2)
-        node1 = MockNode()
+        sched.addnode(MockNode())
-        node2 = MockNode()
+        sched.addnode(MockNode())
-        sched.addnode(node1)
+        node1, node2 = sched.nodes
-        sched.addnode(node2)
+        col = ["xyz"] * (6)
-        sched.ITEM_CHUNKSIZE = 2
-        col = ["xyz"] * (2*sched.ITEM_CHUNKSIZE +1)
         sched.addnode_collection(node1, col)
         sched.addnode_collection(node2, col)
         sched.init_distribute()
         #assert not sched.tests_finished()
         sent1 = node1.sent
         sent2 = node2.sent
-        chunkitems = col[:sched.ITEM_CHUNKSIZE]
+        assert sent1 == [0, 1]
-        assert sent1 == chunkitems
+        assert sent2 == [2, 3]
-        assert sent2 == chunkitems
+        assert sched.pending == [4, 5]
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
-        assert len(sched.pending) == 1
+        assert len(sched.pending) == 2
-        for node in (node1, node2):
+        sched.remove_item(node1, 0)
-            for i in range(sched.ITEM_CHUNKSIZE):
+        assert node1.sent == [0, 1, 4]
-                sched.remove_item(node, "xyz")
+        assert sched.pending == [5]
+        assert node2.sent == [2, 3]
+        sched.remove_item(node1, 1)
+        assert node1.sent == [0, 1, 4, 5]
         assert not sched.pending
 
     def test_add_remove_node(self):
@@ -135,6 +144,37 @@ class TestLoadScheduling:
         crashitem = sched.remove_node(node)
         assert crashitem == collection[0]
 
+    def test_different_tests_collected(self, testdir):
+        """
+        Test that LoadScheduling is reporting collection errors when
+        different test ids are collected by slaves.
+        """
+        class CollectHook(object):
+            """
+            Dummy hook that stores collection reports.
+            """
+
+            def __init__(self):
+                self.reports = []
+
+            def pytest_collectreport(self, report):
+                self.reports.append(report)
+
+        collect_hook = CollectHook()
+        config = testdir.parseconfig()
+        config.pluginmanager.register(collect_hook, "collect_hook")
+        node1 = MockNode()
+        node2 = MockNode()
+        sched = LoadScheduling(2, config=config)
+        sched.addnode(node1)
+        sched.addnode(node2)
+        sched.addnode_collection(node1, ["a.py::test_1"])
+        sched.addnode_collection(node2, ["a.py::test_2"])
+        sched.init_distribute()
+        assert len(collect_hook.reports) == 1
+        rep = collect_hook.reports[0]
+        assert 'Different tests were collected between' in rep.longrepr
+
 
 class TestDistReporter:
 
@@ -165,3 +205,46 @@ class TestDistReporter:
         linecomp.assert_contains_lines([
             "[X1,X2] rsyncing: hello",
         ])
+
+
+def test_report_collection_diff_equal():
+    """Test reporting of equal collections."""
+    from_collection = to_collection = ['aaa', 'bbb', 'ccc']
+    assert report_collection_diff(from_collection, to_collection, 1, 2) is None
+
+
+def test_report_collection_diff_different():
+    """Test reporting of different collections."""
+    from_collection = ['aaa', 'bbb', 'ccc', 'YYY']
+    to_collection = ['aZa', 'bbb', 'XXX', 'ccc']
+    error_message = (
+        'Different tests were collected between 1 and 2. The difference is:\n'
+        '--- 1\n'
+        '\n'
+        '+++ 2\n'
+        '\n'
+        '@@ -1,4 +1,4 @@\n'
+        '\n'
+        '-aaa\n'
+        '+aZa\n'
+        ' bbb\n'
+        '+XXX\n'
+        ' ccc\n'
+        '-YYY'
+    )
+
+    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("""
+        import pytest
+
+        @pytest.mark.parametrize('birth_year', [1988, 1988, ])
+        def test_2011_table(birth_year):
+            pass
+    """)
+    reprec = testdir.inline_run("-n1")
+    reprec.assertoutcome(passed=2)
+    assert 0

testing/test_looponfail.py

@@ -14,6 +14,10 @@ class TestStatRecorder:
         changed = sd.check()
         assert changed
 
+        (hello + "c").write("hello")
+        changed = sd.check()
+        assert not changed
+
         p = tmp.ensure("new.py")
         changed = sd.check()
         assert changed
@@ -36,6 +40,12 @@ class TestStatRecorder:
         changed = sd.check()
         assert changed
 
+    def test_dirchange(self, tmpdir):
+        tmp = tmpdir
+        tmp.ensure("dir", "hello.py")
+        sd = StatRecorder([tmp])
+        assert not sd.fil(tmp.join("dir"))
+
     def test_filechange_deletion_race(self, tmpdir, monkeypatch):
         tmp = tmpdir
         sd = StatRecorder([tmp])
@@ -63,11 +73,9 @@ class TestStatRecorder:
 
         pycfile = hello + "c"
         pycfile.ensure()
-        changed = sd.check()
-        assert not changed
-
         hello.write("world")
         changed = sd.check()
+        assert changed
         assert not pycfile.check()
 
     def test_waitonchange(self, tmpdir, monkeypatch):
@@ -206,6 +214,24 @@ class TestLooponFailing:
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 1
 
+    def test_looponfail_multiple_errors(self, testdir, monkeypatch):
+        modcol = testdir.getmodulecol("""
+            def test_one():
+                assert 0
+        """)
+        remotecontrol = RemoteControl(modcol.config)
+        orig_runsession = remotecontrol.runsession
+
+        def runsession_dups():
+            # twisted.trial test cases may report multiple errors.
+            failures, reports, collection_failed = orig_runsession()
+            print (failures)
+            return failures * 2, reports, collection_failed
+
+        monkeypatch.setattr(remotecontrol, 'runsession', runsession_dups)
+        remotecontrol.loop_once()
+        assert len(remotecontrol.failures) == 1
+
 
 class TestFunctional:
     def test_fail_to_ok(self, testdir):

testing/test_plugin.py

@@ -13,7 +13,7 @@ def test_dist_incompatibility_messages(testdir):
     assert "incompatible" in result.stderr.str()
 
 def test_dist_options(testdir):
-    from xdist.plugin import check_options
+    from xdist.plugin import pytest_cmdline_main as check_options
     config = testdir.parseconfigure("-n 2")
     check_options(config)
     assert config.option.dist == "load"
@@ -46,6 +46,11 @@ class TestDistOptions:
         assert nm.roots
         assert testdir.tmpdir in nm.roots
 
+    def test_getrsyncignore(self, testdir):
+        config = testdir.parseconfigure('--rsyncignore=fo*')
+        nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
+        assert 'fo*' in nm.rsyncoptions['ignores']
+
     def test_getrsyncdirs_with_conftest(self, testdir):
         p = py.path.local()
         for bn in 'x y z'.split():
@@ -62,4 +67,3 @@ class TestDistOptions:
         assert py.path.local('y') in roots
         assert py.path.local('z') in roots
         assert testdir.tmpdir.join('x') in roots
-

testing/test_remote.py

@@ -3,7 +3,6 @@ from xdist.slavemanage import SlaveController, unserialize_report
 from xdist.remote import serialize_report
 import execnet
 queue = py.builtin._tryimport("queue", "Queue")
-from py.builtin import print_
 import marshal
 
 WAIT_TIMEOUT = 10.0
@@ -26,7 +25,7 @@ class SlaveSetup:
     use_callback = False
 
     def __init__(self, request):
-        self.testdir = testdir = request.getfuncargvalue("testdir")
+        self.testdir = request.getfuncargvalue("testdir")
         self.request = request
         self.events = queue.Queue()
 
@@ -88,7 +87,7 @@ class TestReportSerialization:
             assert newrep.passed == rep.passed
             assert newrep.failed == rep.failed
             assert newrep.skipped == rep.skipped
-            if newrep.skipped and 'xfail' not in newrep.keywords:
+            if newrep.skipped and not hasattr(newrep, "wasxfail"):
                 assert len(newrep.longrepr) == 3
             assert newrep.outcome == rep.outcome
             assert newrep.when == rep.when
@@ -121,9 +120,26 @@ class TestReportSerialization:
             if rep.failed:
                 assert newrep.longrepr == str(rep.longrepr)
 
+    def test_extended_report_deserialization(self, testdir):
+        reprec = testdir.inline_runsource("qwe abc")
+        reports = reprec.getreports("pytest_collectreport")
+        assert reports
+        for rep in reports:
+            rep.extra = True
+            d = serialize_report(rep)
+            check_marshallable(d)
+            newrep = unserialize_report("collectreport", d)
+            assert newrep.extra
+            assert newrep.passed == rep.passed
+            assert newrep.failed == rep.failed
+            assert newrep.skipped == rep.skipped
+            if rep.failed:
+                assert newrep.longrepr == str(rep.longrepr)
+
+
 class TestSlaveInteractor:
     def test_basic_collect_and_runtests(self, slave):
-        p = slave.testdir.makepyfile("""
+        slave.testdir.makepyfile("""
             def test_func():
                 pass
         """)
@@ -137,8 +153,11 @@ class TestSlaveInteractor:
         assert ev.kwargs['topdir'] == slave.testdir.tmpdir
         ids = ev.kwargs['ids']
         assert len(ids) == 1
-        slave.sendcommand("runtests", ids=ids)
+        slave.sendcommand("runtests", indices=list(range(len(ids))))
         slave.sendcommand("shutdown")
+        ev = slave.popevent("logstart")
+        assert ev.kwargs["nodeid"].endswith("test_func")
+        assert len(ev.kwargs["location"]) == 3
         ev = slave.popevent("testreport") # setup
         ev = slave.popevent("testreport")
         assert ev.name == "testreport"
@@ -150,7 +169,7 @@ class TestSlaveInteractor:
         assert 'slaveoutput' in ev.kwargs
 
     def test_remote_collect_skip(self, slave):
-        p = slave.testdir.makepyfile("""
+        slave.testdir.makepyfile("""
             import py
             py.test.skip("hello")
         """)
@@ -167,7 +186,7 @@ class TestSlaveInteractor:
         assert not ev.kwargs['ids']
 
     def test_remote_collect_fail(self, slave):
-        p = slave.testdir.makepyfile("""aasd qwe""")
+        slave.testdir.makepyfile("""aasd qwe""")
         slave.setup()
         ev = slave.popevent("collectionstart")
         assert not ev.kwargs
@@ -181,7 +200,7 @@ class TestSlaveInteractor:
         assert not ev.kwargs['ids']
 
     def test_runtests_all(self, slave):
-        p = slave.testdir.makepyfile("""
+        slave.testdir.makepyfile("""
             def test_func(): pass
             def test_func2(): pass
         """)
@@ -208,7 +227,7 @@ class TestSlaveInteractor:
     def test_happy_run_events_converted(self, testdir, slave):
         py.test.xfail("implement a simple test for event production")
         assert not slave.use_callback
-        p = slave.testdir.makepyfile("""
+        slave.testdir.makepyfile("""
             def test_func():
                 pass
         """)

testing/test_slavemanage.py

@@ -1,26 +1,35 @@
 import py
-import os
+import pytest
 import execnet
+from _pytest.pytester import HookRecorder
+from xdist import slavemanage, newhooks
 from xdist.slavemanage import HostRSync, NodeManager
 
-pytest_plugins = "pytester",
+pytest_plugins = "pytester"
 
-def pytest_funcarg__hookrecorder(request):
+def pytest_funcarg__hookrecorder(request, config):
-    _pytest = request.getfuncargvalue('_pytest')
+    hookrecorder = HookRecorder(config.pluginmanager)
-    config = request.getfuncargvalue('config')
+    if hasattr(hookrecorder, "start_recording"):
-    return _pytest.gethookrecorder(config.hook)
+        hookrecorder.start_recording(newhooks)
+    request.addfinalizer(hookrecorder.finish_recording)
+    return hookrecorder
 
-def pytest_funcarg__config(request):
+def pytest_funcarg__config(testdir):
-    testdir = request.getfuncargvalue("testdir")
+    return testdir.parseconfig()
-    config = testdir.parseconfig()
-    return config
 
-class pytest_funcarg__mysetup:
+def pytest_funcarg__mysetup(tmpdir):
-    def __init__(self, request):
+    class mysetup:
-        temp = request.getfuncargvalue("tmpdir")
+        source = tmpdir.mkdir("source")
-        self.source = temp.mkdir("source")
+        dest = tmpdir.mkdir("dest")
-        self.dest = temp.mkdir("dest")
+    return mysetup()
-        request.getfuncargvalue("_pytest")
+
+@pytest.fixture
+def slavecontroller(monkeypatch):
+    class MockController(object):
+        def __init__(self, *args): pass
+        def setup(self): pass
+    monkeypatch.setattr(slavemanage, 'SlaveController', MockController)
+    return MockController
 
 class TestNodeManagerPopen:
     def test_popen_no_default_chdir(self, config):
@@ -34,9 +43,9 @@ class TestNodeManagerPopen:
         for spec in NodeManager(config, l, defaultchdir="abc").specs:
             assert spec.chdir == "abc"
 
-    def test_popen_makegateway_events(self, config, hookrecorder, _pytest):
+    def test_popen_makegateway_events(self, config, hookrecorder, slavecontroller):
         hm = NodeManager(config, ["popen"] * 2)
-        hm.makegateways()
+        hm.setup_nodes(None)
         call = hookrecorder.popcall("pytest_xdist_setupnodes")
         assert len(call.specs) == 2
 
@@ -49,10 +58,10 @@ class TestNodeManagerPopen:
         hm.teardown_nodes()
         assert not len(hm.group)
 
-    def test_popens_rsync(self, config, mysetup):
+    def test_popens_rsync(self, config, mysetup, slavecontroller):
         source = mysetup.source
         hm = NodeManager(config, ["popen"] * 2)
-        hm.makegateways()
+        hm.setup_nodes(None)
         assert len(hm.group) == 2
         for gw in hm.group:
             class pseudoexec:
@@ -63,19 +72,21 @@ class TestNodeManagerPopen:
                     pass
             gw.remote_exec = pseudoexec
         l = []
-        hm.rsync(source, notify=lambda *args: l.append(args))
+        for gw in hm.group:
+            hm.rsync(gw, source, notify=lambda *args: l.append(args))
         assert not l
         hm.teardown_nodes()
         assert not len(hm.group)
         assert "sys.path.insert" in gw.remote_exec.args[0]
 
-    def test_rsync_popen_with_path(self, config, mysetup):
+    def test_rsync_popen_with_path(self, config, mysetup, slavecontroller):
         source, dest = mysetup.source, mysetup.dest
-        hm = NodeManager(config, ["popen//chdir=%s" %dest] * 1)
+        hm = NodeManager(config, ["popen//chdir=%s" % dest] * 1)
-        hm.makegateways()
+        hm.setup_nodes(None)
         source.ensure("dir1", "dir2", "hello")
         l = []
-        hm.rsync(source, notify=lambda *args: l.append(args))
+        for gw in hm.group:
+            hm.rsync(gw, source, notify=lambda *args: l.append(args))
         assert len(l) == 1
         assert l[0] == ("rsyncrootready", hm.group['gw0'].spec, source)
         hm.teardown_nodes()
@@ -84,12 +95,15 @@ class TestNodeManagerPopen:
         assert dest.join("dir1", "dir2").check()
         assert dest.join("dir1", "dir2", 'hello').check()
 
-    def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder):
+    def test_rsync_same_popen_twice(self, config, mysetup,
+                                    hookrecorder, slavecontroller):
         source, dest = mysetup.source, mysetup.dest
-        hm = NodeManager(config, ["popen//chdir=%s" %dest] * 2)
+        hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
-        hm.makegateways()
+        hm.roots = []
+        hm.setup_nodes(None)
         source.ensure("dir1", "dir2", "hello")
-        hm.rsync(source)
+        gw = hm.group[0]
+        hm.rsync(gw, source)
         call = hookrecorder.popcall("pytest_xdist_rsyncstart")
         assert call.source == source
         assert len(call.gateways) == 1
@@ -97,19 +111,13 @@ class TestNodeManagerPopen:
         call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
 
 class TestHRSync:
-    class pytest_funcarg__mysetup:
-        def __init__(self, request):
-            tmp = request.getfuncargvalue('tmpdir')
-            self.source = tmp.mkdir("source")
-            self.dest = tmp.mkdir("dest")
-
     def test_hrsync_filter(self, mysetup):
-        source, dest = mysetup.source, mysetup.dest
+        source, _ = mysetup.source, mysetup.dest  # noqa
         source.ensure("dir", "file.txt")
         source.ensure(".svn", "entries")
         source.ensure(".somedotfile", "moreentries")
         source.ensure("somedir", "editfile~")
-        syncer = HostRSync(source)
+        syncer = HostRSync(source, ignores=NodeManager.DEFAULT_IGNORES)
         l = list(source.visit(rec=syncer.filter,
                                    fil=syncer.filter))
         assert len(l) == 3
@@ -132,10 +140,10 @@ class TestHRSync:
 
 
 class TestNodeManager:
-    @py.test.mark.xfail
+    @py.test.mark.xfail(run=False)
     def test_rsync_roots_no_roots(self, testdir, mysetup):
         mysetup.source.ensure("dir1", "file1").write("hello")
-        config = testdir.parseconfig(source)
+        config = testdir.parseconfig(mysetup.source)
         nodemanager = NodeManager(config, ["popen//chdir=%s" % mysetup.dest])
         #assert nodemanager.config.topdir == source == config.topdir
         nodemanager.makegateways()
@@ -148,7 +156,7 @@ class TestNodeManager:
         assert p.join("dir1").check()
         assert p.join("dir1", "file1").check()
 
-    def test_popen_rsync_subdir(self, testdir, mysetup):
+    def test_popen_rsync_subdir(self, testdir, mysetup, slavecontroller):
         source, dest = mysetup.source, mysetup.dest
         dir1 = mysetup.source.mkdir("dir1")
         dir2 = dir1.mkdir("dir2")
@@ -160,8 +168,7 @@ class TestNodeManager:
                 "--rsyncdir", rsyncroot,
                 source,
             ))
-            nodemanager.makegateways()
+            nodemanager.setup_nodes(None)  # calls .rsync_roots()
-            nodemanager.rsync_roots()
             if rsyncroot == source:
                 dest = dest.join("source")
             assert dest.join("dir1").check()
@@ -169,7 +176,7 @@ class TestNodeManager:
             assert dest.join("dir1", "dir2", 'hello').check()
             nodemanager.teardown_nodes()
 
-    def test_init_rsync_roots(self, testdir, mysetup):
+    def test_init_rsync_roots(self, testdir, mysetup, slavecontroller):
         source, dest = mysetup.source, mysetup.dest
         dir2 = source.ensure("dir1", "dir2", dir=1)
         source.ensure("dir1", "somefile", dir=1)
@@ -181,41 +188,43 @@ class TestNodeManager:
         """))
         config = testdir.parseconfig(source)
         nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
-        nodemanager.makegateways()
+        nodemanager.setup_nodes(None)  # calls .rsync_roots()
-        nodemanager.rsync_roots()
         assert dest.join("dir2").check()
         assert not dest.join("dir1").check()
         assert not dest.join("bogus").check()
 
-    def test_rsyncignore(self, testdir, mysetup):
+    def test_rsyncignore(self, testdir, mysetup, slavecontroller):
         source, dest = mysetup.source, mysetup.dest
         dir2 = source.ensure("dir1", "dir2", dir=1)
-        dir5 = source.ensure("dir5", "dir6", "bogus")
+        source.ensure("dir5", "dir6", "bogus")
-        dirf = source.ensure("dir5", "file")
+        source.ensure("dir5", "file")
         dir2.ensure("hello")
+        source.ensure("foo", "bar")
+        source.ensure("bar", "foo")
         source.join("tox.ini").write(py.std.textwrap.dedent("""
             [pytest]
             rsyncdirs = dir1 dir5
-            rsyncignore = dir1/dir2 dir5/dir6
+            rsyncignore = dir1/dir2 dir5/dir6 foo*
         """))
         config = testdir.parseconfig(source)
+        config.option.rsyncignore = ['bar']
         nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
-        nodemanager.makegateways()
+        nodemanager.setup_nodes(None)  # calls .rsync_roots()
-        nodemanager.rsync_roots()
         assert dest.join("dir1").check()
         assert not dest.join("dir1", "dir2").check()
-        assert dest.join("dir5","file").check()
+        assert dest.join("dir5", "file").check()
         assert not dest.join("dir6").check()
+        assert not dest.join('foo').check()
+        assert not dest.join('bar').check()
 
-    def test_optimise_popen(self, testdir, mysetup):
+    def test_optimise_popen(self, testdir, mysetup, slavecontroller):
-        source, dest = mysetup.source, mysetup.dest
+        source = mysetup.source
         specs = ["popen"] * 3
         source.join("conftest.py").write("rsyncdirs = ['a']")
         source.ensure('a', dir=1)
         config = testdir.parseconfig(source)
         nodemanager = NodeManager(config, specs)
-        nodemanager.makegateways()
+        nodemanager.setup_nodes(None)  # calls .rysnc_roots()
-        nodemanager.rsync_roots()
         for gwspec in nodemanager.specs:
             assert gwspec._samefilesystem()
             assert not gwspec.chdir
@@ -226,8 +235,6 @@ class TestNodeManager:
                 pass
         """)
         reprec = testdir.inline_run("-d", "--rsyncdir=%s" % testdir.tmpdir,
-                "--tx", specssh, testdir.tmpdir)
+                                    "--tx", specssh, testdir.tmpdir)
         rep, = reprec.getreports("pytest_runtest_logreport")
         assert rep.passed
-
-

tox.ini

@@ -1,23 +1,27 @@
 [tox]
-envlist=py26,py31,py27,py25,py24
+envlist=
-indexserver=
+  py{26,33,34,27}-pytest2{4,5,6,7},py{27,34}-pytest27-pexpect,flakes
-    testrun = http://pypi.testrun.org
+
-    pypi = http://pypi.python.org/simple
 
 [testenv]
 changedir=testing
-deps=:testrun:pytest>=2.2.0.dev2
+deps =
-commands= py.test --junitxml={envlogdir}/junit-{envname}.xml []
+    pycmd
+    pytest24: pytest~=2.4.0
+    pytest25: pytest~=2.5.0
 
-#[testenv:py31]
+    pytest26: pytest~=2.6.1
-#deps=:pypi:pytest  # XXX needed because ClueRelease/pip broken
+    pytest27: pytest~=2.7.2
-#[testenv:py32]
+    pexpect: pexpect
-#deps=:pypi:pytest  # XXX needed because ClueRelease/pip broken
+commands=
+  # always clean to avoid code unmarshal mismatch on old python/pytest
+  py.cleanup -aq
+  py.test {posargs}
 
-[testenv:py26]
+[testenv:flakes]
-deps=
+changedir=
-    :testrun:pytest
+deps = pytest-flakes>=0.2
-    :pypi:pexpect
+commands = py.test --flakes -m flakes testing xdist
 
 [pytest]
 addopts = -rsfxX

xdist/__init__.py

@@ -1,2 +1,2 @@
-#
+__all__ = ['__version__']
-__version__ = '1.8'
+from xdist._version import version as __version__

xdist/boxed.py

@@ -0,0 +1,56 @@
+
+import py
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("xdist", "distributed and subprocess testing")
+    group.addoption('--boxed',
+           action="store_true", dest="boxed", default=False,
+           help="box each test run in a separate process (unix)")
+
+
+def pytest_runtest_protocol(item):
+    if item.config.getvalue("boxed"):
+        reports = forked_run_report(item)
+        for rep in reports:
+            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
+    from _pytest.runner import runtestprotocol
+    EXITSTATUS_TESTEXIT = 4
+    import marshal
+    from xdist.remote import serialize_report
+    from xdist.slavemanage import unserialize_report
+    def runforked():
+        try:
+            reports = runtestprotocol(item, log=False)
+        except KeyboardInterrupt:
+            py.std.os._exit(EXITSTATUS_TESTEXIT)
+        return marshal.dumps([serialize_report(x) for x in reports])
+
+    ff = py.process.ForkedFunc(runforked)
+    result = ff.waitfinish()
+    if result.retval is not None:
+        report_dumps = marshal.loads(result.retval)
+        return [unserialize_report("testreport", x) for x in report_dumps]
+    else:
+        if result.exitstatus == EXITSTATUS_TESTEXIT:
+            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" %
+            (path, lineno, result.signal))
+    from _pytest import runner
+    call = runner.CallInfo(lambda: 0/0, "???")
+    call.excinfo = info
+    rep = runner.pytest_runtest_makereport(item, call)
+    if result.out:
+        rep.sections.append(("captured stdout", result.out))
+    if result.err:
+        rep.sections.append(("captured stderr", result.err))
+    return rep

xdist/dsession.py

@@ -1,155 +1,451 @@
-import pytest, py
+import difflib
-import sys
+from _pytest.runner import CollectReport
+
+import pytest
+import py
 from xdist.slavemanage import NodeManager
+
+
 queue = py.builtin._tryimport('queue', 'Queue')
 
+
 class EachScheduling:
+    """Implement scheduling of test items on all nodes
+
+    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
+    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
+    finished all it's pending items.  The new node will then be
+    assigned the remaining items from the removed node.
+    """
 
     def __init__(self, numnodes, log=None):
         self.numnodes = numnodes
         self.node2collection = {}
         self.node2pending = {}
+        self._started = []
+        self._removed2pending = {}
         if log is None:
             self.log = py.log.Producer("eachsched")
         else:
-            self.log = log.loadsched
+            self.log = log.eachsched
         self.collection_is_completed = False
 
+    @property
+    def nodes(self):
+        """A list of all nodes in the scheduler."""
+        return list(self.node2pending.keys())
+
     def hasnodes(self):
         return bool(self.node2pending)
 
+    def haspending(self):
+        """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
+        scheduler is active".
+        """
+        for pending in self.node2pending.values():
+            if pending:
+                return True
+        return False
+
     def addnode(self, node):
-        self.node2collection[node] = None
+        assert node not in self.node2pending
+        self.node2pending[node] = []
 
     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 addnode_collection(self, node, collection):
-        assert not self.collection_is_completed
+        """Add the collected test items from a node
-        assert self.node2collection[node] is None
-        self.node2collection[node] = list(collection)
-        self.node2pending[node] = []
-        if len(self.node2pending) >= self.numnodes:
-            self.collection_is_completed = True
 
-    def remove_item(self, node, item):
+        Collection is complete once all nodes have submitted their
-        self.node2pending[node].remove(item)
+        collection.  In this case it's peding 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
+        here.  In either case ``.init_distribute()`` will instruct the
+        node to start running the required tests.
+        """
+        assert node in self.node2pending
+        if not self.collection_is_completed:
+            self.node2collection[node] = list(collection)
+            self.node2pending[node] = []
+            if len(self.node2collection) >= self.numnodes:
+                self.collection_is_completed = True
+        elif self._removed2pending:
+            for deadnode in self._removed2pending:
+                if deadnode.gateway.spec == node.gateway.spec:
+                    dead_collection = self.node2collection[deadnode]
+                    if collection != dead_collection:
+                        msg = report_collection_diff(dead_collection,
+                                                     collection,
+                                                     deadnode.gateway.id,
+                                                     node.gateway.id)
+                        self.log(msg)
+                        return
+                    pending = self._removed2pending.pop(deadnode)
+                    self.node2pending[node] = pending
+                    break
+
+    def remove_item(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
         pending = self.node2pending.pop(node)
         if not pending:
             return
-        crashitem = pending.pop(0)
+        crashitem = self.node2collection[node][pending.pop(0)]
-        # XXX what about the rest of pending?
+        if pending:
+            self._removed2pending[node] = pending
         return crashitem
 
     def init_distribute(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
+        died node and we only need to run those tests.
+        """
         assert self.collection_is_completed
         for node, pending in self.node2pending.items():
-            node.send_runtest_all()
+            if node in self._started:
-            pending[:] = self.node2collection[node]
+                continue
+            if not pending:
+                pending[:] = range(len(self.node2collection[node]))
+                node.send_runtest_all()
+            else:
+                node.send_runtest_some(pending)
+            self._started.append(node)
+
 
 class LoadScheduling:
-    LOAD_THRESHOLD_NEWITEMS = 5
+    """Implement load scheduling accross nodes.
-    ITEM_CHUNKSIZE = 10
 
-    def __init__(self, numnodes, log=None):
+    This distributes the tests collected across all nodes so each test
+    is run just once.  All nodes collect and submit the test suit and
+    when all collections are received it is verified they are
+    identical collections.  Then the collection gets devided up in
+    chunks and chunks get submitted to nodes.  Whenver a node finishes
+    an item they call ``.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
+    submit a collection, this is used to know when all nodes have
+    finished collection or how large the chunks need to be created.
+
+    Attributes:
+
+    :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
+       number is primarily used to know when the initial collection is
+       completed.
+
+    :node2collection: Map of nodes and their test collection.  All
+       collections should always be identical.
+
+    :node2pending: Map of nodes and the indices of their pending
+       tests.  The indices are an index into ``.pending`` (which is
+       identical to their own collection stored in
+       ``.node2collection``).
+
+    :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.
+
+    :pending: List of indices of globally pending tests.  These are
+       tests which have not yet been allocated to a chunk for a node
+       to process.
+
+    :log: A py.log.Producer instance.
+
+    :config: Config object, used for handling hooks.
+    """
+
+    def __init__(self, numnodes, log=None, config=None):
         self.numnodes = numnodes
-        self.node2pending = {}
         self.node2collection = {}
+        self.node2pending = {}
         self.pending = []
+        self.collection = None
         if log is None:
             self.log = py.log.Producer("loadsched")
         else:
             self.log = log.loadsched
-        self.collection_is_completed = False
+        self.config = config
+
+    @property
+    def nodes(self):
+        """A list of all nodes in the scheduler."""
+        return list(self.node2pending.keys())
+
+    @property
+    def collection_is_completed(self):
+        """Boolean indication initial test collection is complete.
+
+        This is a boolean indicating all initial participating nodes
+        have finished collection.  The required number of initial
+        nodes is defined by ``.numnodes``.
+        """
+        return len(self.node2collection) >= self.numnodes
+
+    def haspending(self):
+        """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
+        scheduler is active".
+        """
+        if self.pending:
+            return True
+        for pending in self.node2pending.values():
+            if pending:
+                return True
+        return False
 
     def hasnodes(self):
+        """Return True if nodes exist in the scheduler."""
         return bool(self.node2pending)
 
     def addnode(self, node):
+        """Add a new node in 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.
+        """
+        assert node not in self.node2pending
         self.node2pending[node] = []
 
     def tests_finished(self):
-        if not self.collection_is_completed or self.pending:
+        """Return True if all tests have been executed by the nodes."""
+        if not self.collection_is_completed:
             return False
-        #for items in self.node2pending.values():
+        if self.pending:
-        #    if items:
+            return False
-        #        return False
+        for pending in self.node2pending.values():
+            if len(pending) >= 2:
+                return False
         return True
 
     def addnode_collection(self, node, collection):
-        assert not self.collection_is_completed
+        """Add the collected test items from a node
-        assert node in self.node2pending
-        self.node2collection[node] = list(collection)
-        if len(self.node2collection) >= self.numnodes:
-            self.collection_is_completed = True
 
-    def remove_item(self, node, item):
+        The collection is stored in the ``.node2collection`` map.
-        if item not in self.item2nodes:
+        Called by the ``DSession.slave_collectionfinish`` hook.
-            raise AssertionError(item, self.item2nodes)
+        """
-        nodes = self.item2nodes[item]
+        assert node in self.node2pending
-        if node in nodes: # the node might have gone down already
+        if self.collection_is_completed:
-            nodes.remove(node)
+            # A new node has been added later, perhaps an original one died.
-        #if not nodes:
+            assert self.collection  # .init_distribute() should have
-        #    del self.item2nodes[item]
+                                    # been called by now
-        pending = self.node2pending[node]
+            if collection != self.collection:
-        pending.remove(item)
+                other_node = next(iter(self.node2collection.keys()))
-        # pre-load items-to-test if the node may become ready
+                msg = report_collection_diff(self.collection,
-        if self.pending and len(pending) < self.LOAD_THRESHOLD_NEWITEMS:
+                                             collection,
-            item = self.pending.pop(0)
+                                             other_node.gateway.id,
-            pending.append(item)
+                                             node.gateway.id)
-            self.item2nodes.setdefault(item, []).append(node)
+                self.log(msg)
-            node.send_runtest(item)
+                return
-        self.log("items waiting for node: %d" %(len(self.pending)))
+        self.node2collection[node] = list(collection)
-        #self.log("item2pending still executing: %s" %(self.item2nodes,))
+
-        #self.log("node2pending: %s" %(self.node2pending,))
+    def remove_item(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
+        the scheduler.
+
+        This is called by the ``DSession.slave_testreport`` hook.
+        """
+        self.node2pending[node].remove(item_index)
+        self.check_schedule(node, duration=duration)
+
+    def check_schedule(self, node, duration=0):
+        """Maybe schedule new items on the node
+
+        If there are any globally pending nodes left then this will
+        check if the given node should be given any more tests.  The
+        ``duration`` of the last test is optionally used as a
+        heuristic to influence how many tests the node is assigned.
+        """
+        if self.pending:
+            # how many nodes do we have?
+            num_nodes = len(self.node2pending)
+            # if our node goes below a heuristic minimum, fill it out to
+            # heuristic maximum
+            items_per_node_min = max(2, len(self.pending) // num_nodes // 4)
+            items_per_node_max = max(2, len(self.pending) // num_nodes // 2)
+            node_pending = self.node2pending[node]
+            if len(node_pending) < items_per_node_min:
+                if duration >= 0.1 and len(node_pending) >= 2:
+                    # seems the node is doing long-running tests
+                    # and has enough items to continue
+                    # so let's rather wait with sending new items
+                    return
+                num_send = items_per_node_max - len(node_pending)
+                self._send_tests(node, num_send)
+        self.log("num items waiting for node:", len(self.pending))
 
     def remove_node(self, node):
+        """Remove an 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
+        to the node will be re-scheduled.  Called by the
+        ``DSession.slave_slavefinished`` and
+        ``DSession.slave_errordown`` hooks.
+
+        Return the item which was being executing while the node
+        crashed or None if the node has no more pending items.
+
+        """
         pending = self.node2pending.pop(node)
-        # KeyError if we didn't get an addnode() yet
-        for item in pending:
-            l = self.item2nodes[item]
-            l.remove(node)
-            if not l:
-                del self.item2nodes[item]
         if not pending:
             return
-        crashitem = pending.pop(0)
+
+        # The node crashed, reassing pending items
+        crashitem = self.collection[pending.pop(0)]
         self.pending.extend(pending)
+        for node in self.node2pending:
+            self.check_schedule(node)
         return crashitem
 
     def init_distribute(self):
+        """Initiate distribution of the test collection
+
+        Initiate scheduling of the items across the nodes.  If this
+        gets called again later it behaves the same as calling
+        ``.check_schedule()`` on all nodes so that newly added nodes
+        will start to be used.
+
+        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
-        assert not hasattr(self, 'item2nodes')
+
-        self.item2nodes = {}
+        # Initial distribution already happend, reschedule on all nodes
-        # XXX allow nodes to have different collections
+        if self.collection is not None:
-        col = list(self.node2collection.values())[0]
+            for node in self.nodes:
-        for node, collection in self.node2collection.items():
+                self.check_schedule(node)
-            assert collection == col
-        self.pending = col
-        if not col:
             return
-        available = list(self.node2pending.items())
+
-        num_available = self.numnodes
+        # XXX allow nodes to have different collections
-        max_one_round = num_available * self.ITEM_CHUNKSIZE -1
+        if not self._check_nodes_have_same_collection():
-        for i, item in enumerate(self.pending):
+            self.log('**Different tests collected, aborting run**')
-            nodeindex = i % num_available
+            return
-            node, pending = available[nodeindex]
+
-            node.send_runtest(item)
+        # Collections are identical, create the index of pending items.
-            self.item2nodes.setdefault(item, []).append(node)
+        self.collection = list(self.node2collection.values())[0]
-            pending.append(item)
+        self.pending[:] = range(len(self.collection))
-            if i >= max_one_round:
+        if not self.collection:
-                break
+            return
-        del self.pending[:i+1]
+
+        # how many items per node do we have about?
+        items_per_node = len(self.collection) // len(self.node2pending)
+        # take a fraction of tests for initial distribution
+        node_chunksize = max(items_per_node // 4, 2)
+        # and initialize each node with a chunk of tests
+        for node in self.nodes:
+            self._send_tests(node, node_chunksize)
+
+    def _send_tests(self, node, num):
+        tests_per_node = self.pending[:num]
+        if tests_per_node:
+            del self.pending[:num]
+            self.node2pending[node].extend(tests_per_node)
+            node.send_runtest_some(tests_per_node)
+
+    def _check_nodes_have_same_collection(self):
+        """Return True if all nodes have collected the same items.
+
+        If collections differ, this method returns False while logging
+        the collection differences and posting collection errors to
+        pytest_collectreport hook.
+        """
+        node_collection_items = list(self.node2collection.items())
+        first_node, col = node_collection_items[0]
+        same_collection = True
+        for node, collection in node_collection_items[1:]:
+            msg = report_collection_diff(
+                col,
+                collection,
+                first_node.gateway.id,
+                node.gateway.id,
+            )
+            if msg:
+                same_collection = False
+                self.log(msg)
+                if self.config is not None:
+                    rep = CollectReport(node.gateway.id, 'failed', longrepr=msg,
+                                        result=[])
+                    self.config.hook.pytest_collectreport(report=rep)
+
+        return same_collection
+
+
+def report_collection_diff(from_collection, to_collection, from_id, to_id):
+    """Report the collected test difference between two nodes.
+
+    :returns: detailed message describing the difference between the given
+    collections, or None if they are equal.
+    """
+    if from_collection == to_collection:
+        return None
+
+    diff = difflib.unified_diff(
+        from_collection,
+        to_collection,
+        fromfile=from_id,
+        tofile=to_id,
+    )
+    error_message = py.builtin._totext(
+        'Different tests were collected between {from_id} and {to_id}. '
+        'The difference is:\n'
+        '{diff}'
+    ).format(from_id=from_id, to_id=to_id, diff='\n'.join(diff))
+    msg = "\n".join([x.rstrip() for x in error_message.split("\n")])
+    return msg
+
 
 class Interrupted(KeyboardInterrupt):
     """ signals an immediate interruption. """
 
+
 class DSession:
+    """A py.test plugin which runs a distributed test session
+
+    At the beginning of the test session this creates a NodeManager
+    instance which creates and starts all nodes.  Nodes then emit
+    events processed in the pytest_runtestloop hook using the slave_*
+    methods.
+
+    Once a node is started it will automatically start running the
+    py.test mainloop with some custom hooks.  This means a node
+    automatically starts collecting tests.  Once tests are collected
+    it will wait for instructions.
+    """
     def __init__(self, config):
         self.config = config
         self.log = py.log.Producer("dsession")
@@ -159,6 +455,13 @@ class DSession:
         self.countfailures = 0
         self.maxfail = config.getvalue("maxfail")
         self.queue = queue.Queue()
+        self._session = None
+        self._failed_collection_errors = {}
+        self._active_nodes = set()
+        self._failed_nodes_count = 0
+        self._max_slave_restart = self.config.getoption('max_slave_restart')
+        if self._max_slave_restart is not None:
+            self._max_slave_restart = int(self._max_slave_restart)
         try:
             self.terminal = config.pluginmanager.getplugin("terminalreporter")
         except KeyError:
@@ -167,20 +470,37 @@ class DSession:
             self.trdist = TerminalDistReporter(config)
             config.pluginmanager.register(self.trdist, "terminaldistreporter")
 
+    @property
+    def session_finished(self):
+        """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.
+        """
+        return bool(self.shuttingdown and not self._active_nodes)
+
     def report_line(self, line):
         if self.terminal and self.config.option.verbose >= 0:
             self.terminal.write_line(line)
 
     @pytest.mark.trylast
     def pytest_sessionstart(self, session):
+        """Creates and starts the nodes.
+
+        The nodes are setup to put their events onto self.queue.  As
+        soon as nodes start they will emit the slave_slaveready event.
+        """
         self.nodemanager = NodeManager(self.config)
-        self.nodemanager.setup_nodes(putevent=self.queue.put)
+        nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
+        self._active_nodes.update(nodes)
+        self._session = session
 
     def pytest_sessionfinish(self, session):
-        """ teardown any resources after a test run. """
+        """Shutdown all nodes."""
-        nm = getattr(self, 'nodemanager', None) # if not fully initialized
+        nm = getattr(self, 'nodemanager', None)  # if not fully initialized
         if nm is not None:
             nm.teardown_nodes()
+        self._session = None
 
     def pytest_collection(self):
         # prohibit collection of test items in master process
@@ -190,13 +510,13 @@ class DSession:
         numnodes = len(self.nodemanager.specs)
         dist = self.config.getvalue("dist")
         if dist == "load":
-            self.sched = LoadScheduling(numnodes, log=self.log)
+            self.sched = LoadScheduling(numnodes, log=self.log,
+                                        config=self.config)
         elif dist == "each":
             self.sched = EachScheduling(numnodes, log=self.log)
         else:
             assert 0, dist
         self.shouldstop = False
-        self.session_finished = False
         while not self.session_finished:
             self.loop_once()
             if self.shouldstop:
@@ -204,7 +524,7 @@ class DSession:
         return True
 
     def loop_once(self):
-        """ process one callback from one of the slaves. """
+        """Process one callback from one of the slaves."""
         while 1:
             try:
                 eventcall = self.queue.get(timeout=2.0)
@@ -215,7 +535,7 @@ class DSession:
         assert callname, kwargs
         method = "slave_" + callname
         call = getattr(self, method)
-        self.log("calling method: %s(**%s)" % (method, kwargs))
+        self.log("calling method", method, kwargs)
         call(**kwargs)
         if self.sched.tests_finished():
             self.triggershutdown()
@@ -225,26 +545,40 @@ class DSession:
     #
 
     def slave_slaveready(self, node, slaveinfo):
+        """Emitted when a node first starts up.
+
+        This adds the node to the scheduler, nodes continue with
+        collection without any further input.
+        """
         node.slaveinfo = slaveinfo
         node.slaveinfo['id'] = node.gateway.id
         node.slaveinfo['spec'] = node.gateway.spec
         self.config.hook.pytest_testnodeready(node=node)
-        self.sched.addnode(node)
         if self.shuttingdown:
             node.shutdown()
+        else:
+            self.sched.addnode(node)
 
     def slave_slavefinished(self, node):
+        """Emitted when node executes its pytest_sessionfinish hook.
+
+        Removes the node from the scheduler.
+
+        The node might not be the scheduler if it had not emitted
+        slaveready before shutdown was triggered.
+        """
         self.config.hook.pytest_testnodedown(node=node, error=None)
-        if node.slaveoutput['exitstatus'] == 2: # keyboard-interrupt
+        if node.slaveoutput['exitstatus'] == 2:  # keyboard-interrupt
             self.shouldstop = "%s received keyboard-interrupt" % (node,)
             self.slave_errordown(node, "keyboard-interrupt")
             return
-        crashitem = self.sched.remove_node(node)
+        if node in self.sched.nodes:
-        #assert not crashitem, (crashitem, node)
+            crashitem = self.sched.remove_node(node)
-        if self.shuttingdown and not self.sched.hasnodes():
+            assert not crashitem, (crashitem, node)
-            self.session_finished = True
+        self._active_nodes.remove(node)
 
     def slave_errordown(self, node, error):
+        """Emitted by the SlaveController when a node dies."""
         self.config.hook.pytest_testnodedown(node=node, error=error)
         try:
             crashitem = self.sched.remove_node(node)
@@ -253,41 +587,92 @@ class DSession:
         else:
             if crashitem:
                 self.handle_crashitem(crashitem, node)
-                #self.report_line("item crashed on node: %s" % crashitem)
+
-        if not self.sched.hasnodes():
+        self._failed_nodes_count += 1
-            self.session_finished = True
+        maximum_reached = (self._max_slave_restart is not None and
+                           self._failed_nodes_count > self._max_slave_restart)
+        if maximum_reached:
+            if self._max_slave_restart == 0:
+                msg = 'Slave restarting disabled'
+            else:
+                msg = "Maximum crashed slaves reached: %d" % \
+                      self._max_slave_restart
+            self.report_line(msg)
+        else:
+            self.report_line("Replacing crashed slave %s" % node.gateway.id)
+            self._clone_node(node)
+        self._active_nodes.remove(node)
 
     def slave_collectionfinish(self, node, ids):
+        """Slave has finished test collection.
+
+        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
+        scheduler to schedule the collected items.  When initiating
+        scheduling the first time it logs which scheduler is in use.
+        """
+        if self.shuttingdown:
+            return
+        # 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)
         if self.terminal:
-            self.trdist.setstatus(node.gateway.spec, "[%d]" %(len(ids)))
+            self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
-
         if self.sched.collection_is_completed:
-            if self.terminal:
+            if self.terminal and not self.sched.haspending():
                 self.trdist.ensure_show_status()
                 self.terminal.write_line("")
-                self.terminal.write_line("scheduling tests via %s" %(
+                self.terminal.write_line("scheduling tests via %s" % (
                     self.sched.__class__.__name__))
-
             self.sched.init_distribute()
 
     def slave_logstart(self, node, nodeid, location):
+        """Emitted when a node calls the pytest_runtest_logstart hook."""
         self.config.hook.pytest_runtest_logstart(
             nodeid=nodeid, location=location)
 
     def slave_testreport(self, node, rep):
-        if not (rep.passed and rep.when != "call"):
+        """Emitted when a node calls the pytest_runtest_logreport hook.
-            if rep.when in ("setup", "call"):
+
-                self.sched.remove_item(node, rep.nodeid)
+        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):
+            self.sched.remove_item(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)
         self._handlefailures(rep)
 
     def slave_collectreport(self, node, rep):
-        #self.report_line("collectreport %s: %s" %(rep.id, rep.status))
+        """Emitted when a node calls the pytest_collectreport hook."""
-        #rep.node = node
+        if rep.failed:
-        self._handlefailures(rep)
+            self._failed_slave_collectreport(node, rep)
+
+    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
+        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
+        "slave_*" hooks and do work soon.
+        """
+        spec = node.gateway.spec
+        spec.id = None
+        self.nodemanager.group.allocate_id(spec)
+        node = self.nodemanager.setup_node(spec, self.queue.put)
+        self._active_nodes.add(node)
+        return node
+
+    def _failed_slave_collectreport(self, node, rep):
+        # Check we haven't already seen this report (from
+        # another slave).
+        if rep.longrepr not in self._failed_collection_errors:
+            self._failed_collection_errors[rep.longrepr] = True
+            self.config.hook.pytest_collectreport(report=rep)
+            self._handlefailures(rep)
 
     def _handlefailures(self, rep):
         if rep.failed:
@@ -299,19 +684,21 @@ class DSession:
     def triggershutdown(self):
         self.log("triggering shutdown")
         self.shuttingdown = True
-        for node in self.sched.node2pending:
+        for node in self.sched.nodes:
             node.shutdown()
 
     def handle_crashitem(self, nodeid, slave):
         # XXX get more reporting info by recording pytest_runtest_logstart?
+        # XXX count no of failures and retry N times
         runner = self.config.pluginmanager.getplugin("runner")
         fspath = nodeid.split("::")[0]
-        msg = "Slave %r crashed while running %r" %(slave.gateway.id, nodeid)
+        msg = "Slave %r crashed while running %r" % (slave.gateway.id, nodeid)
-        rep = runner.TestReport(nodeid, (fspath, None, fspath), (),
+        rep = runner.TestReport(nodeid, (fspath, None, fspath),
-            "failed", msg, "???")
+                                (), "failed", msg, "???")
         rep.node = slave
         self.config.hook.pytest_runtest_logreport(report=rep)
 
+
 class TerminalDistReporter:
     def __init__(self, config):
         self.config = config

xdist/looponfail.py

@@ -11,6 +11,21 @@ import py, pytest
 import sys
 import execnet
 
+def pytest_addoption(parser):
+    group = parser.getgroup("xdist", "distributed and subprocess testing")
+    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"):
+        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")
@@ -87,9 +102,13 @@ class RemoteControl(object):
         result = self.runsession()
         failures, reports, collection_failed = result
         if collection_failed:
-            reports = ["Collection failed, keeping previous failure set"]
+            pass # "Collection failed, keeping previous failure set"
         else:
-            self.failures = failures
+            uniq_failures = []
+            for failure in failures:
+                if failure not in uniq_failures:
+                    uniq_failures.append(failure)
+            self.failures = uniq_failures
 
 def repr_pytest_looponfailinfo(failreports, rootdirs):
     tr = py.io.TerminalWriter()
@@ -105,7 +124,6 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
 
 def init_slave_session(channel, args, option_dict):
     import os, sys
-    import py
     outchannel = channel.gateway.newchannel()
     sys.stdout = sys.stderr = outchannel.makefile('w')
     channel.send(outchannel)
@@ -184,7 +202,7 @@ class StatRecorder:
         self.check() # snapshot state
 
     def fil(self, p):
-        return p.ext in ('.py', '.txt', '.c', '.h')
+        return p.check(file=1, dotfile=0) and p.ext != ".pyc"
     def rec(self, p):
         return p.check(dotfile=0)
 
@@ -224,4 +242,3 @@ class StatRecorder:
             changed = True
         self.statcache = newstat
         return changed
-

xdist/plugin.py

@@ -1,18 +1,26 @@
-import sys
+import py
-import py, pytest
+import pytest
+
+
+def parse_numprocesses(s):
+    if s == 'auto':
+        import multiprocessing
+        return multiprocessing.cpu_count()
+    else:
+        return int(s)
+
 
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
-    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.")
     group._addoption('-n', dest="numprocesses", metavar="numprocesses",
-           action="store", type="int",
+           action="store",
-           help="shortcut for '--dist=load --tx=NUM*popen'")
+           type=parse_numprocesses,
-    group.addoption('--boxed',
+           help="shortcut for '--dist=load --tx=NUM*popen', "
-           action="store_true", dest="boxed", default=False,
+                "you can use 'auto' here for auto detection CPUs number on "
-           help="box each test run in a separate process (unix)")
+                "host system")
+    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",
            action="store", choices=['load', 'each', 'no'],
            type="choice", dest="dist", default="no",
@@ -28,12 +36,14 @@ def pytest_addoption(parser):
     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="dir1",
+    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",
+           help="add expression for ignores when rsyncing to remote tx nodes.")
 
     parser.addini('rsyncdirs', 'list of (relative) paths to be rsynced for'
          ' remote distributed testing.', type="pathlist")
-    parser.addini('rsyncignore', 'list of (relative) 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",
         help="directories to check for changes", default=[py.path.local()])
@@ -43,30 +53,31 @@ def pytest_addoption(parser):
 # -------------------------------------------------------------------------
 def pytest_addhooks(pluginmanager):
     from xdist import newhooks
-    pluginmanager.addhooks(newhooks)
+    # avoid warnings with pytest-2.8
+    method = getattr(pluginmanager, "add_hookspecs", None)
+    if method is None:
+        method = pluginmanager.addhooks
+    method(newhooks)
 
 # -------------------------------------------------------------------------
 # distributed testing initialization
 # -------------------------------------------------------------------------
 
-def pytest_cmdline_main(config):
-    check_options(config)
-    if config.getvalue("looponfail"):
-        from xdist.looponfail import looponfail_main
-        looponfail_main(config)
-        return 2 # looponfail only can get stop with ctrl-C anyway
 
-def pytest_configure(config, __multicall__):
+@pytest.mark.trylast
-    __multicall__.execute()
+def pytest_configure(config):
-    if config.getvalue("dist") != "no":
+    if config.getoption("dist") != "no":
         from xdist.dsession import DSession
         session = DSession(config)
         config.pluginmanager.register(session, "dsession")
+        tr = config.pluginmanager.getplugin("terminalreporter")
+        tr.showfspath = False
 
-def check_options(config):
+@pytest.mark.tryfirst
+def pytest_cmdline_main(config):
     if config.option.numprocesses:
         config.option.dist = "load"
-        config.option.tx = ['popen'] * int(config.option.numprocesses)
+        config.option.tx = ['popen'] * config.option.numprocesses
     if config.option.distload:
         config.option.dist = "load"
     val = config.getvalue
@@ -78,46 +89,3 @@ def check_options(config):
         elif val("dist") != "no":
             if usepdb:
                 raise pytest.UsageError("--pdb incompatible with distributing tests.")
-
-
-def pytest_runtest_protocol(item):
-    if item.config.getvalue("boxed"):
-        reports = forked_run_report(item)
-        for rep in reports:
-            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
-    from _pytest.runner import runtestprotocol
-    EXITSTATUS_TESTEXIT = 4
-    import marshal
-    from xdist.remote import serialize_report
-    from xdist.slavemanage import unserialize_report
-    def runforked():
-        try:
-            reports = runtestprotocol(item, log=False)
-        except KeyboardInterrupt:
-            py.std.os._exit(EXITSTATUS_TESTEXIT)
-        return marshal.dumps([serialize_report(x) for x in reports])
-
-    ff = py.process.ForkedFunc(runforked)
-    result = ff.waitfinish()
-    if result.retval is not None:
-        report_dumps = marshal.loads(result.retval)
-        return [unserialize_report("testreport", x) for x in report_dumps]
-    else:
-        if result.exitstatus == EXITSTATUS_TESTEXIT:
-            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" %(
-            path, lineno, result.signal)
-    from _pytest import runner
-    call = runner.CallInfo(lambda: 0/0, "???")
-    call.excinfo = info
-    rep = runner.pytest_runtest_makereport(item, call)
-    return rep

xdist/remote.py

@@ -24,7 +24,7 @@ class SlaveInteractor:
 
     def pytest_internalerror(self, excrepr):
         for line in str(excrepr).split("\n"):
-            self.log("IERROR> " + line)
+            self.log("IERROR>", line)
 
     def pytest_sessionstart(self, session):
         self.session = session
@@ -45,41 +45,44 @@ class SlaveInteractor:
         torun = []
         while 1:
             name, kwargs = self.channel.receive()
-            self.log("received command %s(**%s)" % (name, kwargs))
+            self.log("received command", name, kwargs)
             if name == "runtests":
-                ids = kwargs['ids']
+                torun.extend(kwargs['indices'])
-                for nodeid in ids:
-                    torun.append(self._id2item[nodeid])
             elif name == "runtests_all":
-                torun.extend(session.items)
+                torun.extend(range(len(session.items)))
-            self.log("items to run: %s" %(len(torun)))
+            self.log("items to run:", torun)
+            # only run if we have an item and a next item
             while len(torun) >= 2:
-                item = torun.pop(0)
+                self.run_tests(torun)
-                nextitem = torun[0]
-                self.config.hook.pytest_runtest_protocol(item=item,
-                    nextitem=nextitem)
             if name == "shutdown":
-                while torun:
+                if torun:
-                    self.config.hook.pytest_runtest_protocol(
+                    self.run_tests(torun)
-                        item=torun.pop(0), nextitem=None)
                 break
         return True
 
+    def run_tests(self, torun):
+        items = self.session.items
+        self.item_index = torun.pop(0)
+        if torun:
+            nextitem = items[torun[0]]
+        else:
+            nextitem = None
+        self.config.hook.pytest_runtest_protocol(
+            item=items[self.item_index],
+            nextitem=nextitem)
+
     def pytest_collection_finish(self, session):
-        self._id2item = {}
-        ids = []
-        for item in session.items:
-            self._id2item[item.nodeid] = item
-            ids.append(item.nodeid)
         self.sendevent("collectionfinish",
             topdir=str(session.fspath),
-            ids=ids)
+            ids=[item.nodeid for item in session.items])
 
-    #def pytest_runtest_logstart(self, nodeid, location, fspath):
+    def pytest_runtest_logstart(self, nodeid, location):
-    #    self.sendevent("logstart", nodeid=nodeid, location=location)
+        self.sendevent("logstart", nodeid=nodeid, location=location)
 
     def pytest_runtest_logreport(self, report):
         data = serialize_report(report)
+        data["item_index"] = self.item_index
+        assert self.session.items[self.item_index].nodeid == report.nodeid
         self.sendevent("testreport", data=data)
 
     def pytest_collectreport(self, report):
@@ -125,6 +128,11 @@ def remote_initconfig(option_dict, args):
 
 
 if __name__ == '__channelexec__':
+    channel = channel  # noqa
+    # python3.2 is not concurrent import safe, so let's play it safe
+    # https://bitbucket.org/hpk42/pytest/issue/347/pytest-xdist-and-python-32
+    if sys.version_info[:2] == (3,2):
+        os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
     slaveinput,args,option_dict = channel.receive()
     importpath = os.getcwd()
     sys.path.insert(0, importpath) # XXX only for remote situations

xdist/slavemanage.py

@@ -1,5 +1,8 @@
-import py, pytest
+import fnmatch
-import sys, os
+import os
+
+import py
+import pytest
 import execnet
 import xdist.remote
 
@@ -7,6 +10,7 @@ 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()
@@ -23,38 +27,33 @@ class NodeManager(object):
             self.group.allocate_id(spec)
             self.specs.append(spec)
         self.roots = self._getrsyncdirs()
+        self.rsyncoptions = self._getrsyncoptions()
+        self._rsynced_specs = py.builtin.set()
 
-    def rsync_roots(self):
+    def rsync_roots(self, gateway):
-        """ make sure that all remote gateways
+        """Rsync the set of roots to the node's gateway cwd."""
-            have the same set of roots in their
-            current directory.
-        """
-        options = {
-            'ignores': self.config.getini("rsyncignore"),
-            'verbose': self.config.option.verbose,
-        }
         if self.roots:
-            # send each rsync root
             for root in self.roots:
-                self.rsync(root, **options)
+                self.rsync(gateway, root, **self.rsyncoptions)
-
-    def makegateways(self):
-        assert not list(self.group)
-        self.config.hook.pytest_xdist_setupnodes(config=self.config,
-            specs=self.specs)
-        for spec in self.specs:
-            gw = self.group.makegateway(spec)
-            self.config.hook.pytest_xdist_newgateway(gateway=gw)
 
     def setup_nodes(self, putevent):
-        self.makegateways()
+        self.config.hook.pytest_xdist_setupnodes(config=self.config,
-        self.rsync_roots()
+                                                 specs=self.specs)
         self.trace("setting up nodes")
-        for gateway in self.group:
+        nodes = []
-            node = SlaveController(self, gateway, self.config, putevent)
+        for spec in self.specs:
-            gateway.node = node  # to keep node alive
+            nodes.append(self.setup_node(spec, putevent))
-            node.setup()
+        return nodes
-            self.trace("started node %r" % node)
+
+    def setup_node(self, spec, putevent):
+        gw = self.group.makegateway(spec)
+        self.config.hook.pytest_xdist_newgateway(gateway=gw)
+        self.rsync_roots(gw)
+        node = SlaveController(self, gw, self.config, putevent)
+        gw.node = node          # keep the node alive
+        node.setup()
+        self.trace("started node %r" % node)
+        return node
 
     def teardown_nodes(self):
         self.group.terminate(self.EXIT_TIMEOUT)
@@ -98,60 +97,64 @@ class NodeManager(object):
                 roots.append(root)
         return roots
 
-    def rsync(self, source, notify=None, verbose=False, ignores=None):
+    def _getrsyncoptions(self):
-        """ perform rsync to all remote hosts.
+        """Get options to be passed for rsync."""
-        """
+        ignores = list(self.DEFAULT_IGNORES)
+        ignores += self.config.option.rsyncignore
+        ignores += self.config.getini("rsyncignore")
+
+        return {
+            'ignores': ignores,
+            'verbose': self.config.option.verbose,
+        }
+
+    def rsync(self, gateway, source, notify=None, verbose=False, ignores=None):
+        """Perform rsync to remote hosts for node."""
+        # XXX This changes the calling behaviour of
+        #     pytest_xdist_rsyncstart and pytest_xdist_rsyncfinish to
+        #     be called once per rsync target.
         rsync = HostRSync(source, verbose=verbose, ignores=ignores)
-        seen = py.builtin.set()
+        spec = gateway.spec
-        gateways = []
+        if spec.popen and not spec.chdir:
-        for gateway in self.group:
+            # XXX This assumes that sources are python-packages
-            spec = gateway.spec
+            #     and that adding the basedir does not hurt.
-            if spec.popen and not spec.chdir:
+            gateway.remote_exec("""
-                # XXX this assumes that sources are python-packages
+                import sys ; sys.path.insert(0, %r)
-                # and that adding the basedir does not hurt
+            """ % os.path.dirname(str(source))).waitclose()
-                gateway.remote_exec("""
+            return
-                    import sys ; sys.path.insert(0, %r)
+        if (spec, source) in self._rsynced_specs:
-                """ % os.path.dirname(str(source))).waitclose()
+            return
-                continue
+        def finished():
-            if spec not in seen:
+            if notify:
-                def finished():
+                notify("rsyncrootready", spec, source)
-                    if notify:
+        rsync.add_target_host(gateway, finished=finished)
-                        notify("rsyncrootready", spec, source)
+        self._rsynced_specs.add((spec, source))
-                rsync.add_target_host(gateway, finished=finished)
+        self.config.hook.pytest_xdist_rsyncstart(
-                seen.add(spec)
+            source=source,
-                gateways.append(gateway)
+            gateways=[gateway],
-        if seen:
+        )
-            self.config.hook.pytest_xdist_rsyncstart(
+        rsync.send()
-                source=source,
+        self.config.hook.pytest_xdist_rsyncfinish(
-                gateways=gateways,
+            source=source,
-            )
+            gateways=[gateway],
-            rsync.send()
+        )
-            self.config.hook.pytest_xdist_rsyncfinish(
-                source=source,
-                gateways=gateways,
-            )
 
 class HostRSync(execnet.RSync):
     """ RSyncer that filters out common files
     """
     def __init__(self, sourcedir, *args, **kwargs):
         self._synced = {}
-        ignores= None
+        self._ignores = kwargs.pop('ignores', None) or []
-        if 'ignores' in kwargs:
-            ignores = kwargs.pop('ignores')
-        self._ignores = ignores or []
         super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
 
     def filter(self, path):
         path = py.path.local(path)
-        if not path.ext in ('.pyc', '.pyo'):
+        for x in self._ignores:
-            if not path.basename.endswith('~'):
+            x = getattr(x, 'strpath', x)
-                if path.check(dotfile=0):
+            if fnmatch.fnmatch(path.basename, x) or fnmatch.fnmatch(path.strpath, x):
-                    for x in self._ignores:
+                return False
-                        if path == x:
+        else:
-                            break
+            return True
-                    else:
-                        return True
 
     def add_target_host(self, gateway, finished=None):
         remotepath = os.path.basename(self._sourcedir)
@@ -230,8 +233,8 @@ class SlaveController(object):
             self.gateway.exit()
             #del self.gateway
 
-    def send_runtest(self, nodeid):
+    def send_runtest_some(self, indices):
-        self.sendcommand("runtests", ids=[nodeid])
+        self.sendcommand("runtests", indices=indices)
 
     def send_runtest_all(self):
         self.sendcommand("runtests_all",)
@@ -278,10 +281,13 @@ class SlaveController(object):
                 self._down = True
                 self.slaveoutput = kwargs['slaveoutput']
                 self.notify_inproc("slavefinished", node=self)
-            #elif eventname == "logstart":
+            elif eventname == "logstart":
-            #    self.notify_inproc(eventname, node=self, **kwargs)
+                self.notify_inproc(eventname, node=self, **kwargs)
             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:
+                    rep.item_index = item_index
                 self.notify_inproc(eventname, node=self, rep=rep)
             elif eventname == "collectionfinish":
                 self.notify_inproc(eventname, node=self, ids=kwargs['ids'])
@@ -296,8 +302,7 @@ class SlaveController(object):
             self.config.pluginmanager.notify_exception(excinfo)
 
 def unserialize_report(name, reportdict):
-    d = reportdict
     if name == "testreport":
-        return runner.TestReport(**d)
+        return runner.TestReport(**reportdict)
     elif name == "collectreport":
-        return runner.CollectReport(**d)
+        return runner.CollectReport(**reportdict)