flake8 fixes

Add --max-slave-restart option

.hgignore

@@ -22,6 +22,7 @@ dist/
 include/
 lib/
 bin/
+xdist/_version.py
 pytest_xdist.egg-info
 issue/
 3rdparty/

.hgtags

@@ -15,3 +15,5 @@ cd44a941c833c098e4899fe3d42a96703754d0d5 1.5
 1d27987c267577899350a25ba5828d55d87083ad 1.8
 5c5cb6d59e12e566fbb0217aea718dc31578bee1 1.9
 4406fc2a6427fadc021ed7e43e7aa5032b1ea91f 1.10
+220f6e46eb71a6212ccbe6b67b9e6edcf8ee4fa5 1.11
+39ef85dbc893cc63dede11601208098a667b58e9 1.12

CHANGELOG

@@ -1,3 +1,37 @@
+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
 -------------------------
 
@@ -136,4 +170,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
-

README.txt

@@ -1,5 +1,10 @@
+.. 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:
@@ -54,7 +59,13 @@ 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
@@ -201,12 +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 .
-
-

setup.cfg

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

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup
 
 setup(
     name="pytest-xdist",
-    version='1.11',
+    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='MIT',
@@ -11,9 +11,16 @@ setup(
     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.1', 'pytest>=2.4.2', 'py>=1.4.22'],
+    setup_requires=['setuptools_scm'],
     classifiers=[
         'Development Status :: 5 - Production/Stable',
         'Intended Audience :: Developers',

testing/acceptance_test.py

@@ -113,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*",
@@ -243,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
@@ -256,9 +256,8 @@ class TestDistribution:
                 import time
                 time.sleep(10)
         """)
-        child = testdir.spawn_pytest("-n1")
+        child = testdir.spawn_pytest("-n1 -v")
-        py.std.time.sleep(0.1)
+        child.expect(".*test_sleep.*")
-        child.expect(".*test session starts.*")
         child.kill(2) # keyboard interrupt
         child.expect(".*KeyboardInterrupt.*")
         #child.expect(".*seconds.*")
@@ -271,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*"])
 
@@ -364,6 +363,19 @@ 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("""
@@ -398,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):
@@ -408,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*",
@@ -455,7 +492,7 @@ 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*",
@@ -485,6 +522,28 @@ def test_fixture_scope_caching_issue503(testdir):
     ])
 
 
+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):
@@ -495,7 +554,7 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '-n1')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*1 passed*",
         ])
@@ -510,7 +569,7 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '-n2')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*3 passed*",
         ])
@@ -523,7 +582,7 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '--dist=each', '--tx=popen')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*1 passed*",
         ])
@@ -536,7 +595,39 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*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,10 +1,29 @@
 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",
@@ -13,6 +32,13 @@ def pytest_addoption(parser):
 def pytest_funcarg__specssh(request):
     return getspecssh(request.config)
 
+@pytest.fixture
+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):
     return [execnet.XSpec(spec)

testing/test_dsession.py

@@ -144,24 +144,36 @@ class TestLoadScheduling:
         crashitem = sched.remove_node(node)
         assert crashitem == collection[0]
 
-    def test_schedule_different_tests_collected(self):
+    def test_different_tests_collected(self, testdir):
         """
-        Test that LoadScheduling is logging different tests were
+        Test that LoadScheduling is reporting collection errors when
-        collected by slaves when that happens.
+        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)
+        sched = LoadScheduling(2, config=config)
-        logged_messages = []
-        py.log.setconsumer('loadsched', logged_messages.append)
         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()
-        logged_content = ''.join(x.content() for x in logged_messages)
+        assert len(collect_hook.reports) == 1
-        assert 'Different tests were collected between' in logged_content
+        rep = collect_hook.reports[0]
-        assert 'Different tests collected, aborting run' in logged_content
+        assert 'Different tests were collected between' in rep.longrepr
 
 
 class TestDistReporter:

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"
@@ -67,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_slavemanage.py

@@ -1,29 +1,27 @@
 import py
 import pytest
 import execnet
-from xdist import slavemanage
+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
 
-def pytest_funcarg__mysetup(request):
+def pytest_funcarg__mysetup(tmpdir):
     class mysetup:
-        def __init__(self, request):
+        source = tmpdir.mkdir("source")
-            temp = request.getfuncargvalue("tmpdir")
+        dest = tmpdir.mkdir("dest")
-            self.source = temp.mkdir("source")
+    return mysetup()
-            self.dest = temp.mkdir("dest")
-            request.getfuncargvalue("_pytest")
-    return mysetup(request)
 
 @pytest.fixture
 def slavecontroller(monkeypatch):
@@ -45,8 +43,7 @@ class TestNodeManagerPopen:
         for spec in NodeManager(config, l, defaultchdir="abc").specs:
             assert spec.chdir == "abc"
 
-    def test_popen_makegateway_events(self, config,
+    def test_popen_makegateway_events(self, config, hookrecorder, slavecontroller):
-                                      hookrecorder, _pytest, slavecontroller):
         hm = NodeManager(config, ["popen"] * 2)
         hm.setup_nodes(None)
         call = hookrecorder.popcall("pytest_xdist_setupnodes")
@@ -114,14 +111,6 @@ class TestNodeManagerPopen:
         call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
 
 class TestHRSync:
-    def pytest_funcarg__mysetup(self, request):
-        class mysetup:
-            def __init__(self, request):
-                tmp = request.getfuncargvalue('tmpdir')
-                self.source = tmp.mkdir("source")
-                self.dest = tmp.mkdir("dest")
-        return mysetup(request)
-
     def test_hrsync_filter(self, mysetup):
         source, _ = mysetup.source, mysetup.dest  # noqa
         source.ensure("dir", "file.txt")
@@ -151,7 +140,7 @@ 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(mysetup.source)

tox.ini

@@ -1,43 +1,28 @@
 [tox]
-envlist=py26,py33,py34,py27,py27-pexpect,py33-pexpect,py26,py26-old,py33-old,flakes
+envlist=
+  py{26,33,34,27}-pytest2{4,5,6,7},py{27,34}-pytest27-pexpect,flakes
+
 
 [testenv]
 changedir=testing
-deps=pytest>=2.5.1
+deps =
-commands= py.test {posargs}
+    pycmd
+    pytest24: pytest~=2.4.0
+    pytest25: pytest~=2.5.0
 
-[testenv:py27-pexpect]
+    pytest26: pytest~=2.6.1
-deps={[testenv]deps}
+    pytest27: pytest~=2.7.2
-    pexpect
+    pexpect: pexpect
-[testenv:py33-pexpect]
+commands=
-deps={[testenv]deps}
+  # always clean to avoid code unmarshal mismatch on old python/pytest
-    pexpect
+  py.cleanup -aq
+  py.test {posargs}
 
 [testenv:flakes]
 changedir=
 deps = pytest-flakes>=0.2
 commands = py.test --flakes -m flakes testing xdist
 
-[testenv:py26-old]
-basepython = python2.6
-deps=
-    pytest==2.5.2
-    pycmd
-
-commands= 
-    py.cleanup -a
-    py.test {posargs}
-
-[testenv:py33-old]
-basepython = python3.3
-deps=
-    pytest==2.5.2
-    pycmd
-
-commands= 
-    py.cleanup -a
-    py.test {posargs}
-
 [pytest]
 addopts = -rsfxX
 ;; hello

xdist/__init__.py

@@ -1,2 +1,2 @@
-#
+__all__ = ['__version__']
-__version__ = '1.11'
+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,4 +1,5 @@
 import difflib
+from _pytest.runner import CollectReport
 
 import pytest
 import py
@@ -88,8 +89,9 @@ class EachScheduling:
         elif self._removed2pending:
             for deadnode in self._removed2pending:
                 if deadnode.gateway.spec == node.gateway.spec:
-                    if collection != self.node2collection[deadnode]:
+                    dead_collection = self.node2collection[deadnode]
-                        msg = report_collection_diff(self.collection,
+                    if collection != dead_collection:
+                        msg = report_collection_diff(dead_collection,
                                                      collection,
                                                      deadnode.gateway.id,
                                                      node.gateway.id)
@@ -175,9 +177,10 @@ class LoadScheduling:
 
     :log: A py.log.Producer instance.
 
+    :config: Config object, used for handling hooks.
     """
 
-    def __init__(self, numnodes, log=None):
+    def __init__(self, numnodes, log=None, config=None):
         self.numnodes = numnodes
         self.node2collection = {}
         self.node2pending = {}
@@ -187,6 +190,7 @@ class LoadScheduling:
             self.log = py.log.Producer("loadsched")
         else:
             self.log = log.loadsched
+        self.config = config
 
     @property
     def nodes(self):
@@ -376,8 +380,9 @@ class LoadScheduling:
     def _check_nodes_have_same_collection(self):
         """Return True if all nodes have collected the same items.
 
-        If collections differ this returns False and logs the
+        If collections differ, this method returns False while logging
-        collection differences as they are found.
+        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]
@@ -390,8 +395,12 @@ class LoadScheduling:
                 node.gateway.id,
             )
             if msg:
-                self.log(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
 
@@ -446,8 +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:
@@ -479,12 +493,14 @@ class DSession:
         self.nodemanager = NodeManager(self.config)
         nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
         self._active_nodes.update(nodes)
+        self._session = session
 
     def pytest_sessionfinish(self, session):
         """Shutdown all nodes."""
         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
@@ -494,7 +510,8 @@ 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:
@@ -570,7 +587,19 @@ class DSession:
         else:
             if crashitem:
                 self.handle_crashitem(crashitem, node)
-        self.report_line("Replacing failed node %s" % node.gateway.id)
+
+        self._failed_nodes_count += 1
+        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)
 
@@ -585,6 +614,9 @@ class DSession:
         """
         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)))

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")
@@ -227,4 +242,3 @@ class StatRecorder:
             changed = True
         self.statcache = newstat
         return changed
-

xdist/plugin.py

@@ -1,18 +1,18 @@
+import multiprocessing
+
 import py
 import pytest
 
 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'")
+           help="shortcut for '--dist=load --tx=NUM*popen', "
-    group.addoption('--boxed',
+                "you can use 'auto' here for auto detection CPUs number on "
-           action="store_true", dest="boxed", default=False,
+                "host system")
-           help="box each test run in a separate process (unix)")
+    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",
@@ -45,21 +45,19 @@ 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.getoption("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.getoption("dist") != "no":
         from xdist.dsession import DSession
         session = DSession(config)
@@ -67,10 +65,18 @@ def pytest_configure(config, __multicall__):
         tr = config.pluginmanager.getplugin("terminalreporter")
         tr.showfspath = False
 
-def check_options(config):
+@pytest.mark.tryfirst
+def pytest_cmdline_main(config):
     if config.option.numprocesses:
+        if config.option.numprocesses == 'auto':
+            config.option.numprocesses = multiprocessing.cpu_count()
+        else:
+            try:
+                config.option.numprocesses = int(config.option.numprocesses)
+            except ValueError:
+                config.option.numprocesses = 1
         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
@@ -82,50 +88,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)
-    if result.out:
-        rep.sections.append(("captured stdout", result.out))
-    if result.err:
-        rep.sections.append(("captured stderr", result.err))
-    return rep

xdist/slavemanage.py

@@ -144,10 +144,7 @@ class HostRSync(execnet.RSync):
     """
     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):