Prepare release 1.19.1

Fix crash when transferring internal pytest warnings from workers to the master node

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,17 @@
+Thanks for submitting a PR, your contribution is really appreciated!
+
+Here's a quick checklist that should be present in PRs:
+
+- [ ] Make sure to include reasonable tests for your change if necessary
+
+- [ ] We use [towncrier](https://pypi.python.org/pypi/towncrier) for changelog management, so please add a *news* file into the `changelog` folder following these guidelines:
+  * Name it `$issue_id.$type` for example `588.bugfix`;
+  * If you don't have an issue_id change it to the PR id after creating it
+  * Ensure type is one of `removal`, `feature`, `bugfix`, `vendor`, `doc` or `trivial`
+  * Make sure to use full sentences with correct case and punctuation, for example:
+
+    ```
+    Fix issue with non-ascii contents in doctest text files.
+    ```
+
+

.travis.yml

@@ -1,30 +1,25 @@
 sudo: false
 language: python
 python:
+- '2.6'
+- '2.7'
+- '3.4'
 - '3.5'
-install: pip install -U tox setuptools_scm
+- '3.6'
 env:
-  matrix:
-  - TESTENV=py26-pytest26
-  - TESTENV=py26-pytest27
-  - TESTENV=py26-pytest28
-  - TESTENV=py26-pytest29
-  - TESTENV=py27-pytest26
-  - TESTENV=py27-pytest27
-  - TESTENV=py27-pytest28
-  - TESTENV=py27-pytest29
-  - TESTENV=py34-pytest26
-  - TESTENV=py34-pytest27
-  - TESTENV=py34-pytest28
-  - TESTENV=py34-pytest29
-  - TESTENV=py35-pytest27
-  - TESTENV=py35-pytest28
-  - TESTENV=py35-pytest29
-  - TESTENV=py27-pytest28-pexpect
-  - TESTENV=py35-pytest28-pexpect
-  - TESTENV=flakes
-  - TESTENV=readme
-script: tox --recreate -e $TESTENV
+- TOXENV=py-pytest30
+- TOXENV=py-pytest31
+- TOXENV=py-pytest32
+install: pip install tox setuptools_scm
+matrix:
+  include:
+  # note: please use "tox --listenvs" to populate the build matrix
+  - python: "3.6"
+    env: TOXENV=flakes
+  - python: "3.6"
+    env: TOXENV=readme
+
+script: tox
 notifications:
   irc:
     channels:
@@ -37,6 +32,7 @@ notifications:
 deploy:
   provider: pypi
   user: ronny
+  skip_upload_docs: true
   password:
     secure: cxmSDho5d+PYKEM4ZCg8ms1P4lzhYkrw6fEOm2HtTcsuCyY6aZMSgImWAnEYbJHSkdzgcxlXK9UKJ9B0YenXmBCkAr7UjdnpNXNmkySr0sYzlH/sfqt/dDATCHFaRKxnkOSOVywaDYhT9n8YudbXI77pXwD12i/CeSSJDbHhsu0JYUfAcb+D6YjRYoA2SEGCnzSzg+gDDfwXZx4ZiODCGLVwieNp1klCg88YROUE1BaYYNuUOONvfXX8+TWowbCF6ChH1WL/bZ49OStEYQNuYxZQZr4yClIqu9VJbchrU8j860K9ott2kkGTgfB/dDrQB/XncBubyIX9ikzCQAmmBXWAI3eyvWLPDk2Jz7kW2l2RT7syct80tCq3JhvQ1qdwr5ap7siocTLgnBW0tF4tkHSTFN3510fkc43npnp6FThebESQpnI24vqpwJ9hI/kW5mYi014Og2E/cpCXnz2XO8iZPDbqAMQpDsqEQoyhfGNgPTGp4K30TxRtwZBI5hHhDKnnR16fXtRgt1gYPvz/peUQvvpOm4JzIzGXPzluuutpnCBy75v5+oiwT3YRrLL/Meims9FtDDXL3qQubAE/ezIOOpm0N5XXV8DxIom8EN71yq5ab1tqhM+tBX7owRjy4FR4If2Q8feBdmTuh26DIQt/y+qSG8VkB9Sw/JCjc7c=
   on:

CHANGELOG.rst

@@ -1,3 +1,139 @@
+pytest-xdist 1.19.1 (2017-08-10)
+================================
+
+Bug Fixes
+---------
+
+- Fix crash when transferring internal pytest warnings from workers to the
+  master node. (`#214 <https://github.com/pytest-dev/pytest-
+  xdist/issues/214>`_)
+
+
+pytest-xdist 1.19.0 (2017-08-09)
+================================
+
+Deprecations and Removals
+-------------------------
+
+- ``--boxed`` functionality has been moved to a separate plugin, `pytest-forked
+  <https://github.com/pytest-dev/pytest-forked>`_. This release now depends on
+  `` pytest-forked`` and provides ``--boxed`` as a backward compatibility
+  option. (`#1 <https://github.com/pytest-dev/pytest-xdist/issues/1>`_)
+
+
+Features
+--------
+
+- New ``--dist=loadscope`` option: sends group of related tests to the same
+  worker. Tests are grouped by module for test functions and by class for test
+  methods. See ``README.rst`` for more information. (`#191 <https://github.com
+  /pytest-dev/pytest-xdist/issues/191>`_)
+
+- Warnings are now properly transferred from workers to the master node. (`#92
+  <https://github.com/pytest-dev/pytest-xdist/issues/92>`_)
+
+
+Bug Fixes
+---------
+
+- Fix serialization of native tracebacks (``--tb=native``). (`#196
+  <https://github.com/pytest-dev/pytest-xdist/issues/196>`_)
+
+
+pytest-xdist 1.18.2 (2017-07-28)
+================================
+
+Bug Fixes
+---------
+
+- Removal of unnecessary dependency on incorrect version of py. (`#105
+  <https://github.com/pytest-dev/pytest-xdist/issues/105>`_)
+
+- Fix bug in internal event-loop error handler in the master node. This bug
+  would shadow the original errors making extremely hard/impossible for users
+  to diagnose the problem properly. (`#175 <https://github.com/pytest-
+  dev/pytest-xdist/issues/175>`_)
+
+
+pytest-xdist 1.18.1 (2017-07-05)
+================================
+
+Bug Fixes
+---------
+
+- Fixed serialization of ``longrepr.sections`` during error reporting from
+  workers. (`#171 <https://github.com/pytest-dev/pytest-xdist/issues/171>`_)
+
+- Fix ``ReprLocal`` not being unserialized breaking --showlocals usages. (`#176
+  <https://github.com/pytest-dev/pytest-xdist/issues/176>`_)
+
+
+pytest-xdist 1.18.0 (2017-06-26)
+================================
+
+- ``pytest-xdist`` now requires ``pytest>=3.0.0``.
+
+Features
+--------
+
+- Add long option `--numprocesses` as alternative for `-n`. (#168)
+
+
+Bug Fixes
+---------
+
+- Fix serialization and deserialization dropping longrepr details. (#133)
+
+
+pytest-xdist 1.17.1 (2017-06-10)
+================================
+
+Bug Fixes
+---------
+
+- Hot fix release reverting the change introduced by #124, unfortunately it
+  broke a number of test suites so we are reversing this change while we
+  investigate the problem. (#157)
+
+
+Improved Documentation
+----------------------
+
+- Introduced ``towncrier`` for ``CHANGELOG`` management. (#154)
+
+- Added ``HOWTORELEASE`` documentation. (#155)
+
+
+..
+    You should *NOT* be adding new change log entries to this file, this
+    file is managed by towncrier. You *may* edit previous change logs to
+    fix problems like typo corrections or such.
+    To add a new change log entry, please see
+    https://pip.pypa.io/en/latest/development/#adding-a-news-entry
+    We named the news folder ``changelog``
+
+.. towncrier release notes start
+
+
+1.17.0
+------
+
+- fix #124: xdist would mark test as complete after 'call' step. As a result,
+  xdist could identify the wrong test as failing when test crashes at teardown.
+  To address this issue, xdist now marks test as complete at teardown.
+
+1.16.0
+------
+
+- ``pytest-xdist`` now requires pytest 2.7 or later.
+
+- Add ``worker_id`` attribute in the TestReport
+
+- new hook: ``pytest_xdist_make_scheduler(config, log)``, can return custom tests items
+  distribution logic implementation. You can take a look at built-in ``LoadScheduling``
+  and ``EachScheduling`` implementations. Note that required scheduler class public
+  API may change in next ``pytest-xdist`` versions.
+
 1.15.0
 ------
 

HOWTORELEASE.rst

@@ -0,0 +1,45 @@
+======================
+Releasing pytest-xdist
+======================
+
+This document describes the steps to make a new ``pytest-xdist`` release.
+
+Version
+-------
+
+``master`` should always be green and a potential release candidate. ``pytest-xdist`` follows
+semantic versioning, so given that the current version is ``X.Y.Z``, to find the next version number
+one needs to look at the ``changelog`` folder:
+
+- If there is any file named ``*.feature``, then we must make a new **minor** release: next
+  release will be ``X.Y+1.0``.
+
+- Otherwise it is just a **bug fix** release: ``X.Y.Z+1``.
+
+
+Steps
+-----
+
+To publish a new release ``X.Y.Z``, the steps are as follows:
+
+#. Create a new branch named ``release-X.Y.Z`` from the latest ``master``.
+
+#. Install ``pytest-xdist`` and dev requirements in a virtualenv::
+
+    $ pip install -e . -r dev-requirements.txt
+
+#. Update ``CHANGELOG.rst`` file by running::
+
+    $ towncrier --version X.Y.Z
+
+   It might ask for confirmation to remove news fragments; answer yes.
+
+#. Commit and push the branch for review.
+
+#. Once PR is **green** and **approved**, create and push a tag::
+
+    $ export VERSION=X.Y.Z
+    $ git tag v$VERSION release-$VERSION
+    $ git push git@github.com:pytest-dev/pytest-xdist.git v$VERSION
+
+That will build the package and publish it on ``PyPI`` automatically.

MANIFEST.in

@@ -1,7 +0,0 @@
-include CHANGELOG
-include LICENSE 
-include README.txt
-include setup.py
-include tox.ini
-graft testing
-prune .git

OVERVIEW.md

@@ -35,6 +35,8 @@ The execution flow is:
    fashion. The rest of the tests will be distributed later as **workers**
    finish tests (see below).
 
+1. Note that `pytest_xdist_make_scheduler` hook can be used to implement custom tests distribution logic.
+   
 1. **workers** re-implement `pytest_runtestloop`: pytest's default implementation
    basically loops over all collected items in the `session` object and executes
    the `pytest_runtest_protocol` for each test item, but in xdist **workers** sit idly 

README.rst

@@ -1,10 +1,14 @@
 
+
+.. image:: http://img.shields.io/pypi/v/pytest-xdist.svg
+    :target: https://pypi.python.org/pypi/pytest-xdist
+
+.. image:: https://anaconda.org/conda-forge/pytest-xdist/badges/version.svg
+    :target: https://anaconda.org/conda-forge/pytest-xdist
+
 .. image:: https://travis-ci.org/pytest-dev/pytest-xdist.svg?branch=master
     :target: https://travis-ci.org/pytest-dev/pytest-xdist
 
-.. image:: http://img.shields.io/pypi/v/pytest-xdist.svg
-   :target: https://pypi.python.org/pypi/pytest-xdist
-
 .. image:: https://ci.appveyor.com/api/projects/status/56eq1a1avd4sdd7e/branch/master?svg=true
     :target: https://ci.appveyor.com/project/pytestbot/pytest-xdist
 
@@ -18,8 +22,6 @@ test execution modes:
   those for a combined test run.  This allows to speed up
   development or to use special resources of `remote machines`_.
 
-* ``--boxed``: (not available on Windows) run each test in a boxed_
-  subprocess to survive ``SEGFAULTS`` or otherwise dying processes
 
 * ``--looponfail``: run your tests repeatedly in a subprocess.  After each run
   py.test waits until a file in your project changes and then re-runs
@@ -40,13 +42,9 @@ Installation
 
 Install the plugin with::
 
-    easy_install pytest-xdist
-
-    # or
-
     pip install pytest-xdist
 
-or use the package in develope/in-place mode with
+or use the package in develop/in-place mode with
 a checkout of the `pytest-xdist repository`_ ::
 
     python setup.py develop
@@ -188,6 +186,7 @@ at once.   The specifications strings use the `xspec syntax`_.
 Identifying the worker process during a test
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
+*New in version 1.15.*
 
 If you need to determine the identity of a worker process in
 a test or fixture, you may use the ``worker_id`` fixture to do so:
@@ -209,7 +208,9 @@ defined:
 * ``PYTEST_XDIST_WORKER_COUNT``: the total number of workers in this session,
   e.g., ``"4"`` when ``-n 4`` is given in the command-line.
 
-*New in version 1.15.*
+The information about the worker_id in a test is stored in the TestReport as
+well, under worker_id attribute.
+
 
 Specifying test exec environments in an ini file
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -236,6 +237,83 @@ and then just type::
 
 to run tests in each of the environments.
 
+
+Sending groups of related tests to the same worker
+++++++++++++++++++++++++++++++++++++++++++++++++++
+
+*New in version 1.19.*
+
+.. note::
+    This is an **experimental** feature: the actual functionality will
+    likely stay the same, but the CLI might change slightly in future versions.
+
+You can send groups of related tests to the same worker by using the
+``--dist=loadscope`` option. Tests will be grouped by **module**
+for *test functions* and by **class** for *test methods*.
+
+For example, consider this two test files:
+
+.. code-block:: python
+
+    # content of test_container.py
+    import pytest
+
+    def test_container_startup():
+        pass
+
+    def test_container_logging():
+        pass
+
+    @pytest.mark.parametrize('methods', ['ssh', 'http'])
+    def test_container_communication(methods):
+        pass
+
+    # content of test_io.py
+    class TestHDF:
+
+        def test_listing(self):
+            pass
+
+        def test_search(self):
+            pass
+
+
+    class TestXML:
+
+        def test_listing(self):
+            pass
+
+        def test_search(self):
+            pass
+
+
+By executing ``pytest -v --dist=loadscope -n4`` you might get this output
+(sorted by worker for readability)::
+
+    ============================= test session starts =============================
+    <skip header>
+    gw0 [8] / gw1 [8] / gw2 [8] / gw3 [8]
+    scheduling tests via LoadScopeScheduling
+
+    [gw0] PASSED test_container.py::test_container_communication[http]
+    [gw0] PASSED test_container.py::test_container_communication[ssh]
+    [gw0] PASSED test_container.py::test_container_logging
+    [gw0] PASSED test_container.py::test_container_startup
+    [gw1] PASSED test_io.py::TestHDF::test_listing
+    [gw1] PASSED test_io.py::TestHDF::test_search
+    [gw2] PASSED test_io.py::TestXML::test_listing
+    [gw2] PASSED test_io.py::TestXML::test_search
+
+    ========================== 8 passed in 0.56 seconds ===========================
+
+As you can see, all test functions from ``test_container.py`` executed on
+the same worker ``gw0``, while the test methods from classes ``TestHDF`` and
+``TestXML`` executed in workers ``gw1`` and ``gw2`` respectively.
+
+Currently the groupings can't be customized, with grouping by class takes
+priority over grouping by module.
+
+
 Specifying "rsync" dirs in an ini-file
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
@@ -254,9 +332,3 @@ where the configuration file was found.
 .. _`pytest-xdist`: http://pypi.python.org/pypi/pytest-xdist
 .. _`pytest-xdist repository`: https://github.com/pytest-dev/pytest-xdist
 .. _`pytest`: http://pytest.org
-
-Issue and Bug Tracker
-------------------------
-
-Please use the `pytest issue tracker <https://github.com/pytest-dev/pytest/issues>`_
-for reporting bugs in this plugin.

appveyor.yml

@@ -1,3 +1,16 @@
+environment:
+  matrix:
+  # note: please use "tox --listenvs" to populate the build matrix
+  - TOXENV: "py26-pytest30"
+  - TOXENV: "py27-pytest30"
+  - TOXENV: "py34-pytest30"
+  - TOXENV: "py35-pytest30"
+  - TOXENV: "py36-pytest30"
+  - TOXENV: "py27-pytest30-pexpect"
+  - TOXENV: "py35-pytest30-pexpect"
+  - TOXENV: "flakes"
+  - TOXENV: "readme"
+
 install:
   - C:\Python35\python -m pip install tox setuptools_scm
 

changelog/_template.rst

@@ -0,0 +1,39 @@
+{% for section in sections %}
+{% set underline = "-" %}
+{% if section %}
+{{section}}
+{{ underline * section|length }}{% set underline = "~" %}
+
+{% endif %}
+{% if sections[section] %}
+{% for category, val in definitions.items() if category in sections[section] %}
+
+{{ definitions[category]['name'] }}
+{{ underline * definitions[category]['name']|length }}
+
+{% 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 %}
+
+
+{% endfor %}
+{% else %}
+- {{ sections[section][category]['']|sort|join(', ') }}
+
+
+{% endif %}
+{% if sections[section][category]|length == 0 %}
+
+No significant changes.
+
+
+{% else %}
+{% endif %}
+{% endfor %}
+{% else %}
+
+No significant changes.
+
+
+{% endif %}
+{% endfor %}

dev-requirements.txt

@@ -0,0 +1 @@
+towncrier

example/boxed.txt

@@ -1,3 +1,9 @@
+.. note::
+
+  Since 1.19.0, the actual implementation of the ``--boxed`` option has been moved to a 
+  separate plugin, `pytest-forked <https://github.com/pytest-dev/pytest-forked>`_ 
+  which can be installed independently. The ``--boxed`` command-line options remains 
+  for backward compatibility reasons.
 
 
 If your testing involves C or C++ libraries you might have to deal
@@ -19,7 +25,7 @@ to run each test in a controlled subprocess.  Here is a basic example::
 
 If you run this with::
 
-    $ py.test --boxed
+    $ py.test -n1
     =========================== test session starts ============================
     platform linux2 -- Python 2.7.3 -- pytest-2.3.0.dev8
     plugins: xdist, bugzilla, cache, oejskit, cli, pep8, cov
@@ -40,7 +46,7 @@ You'll see that a couple of tests are reported as crashing, indicated
 by lower-case ``f`` and the respective failure summary. You can also use
 the xdist-provided parallelization feature to speed up your testing::
 
-    $ py.test --boxed -n3
+    $ py.test -n3
     =========================== test session starts ============================
     platform linux2 -- Python 2.7.3 -- pytest-2.3.0.dev8
     plugins: xdist, bugzilla, cache, oejskit, cli, pep8, cov

example/loadscope/epsilon/__init__.py

@@ -0,0 +1,37 @@
+def epsilon1(arg1, arg2=1000):
+    """Do epsilon1
+
+    Usage:
+
+    >>> epsilon1(10, 20)
+    40
+    >>> epsilon1(30)
+    1040
+    """
+    return arg1 + arg2 + 10
+
+
+def epsilon2(arg1, arg2=1000):
+    """Do epsilon2
+
+    Usage:
+
+    >>> epsilon2(10, 20)
+    -20
+    >>> epsilon2(30)
+    -980
+    """
+    return arg1 - arg2 - 10
+
+
+def epsilon3(arg1, arg2=1000):
+    """Do epsilon3
+
+    Usage:
+
+    >>> epsilon3(10, 20)
+    200
+    >>> epsilon3(30)
+    30000
+    """
+    return arg1 * arg2

example/loadscope/requirements.txt

@@ -0,0 +1,3 @@
+ipdb
+pytest
+../../

example/loadscope/test/test_alpha.py

@@ -0,0 +1,51 @@
+from time import sleep
+
+
+def test_alpha0():
+    sleep(5)
+    assert True
+
+
+def test_alpha1():
+    sleep(5)
+    assert True
+
+
+def test_alpha2():
+    sleep(5)
+    assert True
+
+
+def test_alpha3():
+    sleep(5)
+    assert True
+
+
+def test_alpha4():
+    sleep(5)
+    assert True
+
+
+def test_alpha5():
+    sleep(5)
+    assert True
+
+
+def test_alpha6():
+    sleep(5)
+    assert True
+
+
+def test_alpha7():
+    sleep(5)
+    assert True
+
+
+def test_alpha8():
+    sleep(5)
+    assert True
+
+
+def test_alpha9():
+    sleep(5)
+    assert True

example/loadscope/test/test_beta.py

@@ -0,0 +1,51 @@
+from time import sleep
+
+
+def test_beta0():
+    sleep(5)
+    assert True
+
+
+def test_beta1():
+    sleep(5)
+    assert True
+
+
+def test_beta2():
+    sleep(5)
+    assert True
+
+
+def test_beta3():
+    sleep(5)
+    assert True
+
+
+def test_beta4():
+    sleep(5)
+    assert True
+
+
+def test_beta5():
+    sleep(5)
+    assert True
+
+
+def test_beta6():
+    sleep(5)
+    assert True
+
+
+def test_beta7():
+    sleep(5)
+    assert True
+
+
+def test_beta8():
+    sleep(5)
+    assert True
+
+
+def test_beta9():
+    sleep(5)
+    assert True

example/loadscope/test/test_delta.py

@@ -0,0 +1,88 @@
+from time import sleep
+from unittest import TestCase
+
+
+class Delta1(TestCase):
+
+    def test_delta0(self):
+        sleep(5)
+        assert True
+
+    def test_delta1(self):
+        sleep(5)
+        assert True
+
+    def test_delta2(self):
+        sleep(5)
+        assert True
+
+    def test_delta3(self):
+        sleep(5)
+        assert True
+
+    def test_delta4(self):
+        sleep(5)
+        assert True
+
+    def test_delta5(self):
+        sleep(5)
+        assert True
+
+    def test_delta6(self):
+        sleep(5)
+        assert True
+
+    def test_delta7(self):
+        sleep(5)
+        assert True
+
+    def test_delta8(self):
+        sleep(5)
+        assert True
+
+    def test_delta9(self):
+        sleep(5)
+        assert True
+
+
+class Delta2(TestCase):
+
+    def test_delta0(self):
+        sleep(5)
+        assert True
+
+    def test_delta1(self):
+        sleep(5)
+        assert True
+
+    def test_delta2(self):
+        sleep(5)
+        assert True
+
+    def test_delta3(self):
+        sleep(5)
+        assert True
+
+    def test_delta4(self):
+        sleep(5)
+        assert True
+
+    def test_delta5(self):
+        sleep(5)
+        assert True
+
+    def test_delta6(self):
+        sleep(5)
+        assert True
+
+    def test_delta7(self):
+        sleep(5)
+        assert True
+
+    def test_delta8(self):
+        sleep(5)
+        assert True
+
+    def test_delta9(self):
+        sleep(5)
+        assert True

example/loadscope/test/test_gamma.py

@@ -0,0 +1,51 @@
+from time import sleep
+
+
+def test_gamma0():
+    sleep(5)
+    assert True
+
+
+def test_gamma1():
+    sleep(5)
+    assert True
+
+
+def test_gamma2():
+    sleep(5)
+    assert True
+
+
+def test_gamma3():
+    sleep(5)
+    assert True
+
+
+def test_gamma4():
+    sleep(5)
+    assert True
+
+
+def test_gamma5():
+    sleep(5)
+    assert True
+
+
+def test_gamma6():
+    sleep(5)
+    assert True
+
+
+def test_gamma7():
+    sleep(5)
+    assert True
+
+
+def test_gamma8():
+    sleep(5)
+    assert True
+
+
+def test_gamma9():
+    sleep(5)
+    assert True

example/loadscope/tox.ini

@@ -0,0 +1,17 @@
+[tox]
+envlist = test
+setupdir = {toxinidir}/../../
+
+[testenv:test]
+basepython = python3
+passenv = http_proxy https_proxy
+deps = -rrequirements.txt
+changedir = {envtmpdir}
+commands =
+    py.test -s -v \
+        --doctest-modules \
+        --junitxml=tests.xml \
+        --dist=loadscope \
+        --tx=8*popen \
+        {toxinidir}/test \
+        {toxinidir}/epsilon

pyproject.toml

@@ -0,0 +1,36 @@
+[tool.towncrier]
+package = "xdist"
+filename = "CHANGELOG.rst"
+directory = "changelog/"
+title_format = "pytest-xdist {version} ({project_date})"
+template = "changelog/_template.rst"
+
+  [[tool.towncrier.type]]
+  directory = "removal"
+  name = "Deprecations and Removals"
+  showcontent = true
+
+  [[tool.towncrier.type]]
+  directory = "feature"
+  name = "Features"
+  showcontent = true
+
+  [[tool.towncrier.type]]
+  directory = "bugfix"
+  name = "Bug Fixes"
+  showcontent = true
+
+  [[tool.towncrier.type]]
+  directory = "vendor"
+  name = "Vendored Libraries"
+  showcontent = true
+
+  [[tool.towncrier.type]]
+  directory = "doc"
+  name = "Improved Documentation"
+  showcontent = true
+
+  [[tool.towncrier.type]]
+  directory = "trivial"
+  name = "Trivial Changes"
+  showcontent = false

setup.cfg

@@ -1,2 +1,8 @@
 [bdist_wheel]
 universal = 1
+
+[metadata]
+license_file = LICENSE
+
+[flake8]
+max-line-length = 100

setup.py

@@ -1,4 +1,12 @@
-from setuptools import setup
+from sys import version_info
+
+from setuptools import setup, find_packages
+
+install_requires = ['execnet>=1.1', 'pytest>=3.0.0', 'pytest-forked']
+
+if version_info < (2, 7):
+    install_requires.append('ordereddict')
+
 
 setup(
     name="pytest-xdist",
@@ -11,16 +19,15 @@ setup(
     author_email='pytest-dev@python.org,holger@merlinux.eu',
     url='https://github.com/pytest-dev/pytest-xdist',
     platforms=['linux', 'osx', 'win32'],
-    packages=['xdist'],
+    packages=find_packages(exclude=['testing', 'example']),
     entry_points={
         'pytest11': [
             'xdist = xdist.plugin',
             'xdist.looponfail = xdist.looponfail',
-            'xdist.boxed = xdist.boxed',
         ],
     },
     zip_safe=False,
-    install_requires=['execnet>=1.1', 'pytest>=2.4.2', 'py>=1.4.22'],
+    install_requires=install_requires,
     setup_requires=['setuptools_scm'],
     classifiers=[
         'Development Status :: 5 - Production/Stable',

testing/acceptance_test.py

@@ -1,3 +1,5 @@
+import re
+
 import py
 import pytest
 
@@ -10,7 +12,9 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines(["*1 passed*", ])
+        result.stdout.fnmatch_lines([
+            "*1 passed*",
+        ])
 
     def test_n1_fail(self, testdir):
         p1 = testdir.makepyfile("""
@@ -19,7 +23,9 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 1
-        result.stdout.fnmatch_lines(["*1 failed*", ])
+        result.stdout.fnmatch_lines([
+            "*1 failed*",
+        ])
 
     def test_n1_import_error(self, testdir):
         p1 = testdir.makepyfile("""
@@ -30,7 +36,7 @@ class TestDistribution:
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 1
         result.stdout.fnmatch_lines([
-            "E   ImportError: *__import_of_missing_module*",
+            "E   *Error: No module named *__import_of_missing_module*",
         ])
 
     def test_n2_import_error(self, testdir):
@@ -53,7 +59,9 @@ class TestDistribution:
         """)
         result = testdir.runpytest(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines(["*1 skipped*", ])
+        result.stdout.fnmatch_lines([
+            "*1 skipped*",
+        ])
 
     def test_manytests_to_one_import_error(self, testdir):
         p1 = testdir.makepyfile("""
@@ -62,13 +70,14 @@ class TestDistribution:
                 pass
         """)
         result = testdir.runpytest(p1, '--tx=popen', '--tx=popen')
-        assert result.ret == 1
+        assert result.ret in (1, 2)
         result.stdout.fnmatch_lines([
-            "E   ImportError: *__import_of_missing_module*",
+            "E   *Error: No module named *__import_of_missing_module*",
         ])
 
     def test_manytests_to_one_popen(self, testdir):
-        p1 = testdir.makepyfile("""
+        p1 = testdir.makepyfile(
+            """
                 import py
                 def test_fail0():
                     assert 0
@@ -108,10 +117,13 @@ class TestDistribution:
         """ % str(testdir.tmpdir))
         result = testdir.runpytest_subprocess(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines(["*1 passed*", ])
+        result.stdout.fnmatch_lines([
+            "*1 passed*",
+        ])
 
     def test_dist_ini_specified(self, testdir):
-        p1 = testdir.makepyfile("""
+        p1 = testdir.makepyfile(
+            """
                 import py
                 def test_fail0():
                     assert 0
@@ -207,14 +219,14 @@ class TestDistribution:
         """)
         p1 = testdir.makepyfile("def test_func(): pass")
         result = testdir.runpytest("-v", p1, '-d', '--tx=popen')
-        result.stdout.fnmatch_lines([
-            "*0*Python*", "*calculated result is 49*", "*1 passed*"
-        ])
+        result.stdout.fnmatch_lines(
+            ["*0*Python*", "*calculated result is 49*", "*1 passed*"])
         assert result.ret == 0
 
     def test_keyboardinterrupt_hooks_issue79(self, testdir):
-        testdir.makepyfile(__init__="",
-                           test_one="""
+        testdir.makepyfile(
+            __init__="",
+            test_one="""
             def test_hello():
                 raise KeyboardInterrupt()
         """)
@@ -273,8 +285,9 @@ class TestDistEach:
                 py.test.skip("%s not found" % name)
             interpreters.append(interp)
 
-        testdir.makepyfile(__init__="",
-                           test_one="""
+        testdir.makepyfile(
+            __init__="",
+            test_one="""
             import sys
             def test_hello():
                 print("%s...%s" % sys.version_info[:2])
@@ -326,6 +339,31 @@ class TestTerminalReporting:
             "E       assert 0",
         ])
 
+    @pytest.mark.parametrize('n', ['-n0', '-n1'])
+    @pytest.mark.parametrize('warn_type', ['pytest', 'builtin'])
+    def test_logwarning(self, testdir, n, warn_type):
+        from pkg_resources import parse_version
+        if parse_version(pytest.__version__) < parse_version('3.1'):
+            pytest.skip('pytest warnings requires >= 3.1')
+
+        if warn_type == 'builtin':
+            warn_code = """warnings.warn(UserWarning('this is a warning'))"""
+        elif warn_type == 'pytest':
+            warn_code = """request.config.warn('', 'this is a warning',
+                           fslocation=py.path.local())"""
+        else:
+            assert False
+        testdir.makepyfile("""
+            import warnings, py
+            def test_func(request):
+                {warn_code}
+        """.format(warn_code=warn_code))
+        result = testdir.runpytest(n)
+        result.stdout.fnmatch_lines([
+            "*this is a warning*",
+            "*1 passed, 1 warnings*",
+        ])
+
 
 def test_teardownfails_one_function(testdir):
     p = testdir.makepyfile("""
@@ -335,9 +373,8 @@ def test_teardownfails_one_function(testdir):
             assert 0
     """)
     result = testdir.runpytest(p, '-n1', '--tx=popen')
-    result.stdout.fnmatch_lines([
-        "*def teardown_function(function):*", "*1 passed*1 error*"
-    ])
+    result.stdout.fnmatch_lines(
+        ["*def teardown_function(function):*", "*1 passed*1 error*"])
 
 
 @py.test.mark.xfail
@@ -350,7 +387,9 @@ def test_terminate_on_hangingnode(testdir):
     """)
     result = testdir.runpytest(p, '--dist=each', '--tx=popen//id=my')
     assert result.duration < 2.0
-    result.stdout.fnmatch_lines(["*killed*my*", ])
+    result.stdout.fnmatch_lines([
+        "*killed*my*",
+    ])
 
 
 @pytest.mark.xfail(reason="works if run outside test suite", run=False)
@@ -377,7 +416,10 @@ def test_session_hooks(testdir):
             assert hasattr(sys, 'pytestsessionhooks')
     """)
     result = testdir.runpytest(p, "--dist=each", "--tx=popen")
-    result.stdout.fnmatch_lines(["*ValueError*", "*1 passed*", ])
+    result.stdout.fnmatch_lines([
+        "*ValueError*",
+        "*1 passed*",
+    ])
     assert not result.ret
     d = result.parseoutcomes()
     assert d['passed'] == 1
@@ -422,7 +464,10 @@ def test_funcarg_teardown_failure(testdir):
             pass
     """)
     result = testdir.runpytest_subprocess("--debug", p)  # , "-n1")
-    result.stdout.fnmatch_lines(["*ValueError*42*", "*1 passed*1 error*", ])
+    result.stdout.fnmatch_lines([
+        "*ValueError*42*",
+        "*1 passed*1 error*",
+    ])
     assert result.ret
 
 
@@ -453,7 +498,8 @@ def test_skipping(testdir):
 
 
 def test_issue34_pluginloading_in_subprocess(testdir):
-    testdir.tmpdir.join("plugin123.py").write(py.code.Source("""
+    testdir.tmpdir.join("plugin123.py").write(
+        py.code.Source("""
         def pytest_namespace():
             return {'sample_variable': 'testing'}
     """))
@@ -464,7 +510,9 @@ def test_issue34_pluginloading_in_subprocess(testdir):
     """)
     result = testdir.runpytest_subprocess("-n1", "-p", "plugin123")
     assert result.ret == 0
-    result.stdout.fnmatch_lines(["*1 passed*", ])
+    result.stdout.fnmatch_lines([
+        "*1 passed*",
+    ])
 
 
 def test_fixture_scope_caching_issue503(testdir):
@@ -486,7 +534,9 @@ def test_fixture_scope_caching_issue503(testdir):
     """)
     result = testdir.runpytest(p1, '-v', '-n1')
     assert result.ret == 0
-    result.stdout.fnmatch_lines(["*2 passed*", ])
+    result.stdout.fnmatch_lines([
+        "*2 passed*",
+    ])
 
 
 def test_issue_594_random_parametrize(testdir):
@@ -646,6 +696,20 @@ def test_worker_id_fixture(testdir, n):
         assert worker_ids == set(['gw0', 'gw1'])
 
 
+@pytest.mark.parametrize('tb',
+                         ['auto', 'long', 'short', 'no', 'line', 'native'])
+def test_error_report_styles(testdir, tb):
+    testdir.makepyfile("""
+        import pytest
+        def test_error_report_styles():
+            raise RuntimeError('some failure happened')
+    """)
+    result = testdir.runpytest('-n1', '--tb=%s' % tb)
+    if tb != 'no':
+        result.stdout.fnmatch_lines('*some failure happened*')
+    result.assert_outcomes(failed=1)
+
+
 def test_color_yes_collection_on_non_atty(testdir, request):
     """skip collect progress report when working on non-terminals.
 
@@ -687,3 +751,62 @@ def test_internal_error_with_maxfail(testdir):
     result = testdir.runpytest_subprocess('--maxfail=1', '-n1')
     result.stdout.fnmatch_lines(['* 1 error in *'])
     assert 'INTERNALERROR' not in result.stderr.str()
+
+
+class TestLoadScope:
+
+    def test_by_module(self, testdir):
+        test_file = """
+            import pytest
+            @pytest.mark.parametrize('i', range(10))
+            def test(i):
+                pass
+        """
+        testdir.makepyfile(
+            test_a=test_file,
+            test_b=test_file,
+        )
+        result = testdir.runpytest('-n2', '--dist=loadscope', '-v')
+        assert get_workers_and_test_count_by_prefix(
+            'test_a.py::test', result.outlines) in ({'gw0': 10}, {'gw1': 10})
+        assert get_workers_and_test_count_by_prefix(
+            'test_b.py::test', result.outlines) in ({'gw0': 10}, {'gw1': 10})
+
+    def test_by_class(self, testdir):
+        testdir.makepyfile(test_a="""
+            import pytest
+            class TestA:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+
+            class TestB:
+                @pytest.mark.parametrize('i', range(10))
+                def test(self, i):
+                    pass
+        """)
+        result = testdir.runpytest('-n2', '--dist=loadscope', '-v')
+        assert get_workers_and_test_count_by_prefix(
+            'test_a.py::TestA', result.outlines) in ({'gw0': 10}, {'gw1': 10})
+        assert get_workers_and_test_count_by_prefix(
+            'test_a.py::TestB', result.outlines) in ({'gw0': 10}, {'gw1': 10})
+
+
+def parse_tests_and_workers_from_output(lines):
+    result = []
+    for line in lines:
+        # example match: "[gw0] PASSED test_a.py::test[7]"
+        m = re.match(r'\[(gw\d)\]\s(.*?)\s(.*::.*)', line.strip())
+        if m:
+            worker, status, nodeid = m.groups()
+            result.append((worker, status, nodeid))
+    return result
+
+
+def get_workers_and_test_count_by_prefix(prefix, lines,
+                                         expected_status='PASSED'):
+    result = {}
+    for worker, status, nodeid in parse_tests_and_workers_from_output(lines):
+        if expected_status == status and nodeid.startswith(prefix):
+            result[worker] = result.get(worker, 0) + 1
+    return result

testing/test_boxed.py

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

testing/test_dsession.py

@@ -1,6 +1,10 @@
-from xdist.dsession import (
-    DSession, LoadScheduling, EachScheduling, report_collection_diff,
+from xdist.dsession import DSession
+from xdist.report import report_collection_diff
+from xdist.scheduler import (
+    EachScheduling,
+    LoadScheduling,
 )
+
 import py
 import pytest
 import execnet
@@ -49,80 +53,84 @@ def dumpqueue(queue):
 
 
 class TestEachScheduling:
-    def test_schedule_load_simple(self):
+    def test_schedule_load_simple(self, testdir):
         node1 = MockNode()
         node2 = MockNode()
-        sched = EachScheduling(2)
-        sched.addnode(node1)
-        sched.addnode(node2)
+        config = testdir.parseconfig("--tx=2*popen")
+        sched = EachScheduling(config)
+        sched.add_node(node1)
+        sched.add_node(node2)
         collection = ["a.py::test_1", ]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert not sched.collection_is_completed
-        sched.addnode_collection(node2, collection)
+        sched.add_node_collection(node2, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
         assert sched.node2collection[node2] == collection
-        sched.init_distribute()
-        assert sched.tests_finished()
+        sched.schedule()
+        assert sched.tests_finished
         assert node1.sent == ['ALL']
         assert node2.sent == ['ALL']
-        sched.remove_item(node1, 0)
-        assert sched.tests_finished()
-        sched.remove_item(node2, 0)
-        assert sched.tests_finished()
+        sched.mark_test_complete(node1, 0)
+        assert sched.tests_finished
+        sched.mark_test_complete(node2, 0)
+        assert sched.tests_finished
 
-    def test_schedule_remove_node(self):
+    def test_schedule_remove_node(self, testdir):
         node1 = MockNode()
-        sched = EachScheduling(1)
-        sched.addnode(node1)
+        config = testdir.parseconfig("--tx=popen")
+        sched = EachScheduling(config)
+        sched.add_node(node1)
         collection = ["a.py::test_1", ]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
-        sched.init_distribute()
-        assert sched.tests_finished()
+        sched.schedule()
+        assert sched.tests_finished
         crashitem = sched.remove_node(node1)
         assert crashitem
-        assert sched.tests_finished()
-        assert not sched.hasnodes()
+        assert sched.tests_finished
+        assert not sched.nodes
 
 
 class TestLoadScheduling:
-    def test_schedule_load_simple(self):
-        sched = LoadScheduling(2)
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
+    def test_schedule_load_simple(self, testdir):
+        config = testdir.parseconfig("--tx=2*popen")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2 = sched.nodes
         collection = ["a.py::test_1", "a.py::test_2"]
         assert not sched.collection_is_completed
-        sched.addnode_collection(node1, collection)
+        sched.add_node_collection(node1, collection)
         assert not sched.collection_is_completed
-        sched.addnode_collection(node2, collection)
+        sched.add_node_collection(node2, collection)
         assert sched.collection_is_completed
         assert sched.node2collection[node1] == collection
         assert sched.node2collection[node2] == collection
-        sched.init_distribute()
+        sched.schedule()
         assert not sched.pending
-        assert sched.tests_finished()
+        assert sched.tests_finished
         assert len(node1.sent) == 1
         assert len(node2.sent) == 1
         assert node1.sent == [0]
         assert node2.sent == [1]
-        sched.remove_item(node1, node1.sent[0])
-        assert sched.tests_finished()
+        sched.mark_test_complete(node1, node1.sent[0])
+        assert sched.tests_finished
 
-    def test_init_distribute_batch_size(self):
-        sched = LoadScheduling(2)
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
+    def test_schedule_batch_size(self, testdir):
+        config = testdir.parseconfig("--tx=2*popen")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2 = sched.nodes
         col = ["xyz"] * (6)
-        sched.addnode_collection(node1, col)
-        sched.addnode_collection(node2, col)
-        sched.init_distribute()
-        # assert not sched.tests_finished()
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.schedule()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         assert sent1 == [0, 2]
@@ -131,25 +139,26 @@ class TestLoadScheduling:
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
         assert len(sched.pending) == 2
-        sched.remove_item(node1, 0)
+        sched.mark_test_complete(node1, 0)
         assert node1.sent == [0, 2, 4]
         assert sched.pending == [5]
         assert node2.sent == [1, 3]
-        sched.remove_item(node1, 2)
+        sched.mark_test_complete(node1, 2)
         assert node1.sent == [0, 2, 4, 5]
         assert not sched.pending
 
-    def test_init_distribute_fewer_tests_than_nodes(self):
-        sched = LoadScheduling(2)
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
+    def test_schedule_fewer_tests_than_nodes(self, testdir):
+        config = testdir.parseconfig("--tx=2*popen")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2, node3 = sched.nodes
         col = ["xyz"] * 2
-        sched.addnode_collection(node1, col)
-        sched.addnode_collection(node2, col)
-        sched.init_distribute()
-        # assert not sched.tests_finished()
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.schedule()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         sent3 = node3.sent
@@ -158,17 +167,18 @@ class TestLoadScheduling:
         assert sent3 == []
         assert not sched.pending
 
-    def test_init_distribute_fewer_than_two_tests_per_node(self):
-        sched = LoadScheduling(2)
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
-        sched.addnode(MockNode())
+    def test_schedule_fewer_than_two_tests_per_node(self, testdir):
+        config = testdir.parseconfig("--tx=2*popen")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         node1, node2, node3 = sched.nodes
         col = ["xyz"] * 5
-        sched.addnode_collection(node1, col)
-        sched.addnode_collection(node2, col)
-        sched.init_distribute()
-        # assert not sched.tests_finished()
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.schedule()
+        # assert not sched.tests_finished
         sent1 = node1.sent
         sent2 = node2.sent
         sent3 = node3.sent
@@ -177,14 +187,15 @@ class TestLoadScheduling:
         assert sent3 == [2]
         assert not sched.pending
 
-    def test_add_remove_node(self):
+    def test_add_remove_node(self, testdir):
         node = MockNode()
-        sched = LoadScheduling(1)
-        sched.addnode(node)
+        config = testdir.parseconfig("--tx=popen")
+        sched = LoadScheduling(config)
+        sched.add_node(node)
         collection = ["test_file.py::test_func"]
-        sched.addnode_collection(node, collection)
+        sched.add_node_collection(node, collection)
         assert sched.collection_is_completed
-        sched.init_distribute()
+        sched.schedule()
         assert not sched.pending
         crashitem = sched.remove_node(node)
         assert crashitem == collection[0]
@@ -207,16 +218,16 @@ class TestLoadScheduling:
                 self.reports.append(report)
 
         collect_hook = CollectHook()
-        config = testdir.parseconfig()
+        config = testdir.parseconfig("--tx=2*popen")
         config.pluginmanager.register(collect_hook, "collect_hook")
         node1 = MockNode()
         node2 = MockNode()
-        sched = LoadScheduling(2, config=config)
-        sched.addnode(node1)
-        sched.addnode(node2)
-        sched.addnode_collection(node1, ["a.py::test_1"])
-        sched.addnode_collection(node2, ["a.py::test_2"])
-        sched.init_distribute()
+        sched = LoadScheduling(config)
+        sched.add_node(node1)
+        sched.add_node(node2)
+        sched.add_node_collection(node1, ["a.py::test_1"])
+        sched.add_node_collection(node2, ["a.py::test_2"])
+        sched.schedule()
         assert len(collect_hook.reports) == 1
         rep = collect_hook.reports[0]
         assert 'Different tests were collected between' in rep.longrepr

testing/test_looponfail.py

@@ -193,6 +193,8 @@ class TestLooponFailing:
         assert 'test_one' not in remotecontrol.failures[0]
         assert 'test_two' in remotecontrol.failures[0]
 
+    @py.test.mark.xfail(py.test.__version__ >= "3.1",
+                        reason="broken by pytest 3.1+")
     def test_looponfail_removed_test(self, testdir):
         modcol = testdir.getmodulecol("""
             def test_one():
@@ -268,7 +270,7 @@ class TestFunctional:
         """)
         child = testdir.spawn_pytest("-f %s" % p)
         child.expect("1 xpass")
-        child.expect("### LOOPONFAILING ####")
+        # child.expect("### LOOPONFAILING ####")
         child.expect("waiting for changes")
         child.kill(15)
 

testing/test_newhooks.py

@@ -14,14 +14,19 @@ class TestHooks:
 
     def test_runtest_logreport(self, testdir):
         """Test that log reports from pytest_runtest_logreport when running
-        with xdist contain a "node" attribute. (#8)
+        with xdist contain "node", "nodeid" and "worker_id" attributes. (#8)
         """
         testdir.makeconftest("""
             def pytest_runtest_logreport(report):
                 if hasattr(report, 'node'):
-                    slaveid = report.node.slaveinput['slaveid']
                     if report.when == "call":
-                        print("HOOK: %s %s" % (report.nodeid, slaveid))
+                        slaveid = report.node.slaveinput['slaveid']
+                        if slaveid != report.worker_id:
+                            print("HOOK: Worker id mismatch: %s %s"
+                                   % (slaveid, report.worker_id))
+                        else:
+                            print("HOOK: %s %s"
+                                   % (report.nodeid, report.worker_id))
         """)
         res = testdir.runpytest('-n1', '-s')
         res.stdout.fnmatch_lines([

testing/test_plugin.py

@@ -20,6 +20,10 @@ def test_dist_options(testdir):
     check_options(config)
     assert config.option.dist == "load"
     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
     config = testdir.parseconfigure("-d")
     check_options(config)
     assert config.option.dist == "load"

testing/test_remote.py

@@ -29,9 +29,9 @@ class EventCall:
 class SlaveSetup:
     use_callback = False
 
-    def __init__(self, request):
-        self.testdir = request.getfuncargvalue("testdir")
+    def __init__(self, request, testdir):
         self.request = request
+        self.testdir = testdir
         self.events = queue.Queue()
 
     def setup(self, ):
@@ -65,8 +65,8 @@ class SlaveSetup:
 
 
 @pytest.fixture
-def slave(request):
-    return SlaveSetup(request)
+def slave(request, testdir):
+    return SlaveSetup(request, testdir)
 
 
 @pytest.mark.xfail(reason='#59')
@@ -79,6 +79,84 @@ def test_remoteinitconfig(testdir):
 
 
 class TestReportSerialization:
+    def test_xdist_report_longrepr_reprcrash_130(self, testdir):
+        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"), "*")
+        rep.longrepr.sections.append(added_section)
+        d = serialize_report(rep)
+        check_marshallable(d)
+        a = unserialize_report("testreport", d)
+        # Check assembled == rep
+        assert a.__dict__.keys() == rep.__dict__.keys()
+        for key in rep.__dict__.keys():
+            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 \
+            == a.longrepr.reprtraceback.entrysep
+        assert rep.longrepr.reprtraceback.extraline \
+            == a.longrepr.reprtraceback.extraline
+        assert rep.longrepr.reprtraceback.style \
+            == a.longrepr.reprtraceback.style
+        assert rep.longrepr.sections == a.longrepr.sections
+        assert rep.longrepr.reprtraceback.reprentries \
+            == a.longrepr.reprtraceback.reprentries
+        # 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("""
+                            def test_repr_entry():
+                                x = 0
+                                assert x
+                        """, '--showlocals')
+        reports = reprec.getreports("pytest_runtest_logreport")
+        assert len(reports) == 3
+        rep = reports[1]
+        d = serialize_report(rep)
+        a = unserialize_report("testreport", d)
+
+        rep_entries = rep.longrepr.reprtraceback.reprentries
+        a_entries = a.longrepr.reprtraceback.reprentries
+        assert rep_entries == a_entries
+        for i in range(len(a_entries)):
+            assert isinstance(rep_entries[i], ReprEntry)
+            assert rep_entries[i].lines == a_entries[i].lines
+            assert rep_entries[i].localssep == a_entries[i].localssep
+            assert rep_entries[i].reprfileloc == a_entries[i].reprfileloc
+            assert rep_entries[i].reprfuncargs == a_entries[i].reprfuncargs
+            assert rep_entries[i].reprlocals == a_entries[i].reprlocals
+            assert rep_entries[i].style == a_entries[i].style
+
+    def test_reprentries_serialization_196(self, testdir):
+        from _pytest._code.code import ReprEntryNative
+        reprec = testdir.inline_runsource("""
+                            def test_repr_entry_native():
+                                x = 0
+                                assert x
+                        """, '--tb=native')
+        reports = reprec.getreports("pytest_runtest_logreport")
+        assert len(reports) == 3
+        rep = reports[1]
+        d = serialize_report(rep)
+        a = unserialize_report("testreport", d)
+
+        rep_entries = rep.longrepr.reprtraceback.reprentries
+        a_entries = a.longrepr.reprtraceback.reprentries
+        assert rep_entries == a_entries
+        for i in range(len(a_entries)):
+            assert isinstance(rep_entries[i], ReprEntryNative)
+            assert rep_entries[i].lines == a_entries[i].lines
+
     def test_itemreport_outcomes(self, testdir):
         reprec = testdir.inline_runsource("""
             import py
@@ -108,7 +186,7 @@ class TestReportSerialization:
             assert newrep.when == rep.when
             assert newrep.keywords == rep.keywords
             if rep.failed:
-                assert newrep.longrepr == str(rep.longrepr)
+                assert newrep.longreprtext == rep.longreprtext
 
     def test_collectreport_passed(self, testdir):
         reprec = testdir.inline_runsource("def test_func(): pass")
@@ -263,6 +341,15 @@ class TestSlaveInteractor:
             ("pytest_collectreport", "report.collector.fspath == bbb"),
         ])
 
+    def test_process_from_remote_error_handling(self, slave, capsys):
+        slave.use_callback = True
+        slave.setup()
+        slave.slp.process_from_remote(('<nonono>', ()))
+        out, err = capsys.readouterr()
+        assert 'INTERNALERROR> ValueError: unknown event: <nonono>' in out
+        ev = slave.popevent()
+        assert ev.name == "errordown"
+
 
 def test_remote_env_vars(testdir):
     testdir.makepyfile('''

tox.ini

@@ -1,9 +1,8 @@
 [tox]
 # if you change the envlist, please update .travis.yml file as well
 envlist=
-  py{26,27,34}-pytest2{6,7,8,9}
-  py35-pytest2{7,8,9}
-  py{27,35}-pytest28-pexpect
+  py{26,27,34,35,36}-pytest{30,31,32}
+  py{27,36}-pytest{30,31,32}-pexpect
   flakes
   readme
 
@@ -15,10 +14,9 @@ deps =
   pycmd
   # to avoid .eggs
   setuptools_scm
-  pytest26: pytest~=2.6.1
-  pytest27: pytest~=2.7.2
-  pytest28: pytest~=2.8.7
-  pytest29: pytest~=2.9.1
+  pytest30: pytest~=3.0.5
+  pytest31: pytest~=3.1.0
+  pytest32: pytest~=3.2.0
   pexpect: pexpect
 platform=
   pexpect: linux|darwin

xdist/boxed.py

@@ -1,60 +0,0 @@
-
-import py
-
-
-def pytest_addoption(parser):
-    group = parser.getgroup("xdist", "distributed and subprocess testing")
-    group.addoption(
-        '--boxed',
-        action="store_true", dest="boxed", default=False,
-        help="box each test run in a separate process (unix)")
-
-
-def pytest_runtest_protocol(item):
-    if item.config.getvalue("boxed"):
-        reports = forked_run_report(item)
-        for rep in reports:
-            item.ihook.pytest_runtest_logreport(report=rep)
-        return True
-
-
-def forked_run_report(item):
-    # for now, we run setup/teardown in the subprocess
-    # XXX optionally allow sharing of setup/teardown
-    from _pytest.runner import runtestprotocol
-    EXITSTATUS_TESTEXIT = 4
-    import marshal
-    from xdist.remote import serialize_report
-    from xdist.slavemanage import unserialize_report
-
-    def runforked():
-        try:
-            reports = runtestprotocol(item, log=False)
-        except KeyboardInterrupt:
-            py.std.os._exit(EXITSTATUS_TESTEXIT)
-        return marshal.dumps([serialize_report(x) for x in reports])
-
-    ff = py.process.ForkedFunc(runforked)
-    result = ff.waitfinish()
-    if result.retval is not None:
-        report_dumps = marshal.loads(result.retval)
-        return [unserialize_report("testreport", x) for x in report_dumps]
-    else:
-        if result.exitstatus == EXITSTATUS_TESTEXIT:
-            py.test.exit("forked test item %s raised Exit" % (item,))
-        return [report_process_crash(item, result)]
-
-
-def report_process_crash(item, result):
-    path, lineno = item._getfslineno()
-    info = ("%s:%s: running the test CRASHED with signal %d" %
-            (path, lineno, result.signal))
-    from _pytest import runner
-    call = runner.CallInfo(lambda: 0/0, "???")
-    call.excinfo = info
-    rep = runner.pytest_runtest_makereport(item, call)
-    if result.out:
-        rep.sections.append(("captured stdout", result.out))
-    if result.err:
-        rep.sections.append(("captured stderr", result.err))
-    return rep

xdist/dsession.py

@@ -1,449 +1,17 @@
-import difflib
-import itertools
-from _pytest.runner import CollectReport
-
-import pytest
 import py
+import pytest
+
 from xdist.slavemanage import NodeManager
+from xdist.scheduler import (
+    EachScheduling,
+    LoadScheduling,
+    LoadScopeScheduling,
+)
 
 
 queue = py.builtin._tryimport('queue', 'Queue')
 
 
-class EachScheduling:
-    """Implement scheduling of test items on all nodes
-
-    If a node gets added after the test run is started then it is
-    assumed to replace a node which got removed before it finished
-    its collection.  In this case it will only be used if a node
-    with the same spec got removed earlier.
-
-    Any nodes added after the run is started will only get items
-    assigned if a node with a matching spec was removed before it
-    finished all its pending items.  The new node will then be
-    assigned the remaining items from the removed node.
-    """
-
-    def __init__(self, numnodes, log=None):
-        self.numnodes = numnodes
-        self.node2collection = {}
-        self.node2pending = {}
-        self._started = []
-        self._removed2pending = {}
-        if log is None:
-            self.log = py.log.Producer("eachsched")
-        else:
-            self.log = log.eachsched
-        self.collection_is_completed = False
-
-    @property
-    def nodes(self):
-        """A list of all nodes in the scheduler."""
-        return list(self.node2pending.keys())
-
-    def hasnodes(self):
-        return bool(self.node2pending)
-
-    def haspending(self):
-        """Return True if there are pending test items
-
-        This indicates that collection has finished and nodes are
-        still processing test items, so this can be thought of as
-        "the scheduler is active".
-        """
-        for pending in self.node2pending.values():
-            if pending:
-                return True
-        return False
-
-    def addnode(self, node):
-        assert node not in self.node2pending
-        self.node2pending[node] = []
-
-    def tests_finished(self):
-        if not self.collection_is_completed:
-            return False
-        if self._removed2pending:
-            return False
-        for pending in self.node2pending.values():
-            if len(pending) >= 2:
-                return False
-        return True
-
-    def addnode_collection(self, node, collection):
-        """Add the collected test items from a node
-
-        Collection is complete once all nodes have submitted their
-        collection.  In this case its pending list is set to an empty
-        list.  When the collection is already completed this
-        submission is from a node which was restarted to replace a
-        dead node.  In this case we already assign the pending items
-        here.  In either case ``.init_distribute()`` will instruct the
-        node to start running the required tests.
-        """
-        assert node in self.node2pending
-        if not self.collection_is_completed:
-            self.node2collection[node] = list(collection)
-            self.node2pending[node] = []
-            if len(self.node2collection) >= self.numnodes:
-                self.collection_is_completed = True
-        elif self._removed2pending:
-            for deadnode in self._removed2pending:
-                if deadnode.gateway.spec == node.gateway.spec:
-                    dead_collection = self.node2collection[deadnode]
-                    if collection != dead_collection:
-                        msg = report_collection_diff(dead_collection,
-                                                     collection,
-                                                     deadnode.gateway.id,
-                                                     node.gateway.id)
-                        self.log(msg)
-                        return
-                    pending = self._removed2pending.pop(deadnode)
-                    self.node2pending[node] = pending
-                    break
-
-    def remove_item(self, node, item_index, duration=0):
-        self.node2pending[node].remove(item_index)
-
-    def remove_node(self, node):
-        # KeyError if we didn't get an addnode() yet
-        pending = self.node2pending.pop(node)
-        if not pending:
-            return
-        crashitem = self.node2collection[node][pending.pop(0)]
-        if pending:
-            self._removed2pending[node] = pending
-        return crashitem
-
-    def init_distribute(self):
-        """Schedule the test items on the nodes
-
-        If the node's pending list is empty it is a new node which
-        needs to run all the tests.  If the pending list is already
-        populated (by ``.addnode_collection()``) then it replaces a
-        dead node and we only need to run those tests.
-        """
-        assert self.collection_is_completed
-        for node, pending in self.node2pending.items():
-            if node in self._started:
-                continue
-            if not pending:
-                pending[:] = range(len(self.node2collection[node]))
-                node.send_runtest_all()
-            else:
-                node.send_runtest_some(pending)
-            self._started.append(node)
-
-
-class LoadScheduling:
-    """Implement load scheduling across nodes.
-
-    This distributes the tests collected across all nodes so each test
-    is run just once.  All nodes collect and submit the test suite and
-    when all collections are received it is verified they are
-    identical collections.  Then the collection gets divided up in
-    chunks and chunks get submitted to nodes.  Whenever a node finishes
-    an item, it calls ``.remove_item()`` which will trigger the
-    scheduler to assign more tests if the number of pending tests for
-    the node falls below a low-watermark.
-
-    When created, ``numnodes`` defines how many nodes are expected to
-    submit a collection. This is used to know when all nodes have
-    finished collection or how large the chunks need to be created.
-
-    Attributes:
-
-    :numnodes: The expected number of nodes taking part.  The actual
-       number of nodes will vary during the scheduler's lifetime as
-       nodes are added by the DSession as they are brought up and
-       removed either because of a dead node or normal shutdown.  This
-       number is primarily used to know when the initial collection is
-       completed.
-
-    :node2collection: Map of nodes and their test collection.  All
-       collections should always be identical.
-
-    :node2pending: Map of nodes and the indices of their pending
-       tests.  The indices are an index into ``.pending`` (which is
-       identical to their own collection stored in
-       ``.node2collection``).
-
-    :collection: The one collection once it is validated to be
-       identical between all the nodes.  It is initialised to None
-       until ``.init_distribute()`` is called.
-
-    :pending: List of indices of globally pending tests.  These are
-       tests which have not yet been allocated to a chunk for a node
-       to process.
-
-    :log: A py.log.Producer instance.
-
-    :config: Config object, used for handling hooks.
-    """
-
-    def __init__(self, numnodes, log=None, config=None):
-        self.numnodes = numnodes
-        self.node2collection = {}
-        self.node2pending = {}
-        self.pending = []
-        self.collection = None
-        if log is None:
-            self.log = py.log.Producer("loadsched")
-        else:
-            self.log = log.loadsched
-        self.config = config
-
-    @property
-    def nodes(self):
-        """A list of all nodes in the scheduler."""
-        return list(self.node2pending.keys())
-
-    @property
-    def collection_is_completed(self):
-        """Boolean indication initial test collection is complete.
-
-        This is a boolean indicating all initial participating nodes
-        have finished collection.  The required number of initial
-        nodes is defined by ``.numnodes``.
-        """
-        return len(self.node2collection) >= self.numnodes
-
-    def haspending(self):
-        """Return True if there are pending test items
-
-        This indicates that collection has finished and nodes are
-        still processing test items, so this can be thought of as
-        "the scheduler is active".
-        """
-        if self.pending:
-            return True
-        for pending in self.node2pending.values():
-            if pending:
-                return True
-        return False
-
-    def hasnodes(self):
-        """Return True if nodes exist in the scheduler."""
-        return bool(self.node2pending)
-
-    def addnode(self, node):
-        """Add a new node to the scheduler.
-
-        From now on the node will be allocated chunks of tests to
-        execute.
-
-        Called by the ``DSession.slave_slaveready`` hook when it
-        successfully bootstraps a new node.
-        """
-        assert node not in self.node2pending
-        self.node2pending[node] = []
-
-    def tests_finished(self):
-        """Return True if all tests have been executed by the nodes."""
-        if not self.collection_is_completed:
-            return False
-        if self.pending:
-            return False
-        for pending in self.node2pending.values():
-            if len(pending) >= 2:
-                return False
-        return True
-
-    def addnode_collection(self, node, collection):
-        """Add the collected test items from a node
-
-        The collection is stored in the ``.node2collection`` map.
-        Called by the ``DSession.slave_collectionfinish`` hook.
-        """
-        assert node in self.node2pending
-        if self.collection_is_completed:
-            # A new node has been added later, perhaps an original one died.
-            # .init_distribute() should have
-            # been called by now
-            assert self.collection
-            if collection != self.collection:
-                other_node = next(iter(self.node2collection.keys()))
-                msg = report_collection_diff(self.collection,
-                                             collection,
-                                             other_node.gateway.id,
-                                             node.gateway.id)
-                self.log(msg)
-                return
-        self.node2collection[node] = list(collection)
-
-    def remove_item(self, node, item_index, duration=0):
-        """Mark test item as completed by node
-
-        The duration it took to execute the item is used as a hint to
-        the scheduler.
-
-        This is called by the ``DSession.slave_testreport`` hook.
-        """
-        self.node2pending[node].remove(item_index)
-        self.check_schedule(node, duration=duration)
-
-    def check_schedule(self, node, duration=0):
-        """Maybe schedule new items on the node
-
-        If there are any globally pending nodes left then this will
-        check if the given node should be given any more tests.  The
-        ``duration`` of the last test is optionally used as a
-        heuristic to influence how many tests the node is assigned.
-        """
-        if node.shutting_down:
-            return
-
-        if self.pending:
-            # how many nodes do we have?
-            num_nodes = len(self.node2pending)
-            # if our node goes below a heuristic minimum, fill it out to
-            # heuristic maximum
-            items_per_node_min = max(2, len(self.pending) // num_nodes // 4)
-            items_per_node_max = max(2, len(self.pending) // num_nodes // 2)
-            node_pending = self.node2pending[node]
-            if len(node_pending) < items_per_node_min:
-                if duration >= 0.1 and len(node_pending) >= 2:
-                    # seems the node is doing long-running tests
-                    # and has enough items to continue
-                    # so let's rather wait with sending new items
-                    return
-                num_send = items_per_node_max - len(node_pending)
-                self._send_tests(node, num_send)
-        self.log("num items waiting for node:", len(self.pending))
-
-    def remove_node(self, node):
-        """Remove a node from the scheduler
-
-        This should be called either when the node crashed or at
-        shutdown time.  In the former case any pending items assigned
-        to the node will be re-scheduled.  Called by the
-        ``DSession.slave_slavefinished`` and
-        ``DSession.slave_errordown`` hooks.
-
-        Return the item which was being executing while the node
-        crashed or None if the node has no more pending items.
-
-        """
-        pending = self.node2pending.pop(node)
-        if not pending:
-            return
-
-        # The node crashed, reassing pending items
-        crashitem = self.collection[pending.pop(0)]
-        self.pending.extend(pending)
-        for node in self.node2pending:
-            self.check_schedule(node)
-        return crashitem
-
-    def init_distribute(self):
-        """Initiate distribution of the test collection
-
-        Initiate scheduling of the items across the nodes.  If this
-        gets called again later it behaves the same as calling
-        ``.check_schedule()`` on all nodes so that newly added nodes
-        will start to be used.
-
-        This is called by the ``DSession.slave_collectionfinish`` hook
-        if ``.collection_is_completed`` is True.
-
-        XXX Perhaps this method should have been called ".schedule()".
-        """
-        assert self.collection_is_completed
-
-        # Initial distribution already happened, reschedule on all nodes
-        if self.collection is not None:
-            for node in self.nodes:
-                self.check_schedule(node)
-            return
-
-        # XXX allow nodes to have different collections
-        if not self._check_nodes_have_same_collection():
-            self.log('**Different tests collected, aborting run**')
-            return
-
-        # Collections are identical, create the index of pending items.
-        self.collection = list(self.node2collection.values())[0]
-        self.pending[:] = range(len(self.collection))
-        if not self.collection:
-            return
-
-        # 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,
-                            2 * len(self.nodes))
-
-        # distribute tests round-robin up to the batch size
-        # (or until we run out)
-        nodes = itertools.cycle(self.nodes)
-        for i in range(initial_batch):
-            self._send_tests(next(nodes), 1)
-
-        if not self.pending:
-            # initial distribution sent all tests, start node shutdown
-            for node in self.nodes:
-                node.shutdown()
-
-    def _send_tests(self, node, num):
-        tests_per_node = self.pending[:num]
-        if tests_per_node:
-            del self.pending[:num]
-            self.node2pending[node].extend(tests_per_node)
-            node.send_runtest_some(tests_per_node)
-
-    def _check_nodes_have_same_collection(self):
-        """Return True if all nodes have collected the same items.
-
-        If collections differ, this method returns False while logging
-        the collection differences and posting collection errors to
-        pytest_collectreport hook.
-        """
-        node_collection_items = list(self.node2collection.items())
-        first_node, col = node_collection_items[0]
-        same_collection = True
-        for node, collection in node_collection_items[1:]:
-            msg = report_collection_diff(
-                col,
-                collection,
-                first_node.gateway.id,
-                node.gateway.id,
-            )
-            if msg:
-                same_collection = False
-                self.log(msg)
-                if self.config is not None:
-                    rep = CollectReport(
-                        node.gateway.id, 'failed',
-                        longrepr=msg, result=[])
-                    self.config.hook.pytest_collectreport(report=rep)
-
-        return same_collection
-
-
-def report_collection_diff(from_collection, to_collection, from_id, to_id):
-    """Report the collected test difference between two nodes.
-
-    :returns: detailed message describing the difference between the given
-    collections, or None if they are equal.
-    """
-    if from_collection == to_collection:
-        return None
-
-    diff = difflib.unified_diff(
-        from_collection,
-        to_collection,
-        fromfile=from_id,
-        tofile=to_id,
-    )
-    error_message = py.builtin._totext(
-        'Different tests were collected between {from_id} and {to_id}. '
-        'The difference is:\n'
-        '{diff}'
-    ).format(from_id=from_id, to_id=to_id, diff='\n'.join(diff))
-    msg = "\n".join([x.rstrip() for x in error_message.split("\n")])
-    return msg
-
-
 class Interrupted(KeyboardInterrupt):
     """ signals an immediate interruption. """
 
@@ -461,11 +29,14 @@ class DSession:
     automatically starts collecting tests.  Once tests are collected
     it will wait for instructions.
     """
+
     def __init__(self, config):
         self.config = config
         self.log = py.log.Producer("dsession")
         if not config.option.debug:
             py.log.setconsumer(self.log._keywords, None)
+        self.nodemanager = None
+        self.sched = None
         self.shuttingdown = False
         self.countfailures = 0
         self.maxfail = config.getvalue("maxfail")
@@ -521,16 +92,23 @@ class DSession:
         # prohibit collection of test items in master process
         return True
 
+    @pytest.mark.trylast
+    def pytest_xdist_make_scheduler(self, config, log):
+        dist = config.getvalue('dist')
+        schedulers = {
+            'each': EachScheduling,
+            'load': LoadScheduling,
+            'loadscope': LoadScopeScheduling,
+        }
+        return schedulers[dist](config, log)
+
     def pytest_runtestloop(self):
-        numnodes = len(self.nodemanager.specs)
-        dist = self.config.getvalue("dist")
-        if dist == "load":
-            self.sched = LoadScheduling(numnodes, log=self.log,
-                                        config=self.config)
-        elif dist == "each":
-            self.sched = EachScheduling(numnodes, log=self.log)
-        else:
-            assert 0, dist
+        self.sched = self.config.hook.pytest_xdist_make_scheduler(
+            config=self.config,
+            log=self.log
+        )
+        assert self.sched is not None
+
         self.shouldstop = False
         while not self.session_finished:
             self.loop_once()
@@ -553,7 +131,7 @@ class DSession:
         call = getattr(self, method)
         self.log("calling method", method, kwargs)
         call(**kwargs)
-        if self.sched.tests_finished():
+        if self.sched.tests_finished:
             self.triggershutdown()
 
     #
@@ -573,7 +151,7 @@ class DSession:
         if self.shuttingdown:
             node.shutdown()
         else:
-            self.sched.addnode(node)
+            self.sched.add_node(node)
 
     def slave_slavefinished(self, node):
         """Emitted when node executes its pytest_sessionfinish hook.
@@ -635,16 +213,16 @@ class DSession:
         # tell session which items were effectively collected otherwise
         # the master node will finish the session with EXIT_NOTESTSCOLLECTED
         self._session.testscollected = len(ids)
-        self.sched.addnode_collection(node, ids)
+        self.sched.add_node_collection(node, ids)
         if self.terminal:
             self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
         if self.sched.collection_is_completed:
-            if self.terminal and not self.sched.haspending():
+            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" % (
                     self.sched.__class__.__name__))
-            self.sched.init_distribute()
+            self.sched.schedule()
 
     def slave_logstart(self, node, nodeid, location):
         """Emitted when a node calls the pytest_runtest_logstart hook."""
@@ -658,7 +236,7 @@ class DSession:
         the item from the pending list in the scheduler.
         """
         if rep.when == "call" or (rep.when == "setup" and not rep.passed):
-            self.sched.remove_item(node, rep.item_index, rep.duration)
+            self.sched.mark_test_complete(node, rep.item_index, rep.duration)
         # self.report_line("testreport %s: %s" %(rep.id, rep.status))
         rep.node = node
         self.config.hook.pytest_runtest_logreport(report=rep)
@@ -669,6 +247,11 @@ class DSession:
         if rep.failed:
             self._failed_slave_collectreport(node, rep)
 
+    def slave_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 _clone_node(self, node):
         """Return new node based on an existing one.
 
@@ -743,7 +326,7 @@ class TerminalDistReporter:
         return " / ".join(parts)
 
     def rewrite(self, line, newline=False):
-        pline = line + " " * max(self._lastlen-len(line), 0)
+        pline = line + " " * max(self._lastlen - len(line), 0)
         if newline:
             self._lastlen = 0
             pline += "\n"

xdist/newhooks.py

@@ -9,8 +9,9 @@ with the worker instance that executed the hook originally:
 You can use this hooks just as you would use normal pytest hooks, but some care
 must be taken in plugins in case ``xdist`` is not installed. Please see:
 
-    http://pytest.org/latest/writing_plugins.html#optionally-using-hooks-from-3rd-party-plugins
+    http://pytest.org/en/latest/writing_plugins.html#optionally-using-hooks-from-3rd-party-plugins
 """
+import pytest
 
 
 def pytest_xdist_setupnodes(config, specs):
@@ -44,3 +45,8 @@ def pytest_testnodedown(node, error):
 def pytest_xdist_node_collection_finished(node, ids):
     """called by the master node when a node finishes collecting.
     """
+
+
+@pytest.mark.firstresult
+def pytest_xdist_make_scheduler(config, log):
+    """ return a node scheduler implementation """

xdist/plugin.py

@@ -20,24 +20,27 @@ def parse_numprocesses(s):
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
     group._addoption(
-        '-n', 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 "
              "host system")
-    group._addoption('--max-slave-restart', action="store", default=None,
-                     help="maximum number of slaves that can be restarted "
-                          "when crashed (set to zero to disable this feature)")
-    group._addoption(
+    group.addoption('--max-slave-restart', action="store", default=None,
+                    help="maximum number of slaves that can be restarted "
+                         "when crashed (set to zero to disable this feature)")
+    group.addoption(
         '--dist', metavar="distmode",
-        action="store", choices=['load', 'each', 'no'],
-        type="choice", dest="dist", default="no",
+        action="store", choices=['each', 'load', 'loadscope', 'no'],
+        dest="dist", default="no",
         help=("set mode for distributing tests to exec environments.\n\n"
-              "each: send each test to each available environment.\n\n"
-              "load: send each test to available environment.\n\n"
+              "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"
               "(default) no: run tests inprocess, don't distribute."))
-    group._addoption(
+    group.addoption(
         '--tx', dest="tx", action="append", default=[],
         metavar="xspec",
         help=("add a test execution environment. some examples: "
@@ -54,6 +57,9 @@ def pytest_addoption(parser):
         '--rsyncignore', action="append", default=[], metavar="GLOB",
         help="add expression for ignores when rsyncing to remote tx nodes.")
 
+    group.addoption(
+        "--boxed", action="store_true",
+        help="backward compatibility alias for pytest-forked --forked")
     parser.addini(
         'rsyncdirs', 'list of (relative) paths to be rsynced for'
         ' remote distributed testing.', type="pathlist")
@@ -64,6 +70,7 @@ def pytest_addoption(parser):
         "looponfailroots", type="pathlist",
         help="directories to check for changes", default=[py.path.local()])
 
+
 # -------------------------------------------------------------------------
 # distributed testing hooks
 # -------------------------------------------------------------------------
@@ -90,12 +97,15 @@ def pytest_configure(config):
         config.pluginmanager.register(session, "dsession")
         tr = config.pluginmanager.getplugin("terminalreporter")
         tr.showfspath = False
+    if config.getoption("boxed"):
+        config.option.forked = True
 
 
 @pytest.mark.tryfirst
 def pytest_cmdline_main(config):
     if config.option.numprocesses:
-        config.option.dist = "load"
+        if config.option.dist == 'no':
+            config.option.dist = "load"
         config.option.tx = ['popen'] * config.option.numprocesses
     if config.option.distload:
         config.option.dist = "load"

xdist/remote.py

@@ -8,6 +8,7 @@
 
 import sys
 import os
+import pytest
 
 
 class SlaveInteractor:
@@ -33,11 +34,11 @@ class SlaveInteractor:
         slaveinfo = getinfodict()
         self.sendevent("slaveready", slaveinfo=slaveinfo)
 
-    def pytest_sessionfinish(self, __multicall__, exitstatus):
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_sessionfinish(self, exitstatus):
         self.config.slaveoutput['exitstatus'] = exitstatus
-        res = __multicall__.execute()
+        yield
         self.sendevent("slavefinished", slaveoutput=self.config.slaveoutput)
-        return res
 
     def pytest_collection(self, session):
         self.sendevent("collectionstart")
@@ -88,6 +89,7 @@ class SlaveInteractor:
     def pytest_runtest_logreport(self, report):
         data = serialize_report(report)
         data["item_index"] = self.item_index
+        data["worker_id"] = self.slaveid
         assert self.session.items[self.item_index].nodeid == report.nodeid
         self.sendevent("testreport", data=data)
 
@@ -95,12 +97,43 @@ class SlaveInteractor:
         data = serialize_report(report)
         self.sendevent("collectreport", data=data)
 
+    def pytest_logwarning(self, message, code, nodeid, fslocation):
+        self.sendevent("logwarning", message=message, code=code, nodeid=nodeid,
+                       fslocation=str(fslocation))
+
 
 def serialize_report(rep):
+    def disassembled_report(rep):
+        reprtraceback = rep.longrepr.reprtraceback.__dict__
+        reprcrash = rep.longrepr.reprcrash.__dict__
+
+        new_entries = []
+        for entry in reprtraceback['reprentries']:
+            entry_data = {
+                'type': type(entry).__name__,
+                'data': entry.__dict__,
+            }
+            for key, value in entry_data['data'].items():
+                if hasattr(value, '__dict__'):
+                    entry_data['data'][key] = value.__dict__
+            new_entries.append(entry_data)
+
+        reprtraceback['reprentries'] = new_entries
+
+        return {
+            'reprcrash': reprcrash,
+            'reprtraceback': reprtraceback,
+            'sections': rep.longrepr.sections
+        }
+
     import py
     d = rep.__dict__.copy()
     if hasattr(rep.longrepr, 'toterminal'):
-        d['longrepr'] = str(rep.longrepr)
+        if hasattr(rep.longrepr, 'reprtraceback') \
+                and hasattr(rep.longrepr, 'reprcrash'):
+            d['longrepr'] = disassembled_report(rep)
+        else:
+            d['longrepr'] = str(rep.longrepr)
     else:
         d['longrepr'] = rep.longrepr
     for name in d:

xdist/report.py

@@ -0,0 +1,26 @@
+import py
+from difflib import unified_diff
+
+
+def report_collection_diff(from_collection, to_collection, from_id, to_id):
+    """Report the collected test difference between two nodes.
+
+    :returns: detailed message describing the difference between the given
+    collections, or None if they are equal.
+    """
+    if from_collection == to_collection:
+        return None
+
+    diff = unified_diff(
+        from_collection,
+        to_collection,
+        fromfile=from_id,
+        tofile=to_id,
+    )
+    error_message = py.builtin._totext(
+        'Different tests were collected between {from_id} and {to_id}. '
+        'The difference is:\n'
+        '{diff}'
+    ).format(from_id=from_id, to_id=to_id, diff='\n'.join(diff))
+    msg = "\n".join([x.rstrip() for x in error_message.split("\n")])
+    return msg

xdist/scheduler/__init__.py

@@ -0,0 +1,3 @@
+from xdist.scheduler.each import EachScheduling  # noqa
+from xdist.scheduler.load import LoadScheduling  # noqa
+from xdist.scheduler.loadscope import LoadScopeScheduling  # noqa

xdist/scheduler/each.py

@@ -0,0 +1,129 @@
+from py.log import Producer
+
+from xdist.slavemanage import parse_spec_config
+from xdist.report import report_collection_diff
+
+
+class EachScheduling:
+    """Implement scheduling of test items on all nodes
+
+    If a node gets added after the test run is started then it is
+    assumed to replace a node which got removed before it finished
+    its collection.  In this case it will only be used if a node
+    with the same spec got removed earlier.
+
+    Any nodes added after the run is started will only get items
+    assigned if a node with a matching spec was removed before it
+    finished all its pending items.  The new node will then be
+    assigned the remaining items from the removed node.
+    """
+
+    def __init__(self, config, log=None):
+        self.config = config
+        self.numnodes = len(parse_spec_config(config))
+        self.node2collection = {}
+        self.node2pending = {}
+        self._started = []
+        self._removed2pending = {}
+        if log is None:
+            self.log = Producer("eachsched")
+        else:
+            self.log = log.eachsched
+        self.collection_is_completed = False
+
+    @property
+    def nodes(self):
+        """A list of all nodes in the scheduler."""
+        return list(self.node2pending.keys())
+
+    @property
+    def tests_finished(self):
+        if not self.collection_is_completed:
+            return False
+        if self._removed2pending:
+            return False
+        for pending in self.node2pending.values():
+            if len(pending) >= 2:
+                return False
+        return True
+
+    @property
+    def has_pending(self):
+        """Return True if there are pending test items
+
+        This indicates that collection has finished and nodes are
+        still processing test items, so this can be thought of as
+        "the scheduler is active".
+        """
+        for pending in self.node2pending.values():
+            if pending:
+                return True
+        return False
+
+    def add_node(self, node):
+        assert node not in self.node2pending
+        self.node2pending[node] = []
+
+    def add_node_collection(self, node, collection):
+        """Add the collected test items from a node
+
+        Collection is complete once all nodes have submitted their
+        collection.  In this case its pending list is set to an empty
+        list.  When the collection is already completed this
+        submission is from a node which was restarted to replace a
+        dead node.  In this case we already assign the pending items
+        here.  In either case ``.schedule()`` will instruct the
+        node to start running the required tests.
+        """
+        assert node in self.node2pending
+        if not self.collection_is_completed:
+            self.node2collection[node] = list(collection)
+            self.node2pending[node] = []
+            if len(self.node2collection) >= self.numnodes:
+                self.collection_is_completed = True
+        elif self._removed2pending:
+            for deadnode in self._removed2pending:
+                if deadnode.gateway.spec == node.gateway.spec:
+                    dead_collection = self.node2collection[deadnode]
+                    if collection != dead_collection:
+                        msg = report_collection_diff(dead_collection,
+                                                     collection,
+                                                     deadnode.gateway.id,
+                                                     node.gateway.id)
+                        self.log(msg)
+                        return
+                    pending = self._removed2pending.pop(deadnode)
+                    self.node2pending[node] = pending
+                    break
+
+    def mark_test_complete(self, node, item_index, duration=0):
+        self.node2pending[node].remove(item_index)
+
+    def remove_node(self, node):
+        # KeyError if we didn't get an add_node() yet
+        pending = self.node2pending.pop(node)
+        if not pending:
+            return
+        crashitem = self.node2collection[node][pending.pop(0)]
+        if pending:
+            self._removed2pending[node] = pending
+        return crashitem
+
+    def schedule(self):
+        """Schedule the test items on the nodes
+
+        If the node's pending list is empty it is a new node which
+        needs to run all the tests.  If the pending list is already
+        populated (by ``.add_node_collection()``) then it replaces a
+        dead node and we only need to run those tests.
+        """
+        assert self.collection_is_completed
+        for node, pending in self.node2pending.items():
+            if node in self._started:
+                continue
+            if not pending:
+                pending[:] = range(len(self.node2collection[node]))
+                node.send_runtest_all()
+            else:
+                node.send_runtest_some(pending)
+            self._started.append(node)

xdist/scheduler/load.py

@@ -0,0 +1,288 @@
+from itertools import cycle
+
+from py.log import Producer
+from _pytest.runner import CollectReport
+
+from xdist.slavemanage import parse_spec_config
+from xdist.report import report_collection_diff
+
+
+class LoadScheduling:
+    """Implement load scheduling across nodes.
+
+    This distributes the tests collected across all nodes so each test
+    is run just once.  All nodes collect and submit the test suite and
+    when all collections are received it is verified they are
+    identical collections.  Then the collection gets divided up in
+    chunks and chunks get submitted to nodes.  Whenever a node finishes
+    an item, it calls ``.mark_test_complete()`` which will trigger the
+    scheduler to assign more tests if the number of pending tests for
+    the node falls below a low-watermark.
+
+    When created, ``numnodes`` defines how many nodes are expected to
+    submit a collection. This is used to know when all nodes have
+    finished collection or how large the chunks need to be created.
+
+    Attributes:
+
+    :numnodes: The expected number of nodes taking part.  The actual
+       number of nodes will vary during the scheduler's lifetime as
+       nodes are added by the DSession as they are brought up and
+       removed either because of a dead node or normal shutdown.  This
+       number is primarily used to know when the initial collection is
+       completed.
+
+    :node2collection: Map of nodes and their test collection.  All
+       collections should always be identical.
+
+    :node2pending: Map of nodes and the indices of their pending
+       tests.  The indices are an index into ``.pending`` (which is
+       identical to their own collection stored in
+       ``.node2collection``).
+
+    :collection: The one collection once it is validated to be
+       identical between all the nodes.  It is initialised to None
+       until ``.schedule()`` is called.
+
+    :pending: List of indices of globally pending tests.  These are
+       tests which have not yet been allocated to a chunk for a node
+       to process.
+
+    :log: A py.log.Producer instance.
+
+    :config: Config object, used for handling hooks.
+    """
+
+    def __init__(self, config, log=None):
+        self.numnodes = len(parse_spec_config(config))
+        self.node2collection = {}
+        self.node2pending = {}
+        self.pending = []
+        self.collection = None
+        if log is None:
+            self.log = Producer("loadsched")
+        else:
+            self.log = log.loadsched
+        self.config = config
+
+    @property
+    def nodes(self):
+        """A list of all nodes in the scheduler."""
+        return list(self.node2pending.keys())
+
+    @property
+    def collection_is_completed(self):
+        """Boolean indication initial test collection is complete.
+
+        This is a boolean indicating all initial participating nodes
+        have finished collection.  The required number of initial
+        nodes is defined by ``.numnodes``.
+        """
+        return len(self.node2collection) >= self.numnodes
+
+    @property
+    def tests_finished(self):
+        """Return True if all tests have been executed by the nodes."""
+        if not self.collection_is_completed:
+            return False
+        if self.pending:
+            return False
+        for pending in self.node2pending.values():
+            if len(pending) >= 2:
+                return False
+        return True
+
+    @property
+    def has_pending(self):
+        """Return True if there are pending test items
+
+        This indicates that collection has finished and nodes are
+        still processing test items, so this can be thought of as
+        "the scheduler is active".
+        """
+        if self.pending:
+            return True
+        for pending in self.node2pending.values():
+            if pending:
+                return True
+        return False
+
+    def add_node(self, node):
+        """Add a new node to the scheduler.
+
+        From now on the node will be allocated chunks of tests to
+        execute.
+
+        Called by the ``DSession.slave_slaveready`` hook when it
+        successfully bootstraps a new node.
+        """
+        assert node not in self.node2pending
+        self.node2pending[node] = []
+
+    def add_node_collection(self, node, collection):
+        """Add the collected test items from a node
+
+        The collection is stored in the ``.node2collection`` map.
+        Called by the ``DSession.slave_collectionfinish`` hook.
+        """
+        assert node in self.node2pending
+        if self.collection_is_completed:
+            # A new node has been added later, perhaps an original one died.
+            # .schedule() should have
+            # been called by now
+            assert self.collection
+            if collection != self.collection:
+                other_node = next(iter(self.node2collection.keys()))
+                msg = report_collection_diff(self.collection,
+                                             collection,
+                                             other_node.gateway.id,
+                                             node.gateway.id)
+                self.log(msg)
+                return
+        self.node2collection[node] = list(collection)
+
+    def mark_test_complete(self, node, item_index, duration=0):
+        """Mark test item as completed by node
+
+        The duration it took to execute the item is used as a hint to
+        the scheduler.
+
+        This is called by the ``DSession.slave_testreport`` hook.
+        """
+        self.node2pending[node].remove(item_index)
+        self.check_schedule(node, duration=duration)
+
+    def check_schedule(self, node, duration=0):
+        """Maybe schedule new items on the node
+
+        If there are any globally pending nodes left then this will
+        check if the given node should be given any more tests.  The
+        ``duration`` of the last test is optionally used as a
+        heuristic to influence how many tests the node is assigned.
+        """
+        if node.shutting_down:
+            return
+
+        if self.pending:
+            # how many nodes do we have?
+            num_nodes = len(self.node2pending)
+            # if our node goes below a heuristic minimum, fill it out to
+            # heuristic maximum
+            items_per_node_min = max(2, len(self.pending) // num_nodes // 4)
+            items_per_node_max = max(2, len(self.pending) // num_nodes // 2)
+            node_pending = self.node2pending[node]
+            if len(node_pending) < items_per_node_min:
+                if duration >= 0.1 and len(node_pending) >= 2:
+                    # seems the node is doing long-running tests
+                    # and has enough items to continue
+                    # so let's rather wait with sending new items
+                    return
+                num_send = items_per_node_max - len(node_pending)
+                self._send_tests(node, num_send)
+        self.log("num items waiting for node:", len(self.pending))
+
+    def remove_node(self, node):
+        """Remove a node from the scheduler
+
+        This should be called either when the node crashed or at
+        shutdown time.  In the former case any pending items assigned
+        to the node will be re-scheduled.  Called by the
+        ``DSession.slave_slavefinished`` and
+        ``DSession.slave_errordown`` hooks.
+
+        Return the item which was being executing while the node
+        crashed or None if the node has no more pending items.
+
+        """
+        pending = self.node2pending.pop(node)
+        if not pending:
+            return
+
+        # The node crashed, reassing pending items
+        crashitem = self.collection[pending.pop(0)]
+        self.pending.extend(pending)
+        for node in self.node2pending:
+            self.check_schedule(node)
+        return crashitem
+
+    def schedule(self):
+        """Initiate distribution of the test collection
+
+        Initiate scheduling of the items across the nodes.  If this
+        gets called again later it behaves the same as calling
+        ``.check_schedule()`` on all nodes so that newly added nodes
+        will start to be used.
+
+        This is called by the ``DSession.slave_collectionfinish`` hook
+        if ``.collection_is_completed`` is True.
+        """
+        assert self.collection_is_completed
+
+        # Initial distribution already happened, reschedule on all nodes
+        if self.collection is not None:
+            for node in self.nodes:
+                self.check_schedule(node)
+            return
+
+        # XXX allow nodes to have different collections
+        if not self._check_nodes_have_same_collection():
+            self.log('**Different tests collected, aborting run**')
+            return
+
+        # Collections are identical, create the index of pending items.
+        self.collection = list(self.node2collection.values())[0]
+        self.pending[:] = range(len(self.collection))
+        if not self.collection:
+            return
+
+        # 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,
+                            2 * len(self.nodes))
+
+        # distribute tests round-robin up to the batch size
+        # (or until we run out)
+        nodes = cycle(self.nodes)
+        for i in range(initial_batch):
+            self._send_tests(next(nodes), 1)
+
+        if not self.pending:
+            # initial distribution sent all tests, start node shutdown
+            for node in self.nodes:
+                node.shutdown()
+
+    def _send_tests(self, node, num):
+        tests_per_node = self.pending[:num]
+        if tests_per_node:
+            del self.pending[:num]
+            self.node2pending[node].extend(tests_per_node)
+            node.send_runtest_some(tests_per_node)
+
+    def _check_nodes_have_same_collection(self):
+        """Return True if all nodes have collected the same items.
+
+        If collections differ, this method returns False while logging
+        the collection differences and posting collection errors to
+        pytest_collectreport hook.
+        """
+        node_collection_items = list(self.node2collection.items())
+        first_node, col = node_collection_items[0]
+        same_collection = True
+        for node, collection in node_collection_items[1:]:
+            msg = report_collection_diff(
+                col,
+                collection,
+                first_node.gateway.id,
+                node.gateway.id,
+            )
+            if msg:
+                same_collection = False
+                self.log(msg)
+                if self.config is not None:
+                    rep = CollectReport(
+                        node.gateway.id, 'failed',
+                        longrepr=msg, result=[])
+                    self.config.hook.pytest_collectreport(report=rep)
+
+        return same_collection

xdist/scheduler/loadscope.py

@@ -0,0 +1,428 @@
+try:
+    from collections import OrderedDict
+except ImportError:
+    # Support for Python 2.6
+    from ordereddict import OrderedDict
+
+from py.log import Producer
+from _pytest.runner import CollectReport
+
+from xdist.slavemanage import parse_spec_config
+from xdist.report import report_collection_diff
+
+
+class LoadScopeScheduling:
+    """Implement load scheduling across nodes, but grouping test by scope.
+
+    This distributes the tests collected across all nodes so each test is run
+    just once.  All nodes collect and submit the list of tests and when all
+    collections are received it is verified they are identical collections.
+    Then the collection gets divided up in work units, grouped by test scope,
+    and those work units get submitted to nodes.  Whenever a node finishes an
+    item, it calls ``.mark_test_complete()`` which will trigger the scheduler
+    to assign more work units if the number of pending tests for the node falls
+    below a low-watermark.
+
+    When created, ``numnodes`` defines how many nodes are expected to submit a
+    collection. This is used to know when all nodes have finished collection.
+
+    Attributes:
+
+    :numnodes: The expected number of nodes taking part.  The actual number of
+       nodes will vary during the scheduler's lifetime as nodes are added by
+       the DSession as they are brought up and removed either because of a dead
+       node or normal shutdown.  This number is primarily used to know when the
+       initial collection is completed.
+
+    :collection: The final list of tests collected by all nodes once it is
+       validated to be identical between all the nodes.  It is initialised to
+       None until ``.schedule()`` is called.
+
+    :workqueue: Ordered dictionary that maps all available scopes with their
+       associated tests (nodeid). Nodeids are in turn associated with their
+       completion status. One entry of the workqueue is called a work unit.
+       In turn, a collection of work unit is called a workload.
+
+       ::
+
+            workqueue = {
+                '<full>/<path>/<to>/test_module.py': {
+                    '<full>/<path>/<to>/test_module.py::test_case1': False,
+                    '<full>/<path>/<to>/test_module.py::test_case2': False,
+                    (...)
+                },
+                (...)
+            }
+
+    :assigned_work: Ordered dictionary that maps worker nodes with their
+       assigned work units.
+
+       ::
+
+            assigned_work = {
+                '<worker node A>': {
+                    '<full>/<path>/<to>/test_module.py': {
+                        '<full>/<path>/<to>/test_module.py::test_case1': False,
+                        '<full>/<path>/<to>/test_module.py::test_case2': False,
+                        (...)
+                    },
+                    (...)
+                },
+                (...)
+            }
+
+    :registered_collections: Ordered dictionary that maps worker nodes with
+       their collection of tests gathered during test discovery.
+
+       ::
+
+            registered_collections = {
+                '<worker node A>': [
+                    '<full>/<path>/<to>/test_module.py::test_case1',
+                    '<full>/<path>/<to>/test_module.py::test_case2',
+                ],
+                (...)
+            }
+
+    :log: A py.log.Producer instance.
+
+    :config: Config object, used for handling hooks.
+    """
+
+    def __init__(self, config, log=None):
+        self.numnodes = len(parse_spec_config(config))
+        self.collection = None
+
+        self.workqueue = OrderedDict()
+        self.assigned_work = OrderedDict()
+        self.registered_collections = OrderedDict()
+
+        if log is None:
+            self.log = Producer('loadscopesched')
+        else:
+            self.log = log.loadscopesched
+
+        self.config = config
+
+    @property
+    def nodes(self):
+        """A list of all active nodes in the scheduler."""
+        return list(self.assigned_work.keys())
+
+    @property
+    def collection_is_completed(self):
+        """Boolean indication initial test collection is complete.
+
+        This is a boolean indicating all initial participating nodes have
+        finished collection.  The required number of initial nodes is defined
+        by ``.numnodes``.
+        """
+        return len(self.registered_collections) >= self.numnodes
+
+    @property
+    def tests_finished(self):
+        """Return True if all tests have been executed by the nodes."""
+        if not self.collection_is_completed:
+            return False
+
+        if self.workqueue:
+            return False
+
+        for assigned_unit in self.assigned_work.values():
+            if self._pending_of(assigned_unit) >= 2:
+                return False
+
+        return True
+
+    @property
+    def has_pending(self):
+        """Return True if there are pending test items.
+
+        This indicates that collection has finished and nodes are still
+        processing test items, so this can be thought of as
+        "the scheduler is active".
+        """
+        if self.workqueue:
+            return True
+
+        for assigned_unit in self.assigned_work.values():
+            if self._pending_of(assigned_unit) > 0:
+                return True
+
+        return False
+
+    def add_node(self, node):
+        """Add a new node to the scheduler.
+
+        From now on the node will be assigned work units to be executed.
+
+        Called by the ``DSession.slave_slaveready`` hook when it successfully
+        bootstraps a new node.
+        """
+        assert node not in self.assigned_work
+        self.assigned_work[node] = OrderedDict()
+
+    def remove_node(self, node):
+        """Remove a node from the scheduler.
+
+        This should be called either when the node crashed or at shutdown time.
+        In the former case any pending items assigned to the node will be
+        re-scheduled.
+
+        Called by the hooks:
+
+        - ``DSession.slave_slavefinished``.
+        - ``DSession.slave_errordown``.
+
+        Return the item being executed while the node crashed or None if the
+        node has no more pending items.
+        """
+        workload = self.assigned_work.pop(node)
+        if not self._pending_of(workload):
+            return None
+
+        # The node crashed, identify test that crashed
+        for work_unit in workload.values():
+            for nodeid, completed in work_unit.items():
+                if not completed:
+                    crashitem = nodeid
+                    break
+            else:
+                continue
+            break
+        else:
+            raise RuntimeError(
+                'Unable to identify crashitem on a workload with '
+                'pending items'
+            )
+
+        # Made uncompleted work unit available again
+        self.workqueue.update(workload)
+
+        for node in self.assigned_work:
+            self._reschedule(node)
+
+        return crashitem
+
+    def add_node_collection(self, node, collection):
+        """Add the collected test items from a node.
+
+        The collection is stored in the ``.registered_collections`` dictionary.
+
+        Called by the hook:
+
+        - ``DSession.slave_collectionfinish``.
+        """
+
+        # Check that add_node() was called on the node before
+        assert node in self.assigned_work
+
+        # A new node has been added later, perhaps an original one died.
+        if self.collection_is_completed:
+
+            # Assert that .schedule() should have been called by now
+            assert self.collection
+
+            # Check that the new collection matches the official collection
+            if collection != self.collection:
+
+                other_node = next(iter(self.registered_collections.keys()))
+
+                msg = report_collection_diff(
+                    self.collection,
+                    collection,
+                    other_node.gateway.id,
+                    node.gateway.id
+                )
+                self.log(msg)
+                return
+
+        self.registered_collections[node] = list(collection)
+
+    def mark_test_complete(self, node, item_index, duration=0):
+        """Mark test item as completed by node.
+
+        Called by the hook:
+
+        - ``DSession.slave_testreport``.
+        """
+        nodeid = self.registered_collections[node][item_index]
+        scope = self._split_scope(nodeid)
+
+        self.assigned_work[node][scope][nodeid] = True
+        self._reschedule(node)
+
+    def _assign_work_unit(self, node):
+        """Assign a work unit to a node."""
+        assert self.workqueue
+
+        # Grab a unit of work
+        scope, work_unit = self.workqueue.popitem(last=False)
+
+        # Keep track of the assigned work
+        assigned_to_node = self.assigned_work.setdefault(
+            node, default=OrderedDict()
+        )
+        assigned_to_node[scope] = work_unit
+
+        # Ask the node to execute the workload
+        worker_collection = self.registered_collections[node]
+        nodeids_indexes = [
+            worker_collection.index(nodeid)
+            for nodeid, completed in work_unit.items()
+            if not completed
+        ]
+
+        node.send_runtest_some(nodeids_indexes)
+
+    def _split_scope(self, nodeid):
+        """Determine the scope (grouping) of a nodeid.
+
+        There are usually 3 cases for a nodeid::
+
+            example/loadsuite/test/test_beta.py::test_beta0
+            example/loadsuite/test/test_delta.py::Delta1::test_delta0
+            example/loadsuite/epsilon/__init__.py::epsilon.epsilon
+
+        #. Function in a test module.
+        #. Method of a class in a test module.
+        #. Doctest in a function in a package.
+
+        This function will group tests with the scope determined by splitting
+        the first ``::`` from the right. That is, classes will be grouped in a
+        single work unit, and functions from a test module will be grouped by
+        their module. In the above example, scopes will be::
+
+            example/loadsuite/test/test_beta.py
+            example/loadsuite/test/test_delta.py::Delta1
+            example/loadsuite/epsilon/__init__.py
+        """
+        return nodeid.rsplit('::', 1)[0]
+
+    def _pending_of(self, workload):
+        """Return the number of pending tests in a workload."""
+        pending = sum(
+            list(scope.values()).count(False)
+            for scope in workload.values()
+        )
+        return pending
+
+    def _reschedule(self, node):
+        """Maybe schedule new items on the node.
+
+        If there are any globally pending work units left then this will check
+        if the given node should be given any more tests.
+        """
+
+        # Do not add more work to a node shutting down
+        if node.shutting_down:
+            return
+
+        # Check that more work is available
+        if not self.workqueue:
+            return
+
+        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
+        if self._pending_of(self.assigned_work[node]) > 2:
+            return
+
+        # Pop one unit of work and assign it
+        self._assign_work_unit(node)
+
+    def schedule(self):
+        """Initiate distribution of the test collection.
+
+        Initiate scheduling of the items across the nodes.  If this gets called
+        again later it behaves the same as calling ``._reschedule()`` on all
+        nodes so that newly added nodes will start to be used.
+
+        If ``.collection_is_completed`` is True, this is called by the hook:
+
+        - ``DSession.slave_collectionfinish``.
+        """
+        assert self.collection_is_completed
+
+        # Initial distribution already happened, reschedule on all nodes
+        if self.collection is not None:
+            for node in self.nodes:
+                self._reschedule(node)
+            return
+
+        # Check that all nodes collected the same tests
+        if not self._check_nodes_have_same_collection():
+            self.log('**Different tests collected, aborting run**')
+            return
+
+        # Collections are identical, create the final list of items
+        self.collection = list(
+            next(iter(self.registered_collections.values()))
+        )
+        if not self.collection:
+            return
+
+        # Determine chunks of work (scopes)
+        for nodeid in self.collection:
+            scope = self._split_scope(nodeid)
+            work_unit = self.workqueue.setdefault(scope, default=OrderedDict())
+            work_unit[nodeid] = False
+
+        # Avoid having more workers than work
+        extra_nodes = len(self.nodes) - len(self.workqueue)
+
+        if extra_nodes > 0:
+            self.log('Shuting down {} nodes'.format(extra_nodes))
+
+            for _ in range(extra_nodes):
+                unused_node, assigned = self.assigned_work.popitem(last=True)
+
+                self.log('Shuting down unused node {}'.format(unused_node))
+                unused_node.shutdown()
+
+        # Assign initial workload
+        for node in self.nodes:
+            self._assign_work_unit(node)
+
+        # Initial distribution sent all tests, start node shutdown
+        if not self.workqueue:
+            for node in self.nodes:
+                node.shutdown()
+
+    def _check_nodes_have_same_collection(self):
+        """Return True if all nodes have collected the same items.
+
+        If collections differ, this method returns False while logging
+        the collection differences and posting collection errors to
+        pytest_collectreport hook.
+        """
+        node_collection_items = list(self.registered_collections.items())
+        first_node, col = node_collection_items[0]
+        same_collection = True
+
+        for node, collection in node_collection_items[1:]:
+            msg = report_collection_diff(
+                col,
+                collection,
+                first_node.gateway.id,
+                node.gateway.id,
+            )
+            if not msg:
+                continue
+
+            same_collection = False
+            self.log(msg)
+
+            if self.config is None:
+                continue
+
+            rep = CollectReport(
+                node.gateway.id,
+                'failed',
+                longrepr=msg,
+                result=[]
+            )
+            self.config.hook.pytest_collectreport(report=rep)
+
+        return same_collection

xdist/slavemanage.py

@@ -10,6 +10,22 @@ import xdist.remote
 from _pytest import runner  # XXX load dynamically
 
 
+def parse_spec_config(config):
+    xspeclist = []
+    for xspec in config.getvalue("tx"):
+        i = xspec.find("*")
+        try:
+            num = int(xspec[:i])
+        except ValueError:
+            xspeclist.append(xspec)
+        else:
+            xspeclist.extend([xspec[i + 1:]] * num)
+    if not xspeclist:
+        raise pytest.UsageError(
+            "MISSING test execution (tx) nodes: please specify --tx")
+    return xspeclist
+
+
 class NodeManager(object):
     EXIT_TIMEOUT = 10
     DEFAULT_IGNORES = ['.*', '*.pyc', '*.pyo', '*~']
@@ -62,19 +78,7 @@ class NodeManager(object):
         self.group.terminate(self.EXIT_TIMEOUT)
 
     def _getxspecs(self):
-        xspeclist = []
-        for xspec in self.config.getvalue("tx"):
-            i = xspec.find("*")
-            try:
-                num = int(xspec[:i])
-            except ValueError:
-                xspeclist.append(xspec)
-            else:
-                xspeclist.extend([xspec[i+1:]] * num)
-        if not xspeclist:
-            raise pytest.UsageError(
-                "MISSING test execution (tx) nodes: please specify --tx")
-        return [execnet.XSpec(x) for x in xspeclist]
+        return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
 
     def _getrsyncdirs(self):
         for spec in self.specs:
@@ -292,7 +296,7 @@ class SlaveController(object):
                     self._down = True
                 return
             eventname, kwargs = eventcall
-            if eventname in ("collectionstart"):
+            if eventname in ("collectionstart",):
                 self.log("ignoring %s(%s)" % (eventname, kwargs))
             elif eventname == "slaveready":
                 self.notify_inproc(eventname, node=self, **kwargs)
@@ -311,6 +315,10 @@ class SlaveController(object):
                 self.notify_inproc(eventname, node=self, rep=rep)
             elif eventname == "collectionfinish":
                 self.notify_inproc(eventname, node=self, ids=kwargs['ids'])
+            elif eventname == "logwarning":
+                self.notify_inproc(eventname, message=kwargs['message'],
+                                   code=kwargs['code'], nodeid=kwargs['nodeid'],
+                                   fslocation=kwargs['nodeid'])
             else:
                 raise ValueError("unknown event: %s" % (eventname,))
         except KeyboardInterrupt:
@@ -319,11 +327,87 @@ class SlaveController(object):
         except:
             excinfo = py.code.ExceptionInfo()
             py.builtin.print_("!" * 20, excinfo)
-            self.config.pluginmanager.notify_exception(excinfo)
+            self.config.notify_exception(excinfo)
+            self.shutdown()
+            self.notify_inproc("errordown", node=self, error=excinfo)
 
 
 def unserialize_report(name, reportdict):
+    def assembled_report(reportdict):
+        from _pytest._code.code import (
+            ReprEntry,
+            ReprEntryNative,
+            ReprExceptionInfo,
+            ReprFileLocation,
+            ReprFuncArgs,
+            ReprLocals,
+            ReprTraceback
+        )
+        if reportdict['longrepr']:
+            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']:
+                    data = entry_data['data']
+                    entry_type = entry_data['type']
+                    if entry_type == 'ReprEntry':
+                        reprfuncargs = None
+                        reprfileloc = None
+                        reprlocals = None
+                        if data['reprfuncargs']:
+                            reprfuncargs = ReprFuncArgs(
+                                **data['reprfuncargs'])
+                        if data['reprfileloc']:
+                            reprfileloc = ReprFileLocation(
+                                **data['reprfileloc'])
+                        if data['reprlocals']:
+                            reprlocals = ReprLocals(
+                                data['reprlocals']['lines'])
+
+                        reprentry = ReprEntry(
+                            lines=data['lines'],
+                            reprfuncargs=reprfuncargs,
+                            reprlocals=reprlocals,
+                            filelocrepr=reprfileloc,
+                            style=data['style']
+                        )
+                    elif entry_type == 'ReprEntryNative':
+                        reprentry = ReprEntryNative(data['lines'])
+                    else:
+                        report_unserialization_failure(
+                            entry_type, name, reportdict)
+                    unserialized_entries.append(reprentry)
+                reprtraceback['reprentries'] = unserialized_entries
+
+                exception_info = ReprExceptionInfo(
+                    reprtraceback=ReprTraceback(**reprtraceback),
+                    reprcrash=ReprFileLocation(**reprcrash),
+                )
+
+                for section in reportdict['longrepr']['sections']:
+                    exception_info.addsection(*section)
+                reportdict['longrepr'] = exception_info
+        return reportdict
+
     if name == "testreport":
-        return runner.TestReport(**reportdict)
+        return runner.TestReport(**assembled_report(reportdict))
     elif name == "collectreport":
-        return runner.CollectReport(**reportdict)
+        return runner.CollectReport(**assembled_report(reportdict))
+
+
+def report_unserialization_failure(type_name, report_name, reportdict):
+    from pprint import pprint
+    url = 'https://github.com/pytest-dev/pytest-xdist/issues'
+    stream = py.io.TextIO()
+    pprint('-' * 100, stream=stream)
+    pprint('INTERNALERROR: Unknown entry type returned: %s' % type_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('-' * 100, stream=stream)
+    assert 0, stream.getvalue()