refactor cpu number autodetection to avoid the regression

Add --max-slave-restart option

.hgignore

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

.hgtags

@@ -16,3 +16,5 @@ cd44a941c833c098e4899fe3d42a96703754d0d5 1.5
 5c5cb6d59e12e566fbb0217aea718dc31578bee1 1.9
 4406fc2a6427fadc021ed7e43e7aa5032b1ea91f 1.10
 220f6e46eb71a6212ccbe6b67b9e6edcf8ee4fa5 1.11
+39ef85dbc893cc63dede11601208098a667b58e9 1.12
+4e25f4c568be2d7cb4d1739638a9e66bbf28f588 v1.13

CHANGELOG

@@ -1,3 +1,33 @@
+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
 -------------------------
 
@@ -20,7 +50,7 @@
   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.  
+  of nodes.  Also includes better code documentation.
   Many thanks to Floris Bruynooghe for the complete PR.
 
 
@@ -33,7 +63,7 @@
 - fix pytest issue382 - produce "pytest_runtest_logstart" event again
   in master. Thanks Aron Curzon.
 
-- fix pytest issue419 by sending/receiving indices into the test 
+- 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)
 
@@ -146,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
-

README.txt

@@ -59,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

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.12',
+    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,20 +11,27 @@ 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'],
+    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',
-    'License :: OSI Approved :: MIT License',
-    'Operating System :: POSIX',
-    'Operating System :: Microsoft :: Windows',
-    'Operating System :: MacOS :: MacOS X',
-    'Topic :: Software Development :: Testing',
-    'Topic :: Software Development :: Quality Assurance',
-    'Topic :: Utilities',
-    'Programming Language :: Python',
-    'Programming Language :: Python :: 3',
+        'Development Status :: 5 - Production/Stable',
+        'Intended Audience :: Developers',
+        'License :: OSI Approved :: MIT License',
+        'Operating System :: POSIX',
+        'Operating System :: Microsoft :: Windows',
+        'Operating System :: MacOS :: MacOS X',
+        'Topic :: Software Development :: Testing',
+        'Topic :: Software Development :: Quality Assurance',
+        'Topic :: Utilities',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 3',
     ],
 )

testing/acceptance_test.py

@@ -363,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("""
@@ -397,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):
@@ -516,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*",
         ])
@@ -531,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*",
         ])
@@ -544,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*",
         ])
@@ -557,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/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
-

tox.ini

@@ -1,43 +1,28 @@
 [tox]
-envlist=py26,py33,py34,py27,py27-pexpect,py33-pexpect,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
-commands= py.test {posargs}
+deps =
+    pycmd
+    pytest24: pytest~=2.4.0
+    pytest25: pytest~=2.5.0
 
-[testenv:py27-pexpect]
-deps={[testenv]deps}
-    pexpect
-[testenv:py33-pexpect]
-deps={[testenv]deps}
-    pexpect
+    pytest26: pytest~=2.6.1
+    pytest27: pytest~=2.7.2
+    pexpect: pexpect
+commands=
+  # always clean to avoid code unmarshal mismatch on old python/pytest
+  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 @@
-#
-__version__ = '1.12'
+__all__ = ['__version__']
+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

@@ -455,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:
@@ -488,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
@@ -580,8 +587,20 @@ class DSession:
         else:
             if crashitem:
                 self.handle_crashitem(crashitem, node)
-        self.report_line("Replacing failed node %s" % node.gateway.id)
-        self._clone_node(node)
+
+        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)
 
     def slave_collectionfinish(self, node, ids):
@@ -595,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,26 @@
 import py
 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",
-           help="shortcut for '--dist=load --tx=NUM*popen'")
-    group.addoption('--boxed',
-           action="store_true", dest="boxed", default=False,
-           help="box each test run in a separate process (unix)")
+           action="store",
+           type=parse_numprocesses,
+           help="shortcut for '--dist=load --tx=NUM*popen', "
+                "you can use 'auto' here for auto detection CPUs number on "
+                "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",
@@ -55,12 +63,6 @@ def pytest_addhooks(pluginmanager):
 # 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
 
 @pytest.mark.trylast
 def pytest_configure(config):
@@ -71,10 +73,11 @@ def pytest_configure(config):
         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
@@ -86,50 +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)
-    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
-        if 'ignores' in kwargs:
-            ignores = kwargs.pop('ignores')
-        self._ignores = ignores or []
+        self._ignores = kwargs.pop('ignores', None) or []
         super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
 
     def filter(self, path):