Merge remote-tracking branch 'upstream/master' into release-1.25.0

Fix test_remote_usage_prog which started to break with pytest@features

.github/PULL_REQUEST_TEMPLATE.md

@@ -13,5 +13,3 @@ Here's a quick checklist that should be present in PRs:
     ```
     Fix issue with non-ascii contents in doctest text files.
     ```
-
-

.pre-commit-config.yaml

@@ -0,0 +1,24 @@
+repos:
+-   repo: https://github.com/ambv/black
+    rev: 18.6b4
+    hooks:
+    -   id: black
+        args: [--safe, --quiet]
+        language_version: python3.6
+-   repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v1.3.0
+    hooks:
+    -   id: trailing-whitespace
+    -   id: end-of-file-fixer
+    -   id: check-yaml
+    -   id: debug-statements
+    -   id: flake8
+-   repo: local
+    hooks:
+    -   id: rst
+        name: rst
+        entry: rst-lint --encoding utf-8
+        files: ^(CHANGELOG.rst|HOWTORELEASE.rst|README.rst|changelog/.*)$
+        language: python
+        additional_dependencies: [pygments, restructuredtext_lint]
+        python_version: python3.6

.travis.yml

@@ -11,29 +11,36 @@ notifications:
   email:
   - pytest-commit@python.org
 
-python:
+install:
-- '2.7'
+  - pip install -U pip
-- '3.4'
+  - pip install tox setuptools_scm
-- '3.5'
-- '3.6'
-env:
-- TOXENV=py-pytest30
-- TOXENV=py-pytest31
-- TOXENV=py-pytest32
-- TOXENV=py-pytest33
-
-install: pip install tox setuptools_scm
 script: tox
 
 stages:
+- baseline
 - test
 - name: deploy
   if: repo = pytest-dev/pytest-xdist AND tag IS present
 
 jobs:
   include:
+    - stage: baseline
+      python: '3.6'
+      env: TOXENV=linting
+    - python: '3.6'
+      env: TOXENV=py36-pytestlatest
+    - python: '2.7'
+      env: TOXENV=py27-pytestlatest
+
     - stage: test
-    # python x env above are already included into this stage
+      python: "3.4"
+      env: TOXENV=py34-pytestlatest
+    - python: "3.5"
+      env: TOXENV=py35-pytestlatest
+    - python: "3.7"
+      env: TOXENV=py37-pytestlatest
+      sudo: required
+      dist: xenial
     - python: "2.7"
       env: TOXENV=py27-pytestmaster
     - python: "2.7"
@@ -42,10 +49,6 @@ jobs:
       env: TOXENV=py36-pytestmaster
     - python: "3.6"
       env: TOXENV=py36-pytestfeatures
-    - python: "3.6"
-      env: TOXENV=flakes
-    - python: "3.6"
-      env: TOXENV=readme
 
     - stage: deploy
       python: '3.6'

CHANGELOG.rst

@@ -1,3 +1,111 @@
+pytest-xdist 1.25.0 (2018-12-12)
+================================
+
+Deprecations and Removals
+-------------------------
+
+- `#372 <https://github.com/pytest-dev/pytest-xdist/issues/372>`_: Pytest versions older than 3.6 are no longer supported.
+
+
+Features
+--------
+
+- `#373 <https://github.com/pytest-dev/pytest-xdist/issues/373>`_: Node setup information is hidden when pytest is run in quiet mode to reduce noise on many-core machines.
+
+- `#388 <https://github.com/pytest-dev/pytest-xdist/issues/388>`_: ``mainargv`` is made available in ``workerinput`` from the host's ``sys.argv``.
+
+  This can be used via ``request.config.workerinput["mainargv"]``.
+
+
+Bug Fixes
+---------
+
+- `#332 <https://github.com/pytest-dev/pytest-xdist/issues/332>`_: Fix report of module-level skips (``pytest.skip(reason, allow_module_level=True)``).
+
+- `#378 <https://github.com/pytest-dev/pytest-xdist/issues/378>`_: Fix support for gevent monkeypatching
+
+- `#384 <https://github.com/pytest-dev/pytest-xdist/issues/384>`_: pytest 4.1 support: ``ExceptionInfo`` API changes.
+
+- `#390 <https://github.com/pytest-dev/pytest-xdist/issues/390>`_: pytest 4.1 support: ``pytest_logwarning`` hook removed.
+
+
+pytest-xdist 1.24.1 (2018-11-09)
+================================
+
+Bug Fixes
+---------
+
+- `#349 <https://github.com/pytest-dev/pytest-xdist/issues/349>`_: Correctly handle warnings created with arguments that can't be serialized during the transfer from workers to master node.
+
+
+pytest-xdist 1.24.0 (2018-10-18)
+================================
+
+Features
+--------
+
+- `#337 <https://github.com/pytest-dev/pytest-xdist/issues/337>`_: New ``--maxprocesses`` command-line option that limits the maximum number of workers when using ``--numprocesses=auto``.
+
+
+Bug Fixes
+---------
+
+- `#351 <https://github.com/pytest-dev/pytest-xdist/issues/351>`_: Fix scheduling deadlock in case of inter-test locking.
+
+
+pytest-xdist 1.23.2 (2018-09-28)
+================================
+
+Bug Fixes
+---------
+
+- `#344 <https://github.com/pytest-dev/pytest-xdist/issues/344>`_: Fix issue where Warnings could cause pytest to fail if they do not set the args attribute correctly.
+
+
+pytest-xdist 1.23.1 (2018-09-25)
+================================
+
+Bug Fixes
+---------
+
+- `#341 <https://github.com/pytest-dev/pytest-xdist/issues/341>`_: Fix warnings transfer between workers and master node with pytest >= 3.8.
+
+
+pytest-xdist 1.23.0 (2018-08-23)
+================================
+
+Features
+--------
+
+- `#330 <https://github.com/pytest-dev/pytest-xdist/issues/330>`_: Improve collection performance by reducing the number of events sent to ``master`` node.
+
+
+pytest-xdist 1.22.5 (2018-07-27)
+================================
+
+Bug Fixes
+---------
+
+- `#321 <https://github.com/pytest-dev/pytest-xdist/issues/321>`_: Revert change that dropped support for ``pytest<3.4`` and require ``six``.
+
+  This change caused problems in some installations, and was a mistaken
+  in the first place as we should not change version requirements
+  in bug-fix releases unless they fix an actual bug.
+
+
+pytest-xdist 1.22.4 (2018-07-27)
+================================
+
+Bug Fixes
+---------
+
+- `#305 <https://github.com/pytest-dev/pytest-xdist/issues/305>`_: Remove last references to obsolete ``py.code``.
+
+  Remove some unnecessary references to ``py.builtin``.
+
+- `#316 <https://github.com/pytest-dev/pytest-xdist/issues/316>`_: Workaround cpu detection on Travis CI.
+
+
 pytest-xdist 1.22.3 (2018-07-23)
 ================================
 

LICENSE

@@ -16,4 +16,3 @@
   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.
-

OVERVIEW.md

@@ -74,5 +74,3 @@ serialize collected items to send them through the wire, as workers live in anot
 The problem is that test items are not easily (impossible?) to serialize, as they contain references to
 the test functions, fixture managers, config objects, etc. Even if one manages to serialize it,
 it seems it would be very hard to get it right and easy to break by any small change in pytest.
-  
-

README.rst

@@ -4,14 +4,13 @@
     :alt: PyPI version
     :target: https://pypi.python.org/pypi/pytest-xdist
 
+.. image:: https://img.shields.io/conda/vn/conda-forge/pytest-xdist.svg
+    :target: https://anaconda.org/conda-forge/pytest-xdist
+
 .. image:: https://img.shields.io/pypi/pyversions/pytest-xdist.svg
     :alt: Python versions
     :target: https://pypi.python.org/pypi/pytest-xdist
 
-.. image:: https://anaconda.org/conda-forge/pytest-xdist/badges/version.svg
-    :alt: Anaconda version
-    :target: https://anaconda.org/conda-forge/pytest-xdist
-
 .. image:: https://travis-ci.org/pytest-dev/pytest-xdist.svg?branch=master
     :alt: Travis CI build status
     :target: https://travis-ci.org/pytest-dev/pytest-xdist
@@ -20,6 +19,9 @@
     :alt: AppVeyor build status
     :target: https://ci.appveyor.com/project/pytestbot/pytest-xdist
 
+.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
+    :target: https://github.com/ambv/black
+
 xdist: pytest distributed testing plugin
 ========================================
 

appveyor.yml

@@ -1,19 +1,24 @@
 environment:
   matrix:
-  # note: please use "tox --listenvs" to populate the build matrix
+  - TOXENV: "py27-pytestlatest"
-  - TOXENV: "py27-pytest33"
+  - TOXENV: "py34-pytestlatest"
-  - TOXENV: "py34-pytest33"
+  - TOXENV: "py35-pytestlatest"
-  - TOXENV: "py35-pytest33"
+  - TOXENV: "py36-pytestlatest"
-  - TOXENV: "py36-pytest33"
+  - TOXENV: "py37-pytestlatest"
-  - TOXENV: "py27-pytest33-pexpect"
+  - TOXENV: "py27-pytestmaster"
-  - TOXENV: "py36-pytest33-pexpect"
+  - TOXENV: "py36-pytestmaster"
-  - TOXENV: "flakes"
+  - TOXENV: "py27-pytestfeatures"
-  - TOXENV: "readme"
+  - TOXENV: "py36-pytestfeatures"
 
 install:
-  - C:\Python35\python -m pip install -U tox setuptools_scm pip
+  - C:\Python37\python -m pip install -U pip
+  - C:\Python37\python -m pip install -U tox setuptools_scm
 
 build: false  # Not a C# project, build stuff at the test step instead.
 
 test_script:
-  - C:\Python35\python -m tox
+  - C:\Python37\python -m tox
+
+# We don't deploy anything on tags with AppVeyor, we use Travis instead, so we
+# might as well save resources
+skip_tags: true

changelog/_template.rst

@@ -13,8 +13,7 @@
 
 {% if definitions[category]['showcontent'] %}
 {% for text, values in sections[section][category]|dictsort(by='value') %}
-- {{ text }}{% if category != 'vendor' %} (`{{ values[0] }} <https://github.com/pytest-dev/pytest-xdist/issues/{{ values[0][1:] }}>`_){% endif %}
+- `{{ values[0] }} <https://github.com/pytest-dev/pytest-xdist/issues/{{ values[0][1:] }}>`_: {{ text }}
-
 
 {% endfor %}
 {% else %}

example/loadscope/test/test_delta.py

@@ -3,7 +3,6 @@ from unittest import TestCase
 
 
 class Delta1(TestCase):
-
     def test_delta0(self):
         sleep(5)
         assert True
@@ -46,7 +45,6 @@ class Delta1(TestCase):
 
 
 class Delta2(TestCase):
-
     def test_delta0(self):
         sleep(5)
         assert True

setup.py

@@ -1,47 +1,45 @@
 from setuptools import setup, find_packages
 
-install_requires = ['execnet>=1.1', 'pytest>=3.0.0', 'pytest-forked']
+install_requires = ["execnet>=1.1", "pytest>=3.6.0", "pytest-forked", "six"]
 
 
 setup(
     name="pytest-xdist",
-    use_scm_version={'write_to': 'xdist/_version.py'},
+    use_scm_version={"write_to": "xdist/_version.py"},
-    description='pytest xdist plugin for distributed testing'
+    description="pytest xdist plugin for distributed testing"
-                ' and loop-on-failing modes',
+    " and loop-on-failing modes",
-    long_description=open('README.rst').read(),
+    long_description=open("README.rst").read(),
-    license='MIT',
+    license="MIT",
-    author='holger krekel and contributors',
+    author="holger krekel and contributors",
-    author_email='pytest-dev@python.org,holger@merlinux.eu',
+    author_email="pytest-dev@python.org,holger@merlinux.eu",
-    url='https://github.com/pytest-dev/pytest-xdist',
+    url="https://github.com/pytest-dev/pytest-xdist",
-    platforms=['linux', 'osx', 'win32'],
+    platforms=["linux", "osx", "win32"],
-    packages=find_packages(exclude=['testing', 'example']),
+    packages=find_packages(exclude=["testing", "example"]),
     entry_points={
-        'pytest11': [
+        "pytest11": ["xdist = xdist.plugin", "xdist.looponfail = xdist.looponfail"]
-            'xdist = xdist.plugin',
-            'xdist.looponfail = xdist.looponfail',
-        ],
     },
     zip_safe=False,
-    python_requires='>=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*',
+    python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*",
     install_requires=install_requires,
-    setup_requires=['setuptools_scm'],
+    setup_requires=["setuptools_scm"],
     classifiers=[
-        'Development Status :: 5 - Production/Stable',
+        "Development Status :: 5 - Production/Stable",
-        'Framework :: Pytest',
+        "Framework :: Pytest",
-        'Intended Audience :: Developers',
+        "Intended Audience :: Developers",
-        'License :: OSI Approved :: MIT License',
+        "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 :: 2',
+        "Programming Language :: Python :: 2",
-        'Programming Language :: Python :: 2.7',
+        "Programming Language :: Python :: 2.7",
-        'Programming Language :: Python :: 3',
+        "Programming Language :: Python :: 3",
-        'Programming Language :: Python :: 3.4',
+        "Programming Language :: Python :: 3.4",
-        'Programming Language :: Python :: 3.5',
+        "Programming Language :: Python :: 3.5",
-        'Programming Language :: Python :: 3.6',
+        "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
     ],
 )

testing/conftest.py

@@ -20,6 +20,7 @@ pytest_plugins = "pytester"
 @pytest.fixture(autouse=True)
 def _divert_atexit(request, monkeypatch):
     import atexit
+
     finalizers = []
 
     def finish():
@@ -31,10 +32,12 @@ def _divert_atexit(request, monkeypatch):
 
 
 def pytest_addoption(parser):
-    parser.addoption('--gx',
+    parser.addoption(
-                     action="append",
+        "--gx",
-                     dest="gspecs",
+        action="append",
-                     help="add a global test environment, XSpec-syntax. ")
+        dest="gspecs",
+        help="add a global test environment, XSpec-syntax. ",
+    )
 
 
 @pytest.fixture

testing/test_dsession.py

@@ -1,9 +1,6 @@
 from xdist.dsession import DSession
 from xdist.report import report_collection_diff
-from xdist.scheduler import (
+from xdist.scheduler import EachScheduling, LoadScheduling
-    EachScheduling,
-    LoadScheduling,
-)
 
 import py
 import pytest
@@ -60,7 +57,7 @@ class TestEachScheduling:
         sched = EachScheduling(config)
         sched.add_node(node1)
         sched.add_node(node2)
-        collection = ["a.py::test_1", ]
+        collection = ["a.py::test_1"]
         assert not sched.collection_is_completed
         sched.add_node_collection(node1, collection)
         assert not sched.collection_is_completed
@@ -70,8 +67,8 @@ class TestEachScheduling:
         assert sched.node2collection[node2] == collection
         sched.schedule()
         assert sched.tests_finished
-        assert node1.sent == ['ALL']
+        assert node1.sent == ["ALL"]
-        assert node2.sent == ['ALL']
+        assert node2.sent == ["ALL"]
         sched.mark_test_complete(node1, 0)
         assert sched.tests_finished
         sched.mark_test_complete(node2, 0)
@@ -82,7 +79,7 @@ class TestEachScheduling:
         config = testdir.parseconfig("--tx=popen")
         sched = EachScheduling(config)
         sched.add_node(node1)
-        collection = ["a.py::test_1", ]
+        collection = ["a.py::test_1"]
         assert not sched.collection_is_completed
         sched.add_node_collection(node1, collection)
         assert sched.collection_is_completed
@@ -230,7 +227,7 @@ class TestLoadScheduling:
         sched.schedule()
         assert len(collect_hook.reports) == 1
         rep = collect_hook.reports[0]
-        assert 'Different tests were collected between' in rep.longrepr
+        assert "Different tests were collected between" in rep.longrepr
 
 
 class TestDistReporter:
@@ -238,6 +235,7 @@ class TestDistReporter:
     def test_rsync_printing(self, testdir, linecomp):
         config = testdir.parseconfig()
         from _pytest.pytest_terminal import TerminalReporter
+
         rep = TerminalReporter(config, file=linecomp.stringio)
         config.pluginmanager.register(rep, "terminalreporter")
         dsession = DSession(config)
@@ -249,6 +247,7 @@ class TestDistReporter:
         class gw2:
             id = "X2"
             spec = execnet.XSpec("popen")
+
         # class rinfo:
         #    version_info = (2, 5, 1, 'final', 0)
         #    executable = "hello"
@@ -260,47 +259,50 @@ class TestDistReporter:
         #     "*X1*popen*xyz*2.5*"
         # ])
         dsession.pytest_xdist_rsyncstart(source="hello", gateways=[gw1, gw2])
-        linecomp.assert_contains_lines(["[X1,X2] rsyncing: hello", ])
+        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']
+    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']
+    from_collection = ["aaa", "bbb", "ccc", "YYY"]
-    to_collection = ['aZa', 'bbb', 'XXX', 'ccc']
+    to_collection = ["aZa", "bbb", "XXX", "ccc"]
     error_message = (
-        'Different tests were collected between 1 and 2. The difference is:\n'
+        "Different tests were collected between 1 and 2. The difference is:\n"
-        '--- 1\n'
+        "--- 1\n"
-        '\n'
+        "\n"
-        '+++ 2\n'
+        "+++ 2\n"
-        '\n'
+        "\n"
-        '@@ -1,4 +1,4 @@\n'
+        "@@ -1,4 +1,4 @@\n"
-        '\n'
+        "\n"
-        '-aaa\n'
+        "-aaa\n"
-        '+aZa\n'
+        "+aZa\n"
-        ' bbb\n'
+        " bbb\n"
-        '+XXX\n'
+        "+XXX\n"
-        ' ccc\n'
+        " ccc\n"
-        '-YYY')
+        "-YYY"
+    )
 
-    msg = report_collection_diff(from_collection, to_collection, '1', '2')
+    msg = report_collection_diff(from_collection, to_collection, "1", "2")
     assert msg == error_message
 
 
 @pytest.mark.xfail(reason="duplicate test ids not supported yet")
 def test_pytest_issue419(testdir):
-    testdir.makepyfile("""
+    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

@@ -1,4 +1,7 @@
 import py
+import pytest
+from pkg_resources import parse_version
+
 from xdist.looponfail import RemoteControl
 from xdist.looponfail import StatRecorder
 
@@ -60,7 +63,7 @@ class TestStatRecorder:
         p.remove()
         # make check()'s visit() call return our just removed
         # path as if we were in a race condition
-        monkeypatch.setattr(tmp, 'visit', lambda *args: [p])
+        monkeypatch.setattr(tmp, "visit", lambda *args: [p])
 
         changed = sd.check()
         assert changed
@@ -84,7 +87,7 @@ class TestStatRecorder:
         sd = StatRecorder([tmp])
 
         ret_values = [True, False]
-        monkeypatch.setattr(StatRecorder, 'check', lambda self: ret_values.pop())
+        monkeypatch.setattr(StatRecorder, "check", lambda self: ret_values.pop())
         sd.waitonchange(checkinterval=0.2)
         assert not ret_values
 
@@ -110,19 +113,25 @@ class TestRemoteControl:
         assert not failures
 
     def test_failure_change(self, testdir):
-        modcol = testdir.getitem("""
+        modcol = testdir.getitem(
+            """
             def test_func():
                 assert 0
-        """)
+        """
+        )
         control = RemoteControl(modcol.config)
         control.loop_once()
         assert control.failures
-        modcol.fspath.write(py.code.Source("""
+        modcol.fspath.write(
+            py.code.Source(
+                """
             def test_func():
                 assert 1
             def test_new():
                 assert 0
-        """))
+        """
+            )
+        )
         removepyc(modcol.fspath)
         control.loop_once()
         assert not control.failures
@@ -131,14 +140,17 @@ class TestRemoteControl:
         assert str(control.failures).find("test_new") != -1
 
     def test_failure_subdir_no_init(self, testdir):
-        modcol = testdir.getitem("""
+        modcol = testdir.getitem(
+            """
             def test_func():
                 assert 0
-        """)
+        """
+        )
         parent = modcol.fspath.dirpath().dirpath()
         parent.chdir()
-        modcol.config.args = [py.path.local(x).relto(parent)
+        modcol.config.args = [
-                              for x in modcol.config.args]
+            py.path.local(x).relto(parent) for x in modcol.config.args
+        ]
         control = RemoteControl(modcol.config)
         control.loop_once()
         assert control.failures
@@ -148,70 +160,91 @@ class TestRemoteControl:
 
 class TestLooponFailing:
     def test_looponfail_from_fail_to_ok(self, testdir):
-        modcol = testdir.getmodulecol("""
+        modcol = testdir.getmodulecol(
+            """
             def test_one():
                 x = 0
                 assert x == 1
             def test_two():
                 assert 1
-        """)
+        """
+        )
         remotecontrol = RemoteControl(modcol.config)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 1
 
-        modcol.fspath.write(py.code.Source("""
+        modcol.fspath.write(
+            py.code.Source(
+                """
             def test_one():
                 assert 1
             def test_two():
                 assert 1
-        """))
+        """
+            )
+        )
         removepyc(modcol.fspath)
         remotecontrol.loop_once()
         assert not remotecontrol.failures
 
     def test_looponfail_from_one_to_two_tests(self, testdir):
-        modcol = testdir.getmodulecol("""
+        modcol = testdir.getmodulecol(
+            """
             def test_one():
                 assert 0
-        """)
+        """
+        )
         remotecontrol = RemoteControl(modcol.config)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 1
-        assert 'test_one' in remotecontrol.failures[0]
+        assert "test_one" in remotecontrol.failures[0]
 
-        modcol.fspath.write(py.code.Source("""
+        modcol.fspath.write(
+            py.code.Source(
+                """
             def test_one():
                 assert 1 # passes now
             def test_two():
                 assert 0 # new and fails
-        """))
+        """
+            )
+        )
         removepyc(modcol.fspath)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 0
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 1
-        assert 'test_one' not in remotecontrol.failures[0]
+        assert "test_one" not in remotecontrol.failures[0]
-        assert 'test_two' in remotecontrol.failures[0]
+        assert "test_two" in remotecontrol.failures[0]
 
-    @py.test.mark.xfail(py.test.__version__ >= "3.1",
+    @pytest.mark.xfail(
-                        reason="broken by pytest 3.1+")
+        parse_version(pytest.__version__) >= parse_version("3.1"),
+        reason="broken by pytest 3.1+",
+        strict=True,
+    )
     def test_looponfail_removed_test(self, testdir):
-        modcol = testdir.getmodulecol("""
+        modcol = testdir.getmodulecol(
+            """
             def test_one():
                 assert 0
             def test_two():
                 assert 0
-        """)
+        """
+        )
         remotecontrol = RemoteControl(modcol.config)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 2
 
-        modcol.fspath.write(py.code.Source("""
+        modcol.fspath.write(
+            py.code.Source(
+                """
             def test_xxx(): # renamed test
                 assert 0
             def test_two():
                 assert 1 # pass now
-        """))
+        """
+            )
+        )
         removepyc(modcol.fspath)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 0
@@ -220,10 +253,12 @@ class TestLooponFailing:
         assert len(remotecontrol.failures) == 1
 
     def test_looponfail_multiple_errors(self, testdir, monkeypatch):
-        modcol = testdir.getmodulecol("""
+        modcol = testdir.getmodulecol(
+            """
             def test_one():
                 assert 0
-        """)
+        """
+        )
         remotecontrol = RemoteControl(modcol.config)
         orig_runsession = remotecontrol.runsession
 
@@ -233,18 +268,20 @@ class TestLooponFailing:
             print(failures)
             return failures * 2, reports, collection_failed
 
-        monkeypatch.setattr(remotecontrol, 'runsession', runsession_dups)
+        monkeypatch.setattr(remotecontrol, "runsession", runsession_dups)
         remotecontrol.loop_once()
         assert len(remotecontrol.failures) == 1
 
 
 class TestFunctional:
     def test_fail_to_ok(self, testdir):
-        p = testdir.makepyfile("""
+        p = testdir.makepyfile(
+            """
             def test_one():
                 x = 0
                 assert x == 1
-        """)
+        """
+        )
         # p = testdir.mkdir("sub").join(p1.basename)
         # p1.move(p)
         child = testdir.spawn_pytest("-f %s --traceconfig" % p)
@@ -253,21 +290,27 @@ class TestFunctional:
         child.expect("1 failed")
         child.expect("### LOOPONFAILING ####")
         child.expect("waiting for changes")
-        p.write(py.code.Source("""
+        p.write(
+            py.code.Source(
+                """
             def test_one():
                 x = 1
                 assert x == 1
-        """))
+        """
+            )
+        )
         child.expect(".*1 passed.*")
         child.kill(15)
 
     def test_xfail_passes(self, testdir):
-        p = testdir.makepyfile("""
+        p = testdir.makepyfile(
+            """
             import py
             @py.test.mark.xfail
             def test_one():
                 pass
-        """)
+        """
+        )
         child = testdir.spawn_pytest("-f %s" % p)
         child.expect("1 xpass")
         # child.expect("### LOOPONFAILING ####")

testing/test_newhooks.py

@@ -2,21 +2,23 @@ import pytest
 
 
 class TestHooks:
-
     @pytest.fixture(autouse=True)
     def create_test_file(self, testdir):
-        testdir.makepyfile("""
+        testdir.makepyfile(
+            """
             import os
             def test_a(): pass
             def test_b(): pass
             def test_c(): pass
-        """)
+        """
+        )
 
     def test_runtest_logreport(self, testdir):
         """Test that log reports from pytest_runtest_logreport when running
         with xdist contain "node", "nodeid" and "worker_id" attributes. (#8)
         """
-        testdir.makeconftest("""
+        testdir.makeconftest(
+            """
             def pytest_runtest_logreport(report):
                 if hasattr(report, 'node'):
                     if report.when == "call":
@@ -27,29 +29,31 @@ class TestHooks:
                         else:
                             print("HOOK: %s %s"
                                    % (report.nodeid, report.worker_id))
-        """)
+        """
-        res = testdir.runpytest('-n1', '-s')
+        )
-        res.stdout.fnmatch_lines([
+        res = testdir.runpytest("-n1", "-s")
-            '*HOOK: test_runtest_logreport.py::test_a gw0*',
+        res.stdout.fnmatch_lines(
-            '*HOOK: test_runtest_logreport.py::test_b gw0*',
+            [
-            '*HOOK: test_runtest_logreport.py::test_c gw0*',
+                "*HOOK: test_runtest_logreport.py::test_a gw0*",
-            '*3 passed*',
+                "*HOOK: test_runtest_logreport.py::test_b gw0*",
-        ])
+                "*HOOK: test_runtest_logreport.py::test_c gw0*",
+                "*3 passed*",
+            ]
+        )
 
     def test_node_collection_finished(self, testdir):
         """Test pytest_xdist_node_collection_finished hook (#8).
         """
-        testdir.makeconftest("""
+        testdir.makeconftest(
+            """
             def pytest_xdist_node_collection_finished(node, ids):
                 workerid = node.workerinput['workerid']
                 stripped_ids = [x.split('::')[1] for x in ids]
                 print("HOOK: %s %s" % (workerid, ', '.join(stripped_ids)))
-        """)
+        """
-        res = testdir.runpytest('-n2', '-s')
+        )
-        res.stdout.fnmatch_lines_random([
+        res = testdir.runpytest("-n2", "-s")
-            '*HOOK: gw0 test_a, test_b, test_c',
+        res.stdout.fnmatch_lines_random(
-            '*HOOK: gw1 test_a, test_b, test_c',
+            ["*HOOK: gw0 test_a, test_b, test_c", "*HOOK: gw1 test_a, test_b, test_c"]
-        ])
+        )
-        res.stdout.fnmatch_lines([
+        res.stdout.fnmatch_lines(["*3 passed*"])
-            '*3 passed*',
-        ])

testing/test_plugin.py

@@ -16,14 +16,19 @@ def test_dist_incompatibility_messages(testdir):
 
 def test_dist_options(testdir):
     from xdist.plugin import pytest_cmdline_main as check_options
+
     config = testdir.parseconfigure("-n 2")
     check_options(config)
     assert config.option.dist == "load"
-    assert config.option.tx == ['popen'] * 2
+    assert config.option.tx == ["popen"] * 2
     config = testdir.parseconfigure("--numprocesses", "2")
     check_options(config)
     assert config.option.dist == "load"
-    assert config.option.tx == ['popen'] * 2
+    assert config.option.tx == ["popen"] * 2
+    config = testdir.parseconfigure("--numprocesses", "3", "--maxprocesses", "2")
+    check_options(config)
+    assert config.option.dist == "load"
+    assert config.option.tx == ["popen"] * 2
     config = testdir.parseconfigure("-d")
     check_options(config)
     assert config.option.dist == "load"
@@ -31,23 +36,31 @@ def test_dist_options(testdir):
 
 def test_auto_detect_cpus(testdir, monkeypatch):
     import os
-    if hasattr(os, 'sched_getaffinity'):
+
-        monkeypatch.setattr(os, 'sched_getaffinity', lambda _pid: set(range(99)))
+    if hasattr(os, "sched_getaffinity"):
-    elif hasattr(os, 'cpu_count'):
+        monkeypatch.setattr(os, "sched_getaffinity", lambda _pid: set(range(99)))
-        monkeypatch.setattr(os, 'cpu_count', lambda: 99)
+    elif hasattr(os, "cpu_count"):
+        monkeypatch.setattr(os, "cpu_count", lambda: 99)
     else:
         import multiprocessing
-        monkeypatch.setattr(multiprocessing, 'cpu_count', lambda: 99)
+
+        monkeypatch.setattr(multiprocessing, "cpu_count", lambda: 99)
 
     config = testdir.parseconfigure("-n2")
-    assert config.getoption('numprocesses') == 2
+    assert config.getoption("numprocesses") == 2
 
     config = testdir.parseconfigure("-nauto")
-    assert config.getoption('numprocesses') == 99
+    assert config.getoption("numprocesses") == 99
+
+    monkeypatch.delattr(os, "sched_getaffinity", raising=False)
+    monkeypatch.setenv("TRAVIS", "true")
+    config = testdir.parseconfigure("-nauto")
+    assert config.getoption("numprocesses") == 2
 
 
 def test_boxed_with_collect_only(testdir):
     from xdist.plugin import pytest_cmdline_main as check_options
+
     config = testdir.parseconfigure("-n1", "--boxed")
     check_options(config)
     assert config.option.forked
@@ -87,13 +100,13 @@ class TestDistOptions:
         assert xspecs[1].ssh == "xyz"
 
     def test_xspecs_multiplied(self, testdir):
-        config = testdir.parseconfigure("--tx=3*popen", )
+        config = testdir.parseconfigure("--tx=3*popen")
         xspecs = NodeManager(config)._getxspecs()
         assert len(xspecs) == 3
         assert xspecs[1].popen
 
     def test_getrsyncdirs(self, testdir):
-        config = testdir.parseconfigure('--rsyncdir=' + str(testdir.tmpdir))
+        config = testdir.parseconfigure("--rsyncdir=" + str(testdir.tmpdir))
         nm = NodeManager(config, specs=[execnet.XSpec("popen")])
         assert not nm._getrsyncdirs()
         nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
@@ -101,23 +114,24 @@ class TestDistOptions:
         assert testdir.tmpdir in nm.roots
 
     def test_getrsyncignore(self, testdir):
-        config = testdir.parseconfigure('--rsyncignore=fo*')
+        config = testdir.parseconfigure("--rsyncignore=fo*")
         nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
-        assert 'fo*' in nm.rsyncoptions['ignores']
+        assert "fo*" in nm.rsyncoptions["ignores"]
 
     def test_getrsyncdirs_with_conftest(self, testdir):
         p = py.path.local()
-        for bn in 'x y z'.split():
+        for bn in "x y z".split():
             p.mkdir(bn)
-        testdir.makeini("""
+        testdir.makeini(
+            """
             [pytest]
             rsyncdirs= x
-        """)
+        """
-        config = testdir.parseconfigure(
+        )
-            testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
+        config = testdir.parseconfigure(testdir.tmpdir, "--rsyncdir=y", "--rsyncdir=z")
         nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=xyz")])
         roots = nm._getrsyncdirs()
         # assert len(roots) == 3 + 1 # pylib
-        assert py.path.local('y') in roots
+        assert py.path.local("y") in roots
-        assert py.path.local('z') in roots
+        assert py.path.local("z") in roots
-        assert testdir.tmpdir.join('x') in roots
+        assert testdir.tmpdir.join("x") in roots

testing/test_remote.py

@@ -1,12 +1,14 @@
 import py
 import pprint
 import pytest
+import sys
+
 from xdist.workermanage import WorkerController, unserialize_report
 from xdist.remote import serialize_report
 import execnet
 import marshal
 
-queue = py.builtin._tryimport("queue", "Queue")
+from six.moves.queue import Queue
 
 WAIT_TIMEOUT = 10.0
 
@@ -33,9 +35,9 @@ class WorkerSetup:
     def __init__(self, request, testdir):
         self.request = request
         self.testdir = testdir
-        self.events = queue.Queue()
+        self.events = Queue()
 
-    def setup(self, ):
+    def setup(self,):
         self.testdir.chdir()
         # import os ; os.environ['EXECNET_DEBUG'] = "2"
         self.gateway = execnet.makegateway()
@@ -45,8 +47,7 @@ class WorkerSetup:
         class DummyMananger:
             specs = [0, 1]
 
-        self.slp = WorkerController(DummyMananger, self.gateway, config,
+        self.slp = WorkerController(DummyMananger, self.gateway, config, putevent)
-                                    putevent)
         self.request.addfinalizer(self.slp.ensure_teardown)
         self.slp.setup()
 
@@ -59,7 +60,7 @@ class WorkerSetup:
             ev = EventCall(data)
             if name is None or ev.name == name:
                 return ev
-            print("skipping %s" % (ev, ))
+            print("skipping %s" % (ev,))
 
     def sendcommand(self, name, **kwargs):
         self.slp.sendcommand(name, **kwargs)
@@ -70,9 +71,10 @@ def worker(request, testdir):
     return WorkerSetup(request, testdir)
 
 
-@pytest.mark.xfail(reason='#59')
+@pytest.mark.xfail(reason="#59")
 def test_remoteinitconfig(testdir):
     from xdist.remote import remote_initconfig
+
     config1 = testdir.parseconfig()
     config2 = remote_initconfig(config1.option.__dict__, config1.args)
     assert config2.option.__dict__ == config1.option.__dict__
@@ -81,29 +83,33 @@ def test_remoteinitconfig(testdir):
 
 class TestReportSerialization:
     def test_xdist_longrepr_to_str_issue_241(self, testdir):
-        testdir.makepyfile("""
+        testdir.makepyfile(
+            """
             import os
             def test_a(): assert False
             def test_b(): pass
-        """)
+        """
-        testdir.makeconftest("""
+        )
+        testdir.makeconftest(
+            """
             def pytest_runtest_logreport(report):
                 print(report.longrepr)
-        """)
+        """
-        res = testdir.runpytest('-n1', '-s')
+        )
-        res.stdout.fnmatch_lines([
+        res = testdir.runpytest("-n1", "-s")
-            '*1 failed, 1 passed *'
+        res.stdout.fnmatch_lines(["*1 failed, 1 passed *"])
-        ])
 
     def test_xdist_report_longrepr_reprcrash_130(self, testdir):
-        reprec = testdir.inline_runsource("""
+        reprec = testdir.inline_runsource(
+            """
                     import py
                     def test_fail(): assert False, 'Expected Message'
-                """)
+                """
+        )
         reports = reprec.getreports("pytest_runtest_logreport")
         assert len(reports) == 3
         rep = reports[1]
-        added_section = ('Failure Metadata', str("metadata metadata"), "*")
+        added_section = ("Failure Metadata", str("metadata metadata"), "*")
         rep.longrepr.sections.append(added_section)
         d = serialize_report(rep)
         check_marshallable(d)
@@ -111,28 +117,31 @@ class TestReportSerialization:
         # Check assembled == rep
         assert a.__dict__.keys() == rep.__dict__.keys()
         for key in rep.__dict__.keys():
-            if key != 'longrepr':
+            if key != "longrepr":
                 assert getattr(a, key) == getattr(rep, key)
         assert rep.longrepr.reprcrash.lineno == a.longrepr.reprcrash.lineno
         assert rep.longrepr.reprcrash.message == a.longrepr.reprcrash.message
         assert rep.longrepr.reprcrash.path == a.longrepr.reprcrash.path
-        assert rep.longrepr.reprtraceback.entrysep \
+        assert rep.longrepr.reprtraceback.entrysep == a.longrepr.reprtraceback.entrysep
-            == a.longrepr.reprtraceback.entrysep
+        assert (
-        assert rep.longrepr.reprtraceback.extraline \
+            rep.longrepr.reprtraceback.extraline == a.longrepr.reprtraceback.extraline
-            == a.longrepr.reprtraceback.extraline
+        )
-        assert rep.longrepr.reprtraceback.style \
+        assert rep.longrepr.reprtraceback.style == a.longrepr.reprtraceback.style
-            == a.longrepr.reprtraceback.style
         assert rep.longrepr.sections == a.longrepr.sections
         # Missing section attribute PR171
         assert added_section in a.longrepr.sections
 
     def test_reprentries_serialization_170(self, testdir):
         from _pytest._code.code import ReprEntry
-        reprec = testdir.inline_runsource("""
+
+        reprec = testdir.inline_runsource(
+            """
                             def test_repr_entry():
                                 x = 0
                                 assert x
-                        """, '--showlocals')
+                        """,
+            "--showlocals",
+        )
         reports = reprec.getreports("pytest_runtest_logreport")
         assert len(reports) == 3
         rep = reports[1]
@@ -146,7 +155,9 @@ class TestReportSerialization:
             assert rep_entries[i].lines == a_entries[i].lines
             assert rep_entries[i].localssep == a_entries[i].localssep
             assert rep_entries[i].reprfileloc.lineno == a_entries[i].reprfileloc.lineno
-            assert rep_entries[i].reprfileloc.message == a_entries[i].reprfileloc.message
+            assert (
+                rep_entries[i].reprfileloc.message == a_entries[i].reprfileloc.message
+            )
             assert rep_entries[i].reprfileloc.path == a_entries[i].reprfileloc.path
             assert rep_entries[i].reprfuncargs.args == a_entries[i].reprfuncargs.args
             assert rep_entries[i].reprlocals.lines == a_entries[i].reprlocals.lines
@@ -154,11 +165,15 @@ class TestReportSerialization:
 
     def test_reprentries_serialization_196(self, testdir):
         from _pytest._code.code import ReprEntryNative
-        reprec = testdir.inline_runsource("""
+
+        reprec = testdir.inline_runsource(
+            """
                             def test_repr_entry_native():
                                 x = 0
                                 assert x
-                        """, '--tb=native')
+                        """,
+            "--tb=native",
+        )
         reports = reprec.getreports("pytest_runtest_logreport")
         assert len(reports) == 3
         rep = reports[1]
@@ -172,7 +187,8 @@ class TestReportSerialization:
             assert rep_entries[i].lines == a_entries[i].lines
 
     def test_itemreport_outcomes(self, testdir):
-        reprec = testdir.inline_runsource("""
+        reprec = testdir.inline_runsource(
+            """
             import py
             def test_pass(): pass
             def test_fail(): 0/0
@@ -184,7 +200,8 @@ class TestReportSerialization:
             def test_xfail(): 0/0
             def test_xfail_imperative():
                 py.test.xfail("hello")
-        """)
+        """
+        )
         reports = reprec.getreports("pytest_runtest_logreport")
         assert len(reports) == 17  # with setup/teardown "passed" reports
         for rep in reports:
@@ -246,10 +263,12 @@ class TestReportSerialization:
 
 class TestWorkerInteractor:
     def test_basic_collect_and_runtests(self, worker):
-        worker.testdir.makepyfile("""
+        worker.testdir.makepyfile(
+            """
             def test_func():
                 pass
-        """)
+        """
+        )
         worker.setup()
         ev = worker.popevent()
         assert ev.name == "workerready"
@@ -257,8 +276,8 @@ class TestWorkerInteractor:
         assert ev.name == "collectionstart"
         assert not ev.kwargs
         ev = worker.popevent("collectionfinish")
-        assert ev.kwargs['topdir'] == worker.testdir.tmpdir
+        assert ev.kwargs["topdir"] == worker.testdir.tmpdir
-        ids = ev.kwargs['ids']
+        ids = ev.kwargs["ids"]
         assert len(ids) == 1
         worker.sendcommand("runtests", indices=list(range(len(ids))))
         worker.sendcommand("shutdown")
@@ -268,31 +287,30 @@ class TestWorkerInteractor:
         ev = worker.popevent("testreport")  # setup
         ev = worker.popevent("testreport")
         assert ev.name == "testreport"
-        rep = unserialize_report(ev.name, ev.kwargs['data'])
+        rep = unserialize_report(ev.name, ev.kwargs["data"])
         assert rep.nodeid.endswith("::test_func")
         assert rep.passed
         assert rep.when == "call"
         ev = worker.popevent("workerfinished")
-        assert 'workeroutput' in ev.kwargs
+        assert "workeroutput" in ev.kwargs
 
-    @pytest.mark.skipif(pytest.__version__ >= '3.0',
-                        reason='skip at module level illegal in pytest 3.0')
     def test_remote_collect_skip(self, worker):
-        worker.testdir.makepyfile("""
+        worker.testdir.makepyfile(
-            import py
+            """
-            py.test.skip("hello")
+            import pytest
-        """)
+            pytest.skip("hello", allow_module_level=True)
+        """
+        )
         worker.setup()
         ev = worker.popevent("collectionstart")
         assert not ev.kwargs
         ev = worker.popevent()
         assert ev.name == "collectreport"
-        ev = worker.popevent()
+        rep = unserialize_report(ev.name, ev.kwargs["data"])
-        assert ev.name == "collectreport"
-        rep = unserialize_report(ev.name, ev.kwargs['data'])
         assert rep.skipped
+        assert rep.longrepr[2] == "Skipped: hello"
         ev = worker.popevent("collectionfinish")
-        assert not ev.kwargs['ids']
+        assert not ev.kwargs["ids"]
 
     def test_remote_collect_fail(self, worker):
         worker.testdir.makepyfile("""aasd qwe""")
@@ -301,18 +319,18 @@ class TestWorkerInteractor:
         assert not ev.kwargs
         ev = worker.popevent()
         assert ev.name == "collectreport"
-        ev = worker.popevent()
+        rep = unserialize_report(ev.name, ev.kwargs["data"])
-        assert ev.name == "collectreport"
-        rep = unserialize_report(ev.name, ev.kwargs['data'])
         assert rep.failed
         ev = worker.popevent("collectionfinish")
-        assert not ev.kwargs['ids']
+        assert not ev.kwargs["ids"]
 
     def test_runtests_all(self, worker):
-        worker.testdir.makepyfile("""
+        worker.testdir.makepyfile(
+            """
             def test_func(): pass
             def test_func2(): pass
-        """)
+        """
+        )
         worker.setup()
         ev = worker.popevent()
         assert ev.name == "workerready"
@@ -320,57 +338,126 @@ class TestWorkerInteractor:
         assert ev.name == "collectionstart"
         assert not ev.kwargs
         ev = worker.popevent("collectionfinish")
-        ids = ev.kwargs['ids']
+        ids = ev.kwargs["ids"]
         assert len(ids) == 2
-        worker.sendcommand("runtests_all", )
+        worker.sendcommand("runtests_all")
-        worker.sendcommand("shutdown", )
+        worker.sendcommand("shutdown")
         for func in "::test_func", "::test_func2":
             for i in range(3):  # setup/call/teardown
                 ev = worker.popevent("testreport")
                 assert ev.name == "testreport"
-                rep = unserialize_report(ev.name, ev.kwargs['data'])
+                rep = unserialize_report(ev.name, ev.kwargs["data"])
                 assert rep.nodeid.endswith(func)
         ev = worker.popevent("workerfinished")
-        assert 'workeroutput' in ev.kwargs
+        assert "workeroutput" in ev.kwargs
 
     def test_happy_run_events_converted(self, testdir, worker):
         py.test.xfail("implement a simple test for event production")
         assert not worker.use_callback
-        worker.testdir.makepyfile("""
+        worker.testdir.makepyfile(
+            """
             def test_func():
                 pass
-        """)
+        """
+        )
         worker.setup()
         hookrec = testdir.getreportrecorder(worker.config)
         for data in worker.slp.channel:
             worker.slp.process_from_remote(data)
         worker.slp.process_from_remote(worker.slp.ENDMARK)
         pprint.pprint(hookrec.hookrecorder.calls)
-        hookrec.hookrecorder.contains([
+        hookrec.hookrecorder.contains(
-            ("pytest_collectstart", "collector.fspath == aaa"),
+            [
-            ("pytest_pycollect_makeitem", "name == 'test_func'"),
+                ("pytest_collectstart", "collector.fspath == aaa"),
-            ("pytest_collectreport", "report.collector.fspath == aaa"),
+                ("pytest_pycollect_makeitem", "name == 'test_func'"),
-            ("pytest_collectstart", "collector.fspath == bbb"),
+                ("pytest_collectreport", "report.collector.fspath == aaa"),
-            ("pytest_pycollect_makeitem", "name == 'test_func'"),
+                ("pytest_collectstart", "collector.fspath == bbb"),
-            ("pytest_collectreport", "report.collector.fspath == bbb"),
+                ("pytest_pycollect_makeitem", "name == 'test_func'"),
-        ])
+                ("pytest_collectreport", "report.collector.fspath == bbb"),
+            ]
+        )
 
     def test_process_from_remote_error_handling(self, worker, capsys):
         worker.use_callback = True
         worker.setup()
-        worker.slp.process_from_remote(('<nonono>', ()))
+        worker.slp.process_from_remote(("<nonono>", ()))
         out, err = capsys.readouterr()
-        assert 'INTERNALERROR> ValueError: unknown event: <nonono>' in out
+        assert "INTERNALERROR> ValueError: unknown event: <nonono>" in out
         ev = worker.popevent()
         assert ev.name == "errordown"
 
 
 def test_remote_env_vars(testdir):
-    testdir.makepyfile('''
+    testdir.makepyfile(
+        """
         import os
         def test():
             assert os.environ['PYTEST_XDIST_WORKER'] in ('gw0', 'gw1')
             assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2'
-    ''')
+    """
-    result = testdir.runpytest('-n2', '--max-worker-restart=0')
+    )
+    result = testdir.runpytest("-n2", "--max-worker-restart=0")
     assert result.ret == 0
+
+
+def test_remote_inner_argv(testdir):
+    """Test/document the behavior due to execnet using `python -c`."""
+    testdir.makepyfile(
+        """
+        import sys
+
+        def test_argv():
+            assert sys.argv == ["-c"]
+        """
+    )
+    result = testdir.runpytest("-n1")
+    assert result.ret == 0
+
+
+def test_remote_mainargv(testdir):
+    outer_argv = sys.argv
+
+    testdir.makepyfile(
+        """
+        def test_mainargv(request):
+            assert request.config.workerinput["mainargv"] == {!r}
+        """.format(
+            outer_argv
+        )
+    )
+    result = testdir.runpytest("-n1")
+    assert result.ret == 0
+
+
+def test_remote_usage_prog(testdir, request):
+    if not hasattr(request.config._parser, "prog"):
+        pytest.skip("prog not available in config parser")
+    testdir.makeconftest(
+        """
+        import pytest
+
+        config_parser = None
+
+        @pytest.fixture
+        def get_config_parser():
+            return config_parser
+
+        def pytest_configure(config):
+            global config_parser
+            config_parser = config._parser
+    """
+    )
+    testdir.makepyfile(
+        """
+        import sys
+
+        def test(get_config_parser, request):
+            get_config_parser._getparser().error("my_usage_error")
+    """
+    )
+
+    result = testdir.runpytest_subprocess("-n1")
+    assert result.ret == 1
+    result.stdout.fnmatch_lines(
+        ["usage: pytest.py *", "pytest.py: error: my_usage_error"]
+    )

testing/test_slavemanage.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import py
 import pytest
 import textwrap
@@ -41,7 +42,7 @@ def workercontroller(monkeypatch):
         def setup(self):
             pass
 
-    monkeypatch.setattr(workermanage, 'WorkerController', MockController)
+    monkeypatch.setattr(workermanage, "WorkerController", MockController)
     return MockController
 
 
@@ -57,8 +58,7 @@ class TestNodeManagerPopen:
         for spec in NodeManager(config, specs, defaultchdir="abc").specs:
             assert spec.chdir == "abc"
 
-    def test_popen_makegateway_events(self, config, hookrecorder,
+    def test_popen_makegateway_events(self, config, hookrecorder, workercontroller):
-                                      workercontroller):
         hm = NodeManager(config, ["popen"] * 2)
         hm.setup_nodes(None)
         call = hookrecorder.popcall("pytest_xdist_setupnodes")
@@ -107,15 +107,16 @@ class TestNodeManagerPopen:
         for gw in hm.group:
             hm.rsync(gw, source, notify=lambda *args: notifications.append(args))
         assert len(notifications) == 1
-        assert notifications[0] == ("rsyncrootready", hm.group['gw0'].spec, source)
+        assert notifications[0] == ("rsyncrootready", hm.group["gw0"].spec, source)
         hm.teardown_nodes()
         dest = dest.join(source.basename)
         assert dest.join("dir1").check()
         assert dest.join("dir1", "dir2").check()
-        assert dest.join("dir1", "dir2", 'hello').check()
+        assert dest.join("dir1", "dir2", "hello").check()
 
-    def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
+    def test_rsync_same_popen_twice(
-                                    workercontroller):
+        self, config, mysetup, hookrecorder, workercontroller
+    ):
         source, dest = mysetup.source, mysetup.dest
         hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
         hm.roots = []
@@ -141,9 +142,9 @@ class TestHRSync:
         files = list(source.visit(rec=syncer.filter, fil=syncer.filter))
         assert len(files) == 3
         basenames = [x.basename for x in files]
-        assert 'dir' in basenames
+        assert "dir" in basenames
-        assert 'file.txt' in basenames
+        assert "file.txt" in basenames
-        assert 'somedir' in basenames
+        assert "somedir" in basenames
 
     def test_hrsync_one_host(self, mysetup):
         source, dest = mysetup.source, mysetup.dest
@@ -168,9 +169,10 @@ class TestNodeManager:
         nodemanager.makegateways()
         nodemanager.rsync_roots()
         p, = nodemanager.gwmanager.multi_exec(
-            "import os ; channel.send(os.getcwd())").receive_each()
+            "import os ; channel.send(os.getcwd())"
+        ).receive_each()
         p = py.path.local(p)
-        py.builtin.print_("remote curdir", p)
+        print("remote curdir", p)
         assert p == mysetup.dest.join(config.topdir.basename)
         assert p.join("dir1").check()
         assert p.join("dir1", "file1").check()
@@ -182,15 +184,17 @@ class TestNodeManager:
         dir2.ensure("hello")
         for rsyncroot in (dir1, source):
             dest.remove()
-            nodemanager = NodeManager(testdir.parseconfig(
+            nodemanager = NodeManager(
-                "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot,
+                testdir.parseconfig(
-                source, ))
+                    "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot, source
+                )
+            )
             nodemanager.setup_nodes(None)  # calls .rsync_roots()
             if rsyncroot == source:
                 dest = dest.join("source")
             assert dest.join("dir1").check()
             assert dest.join("dir1", "dir2").check()
-            assert dest.join("dir1", "dir2", 'hello').check()
+            assert dest.join("dir1", "dir2", "hello").check()
             nodemanager.teardown_nodes()
 
     def test_init_rsync_roots(self, testdir, mysetup, workercontroller):
@@ -199,10 +203,14 @@ class TestNodeManager:
         source.ensure("dir1", "somefile", dir=1)
         dir2.ensure("hello")
         source.ensure("bogusdir", "file")
-        source.join("tox.ini").write(textwrap.dedent("""
+        source.join("tox.ini").write(
+            textwrap.dedent(
+                """
             [pytest]
             rsyncdirs=dir1/dir2
-        """))
+        """
+            )
+        )
         config = testdir.parseconfig(source)
         nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
         nodemanager.setup_nodes(None)  # calls .rsync_roots()
@@ -218,27 +226,31 @@ class TestNodeManager:
         dir2.ensure("hello")
         source.ensure("foo", "bar")
         source.ensure("bar", "foo")
-        source.join("tox.ini").write(textwrap.dedent("""
+        source.join("tox.ini").write(
+            textwrap.dedent(
+                """
             [pytest]
             rsyncdirs = dir1 dir5
             rsyncignore = dir1/dir2 dir5/dir6 foo*
-        """))
+        """
+            )
+        )
         config = testdir.parseconfig(source)
-        config.option.rsyncignore = ['bar']
+        config.option.rsyncignore = ["bar"]
         nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
         nodemanager.setup_nodes(None)  # calls .rsync_roots()
         assert dest.join("dir1").check()
         assert not dest.join("dir1", "dir2").check()
         assert dest.join("dir5", "file").check()
         assert not dest.join("dir6").check()
-        assert not dest.join('foo').check()
+        assert not dest.join("foo").check()
-        assert not dest.join('bar').check()
+        assert not dest.join("bar").check()
 
     def test_optimise_popen(self, testdir, mysetup, workercontroller):
         source = mysetup.source
         specs = ["popen"] * 3
         source.join("conftest.py").write("rsyncdirs = ['a']")
-        source.ensure('a', dir=1)
+        source.ensure("a", dir=1)
         config = testdir.parseconfig(source)
         nodemanager = NodeManager(config, specs)
         nodemanager.setup_nodes(None)  # calls .rysnc_roots()
@@ -247,12 +259,15 @@ class TestNodeManager:
             assert not gwspec.chdir
 
     def test_ssh_setup_nodes(self, specssh, testdir):
-        testdir.makepyfile(__init__="",
+        testdir.makepyfile(
-                           test_x="""
+            __init__="",
+            test_x="""
             def test_one():
                 pass
-        """)
+        """,
-        reprec = testdir.inline_run("-d", "--rsyncdir=%s" % testdir.tmpdir,
+        )
-                                    "--tx", specssh, testdir.tmpdir)
+        reprec = testdir.inline_run(
+            "-d", "--rsyncdir=%s" % testdir.tmpdir, "--tx", specssh, testdir.tmpdir
+        )
         rep, = reprec.getreports("pytest_runtest_logreport")
         assert rep.passed

tox.ini

@@ -1,44 +1,26 @@
 [tox]
-# if you change the envlist, please update .travis.yml file as well
 envlist=
-  py{27,34,35,36}-pytest{30,31,32,33}
+  linting
-  py{27,36}-pytest{30,31,32,33}-pexpect
+  py{27,34,35,36,37}-pytestlatest
   py{27,36}-pytest{master,features}
-  flakes
-  readme
-
 
 [testenv]
 changedir=testing
 passenv = USER USERNAME
 deps =
-  pycmd
+  pytestlatest: pytest
-  # to avoid .eggs
-  setuptools_scm
-  pytest30: pytest~=3.0.5
-  pytest31: pytest~=3.1.0
-  pytest32: pytest~=3.2.0
-  pytest33: pytest~=3.3.0
   pytestmaster: git+https://github.com/pytest-dev/pytest.git@master
   pytestfeatures: git+https://github.com/pytest-dev/pytest.git@features
-  pexpect: pexpect
+  filelock
-platform=
-  pexpect: linux|darwin
 commands=
-  # always clean to avoid code unmarshal mismatch on old python/pytest
-  py.cleanup -aq
   pytest {posargs}
 
-[testenv:flakes]
+[testenv:linting]
-changedir=
+skipsdist = True
-deps = flake8
+usedevelop = True
-commands = flake8 setup.py testing xdist
+deps =
-
+  pre-commit
-[testenv:readme]
+commands = pre-commit run --all-files --show-diff-on-failure
-changedir =
-deps = readme
-skip_install = true
-commands = python setup.py check -r -s
 
 [testenv:release]
 changedir=
@@ -53,5 +35,8 @@ commands =
     towncrier --version {posargs} --yes
 
 [pytest]
-addopts = -rsfxX
+addopts = -ra
 
+[flake8]
+max-line-length = 120
+ignore = E203,W503

xdist/__init__.py

@@ -1,3 +1,3 @@
 from xdist._version import version as __version__
 
-__all__ = ['__version__']
+__all__ = ["__version__"]

xdist/dsession.py

@@ -10,7 +10,7 @@ from xdist.scheduler import (
 )
 
 
-queue = py.builtin._tryimport('queue', 'Queue')
+from six.moves.queue import Empty, Queue
 
 
 class Interrupted(KeyboardInterrupt):
@@ -41,7 +41,7 @@ class DSession(object):
         self.shuttingdown = False
         self.countfailures = 0
         self.maxfail = config.getvalue("maxfail")
-        self.queue = queue.Queue()
+        self.queue = Queue()
         self._session = None
         self._failed_collection_errors = {}
         self._active_nodes = set()
@@ -84,7 +84,7 @@ class DSession(object):
 
     def pytest_sessionfinish(self, session):
         """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
@@ -95,19 +95,18 @@ class DSession(object):
 
     @pytest.mark.trylast
     def pytest_xdist_make_scheduler(self, config, log):
-        dist = config.getvalue('dist')
+        dist = config.getvalue("dist")
         schedulers = {
-            'each': EachScheduling,
+            "each": EachScheduling,
-            'load': LoadScheduling,
+            "load": LoadScheduling,
-            'loadscope': LoadScopeScheduling,
+            "loadscope": LoadScopeScheduling,
-            'loadfile': LoadFileScheduling,
+            "loadfile": LoadFileScheduling,
         }
         return schedulers[dist](config, log)
 
     def pytest_runtestloop(self):
         self.sched = self.config.hook.pytest_xdist_make_scheduler(
-            config=self.config,
+            config=self.config, log=self.log
-            log=self.log
         )
         assert self.sched is not None
 
@@ -129,7 +128,7 @@ class DSession(object):
             try:
                 eventcall = self.queue.get(timeout=2.0)
                 break
-            except queue.Empty:
+            except Empty:
                 continue
         callname, kwargs = eventcall
         assert callname, kwargs
@@ -151,8 +150,8 @@ class DSession(object):
         collection without any further input.
         """
         node.workerinfo = workerinfo
-        node.workerinfo['id'] = node.gateway.id
+        node.workerinfo["id"] = node.gateway.id
-        node.workerinfo['spec'] = node.gateway.spec
+        node.workerinfo["spec"] = node.gateway.spec
 
         # TODO: (#234 task) needs this for pytest. Remove when refactor in pytest repo
         node.slaveinfo = node.workerinfo
@@ -172,7 +171,7 @@ class DSession(object):
         workerready before shutdown was triggered.
         """
         self.config.hook.pytest_testnodedown(node=node, error=None)
-        if node.workeroutput['exitstatus'] == 2:  # keyboard-interrupt
+        if node.workeroutput["exitstatus"] == 2:  # keyboard-interrupt
             self.shouldstop = "%s received keyboard-interrupt" % (node,)
             self.worker_errordown(node, "keyboard-interrupt")
             return
@@ -193,14 +192,15 @@ class DSession(object):
                 self.handle_crashitem(crashitem, node)
 
         self._failed_nodes_count += 1
-        maximum_reached = (self._max_worker_restart is not None and
+        maximum_reached = (
-                           self._failed_nodes_count > self._max_worker_restart)
+            self._max_worker_restart is not None
+            and self._failed_nodes_count > self._max_worker_restart
+        )
         if maximum_reached:
             if self._max_worker_restart == 0:
-                msg = 'Worker restarting disabled'
+                msg = "Worker restarting disabled"
             else:
-                msg = "Maximum crashed workers reached: %d" % \
+                msg = "Maximum crashed workers reached: %d" % self._max_worker_restart
-                      self._max_worker_restart
             self.report_line(msg)
         else:
             self.report_line("Replacing crashed worker %s" % node.gateway.id)
@@ -218,8 +218,7 @@ class DSession(object):
         """
         if self.shuttingdown:
             return
-        self.config.hook.pytest_xdist_node_collection_finished(node=node,
+        self.config.hook.pytest_xdist_node_collection_finished(node=node, ids=ids)
-                                                               ids=ids)
         # tell session which items were effectively collected otherwise
         # the master node will finish the session with EXIT_NOTESTSCOLLECTED
         self._session.testscollected = len(ids)
@@ -230,19 +229,19 @@ class DSession(object):
             if self.terminal and not self.sched.has_pending:
                 self.trdist.ensure_show_status()
                 self.terminal.write_line("")
-                self.terminal.write_line("scheduling tests via %s" % (
+                if self.config.option.verbose > 0:
-                    self.sched.__class__.__name__))
+                    self.terminal.write_line(
+                        "scheduling tests via %s" % (self.sched.__class__.__name__)
+                    )
             self.sched.schedule()
 
     def worker_logstart(self, node, nodeid, location):
         """Emitted when a node calls the pytest_runtest_logstart hook."""
-        self.config.hook.pytest_runtest_logstart(
+        self.config.hook.pytest_runtest_logstart(nodeid=nodeid, location=location)
-            nodeid=nodeid, location=location)
 
     def worker_logfinish(self, node, nodeid, location):
         """Emitted when a node calls the pytest_runtest_logfinish hook."""
-        self.config.hook.pytest_runtest_logfinish(
+        self.config.hook.pytest_runtest_logfinish(nodeid=nodeid, location=location)
-            nodeid=nodeid, location=location)
 
     def worker_testreport(self, node, rep):
         """Emitted when a node calls the pytest_runtest_logreport hook."""
@@ -259,15 +258,24 @@ class DSession(object):
         self.sched.mark_test_complete(node, item_index, duration)
 
     def worker_collectreport(self, node, rep):
-        """Emitted when a node calls the pytest_collectreport hook."""
+        """Emitted when a node calls the pytest_collectreport hook.
-        if rep.failed:
+
-            self._failed_worker_collectreport(node, rep)
+        Because we only need the report when there's a failure/skip, as optimization
+        we only expect to receive failed/skipped reports from workers (#330).
+        """
+        assert not rep.passed
+        self._failed_worker_collectreport(node, rep)
 
     def worker_logwarning(self, message, code, nodeid, fslocation):
         """Emitted when a node calls the pytest_logwarning hook."""
         kwargs = dict(message=message, code=code, nodeid=nodeid, fslocation=fslocation)
         self.config.hook.pytest_logwarning.call_historic(kwargs=kwargs)
 
+    def worker_warning_captured(self, warning_message, when, item):
+        """Emitted when a node calls the pytest_logwarning hook."""
+        kwargs = dict(warning_message=warning_message, when=when, item=item)
+        self.config.hook.pytest_warning_captured.call_historic(kwargs=kwargs)
+
     def _clone_node(self, node):
         """Return new node based on an existing one.
 
@@ -295,8 +303,7 @@ class DSession(object):
         if rep.failed:
             self.countfailures += 1
             if self.maxfail and self.countfailures >= self.maxfail:
-                self.shouldstop = "stopping after %d failures" % (
+                self.shouldstop = "stopping after %d failures" % (self.countfailures)
-                    self.countfailures)
 
     def triggershutdown(self):
         self.log("triggering shutdown")
@@ -310,8 +317,9 @@ class DSession(object):
         runner = self.config.pluginmanager.getplugin("runner")
         fspath = nodeid.split("::")[0]
         msg = "Worker %r crashed while running %r" % (worker.gateway.id, nodeid)
-        rep = runner.TestReport(nodeid, (fspath, None, fspath),
+        rep = runner.TestReport(
-                                (), "failed", msg, "???")
+            nodeid, (fspath, None, fspath), (), "failed", msg, "???"
+        )
         rep.node = worker
         self.config.hook.pytest_runtest_logreport(report=rep)
 
@@ -322,7 +330,7 @@ class TerminalDistReporter(object):
         self.tr = config.pluginmanager.getplugin("terminalreporter")
         self._status = {}
         self._lastlen = 0
-        self._isatty = getattr(self.tr, 'isatty', self.tr.hasmarkup)
+        self._isatty = getattr(self.tr, "isatty", self.tr.hasmarkup)
 
     def write_line(self, msg):
         self.tr.write_line(msg)
@@ -337,9 +345,11 @@ class TerminalDistReporter(object):
             self.rewrite(self.getstatus())
 
     def getstatus(self):
-        parts = ["%s %s" % (spec.id, self._status[spec.id])
+        if self.config.option.verbose >= 0:
-                 for spec in self._specs]
+            parts = ["%s %s" % (spec.id, self._status[spec.id]) for spec in self._specs]
-        return " / ".join(parts)
+            return " / ".join(parts)
+        else:
+            return "bringing up nodes..."
 
     def rewrite(self, line, newline=False):
         pline = line + " " * max(self._lastlen - len(line), 0)
@@ -361,17 +371,17 @@ class TerminalDistReporter(object):
         if self.config.option.verbose > 0:
             rinfo = gateway._rinfo()
             version = "%s.%s.%s" % rinfo.version_info[:3]
-            self.rewrite("[%s] %s Python %s cwd: %s" % (
+            self.rewrite(
-                gateway.id, rinfo.platform, version, rinfo.cwd),
+                "[%s] %s Python %s cwd: %s"
-                newline=True)
+                % (gateway.id, rinfo.platform, version, rinfo.cwd),
+                newline=True,
+            )
         self.setstatus(gateway.spec, "C")
 
     def pytest_testnodeready(self, node):
         if self.config.option.verbose > 0:
             d = node.workerinfo
-            infoline = "[%s] Python %s" % (
+            infoline = "[%s] Python %s" % (d["id"], d["version"].replace("\n", " -- "))
-                d['id'],
-                d['version'].replace('\n', ' -- '),)
             self.rewrite(infoline, newline=True)
         self.setstatus(node.gateway.spec, "ok")
 

xdist/looponfail.py

@@ -6,7 +6,7 @@
     processes) otherwise changes to source code can crash
     the controlling process which should best never happen.
 """
-
+from __future__ import print_function
 import py
 import pytest
 import sys
@@ -17,19 +17,22 @@ import execnet
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
     group._addoption(
-        '-f', '--looponfail',
+        "-f",
-        action="store_true", dest="looponfail", default=False,
+        "--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.")
+        "and re-run failing test set until all pass.",
+    )
 
 
 def pytest_cmdline_main(config):
 
     if config.getoption("looponfail"):
-        usepdb = config.getoption('usepdb')  # a core option
+        usepdb = config.getoption("usepdb")  # a core option
         if usepdb:
-            raise pytest.UsageError(
+            raise pytest.UsageError("--pdb is incompatible with --looponfail.")
-                "--pdb incompatible with --looponfail.")
         looponfail_main(config)
         return 2  # looponfail only can get stop with ctrl-C anyway
 
@@ -45,8 +48,8 @@ def looponfail_main(config):
                 # the last failures passed, let's immediately rerun all
                 continue
             repr_pytest_looponfailinfo(
-                failreports=remotecontrol.failures,
+                failreports=remotecontrol.failures, rootdirs=rootdirs
-                rootdirs=rootdirs)
+            )
             statrecorder.waitonchange(checkinterval=2.0)
     except KeyboardInterrupt:
         print()
@@ -60,7 +63,7 @@ class RemoteControl(object):
     def trace(self, *args):
         if self.config.option.debug:
             msg = " ".join([str(x) for x in args])
-            py.builtin.print_("RemoteControl:", msg)
+            print("RemoteControl:", msg)
 
     def initgateway(self):
         return execnet.makegateway("popen")
@@ -68,7 +71,7 @@ class RemoteControl(object):
     def setup(self, out=None):
         if out is None:
             out = py.io.TerminalWriter()
-        if hasattr(self, 'gateway'):
+        if hasattr(self, "gateway"):
             raise ValueError("already have gateway %r" % self.gateway)
         self.trace("setting up worker session")
         self.gateway = self.initgateway()
@@ -82,15 +85,16 @@ class RemoteControl(object):
         def write(s):
             out._file.write(s)
             out._file.flush()
+
         remote_outchannel.setcallback(write)
 
     def ensure_teardown(self):
-        if hasattr(self, 'channel'):
+        if hasattr(self, "channel"):
             if not self.channel.isclosed():
                 self.trace("closing", self.channel)
                 self.channel.close()
             del self.channel
-        if hasattr(self, 'gateway'):
+        if hasattr(self, "gateway"):
             self.trace("exiting", self.gateway)
             self.gateway.exit()
             del self.gateway
@@ -138,8 +142,9 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
 def init_worker_session(channel, args, option_dict):
     import os
     import sys
+
     outchannel = channel.gateway.newchannel()
-    sys.stdout = sys.stderr = outchannel.makefile('w')
+    sys.stdout = sys.stderr = outchannel.makefile("w")
     channel.send(outchannel)
     # prune sys.path to not contain relative paths
     newpaths = []
@@ -152,9 +157,11 @@ def init_worker_session(channel, args, option_dict):
 
     # fullwidth, hasmarkup = channel.receive()
     from _pytest.config import Config
+
     config = Config.fromdictargs(option_dict, list(args))
     config.args = args
     from xdist.looponfail import WorkerFailSession
+
     WorkerFailSession(config, channel).main()
 
 
@@ -181,7 +188,8 @@ class WorkerFailSession(object):
         except pytest.UsageError:
             items = session.perform_collect(None)
         hook.pytest_collection_modifyitems(
-            session=session, config=session.config, items=items)
+            session=session, config=session.config, items=items
+        )
         hook.pytest_collection_finish(session=session)
         return True
 
@@ -207,7 +215,7 @@ class WorkerFailSession(object):
         for rep in self.recorded_failures:
             trails.append(rep.nodeid)
             loc = rep.longrepr
-            loc = str(getattr(loc, 'reprcrash', loc))
+            loc = str(getattr(loc, "reprcrash", loc))
             failreports.append(loc)
         self.channel.send((trails, failreports, self.collection_failed))
 
@@ -245,10 +253,12 @@ class StatRecorder(object):
                         changed = True
                 else:
                     if oldstat:
-                        if oldstat.mtime != curstat.mtime or \
+                        if (
-                           oldstat.size != curstat.size:
+                            oldstat.mtime != curstat.mtime
+                            or oldstat.size != curstat.size
+                        ):
                             changed = True
-                            py.builtin.print_("# MODIFIED", path)
+                            print("# MODIFIED", path)
                             if removepycfiles and path.ext == ".py":
                                 pycfile = path + "c"
                                 if pycfile.check():

xdist/plugin.py

@@ -1,25 +1,35 @@
+import os
+
 import py
 import pytest
 
 
-def parse_numprocesses(s):
+def auto_detect_cpus():
-    if s == 'auto':
+    try:
+        from os import sched_getaffinity
+    except ImportError:
+        if os.environ.get("TRAVIS") == "true":
+            # workaround https://bitbucket.org/pypy/pypy/issues/2375
+            return 2
         try:
-            from os import sched_getaffinity
+            from os import cpu_count
         except ImportError:
-            try:
+            from multiprocessing import cpu_count
-                from os import cpu_count
+    else:
-            except ImportError:
-                from multiprocessing import cpu_count
-        else:
-            def cpu_count():
-                return len(sched_getaffinity(0))
 
-        try:
+        def cpu_count():
-            n = cpu_count()
+            return len(sched_getaffinity(0))
-        except NotImplementedError:
+
-            return 1
+    try:
-        return n if n else 1
+        n = cpu_count()
+    except NotImplementedError:
+        return 1
+    return n if n else 1
+
+
+def parse_numprocesses(s):
+    if s == "auto":
+        return auto_detect_cpus()
     else:
         return int(s)
 
@@ -27,60 +37,109 @@ def parse_numprocesses(s):
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
     group._addoption(
-        '-n', '--numprocesses', dest="numprocesses", metavar="numprocesses",
+        "-n",
+        "--numprocesses",
+        dest="numprocesses",
+        metavar="numprocesses",
         action="store",
         type=parse_numprocesses,
         help="shortcut for '--dist=load --tx=NUM*popen', "
-             "you can use 'auto' here for auto detection CPUs number on "
+        "you can use 'auto' here for auto detection CPUs number on "
-             "host system")
+        "host system",
-    group.addoption('--max-worker-restart', '--max-slave-restart', action="store", default=None,
+    )
-                    dest="maxworkerrestart",
-                    help="maximum number of workers that can be restarted "
-                         "when crashed (set to zero to disable this feature)\n"
-                         "'--max-slave-restart' option is deprecated and will be removed in "
-                         "a future release")
     group.addoption(
-        '--dist', metavar="distmode",
+        "--maxprocesses",
-        action="store", choices=['each', 'load', 'loadscope', 'loadfile', 'no'],
+        dest="maxprocesses",
-        dest="dist", default="no",
+        metavar="maxprocesses",
-        help=("set mode for distributing tests to exec environments.\n\n"
+        action="store",
-              "each: send each test to all available environments.\n\n"
+        type=int,
-              "load: load balance by sending any pending test to any"
+        help="limit the maximum number of workers to process the tests when using --numprocesses=auto",
-              " available environment.\n\n"
+    )
-              "loadscope: load balance by sending pending groups of tests in"
-              " the same scope to any available environment.\n\n"
-              "loadfile: load balance by sending test grouped by file"
-              " to any available environment.\n\n"
-              "(default) no: run tests inprocess, don't distribute."))
     group.addoption(
-        '--tx', dest="tx", action="append", default=[],
+        "--max-worker-restart",
+        "--max-slave-restart",
+        action="store",
+        default=None,
+        dest="maxworkerrestart",
+        help="maximum number of workers that can be restarted "
+        "when crashed (set to zero to disable this feature)\n"
+        "'--max-slave-restart' option is deprecated and will be removed in "
+        "a future release",
+    )
+    group.addoption(
+        "--dist",
+        metavar="distmode",
+        action="store",
+        choices=["each", "load", "loadscope", "loadfile", "no"],
+        dest="dist",
+        default="no",
+        help=(
+            "set mode for distributing tests to exec environments.\n\n"
+            "each: send each test to all available environments.\n\n"
+            "load: load balance by sending any pending test to any"
+            " available environment.\n\n"
+            "loadscope: load balance by sending pending groups of tests in"
+            " the same scope to any available environment.\n\n"
+            "loadfile: load balance by sending test grouped by file"
+            " to any available environment.\n\n"
+            "(default) no: run tests inprocess, don't distribute."
+        ),
+    )
+    group.addoption(
+        "--tx",
+        dest="tx",
+        action="append",
+        default=[],
         metavar="xspec",
-        help=("add a test execution environment. some examples: "
+        help=(
-              "--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
+            "add a test execution environment. some examples: "
-              "--tx ssh=user@codespeak.net//chdir=testcache"))
+            "--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
+            "--tx ssh=user@codespeak.net//chdir=testcache"
+        ),
+    )
     group._addoption(
-        '-d',
+        "-d",
-        action="store_true", dest="distload", default=False,
+        action="store_true",
-        help="load-balance tests.  shortcut for '--dist=load'")
+        dest="distload",
+        default=False,
+        help="load-balance tests.  shortcut for '--dist=load'",
+    )
     group.addoption(
-        '--rsyncdir', action="append", default=[], metavar="DIR",
+        "--rsyncdir",
-        help="add directory for rsyncing to remote tx nodes.")
+        action="append",
+        default=[],
+        metavar="DIR",
+        help="add directory for rsyncing to remote tx nodes.",
+    )
     group.addoption(
-        '--rsyncignore', action="append", default=[], metavar="GLOB",
+        "--rsyncignore",
-        help="add expression for ignores when rsyncing to remote tx nodes.")
+        action="append",
+        default=[],
+        metavar="GLOB",
+        help="add expression for ignores when rsyncing to remote tx nodes.",
+    )
 
     group.addoption(
-        "--boxed", action="store_true",
+        "--boxed",
-        help="backward compatibility alias for pytest-forked --forked")
+        action="store_true",
+        help="backward compatibility alias for pytest-forked --forked",
+    )
     parser.addini(
-        'rsyncdirs', 'list of (relative) paths to be rsynced for'
+        "rsyncdirs",
-        ' remote distributed testing.', type="pathlist")
+        "list of (relative) paths to be rsynced for remote distributed testing.",
+        type="pathlist",
+    )
     parser.addini(
-        'rsyncignore', 'list of (relative) glob-style paths to be ignored '
+        "rsyncignore",
-        'for rsyncing.', type="pathlist")
+        "list of (relative) glob-style paths to be ignored for rsyncing.",
+        type="pathlist",
+    )
     parser.addini(
-        "looponfailroots", type="pathlist",
+        "looponfailroots",
-        help="directories to check for changes", default=[py.path.local()])
+        type="pathlist",
+        help="directories to check for changes",
+        default=[py.path.local()],
+    )
 
 
 # -------------------------------------------------------------------------
@@ -90,12 +149,14 @@ def pytest_addoption(parser):
 
 def pytest_addhooks(pluginmanager):
     from xdist import 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
 # -------------------------------------------------------------------------
@@ -105,6 +166,7 @@ def pytest_addhooks(pluginmanager):
 def pytest_configure(config):
     if config.getoption("dist") != "no" and not config.getvalue("collectonly"):
         from xdist.dsession import DSession
+
         session = DSession(config)
         config.pluginmanager.register(session, "dsession")
         tr = config.pluginmanager.getplugin("terminalreporter")
@@ -116,18 +178,23 @@ def pytest_configure(config):
 @pytest.mark.tryfirst
 def pytest_cmdline_main(config):
     if config.option.numprocesses:
-        if config.option.dist == 'no':
+        if config.option.dist == "no":
             config.option.dist = "load"
-        config.option.tx = ['popen'] * config.option.numprocesses
+        numprocesses = config.option.numprocesses
+        if config.option.maxprocesses:
+            numprocesses = min(numprocesses, config.option.maxprocesses)
+        config.option.tx = ["popen"] * numprocesses
     if config.option.distload:
         config.option.dist = "load"
     val = config.getvalue
     if not val("collectonly"):
-        usepdb = config.getoption('usepdb')  # a core option
+        usepdb = config.getoption("usepdb")  # a core option
         if val("dist") != "no":
             if usepdb:
                 raise pytest.UsageError(
-                    "--pdb is incompatible with distributing tests; try using -n0.")  # noqa: E501
+                    "--pdb is incompatible with distributing tests; try using -n0."
+                )  # noqa: E501
+
 
 # -------------------------------------------------------------------------
 # fixtures
@@ -139,7 +206,7 @@ def worker_id(request):
     """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
     if running on the master node.
     """
-    if hasattr(request.config, 'workerinput'):
+    if hasattr(request.config, "workerinput"):
-        return request.config.workerinput['workerid']
+        return request.config.workerinput["workerid"]
     else:
-        return 'master'
+        return "master"

xdist/remote.py

@@ -12,12 +12,13 @@ import time
 
 import _pytest.hookspec
 import pytest
+from execnet.gateway_base import dumps, DumpError
 
 
 class WorkerInteractor(object):
     def __init__(self, config, channel):
         self.config = config
-        self.workerid = config.workerinput.get('workerid', "?")
+        self.workerid = config.workerinput.get("workerid", "?")
         self.log = py.log.Producer("worker-%s" % self.workerid)
         if not config.option.debug:
             py.log.setconsumer(self.log._keywords, None)
@@ -39,7 +40,7 @@ class WorkerInteractor(object):
 
     @pytest.hookimpl(hookwrapper=True)
     def pytest_sessionfinish(self, exitstatus):
-        self.config.workeroutput['exitstatus'] = exitstatus
+        self.config.workeroutput["exitstatus"] = exitstatus
         yield
         self.sendevent("workerfinished", workeroutput=self.config.workeroutput)
 
@@ -56,7 +57,7 @@ class WorkerInteractor(object):
                 return True
             self.log("received command", name, kwargs)
             if name == "runtests":
-                torun.extend(kwargs['indices'])
+                torun.extend(kwargs["indices"])
             elif name == "runtests_all":
                 torun.extend(range(len(session.items)))
             self.log("items to run:", torun)
@@ -79,24 +80,25 @@ class WorkerInteractor(object):
             nextitem = None
 
         start = time.time()
-        self.config.hook.pytest_runtest_protocol(
+        self.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
-            item=item,
-            nextitem=nextitem)
         duration = time.time() - start
-        self.sendevent("runtest_protocol_complete", item_index=self.item_index,
+        self.sendevent(
-                       duration=duration)
+            "runtest_protocol_complete", item_index=self.item_index, duration=duration
+        )
 
     def pytest_collection_finish(self, session):
         self.sendevent(
             "collectionfinish",
             topdir=str(session.fspath),
-            ids=[item.nodeid for item in session.items])
+            ids=[item.nodeid for item in session.items],
+        )
 
     def pytest_runtest_logstart(self, nodeid, location):
         self.sendevent("logstart", nodeid=nodeid, location=location)
 
     # the pytest_runtest_logfinish hook was introduced in pytest 3.4
-    if hasattr(_pytest.hookspec, 'pytest_runtest_logfinish'):
+    if hasattr(_pytest.hookspec, "pytest_runtest_logfinish"):
+
         def pytest_runtest_logfinish(self, nodeid, location):
             self.sendevent("logfinish", nodeid=nodeid, location=location)
 
@@ -108,12 +110,34 @@ class WorkerInteractor(object):
         self.sendevent("testreport", data=data)
 
     def pytest_collectreport(self, report):
-        data = serialize_report(report)
+        # send only reports that have not passed to master as optimization (#330)
-        self.sendevent("collectreport", data=data)
+        if not report.passed:
+            data = serialize_report(report)
+            self.sendevent("collectreport", data=data)
 
-    def pytest_logwarning(self, message, code, nodeid, fslocation):
+    # the pytest_logwarning hook was removed in pytest 4.1
-        self.sendevent("logwarning", message=message, code=code, nodeid=nodeid,
+    if hasattr(_pytest.hookspec, "pytest_logwarning"):
-                       fslocation=str(fslocation))
+
+        def pytest_logwarning(self, message, code, nodeid, fslocation):
+            self.sendevent(
+                "logwarning",
+                message=message,
+                code=code,
+                nodeid=nodeid,
+                fslocation=str(fslocation),
+            )
+
+    # the pytest_warning_captured hook was introduced in pytest 3.8
+    if hasattr(_pytest.hookspec, "pytest_warning_captured"):
+
+        def pytest_warning_captured(self, warning_message, when, item):
+            self.sendevent(
+                "warning_captured",
+                warning_message_data=serialize_warning_message(warning_message),
+                when=when,
+                # item cannot be serialized and will always be None when used with xdist
+                item=None,
+            )
 
 
 def serialize_report(rep):
@@ -122,34 +146,33 @@ def serialize_report(rep):
         reprcrash = rep.longrepr.reprcrash.__dict__.copy()
 
         new_entries = []
-        for entry in reprtraceback['reprentries']:
+        for entry in reprtraceback["reprentries"]:
-            entry_data = {
+            entry_data = {"type": type(entry).__name__, "data": entry.__dict__.copy()}
-                'type': type(entry).__name__,
+            for key, value in entry_data["data"].items():
-                'data': entry.__dict__.copy(),
+                if hasattr(value, "__dict__"):
-            }
+                    entry_data["data"][key] = value.__dict__.copy()
-            for key, value in entry_data['data'].items():
-                if hasattr(value, '__dict__'):
-                    entry_data['data'][key] = value.__dict__.copy()
             new_entries.append(entry_data)
 
-        reprtraceback['reprentries'] = new_entries
+        reprtraceback["reprentries"] = new_entries
 
         return {
-            'reprcrash': reprcrash,
+            "reprcrash": reprcrash,
-            'reprtraceback': reprtraceback,
+            "reprtraceback": reprtraceback,
-            'sections': rep.longrepr.sections
+            "sections": rep.longrepr.sections,
         }
 
     import py
+
     d = rep.__dict__.copy()
-    if hasattr(rep.longrepr, 'toterminal'):
+    if hasattr(rep.longrepr, "toterminal"):
-        if hasattr(rep.longrepr, 'reprtraceback') \
+        if hasattr(rep.longrepr, "reprtraceback") and hasattr(
-                and hasattr(rep.longrepr, 'reprcrash'):
+            rep.longrepr, "reprcrash"
-            d['longrepr'] = disassembled_report(rep)
+        ):
+            d["longrepr"] = disassembled_report(rep)
         else:
-            d['longrepr'] = str(rep.longrepr)
+            d["longrepr"] = str(rep.longrepr)
     else:
-        d['longrepr'] = rep.longrepr
+        d["longrepr"] = rep.longrepr
     for name in d:
         if isinstance(d[name], py.path.local):
             d[name] = str(d[name])
@@ -158,8 +181,50 @@ def serialize_report(rep):
     return d
 
 
+def serialize_warning_message(warning_message):
+    if isinstance(warning_message.message, Warning):
+        message_module = type(warning_message.message).__module__
+        message_class_name = type(warning_message.message).__name__
+        message_str = str(warning_message.message)
+        # check now if we can serialize the warning arguments (#349)
+        # if not, we will just use the exception message on the master node
+        try:
+            dumps(warning_message.message.args)
+        except DumpError:
+            message_args = None
+        else:
+            message_args = warning_message.message.args
+    else:
+        message_str = warning_message.message
+        message_module = None
+        message_class_name = None
+        message_args = None
+    if warning_message.category:
+        category_module = warning_message.category.__module__
+        category_class_name = warning_message.category.__name__
+    else:
+        category_module = None
+        category_class_name = None
+
+    result = {
+        "message_str": message_str,
+        "message_module": message_module,
+        "message_class_name": message_class_name,
+        "message_args": message_args,
+        "category_module": category_module,
+        "category_class_name": category_class_name,
+    }
+    # access private _WARNING_DETAILS because the attributes vary between Python versions
+    for attr_name in warning_message._WARNING_DETAILS:
+        if attr_name in ("message", "category"):
+            continue
+        result[attr_name] = getattr(warning_message, attr_name)
+    return result
+
+
 def getinfodict():
     import platform
+
     return dict(
         version=sys.version,
         version_info=tuple(sys.version_info),
@@ -172,30 +237,34 @@ def getinfodict():
 
 def remote_initconfig(option_dict, args):
     from _pytest.config import Config
-    option_dict['plugins'].append("no:terminal")
+
+    option_dict["plugins"].append("no:terminal")
     config = Config.fromdictargs(option_dict, args)
     config.option.looponfail = False
     config.option.usepdb = False
     config.option.dist = "no"
     config.option.distload = False
     config.option.numprocesses = None
+    config.option.maxprocesses = None
     config.args = args
     return config
 
 
-if __name__ == '__channelexec__':
+if __name__ == "__channelexec__":
     channel = channel  # noqa
     workerinput, args, option_dict = channel.receive()
     importpath = os.getcwd()
     sys.path.insert(0, importpath)  # XXX only for remote situations
-    os.environ['PYTHONPATH'] = (
+    os.environ["PYTHONPATH"] = (
-        importpath + os.pathsep +
+        importpath + os.pathsep + os.environ.get("PYTHONPATH", "")
-        os.environ.get('PYTHONPATH', ''))
+    )
-    os.environ['PYTEST_XDIST_WORKER'] = workerinput['workerid']
+    os.environ["PYTEST_XDIST_WORKER"] = workerinput["workerid"]
-    os.environ['PYTEST_XDIST_WORKER_COUNT'] = str(workerinput['workercount'])
+    os.environ["PYTEST_XDIST_WORKER_COUNT"] = str(workerinput["workercount"])
     # os.environ['PYTHONPATH'] = importpath
     import py
+
     config = remote_initconfig(option_dict, args)
+    config._parser.prog = os.path.basename(workerinput["mainargv"][0])
     config.workerinput = workerinput
     config.workeroutput = {}
     # TODO: deprecated name, backward compatibility only. Remove it in future

xdist/report.py

@@ -1,4 +1,4 @@
-import py
+from __future__ import print_function
 from difflib import unified_diff
 
 
@@ -11,16 +11,11 @@ def report_collection_diff(from_collection, to_collection, from_id, to_id):
     if from_collection == to_collection:
         return None
 
-    diff = unified_diff(
+    diff = unified_diff(from_collection, to_collection, fromfile=from_id, tofile=to_id)
-        from_collection,
+    error_message = (
-        to_collection,
+        u"Different tests were collected between {from_id} and {to_id}. "
-        fromfile=from_id,
+        u"The difference is:\n"
-        tofile=to_id,
+        u"{diff}"
-    )
+    ).format(from_id=from_id, to_id=to_id, diff="\n".join(diff))
-    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

xdist/scheduler/__init__.py

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

xdist/scheduler/each.py

@@ -86,10 +86,12 @@ class EachScheduling(object):
                 if deadnode.gateway.spec == node.gateway.spec:
                     dead_collection = self.node2collection[deadnode]
                     if collection != dead_collection:
-                        msg = report_collection_diff(dead_collection,
+                        msg = report_collection_diff(
-                                                     collection,
+                            dead_collection,
-                                                     deadnode.gateway.id,
+                            collection,
-                                                     node.gateway.id)
+                            deadnode.gateway.id,
+                            node.gateway.id,
+                        )
                         self.log(msg)
                         return
                     pending = self._removed2pending.pop(deadnode)
@@ -124,6 +126,7 @@ class EachScheduling(object):
             if not pending:
                 pending[:] = range(len(self.node2collection[node]))
                 node.send_runtest_all()
+                node.shutdown()
             else:
                 node.send_runtest_some(pending)
             self._started.append(node)

xdist/scheduler/load.py

@@ -133,10 +133,9 @@ class LoadScheduling(object):
             assert self.collection
             if collection != self.collection:
                 other_node = next(iter(self.node2collection.keys()))
-                msg = report_collection_diff(self.collection,
+                msg = report_collection_diff(
-                                             collection,
+                    self.collection, collection, other_node.gateway.id, node.gateway.id
-                                             other_node.gateway.id,
+                )
-                                             node.gateway.id)
                 self.log(msg)
                 return
         self.node2collection[node] = list(collection)
@@ -179,6 +178,9 @@ class LoadScheduling(object):
                     return
                 num_send = items_per_node_max - len(node_pending)
                 self._send_tests(node, num_send)
+        else:
+            node.shutdown()
+
         self.log("num items waiting for node:", len(self.pending))
 
     def remove_node(self, node):
@@ -226,7 +228,7 @@ class LoadScheduling(object):
 
         # XXX allow nodes to have different collections
         if not self._check_nodes_have_same_collection():
-            self.log('**Different tests collected, aborting run**')
+            self.log("**Different tests collected, aborting run**")
             return
 
         # Collections are identical, create the index of pending items.
@@ -238,8 +240,7 @@ class LoadScheduling(object):
         # Send a batch of tests to run. If we don't have at least two
         # tests per node, we have to send them all so that we can send
         # shutdown signals and get all nodes working.
-        initial_batch = max(len(self.pending) // 4,
+        initial_batch = max(len(self.pending) // 4, 2 * len(self.nodes))
-                            2 * len(self.nodes))
 
         # distribute tests round-robin up to the batch size
         # (or until we run out)
@@ -271,18 +272,15 @@ class LoadScheduling(object):
         same_collection = True
         for node, collection in node_collection_items[1:]:
             msg = report_collection_diff(
-                col,
+                col, collection, first_node.gateway.id, node.gateway.id
-                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',
+                        node.gateway.id, "failed", longrepr=msg, result=[]
-                        longrepr=msg, result=[])
+                    )
                     self.config.hook.pytest_collectreport(report=rep)
 
         return same_collection

xdist/scheduler/loadfile.py

@@ -1,4 +1,4 @@
-from . import LoadScopeScheduling
+from .loadscope import LoadScopeScheduling
 from py.log import Producer
 
 
@@ -23,7 +23,7 @@ class LoadFileScheduling(LoadScopeScheduling):
     def __init__(self, config, log=None):
         super(LoadFileScheduling, self).__init__(config, log)
         if log is None:
-            self.log = Producer('loadfilesched')
+            self.log = Producer("loadfilesched")
         else:
             self.log = log.loadfilesched
 
@@ -49,4 +49,4 @@ class LoadFileScheduling(LoadScopeScheduling):
             example/loadsuite/test/test_delta.py
             example/loadsuite/epsilon/__init__.py
         """
-        return nodeid.split('::', 1)[0]
+        return nodeid.split("::", 1)[0]

xdist/scheduler/loadscope.py

@@ -93,7 +93,7 @@ class LoadScopeScheduling(object):
         self.registered_collections = OrderedDict()
 
         if log is None:
-            self.log = Producer('loadscopesched')
+            self.log = Producer("loadscopesched")
         else:
             self.log = log.loadscopesched
 
@@ -187,8 +187,7 @@ class LoadScopeScheduling(object):
             break
         else:
             raise RuntimeError(
-                'Unable to identify crashitem on a workload with '
+                "Unable to identify crashitem on a workload with pending items"
-                'pending items'
             )
 
         # Made uncompleted work unit available again
@@ -224,10 +223,7 @@ class LoadScopeScheduling(object):
                 other_node = next(iter(self.registered_collections.keys()))
 
                 msg = report_collection_diff(
-                    self.collection,
+                    self.collection, collection, other_node.gateway.id, node.gateway.id
-                    collection,
-                    other_node.gateway.id,
-                    node.gateway.id
                 )
                 self.log(msg)
                 return
@@ -255,9 +251,7 @@ class LoadScopeScheduling(object):
         scope, work_unit = self.workqueue.popitem(last=False)
 
         # Keep track of the assigned work
-        assigned_to_node = self.assigned_work.setdefault(
+        assigned_to_node = self.assigned_work.setdefault(node, default=OrderedDict())
-            node, default=OrderedDict()
-        )
         assigned_to_node[scope] = work_unit
 
         # Ask the node to execute the workload
@@ -292,14 +286,11 @@ class LoadScopeScheduling(object):
             example/loadsuite/test/test_delta.py::Delta1
             example/loadsuite/epsilon/__init__.py
         """
-        return nodeid.rsplit('::', 1)[0]
+        return nodeid.rsplit("::", 1)[0]
 
     def _pending_of(self, workload):
         """Return the number of pending tests in a workload."""
-        pending = sum(
+        pending = sum(list(scope.values()).count(False) for scope in workload.values())
-            list(scope.values()).count(False)
-            for scope in workload.values()
-        )
         return pending
 
     def _reschedule(self, node):
@@ -315,9 +306,10 @@ class LoadScopeScheduling(object):
 
         # Check that more work is available
         if not self.workqueue:
+            node.shutdown()
             return
 
-        self.log('Number of units waiting for node:', len(self.workqueue))
+        self.log("Number of units waiting for node:", len(self.workqueue))
 
         # Check that the node is almost depleted of work
         # 2: Heuristic of minimum tests to enqueue more work
@@ -348,13 +340,11 @@ class LoadScopeScheduling(object):
 
         # Check that all nodes collected the same tests
         if not self._check_nodes_have_same_collection():
-            self.log('**Different tests collected, aborting run**')
+            self.log("**Different tests collected, aborting run**")
             return
 
         # Collections are identical, create the final list of items
-        self.collection = list(
+        self.collection = list(next(iter(self.registered_collections.values())))
-            next(iter(self.registered_collections.values()))
-        )
         if not self.collection:
             return
 
@@ -368,12 +358,12 @@ class LoadScopeScheduling(object):
         extra_nodes = len(self.nodes) - len(self.workqueue)
 
         if extra_nodes > 0:
-            self.log('Shuting down {0} nodes'.format(extra_nodes))
+            self.log("Shuting down {0} nodes".format(extra_nodes))
 
             for _ in range(extra_nodes):
                 unused_node, assigned = self.assigned_work.popitem(last=True)
 
-                self.log('Shuting down unused node {0}'.format(unused_node))
+                self.log("Shuting down unused node {0}".format(unused_node))
                 unused_node.shutdown()
 
         # Assign initial workload
@@ -402,10 +392,7 @@ class LoadScopeScheduling(object):
 
         for node, collection in node_collection_items[1:]:
             msg = report_collection_diff(
-                col,
+                col, collection, first_node.gateway.id, node.gateway.id
-                collection,
-                first_node.gateway.id,
-                node.gateway.id,
             )
             if not msg:
                 continue
@@ -416,12 +403,7 @@ class LoadScopeScheduling(object):
             if self.config is None:
                 continue
 
-            rep = CollectReport(
+            rep = CollectReport(node.gateway.id, "failed", longrepr=msg, result=[])
-                node.gateway.id,
-                'failed',
-                longrepr=msg,
-                result=[]
-            )
             self.config.hook.pytest_collectreport(report=rep)
 
         return same_collection

xdist/workermanage.py

@@ -1,11 +1,14 @@
+from __future__ import print_function
 import fnmatch
 import os
 import re
+import sys
 import threading
 
 import py
 import pytest
 import execnet
+
 import xdist.remote
 
 from _pytest import runner  # XXX load dynamically
@@ -20,16 +23,17 @@ def parse_spec_config(config):
         except ValueError:
             xspeclist.append(xspec)
         else:
-            xspeclist.extend([xspec[i + 1:]] * num)
+            xspeclist.extend([xspec[i + 1 :]] * num)
     if not xspeclist:
         raise pytest.UsageError(
-            "MISSING test execution (tx) nodes: please specify --tx")
+            "MISSING test execution (tx) nodes: please specify --tx"
+        )
     return xspeclist
 
 
 class NodeManager(object):
     EXIT_TIMEOUT = 10
-    DEFAULT_IGNORES = ['.*', '*.pyc', '*.pyo', '*~']
+    DEFAULT_IGNORES = [".*", "*.pyc", "*.pyo", "*~"]
 
     def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
         self.config = config
@@ -48,7 +52,7 @@ class NodeManager(object):
             self.specs.append(spec)
         self.roots = self._getrsyncdirs()
         self.rsyncoptions = self._getrsyncoptions()
-        self._rsynced_specs = py.builtin.set()
+        self._rsynced_specs = set()
 
     def rsync_roots(self, gateway):
         """Rsync the set of roots to the node's gateway cwd."""
@@ -57,8 +61,7 @@ class NodeManager(object):
                 self.rsync(gateway, root, **self.rsyncoptions)
 
     def setup_nodes(self, putevent):
-        self.config.hook.pytest_xdist_setupnodes(config=self.config,
+        self.config.hook.pytest_xdist_setupnodes(config=self.config, specs=self.specs)
-                                                 specs=self.specs)
         self.trace("setting up nodes")
         nodes = []
         for spec in self.specs:
@@ -70,7 +73,7 @@ class NodeManager(object):
         self.config.hook.pytest_xdist_newgateway(gateway=gw)
         self.rsync_roots(gw)
         node = WorkerController(self, gw, self.config, putevent)
-        gw.node = node          # keep the node alive
+        gw.node = node  # keep the node alive
         node.setup()
         self.trace("started node %r" % node)
         return node
@@ -89,6 +92,7 @@ class NodeManager(object):
             return []
         import pytest
         import _pytest
+
         pytestpath = pytest.__file__.rstrip("co")
         pytestdir = py.path.local(_pytest.__file__).dirpath()
         config = self.config
@@ -112,10 +116,7 @@ class NodeManager(object):
         ignores += self.config.option.rsyncignore
         ignores += self.config.getini("rsyncignore")
 
-        return {
+        return {"ignores": ignores, "verbose": self.config.option.verbose}
-            '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."""
@@ -127,9 +128,12 @@ class NodeManager(object):
         if spec.popen and not spec.chdir:
             # XXX This assumes that sources are python-packages
             #     and that adding the basedir does not hurt.
-            gateway.remote_exec("""
+            gateway.remote_exec(
+                """
                 import sys ; sys.path.insert(0, %r)
-            """ % os.path.dirname(str(source))).waitclose()
+            """
+                % os.path.dirname(str(source))
+            ).waitclose()
             return
         if (spec, source) in self._rsynced_specs:
             return
@@ -137,28 +141,24 @@ class NodeManager(object):
         def finished():
             if notify:
                 notify("rsyncrootready", spec, source)
+
         rsync.add_target_host(gateway, finished=finished)
         self._rsynced_specs.add((spec, source))
-        self.config.hook.pytest_xdist_rsyncstart(
+        self.config.hook.pytest_xdist_rsyncstart(source=source, gateways=[gateway])
-            source=source,
-            gateways=[gateway],
-        )
         rsync.send()
-        self.config.hook.pytest_xdist_rsyncfinish(
+        self.config.hook.pytest_xdist_rsyncfinish(source=source, gateways=[gateway])
-            source=source,
-            gateways=[gateway],
-        )
 
 
 class HostRSync(execnet.RSync):
     """ RSyncer that filters out common files
     """
+
     def __init__(self, sourcedir, *args, **kwargs):
         self._synced = {}
         self._ignores = []
-        ignores = kwargs.pop('ignores', None) or []
+        ignores = kwargs.pop("ignores", None) or []
         for x in ignores:
-            x = getattr(x, 'strpath', x)
+            x = getattr(x, "strpath", x)
             self._ignores.append(re.compile(fnmatch.translate(x)))
         super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
 
@@ -172,16 +172,15 @@ class HostRSync(execnet.RSync):
 
     def add_target_host(self, gateway, finished=None):
         remotepath = os.path.basename(self._sourcedir)
-        super(HostRSync, self).add_target(gateway, remotepath,
+        super(HostRSync, self).add_target(
-                                          finishedcallback=finished,
+            gateway, remotepath, finishedcallback=finished, delete=True
-                                          delete=True,)
+        )
 
     def _report_send_file(self, gateway, modified_rel_path):
         if self._verbose:
             path = os.path.basename(self._sourcedir) + "/" + modified_rel_path
             remotepath = gateway.spec.chdir
-            py.builtin.print_('%s:%s <= %s' %
+            print("%s:%s <= %s" % (gateway.spec, remotepath, path))
-                              (gateway.spec, remotepath, path))
 
 
 def make_reltoroot(roots, args):
@@ -210,11 +209,13 @@ class WorkerController(object):
         self.putevent = putevent
         self.gateway = gateway
         self.config = config
-        self.workerinput = {'workerid': gateway.id,
+        self.workerinput = {
-                            'workercount': len(nodemanager.specs),
+            "workerid": gateway.id,
-                            'slaveid': gateway.id,
+            "workercount": len(nodemanager.specs),
-                            'slavecount': len(nodemanager.specs)
+            "slaveid": gateway.id,
-                            }
+            "slavecount": len(nodemanager.specs),
+            "mainargv": sys.argv,
+        }
         # TODO: deprecated name, backward compatibility only. Remove it in future
         self.slaveinput = self.workerinput
         self._down = False
@@ -224,7 +225,7 @@ class WorkerController(object):
             py.log.setconsumer(self.log._keywords, None)
 
     def __repr__(self):
-        return "<%s %s>" % (self.__class__.__name__, self.gateway.id,)
+        return "<%s %s>" % (self.__class__.__name__, self.gateway.id)
 
     @property
     def shutting_down(self):
@@ -239,24 +240,22 @@ class WorkerController(object):
         option_dict = vars(self.config.option)
         if spec.popen:
             name = "popen-%s" % self.gateway.id
-            if hasattr(self.config, '_tmpdirhandler'):
+            if hasattr(self.config, "_tmpdirhandler"):
                 basetemp = self.config._tmpdirhandler.getbasetemp()
-                option_dict['basetemp'] = str(basetemp.join(name))
+                option_dict["basetemp"] = str(basetemp.join(name))
         self.config.hook.pytest_configure_node(node=self)
         self.channel = self.gateway.remote_exec(xdist.remote)
         self.channel.send((self.workerinput, args, option_dict))
         if self.putevent:
-            self.channel.setcallback(
+            self.channel.setcallback(self.process_from_remote, endmarker=self.ENDMARK)
-                self.process_from_remote,
-                endmarker=self.ENDMARK)
 
     def ensure_teardown(self):
-        if hasattr(self, 'channel'):
+        if hasattr(self, "channel"):
             if not self.channel.isclosed():
                 self.log("closing", self.channel)
                 self.channel.close()
             # del self.channel
-        if hasattr(self, 'gateway'):
+        if hasattr(self, "gateway"):
             self.log("exiting", self.gateway)
             self.gateway.exit()
             # del self.gateway
@@ -265,13 +264,13 @@ class WorkerController(object):
         self.sendcommand("runtests", indices=indices)
 
     def send_runtest_all(self):
-        self.sendcommand("runtests_all",)
+        self.sendcommand("runtests_all")
 
     def shutdown(self):
         if not self._down:
             try:
                 self.sendcommand("shutdown")
-            except IOError:
+            except (IOError, OSError):
                 pass
             self._shutdown_sent = True
 
@@ -308,33 +307,52 @@ class WorkerController(object):
                 self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "workerfinished":
                 self._down = True
-                self.workeroutput = self.slaveoutput = kwargs['workeroutput']
+                self.workeroutput = self.slaveoutput = kwargs["workeroutput"]
                 self.notify_inproc("workerfinished", node=self)
             elif eventname in ("logstart", "logfinish"):
                 self.notify_inproc(eventname, node=self, **kwargs)
-            elif eventname in (
+            elif eventname in ("testreport", "collectreport", "teardownreport"):
-                    "testreport", "collectreport", "teardownreport"):
                 item_index = kwargs.pop("item_index", None)
-                rep = unserialize_report(eventname, kwargs['data'])
+                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'])
+                self.notify_inproc(eventname, node=self, ids=kwargs["ids"])
             elif eventname == "runtest_protocol_complete":
                 self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "logwarning":
-                self.notify_inproc(eventname, message=kwargs['message'],
+                self.notify_inproc(
-                                   code=kwargs['code'], nodeid=kwargs['nodeid'],
+                    eventname,
-                                   fslocation=kwargs['nodeid'])
+                    message=kwargs["message"],
+                    code=kwargs["code"],
+                    nodeid=kwargs["nodeid"],
+                    fslocation=kwargs["nodeid"],
+                )
+            elif eventname == "warning_captured":
+                warning_message = unserialize_warning_message(
+                    kwargs["warning_message_data"]
+                )
+                self.notify_inproc(
+                    eventname,
+                    warning_message=warning_message,
+                    when=kwargs["when"],
+                    item=kwargs["item"],
+                )
             else:
                 raise ValueError("unknown event: %s" % (eventname,))
         except KeyboardInterrupt:
             # should not land in receiver-thread
             raise
         except:  # noqa
-            excinfo = py.code.ExceptionInfo()
+            from _pytest._code import ExceptionInfo
-            py.builtin.print_("!" * 20, excinfo)
+
+            # ExceptionInfo API changed in pytest 4.1
+            if hasattr(ExceptionInfo, "from_current"):
+                excinfo = ExceptionInfo.from_current()
+            else:
+                excinfo = ExceptionInfo()
+            print("!" * 20, excinfo)
             self.config.notify_exception(excinfo)
             self.shutdown()
             self.notify_inproc("errordown", node=self, error=excinfo)
@@ -349,56 +367,56 @@ def unserialize_report(name, reportdict):
             ReprFileLocation,
             ReprFuncArgs,
             ReprLocals,
-            ReprTraceback
+            ReprTraceback,
         )
-        if reportdict['longrepr']:
-            if 'reprcrash' in reportdict['longrepr'] and 'reprtraceback' in reportdict['longrepr']:
 
-                reprtraceback = reportdict['longrepr']['reprtraceback']
+        if reportdict["longrepr"]:
-                reprcrash = reportdict['longrepr']['reprcrash']
+            if (
+                "reprcrash" in reportdict["longrepr"]
+                and "reprtraceback" in reportdict["longrepr"]
+            ):
+
+                reprtraceback = reportdict["longrepr"]["reprtraceback"]
+                reprcrash = reportdict["longrepr"]["reprcrash"]
 
                 unserialized_entries = []
                 reprentry = None
-                for entry_data in reprtraceback['reprentries']:
+                for entry_data in reprtraceback["reprentries"]:
-                    data = entry_data['data']
+                    data = entry_data["data"]
-                    entry_type = entry_data['type']
+                    entry_type = entry_data["type"]
-                    if entry_type == 'ReprEntry':
+                    if entry_type == "ReprEntry":
                         reprfuncargs = None
                         reprfileloc = None
                         reprlocals = None
-                        if data['reprfuncargs']:
+                        if data["reprfuncargs"]:
-                            reprfuncargs = ReprFuncArgs(
+                            reprfuncargs = ReprFuncArgs(**data["reprfuncargs"])
-                                **data['reprfuncargs'])
+                        if data["reprfileloc"]:
-                        if data['reprfileloc']:
+                            reprfileloc = ReprFileLocation(**data["reprfileloc"])
-                            reprfileloc = ReprFileLocation(
+                        if data["reprlocals"]:
-                                **data['reprfileloc'])
+                            reprlocals = ReprLocals(data["reprlocals"]["lines"])
-                        if data['reprlocals']:
-                            reprlocals = ReprLocals(
-                                data['reprlocals']['lines'])
 
                         reprentry = ReprEntry(
-                            lines=data['lines'],
+                            lines=data["lines"],
                             reprfuncargs=reprfuncargs,
                             reprlocals=reprlocals,
                             filelocrepr=reprfileloc,
-                            style=data['style']
+                            style=data["style"],
                         )
-                    elif entry_type == 'ReprEntryNative':
+                    elif entry_type == "ReprEntryNative":
-                        reprentry = ReprEntryNative(data['lines'])
+                        reprentry = ReprEntryNative(data["lines"])
                     else:
-                        report_unserialization_failure(
+                        report_unserialization_failure(entry_type, name, reportdict)
-                            entry_type, name, reportdict)
                     unserialized_entries.append(reprentry)
-                reprtraceback['reprentries'] = unserialized_entries
+                reprtraceback["reprentries"] = unserialized_entries
 
                 exception_info = ReprExceptionInfo(
                     reprtraceback=ReprTraceback(**reprtraceback),
                     reprcrash=ReprFileLocation(**reprcrash),
                 )
 
-                for section in reportdict['longrepr']['sections']:
+                for section in reportdict["longrepr"]["sections"]:
                     exception_info.addsection(*section)
-                reportdict['longrepr'] = exception_info
+                reportdict["longrepr"] = exception_info
         return reportdict
 
     if name == "testreport":
@@ -407,15 +425,57 @@ def unserialize_report(name, reportdict):
         return runner.CollectReport(**assembled_report(reportdict))
 
 
+def unserialize_warning_message(data):
+    import warnings
+    import importlib
+
+    if data["message_module"]:
+        mod = importlib.import_module(data["message_module"])
+        cls = getattr(mod, data["message_class_name"])
+        message = None
+        if data["message_args"] is not None:
+            try:
+                message = cls(*data["message_args"])
+            except TypeError:
+                pass
+        if message is None:
+            # could not recreate the original warning instance;
+            # create a generic Warning instance with the original
+            # message at least
+            message_text = "{mod}.{cls}: {msg}".format(
+                mod=data["message_module"],
+                cls=data["message_class_name"],
+                msg=data["message_str"],
+            )
+            message = Warning(message_text)
+    else:
+        message = data["message_str"]
+
+    if data["category_module"]:
+        mod = importlib.import_module(data["category_module"])
+        category = getattr(mod, data["category_class_name"])
+    else:
+        category = None
+
+    kwargs = {"message": message, "category": category}
+    # access private _WARNING_DETAILS because the attributes vary between Python versions
+    for attr_name in warnings.WarningMessage._WARNING_DETAILS:
+        if attr_name in ("message", "category"):
+            continue
+        kwargs[attr_name] = data[attr_name]
+
+    return warnings.WarningMessage(**kwargs)
+
+
 def report_unserialization_failure(type_name, report_name, reportdict):
     from pprint import pprint
-    url = 'https://github.com/pytest-dev/pytest-xdist/issues'
+
+    url = "https://github.com/pytest-dev/pytest-xdist/issues"
     stream = py.io.TextIO()
-    pprint('-' * 100, stream=stream)
+    pprint("-" * 100, stream=stream)
-    pprint('INTERNALERROR: Unknown entry type returned: %s' % type_name,
+    pprint("INTERNALERROR: Unknown entry type returned: %s" % type_name, stream=stream)
-           stream=stream)
+    pprint("report_name: %s" % report_name, stream=stream)
-    pprint('report_name: %s' % report_name, stream=stream)
     pprint(reportdict, stream=stream)
-    pprint('Please report this bug at %s' % url, stream=stream)
+    pprint("Please report this bug at %s" % url, stream=stream)
-    pprint('-' * 100, stream=stream)
+    pprint("-" * 100, stream=stream)
     assert 0, stream.getvalue()