Merge pull request #122 from wronglink/custom_scheduler_notes

Added notes about new pytest_xdist_make_scheduler hook

.gitignore

@@ -22,9 +22,11 @@ dist/
 include/
 lib/
 bin/
+xdist/_version.py*
 pytest_xdist.egg-info
 issue/
 3rdparty/
 pytestdebug.log
-.tox
-.cache
+.tox/
+.cache/
+.eggs/

.hgtags

@@ -1,17 +0,0 @@
-42c6503ee48fae9c4c96d406afb12bfc86f15803 1.0
-eca7ce17eabf296983c36812c8b8be901e7055a3 1.1
-56d8e5280be224a0ad3220a9deed55334710bd23 1.2
-e6c4ce20db4bf65086ff55807a3c306cad7ca393 1.3
-e6c4ce20db4bf65086ff55807a3c306cad7ca393 1.3
-eaf8b1cb7c312883598677231be5bbeea3b5c127 1.3
-a423748bf17ee778a37853225210257699cad9c1 1.4
-cd44a941c833c098e4899fe3d42a96703754d0d5 1.5
-4815040bdad8f182a5487f57a9da385483836e75 1.6
-20875fed94e7f3dff50bdf762df91153b15ceca6 1.7
-20875fed94e7f3dff50bdf762df91153b15ceca6 1.7
-29c38e195526f5f0fdd651fb51f59d6efaaafbb0 1.7
-0d1c00018008433956aa7d93007bab6ea7de96e4 1.8
-0d1c00018008433956aa7d93007bab6ea7de96e4 1.8
-1d27987c267577899350a25ba5828d55d87083ad 1.8
-5c5cb6d59e12e566fbb0217aea718dc31578bee1 1.9
-4406fc2a6427fadc021ed7e43e7aa5032b1ea91f 1.10

.travis.yml

@@ -0,0 +1,47 @@
+sudo: false
+language: python
+python:
+- '3.5'
+install: pip install -U tox setuptools_scm
+env:
+  matrix:
+  # note: please use "tox --listenvs" to populate the build matrix
+  - TOXENV=py26-pytest27
+  - TOXENV=py26-pytest28
+  - TOXENV=py26-pytest29
+  - TOXENV=py26-pytest30
+  - TOXENV=py27-pytest27
+  - TOXENV=py27-pytest28
+  - TOXENV=py27-pytest29
+  - TOXENV=py27-pytest30
+  - TOXENV=py34-pytest27
+  - TOXENV=py34-pytest28
+  - TOXENV=py34-pytest29
+  - TOXENV=py34-pytest30
+  - TOXENV=py35-pytest27
+  - TOXENV=py35-pytest28
+  - TOXENV=py35-pytest29
+  - TOXENV=py27-pytest28-pexpect
+  - TOXENV=py35-pytest28-pexpect
+  - TOXENV=flakes
+  - TOXENV=readme
+
+script: tox
+notifications:
+  irc:
+    channels:
+    - chat.freenode.net#pytest
+    on_success: change
+    on_failure: change
+    skip_join: true
+  email:
+  - pytest-commit@python.org
+deploy:
+  provider: pypi
+  user: ronny
+  password:
+    secure: cxmSDho5d+PYKEM4ZCg8ms1P4lzhYkrw6fEOm2HtTcsuCyY6aZMSgImWAnEYbJHSkdzgcxlXK9UKJ9B0YenXmBCkAr7UjdnpNXNmkySr0sYzlH/sfqt/dDATCHFaRKxnkOSOVywaDYhT9n8YudbXI77pXwD12i/CeSSJDbHhsu0JYUfAcb+D6YjRYoA2SEGCnzSzg+gDDfwXZx4ZiODCGLVwieNp1klCg88YROUE1BaYYNuUOONvfXX8+TWowbCF6ChH1WL/bZ49OStEYQNuYxZQZr4yClIqu9VJbchrU8j860K9ott2kkGTgfB/dDrQB/XncBubyIX9ikzCQAmmBXWAI3eyvWLPDk2Jz7kW2l2RT7syct80tCq3JhvQ1qdwr5ap7siocTLgnBW0tF4tkHSTFN3510fkc43npnp6FThebESQpnI24vqpwJ9hI/kW5mYi014Og2E/cpCXnz2XO8iZPDbqAMQpDsqEQoyhfGNgPTGp4K30TxRtwZBI5hHhDKnnR16fXtRgt1gYPvz/peUQvvpOm4JzIzGXPzluuutpnCBy75v5+oiwT3YRrLL/Meims9FtDDXL3qQubAE/ezIOOpm0N5XXV8DxIom8EN71yq5ab1tqhM+tBX7owRjy4FR4If2Q8feBdmTuh26DIQt/y+qSG8VkB9Sw/JCjc7c=
+  on:
+    tags: true
+    distributions: sdist bdist_wheel
+    repo: pytest-dev/pytest-xdist

CHANGELOG

@@ -1,3 +1,84 @@
+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
+------
+
+- new ``worker_id`` fixture, returns the id of the worker in a test or fixture.
+  Thanks Jared Hellman for the PR.
+
+- display progress during collection only when in a terminal, similar to pytest #1397 issue.
+  Thanks Bruno Oliveira for the PR.
+
+- fix internal error message when ``--maxfail`` is used (#62, #65).
+  Thanks Collin RM Stocks and Bryan A. Jones for reports and Bruno Oliveira for the PR.
+
+
+1.14
+----
+
+- new hook: ``pytest_xdist_node_collection_finished(node, ids)``, called when
+  a worker has finished collection. Thanks Omer Katz for the request and
+  Bruno Oliveira for the PR.
+
+- fix README display on pypi
+
+- fix #22: xdist now works if the internal tmpdir plugin is disabled.
+  Thanks Bruno Oliveira for the PR.
+
+- fix #32: xdist now works if looponfail or boxed are disabled.
+  Thanks Bruno Oliveira for the PR.
+
+
+1.13.1
+-------
+
+- fix a regression -n 0 now disables xdist again
+
+
+1.13
+-------------------------
+
+- extended the tox matrix with the supported py.test versions
+
+- split up the plugin into 3 plugin's
+  to prepare the departure of boxed and looponfail.
+
+  looponfail will be a part of core
+  and forked boxed will be replaced
+  with a more reliable primitive based on xdist
+
+- conforming with new pytest-2.8 behavior of returning non-zero when all
+  tests were skipped or deselected.
+
+- new "--max-slave-restart" option that can be used to control maximum
+  number of times pytest-xdist can restart slaves due to crashes. Thanks to
+  Anatoly Bubenkov for the report and Bruno Oliveira for the PR.
+
+- release as wheel
+
+- "-n" option now can be set to "auto" for automatic detection of number
+  of cpus in the host system. Thanks Suloev Dmitry for the PR.
+
+1.12
+-------------------------
+
+- fix issue594: properly report errors when the test collection
+  is random.  Thanks Bruno Oliveira.
+
+- some internal test suite adaptation (to become forward
+  compatible with the upcoming pytest-2.8)
+
+
 1.11
 -------------------------
 
@@ -10,7 +91,7 @@
   Thanks to Andreas Pelme for bug analysis and failing test.
 
 - restart crashed nodes by internally refactoring setup handling
-  of nodes.  Also includes better code documentation.  
+  of nodes.  Also includes better code documentation.
   Many thanks to Floris Bruynooghe for the complete PR.
 
 
@@ -23,7 +104,7 @@
 - fix pytest issue382 - produce "pytest_runtest_logstart" event again
   in master. Thanks Aron Curzon.
 
-- fix pytest issue419 by sending/receiving indices into the test 
+- fix pytest issue419 by sending/receiving indices into the test
   collection instead of node ids (which are not neccessarily unique
   for functions parametrized with duplicate values)
 
@@ -136,4 +217,3 @@
 - cleaned up termination handling
 - make -x cause hard killing of test nodes to decrease wait time
   until the traceback shows up on first failure
-

MANIFEST.in

@@ -4,4 +4,4 @@ include README.txt
 include setup.py
 include tox.ini
 graft testing
-prune .hg 
+prune .git

OVERVIEW.md

@@ -0,0 +1,78 @@
+# Overview #
+
+`xdist` works by spawning one or more **workers**, which are controlled
+by the **master**. Each **worker** is responsible for performing 
+a full test collection and afterwards running tests as dictated by the **master**.
+
+The execution flow is:
+
+1. **master** spawns one or more **workers** at the beginning of
+   the test session. The communication between **master** and **worker** nodes makes use of 
+   [execnet](http://codespeak.net/execnet/) and its [gateways](http://codespeak.net/execnet/basics.html#gateways-bootstrapping-python-interpreters).
+   The actual interpreters executing the code for the **workers** might
+   be remote or local. 
+  
+1. Each **worker** itself is a mini pytest runner. **workers** at this
+   point perform a full test collection, sending back the collected 
+   test-ids back to the **master** which does not
+   perform any collection itself.
+     
+1. The **master** receives the result of the collection from all nodes.
+   At this point the **master** performs some sanity check to ensure that
+   all **workers** collected the same tests (including order), bailing out otherwise.
+   If all is well, it converts the list of test-ids into a list of simple
+   indexes, where each index corresponds to the position of that test in the
+   original collection list. This works because all nodes have the same 
+   collection list, and saves bandwidth because the **master** can now tell
+   one of the workers to just *execute test index 3* index of passing the
+   full test id.
+   
+1. If **dist-mode** is **each**: the **master** just sends the full list
+   of test indexes to each node at this moment.
+   
+1. If **dist-mode** is **load**: the **master** takes around 25% of the
+   tests and sends them one by one to each **worker** in a round robin
+   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 
+   waiting for **master** to send tests for execution. As tests are
+   received by **workers**, `pytest_runtest_protocol` is executed for each test. 
+   Here it worth noting an implementation detail: **workers** always must keep at 
+   least one test item on their queue due to how the `pytest_runtest_protocol(item, nextitem)` 
+   hook is defined: in order to pass the `nextitem` to the hook, the worker must wait for more 
+   instructions from master before executing that remaining test. If it receives more tests, 
+   then it can safely call `pytest_runtest_protocol` because it knows what the `nextitem` parameter will be. 
+   If it receives a "shutdown" signal, then it can execute the hook passing `nextitem` as `None`. 
+   
+1. As tests are started and completed at the **workers**, the results are sent
+   back to the **master**, which then just forwards the results to 
+   the appropriate pytest hooks: `pytest_runtest_logstart` and 
+   `pytest_runtest_logreport`. This way other plugins (for example `junitxml`)
+   can work normally. The **master** (when in dist-mode **load**) 
+   decides to send more tests to a node when a test completes, using
+   some heuristics such as test durations and how many tests each **worker**
+   still has to run.
+   
+1. When the **master** has no more pending tests it will
+   send a "shutdown" signal to all **workers**, which will then run their 
+   remaining tests to completion and shut down. At this point the 
+   **master** will sit waiting for **workers** to shut down, still
+   processing events such as `pytest_runtest_logreport`.
+ 
+## FAQ ##
+
+> Why does each worker do its own collection, as opposed to having 
+the master collect once and distribute from that collection to the workers?
+
+If collection was performed by master then it would have to 
+serialize collected items to send them through the wire, as workers live in another process. 
+The problem is that test items are not easily (impossible?) to serialize, as they contain references to 
+the test functions, fixture managers, config objects, etc. Even if one manages to serialize it, 
+it seems it would be very hard to get it right and easy to break by any small change in pytest. 
+  
+

README.rst

@@ -1,212 +1,266 @@
-xdist: pytest distributed testing plugin
-===============================================================
-
-The `pytest-xdist`_ plugin extends py.test with some unique
-test execution modes:
-
-* test run parallelization_: if you have multiple CPUs or hosts you can use
-  those for a combined test run.  This allows to speed up
-  development or to use special resources of `remote machines`_.
-
-* ``--boxed``: (not available on Windows) run each test in a boxed_
-  subprocess to survive ``SEGFAULTS`` or otherwise dying processes
-
-* ``--looponfail``: run your tests repeatedly in a subprocess.  After each run
-  py.test waits until a file in your project changes and then re-runs
-  the previously failing tests.  This is repeated until all tests pass
-  after which again a full run is performed.
-
-* `Multi-Platform`_ coverage: you can specify different Python interpreters
-  or different platforms and run tests in parallel on all of them.
-
-Before running tests remotely, ``py.test`` efficiently "rsyncs" your
-program source code to the remote place.  All test results
-are reported back and displayed to your local terminal.
-You may specify different Python versions and interpreters.
-
-
-Installation
------------------------
-
-Install the plugin with::
-
-    easy_install pytest-xdist
-
-    # or
-
-    pip install pytest-xdist
-
-or use the package in develope/in-place mode with
-a checkout of the `pytest-xdist repository`_ ::
-
-    python setup.py develop
-
-Usage examples
----------------------
-
-.. _parallelization:
-
-Speed up test runs by sending tests to multiple CPUs
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-To send tests to multiple CPUs, type::
-
-    py.test -n NUM
-
-Especially for longer running tests or tests requiring
-a lot of IO this can lead to considerable speed ups.
-
-
-Running tests in a Python subprocess
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-To instantiate a python2.5 sub process and send tests to it, you may type::
-
-    py.test -d --tx popen//python=python2.5
-
-This will start a subprocess which is run with the "python2.5"
-Python interpreter, found in your system binary lookup path.
-
-If you prefix the --tx option value like this::
-
-    --tx 3*popen//python=python2.5
-
-then three subprocesses would be created and tests
-will be load-balanced across these three processes.
-
-.. _boxed:
-
-Running tests in a boxed subprocess
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-If you have tests involving C or C++ libraries you might have to deal
-with tests crashing the process.  For this case you may use the boxing
-options::
-
-    py.test --boxed
-
-which will run each test in a subprocess and will report if a test
-crashed the process.  You can also combine this option with
-running multiple processes to speed up the test run and use your CPU cores::
-
-    py.test -n3 --boxed
-
-this would run 3 testing subprocesses in parallel which each
-create new boxed subprocesses for each test.
-
-
-.. _`remote machines`:
-
-Sending tests to remote SSH accounts
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-Suppose you have a package ``mypkg`` which contains some
-tests that you can successfully run locally. And you
-have a ssh-reachable machine ``myhost``.  Then
-you can ad-hoc distribute your tests by typing::
-
-    py.test -d --tx ssh=myhostpopen --rsyncdir mypkg mypkg
-
-This will synchronize your ``mypkg`` package directory
-to an remote ssh account and then locally collect tests
-and send them to remote places for execution.
-
-You can specify multiple ``--rsyncdir`` directories
-to be sent to the remote side.
-
-**NOTE:** For py.test to collect and send tests correctly
-you not only need to make sure all code and tests
-directories are rsynced, but that any test (sub) directory
-also has an ``__init__.py`` file because internally
-py.test references tests as a fully qualified python
-module path.  **You will otherwise get strange errors**
-during setup of the remote side.
-
-You can specify multiple ``--rsyncignore`` glob-patterns
-to be ignored when file are sent to the remote side.
-There are also internal ignores: .*, *.pyc, *.pyo, *~
-Those you cannot override using rsyncignore command-line or
-ini-file option(s).
-
-
-Sending tests to remote Socket Servers
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-Download the single-module `socketserver.py`_ Python program
-and run it like this::
-
-    python socketserver.py
-
-It will tell you that it starts listening on the default
-port.  You can now on your home machine specify this
-new socket host with something like this::
-
-    py.test -d --tx socket=192.168.1.102:8888 --rsyncdir mypkg mypkg
-
-
-.. _`atonce`:
-.. _`Multi-Platform`:
-
-
-Running tests on many platforms at once
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-The basic command to run tests on multiple platforms is::
-
-    py.test --dist=each --tx=spec1 --tx=spec2
-
-If you specify a windows host, an OSX host and a Linux
-environment this command will send each tests to all
-platforms - and report back failures from all platforms
-at once.   The specifications strings use the `xspec syntax`_.
-
-.. _`xspec syntax`: http://codespeak.net/execnet/trunk/basics.html#xspec
-
-.. _`socketserver.py`: http://bitbucket.org/hpk42/execnet/raw/2af991418160/execnet/script/socketserver.py
-
-.. _`execnet`: http://codespeak.net/execnet
-
-Specifying test exec environments in an ini file
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-pytest (since version 2.0) supports ini-style cofiguration.
-You can for example make running with three subprocesses
-your default like this::
-
-    [pytest]
-    addopts = -n3
-
-You can also add default environments like this::
-
-    [pytest]
-    addopts = --tx ssh=myhost//python=python2.5 --tx ssh=myhost//python=python2.6
-
-and then just type::
-
-    py.test --dist=each
-
-to run tests in each of the environments.
-
-Specifying "rsync" dirs in an ini-file
-+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
-
-In a ``tox.ini`` or ``setup.cfg`` file in your root project directory
-you may specify directories to include or to exclude in synchronisation::
-
-    [pytest]
-    rsyncdirs = . mypkg helperpkg
-    rsyncignore = .hg
-
-These directory specifications are relative to the directory
-where the configuration file was found.
-
-.. _`pytest-xdist`: http://pypi.python.org/pypi/pytest-xdist
-.. _`pytest-xdist repository`: http://bitbucket.org/hpk42/pytest-xdist
-.. _`pytest`: http://pytest.org
-
-Issue and Bug Tracker
-------------------------
-
-Please use the pytest issue tracker for bugs in this plugin, see https://bitbucket.org/hpk42/pytest/issues .
-
-
+
+.. 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
+
+xdist: pytest distributed testing plugin
+=========================================
+
+The `pytest-xdist`_ plugin extends py.test with some unique
+test execution modes:
+
+* test run parallelization_: if you have multiple CPUs or hosts you can use
+  those for a combined test run.  This allows to speed up
+  development or to use special resources of `remote machines`_.
+
+* ``--boxed``: (not available on Windows) run each test in a boxed_
+  subprocess to survive ``SEGFAULTS`` or otherwise dying processes
+
+* ``--looponfail``: run your tests repeatedly in a subprocess.  After each run
+  py.test waits until a file in your project changes and then re-runs
+  the previously failing tests.  This is repeated until all tests pass
+  after which again a full run is performed.
+
+* `Multi-Platform`_ coverage: you can specify different Python interpreters
+  or different platforms and run tests in parallel on all of them.
+
+Before running tests remotely, ``py.test`` efficiently "rsyncs" your
+program source code to the remote place.  All test results
+are reported back and displayed to your local terminal.
+You may specify different Python versions and interpreters.
+
+
+Installation
+-----------------------
+
+Install the plugin with::
+
+    easy_install pytest-xdist
+
+    # or
+
+    pip install pytest-xdist
+
+or use the package in develop/in-place mode with
+a checkout of the `pytest-xdist repository`_ ::
+
+    python setup.py develop
+
+Usage examples
+---------------------
+
+.. _parallelization:
+
+Speed up test runs by sending tests to multiple CPUs
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+To send tests to multiple CPUs, type::
+
+    py.test -n NUM
+
+Especially for longer running tests or tests requiring
+a lot of IO this can lead to considerable speed ups. This option can
+also be set to ``auto`` for automatic detection of the number of CPUs.
+
+If a test crashes the interpreter, pytest-xdist will automatically restart
+that slave and report the failure as usual. You can use the
+``--max-slave-restart`` option to limit the number of slaves that can
+be restarted, or disable restarting altogether using ``--max-slave-restart=0``.
+
+
+Running tests in a Python subprocess
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+To instantiate a python2.5 sub process and send tests to it, you may type::
+
+    py.test -d --tx popen//python=python2.5
+
+This will start a subprocess which is run with the "python2.5"
+Python interpreter, found in your system binary lookup path.
+
+If you prefix the --tx option value like this::
+
+    --tx 3*popen//python=python2.5
+
+then three subprocesses would be created and tests
+will be load-balanced across these three processes.
+
+.. _boxed:
+
+Running tests in a boxed subprocess
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+If you have tests involving C or C++ libraries you might have to deal
+with tests crashing the process.  For this case you may use the boxing
+options::
+
+    py.test --boxed
+
+which will run each test in a subprocess and will report if a test
+crashed the process.  You can also combine this option with
+running multiple processes to speed up the test run and use your CPU cores::
+
+    py.test -n3 --boxed
+
+this would run 3 testing subprocesses in parallel which each
+create new boxed subprocesses for each test.
+
+
+.. _`remote machines`:
+
+Sending tests to remote SSH accounts
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+Suppose you have a package ``mypkg`` which contains some
+tests that you can successfully run locally. And you
+have a ssh-reachable machine ``myhost``.  Then
+you can ad-hoc distribute your tests by typing::
+
+    py.test -d --tx ssh=myhostpopen --rsyncdir mypkg mypkg
+
+This will synchronize your :code:`mypkg` package directory
+to an remote ssh account and then locally collect tests
+and send them to remote places for execution.
+
+You can specify multiple :code:`--rsyncdir` directories
+to be sent to the remote side.
+
+.. note::
+
+  For py.test to collect and send tests correctly
+  you not only need to make sure all code and tests
+  directories are rsynced, but that any test (sub) directory
+  also has an :code:`__init__.py` file because internally
+  py.test references tests as a fully qualified python
+  module path.  **You will otherwise get strange errors**
+  during setup of the remote side.
+
+
+You can specify multiple :code:`--rsyncignore` glob patterns
+to be ignored when file are sent to the remote side.
+There are also internal ignores: :code:`.*, *.pyc, *.pyo, *~`
+Those you cannot override using rsyncignore command-line or
+ini-file option(s).
+
+
+Sending tests to remote Socket Servers
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+Download the single-module `socketserver.py`_ Python program
+and run it like this::
+
+    python socketserver.py
+
+It will tell you that it starts listening on the default
+port.  You can now on your home machine specify this
+new socket host with something like this::
+
+    py.test -d --tx socket=192.168.1.102:8888 --rsyncdir mypkg mypkg
+
+
+.. _`atonce`:
+.. _`Multi-Platform`:
+
+
+Running tests on many platforms at once
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+The basic command to run tests on multiple platforms is::
+
+    py.test --dist=each --tx=spec1 --tx=spec2
+
+If you specify a windows host, an OSX host and a Linux
+environment this command will send each tests to all
+platforms - and report back failures from all platforms
+at once.   The specifications strings use the `xspec syntax`_.
+
+.. _`xspec syntax`: http://codespeak.net/execnet/basics.html#xspec
+
+.. _`socketserver.py`: http://bitbucket.org/hpk42/execnet/raw/2af991418160/execnet/script/socketserver.py
+
+.. _`execnet`: http://codespeak.net/execnet
+
+Identifying the worker process during a test
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+
+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:
+
+.. code-block:: python
+
+    @pytest.fixture()
+    def user_account(worker_id):
+        """ use a different account in each xdist worker """
+        return "account_%s" % worker_id
+
+When ``xdist`` is disabled (running with ``-n0`` for example), then
+``worker_id`` will return ``"master"``.
+
+Additionally, worker processes have the following environment variables
+defined:
+
+* ``PYTEST_XDIST_WORKER``: the name of the worker, e.g., ``"gw2"``.
+* ``PYTEST_XDIST_WORKER_COUNT``: the total number of workers in this session,
+  e.g., ``"4"`` when ``-n 4`` is given in the command-line.
+
+The information about the worker_id in a test is stored in the TestReport as
+well, under worker_id attribute.
+
+
+*New in version 1.15.*
+
+Specifying test exec environments in an ini file
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+pytest (since version 2.0) supports ini-style cofiguration.
+You can for example make running with three subprocesses
+your default like this:
+
+.. code-block:: ini
+
+    [pytest]
+    addopts = -n3
+
+You can also add default environments like this:
+
+.. code-block:: ini
+
+    [pytest]
+    addopts = --tx ssh=myhost//python=python2.5 --tx ssh=myhost//python=python2.6
+
+and then just type::
+
+    py.test --dist=each
+
+to run tests in each of the environments.
+
+Specifying "rsync" dirs in an ini-file
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+In a ``tox.ini`` or ``setup.cfg`` file in your root project directory
+you may specify directories to include or to exclude in synchronisation:
+
+.. code-block:: ini
+
+    [pytest]
+    rsyncdirs = . mypkg helperpkg
+    rsyncignore = .hg
+
+These directory specifications are relative to the directory
+where the configuration file was found.
+
+.. _`pytest-xdist`: http://pypi.python.org/pypi/pytest-xdist
+.. _`pytest-xdist repository`: 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

@@ -0,0 +1,30 @@
+environment:
+  matrix:
+  # note: please use "tox --listenvs" to populate the build matrix
+  - TOXENV: "py26-pytest27"
+  - TOXENV: "py26-pytest28"
+  - TOXENV: "py26-pytest29"
+  - TOXENV: "py26-pytest30"
+  - TOXENV: "py27-pytest27"
+  - TOXENV: "py27-pytest28"
+  - TOXENV: "py27-pytest29"
+  - TOXENV: "py27-pytest30"
+  - TOXENV: "py34-pytest27"
+  - TOXENV: "py34-pytest28"
+  - TOXENV: "py34-pytest29"
+  - TOXENV: "py34-pytest30"
+  - TOXENV: "py35-pytest27"
+  - TOXENV: "py35-pytest28"
+  - TOXENV: "py35-pytest29"
+  - TOXENV: "py27-pytest28-pexpect"
+  - TOXENV: "py35-pytest28-pexpect"
+  - TOXENV: "flakes"
+  - TOXENV: "readme"
+
+install:
+  - C:\Python35\python -m pip install tox setuptools_scm
+
+build: false  # Not a C# project, build stuff at the test step instead.
+
+test_script:
+  - C:\Python35\python -m tox

example/boxed.txt

@@ -2,7 +2,7 @@
 
 If your testing involves C or C++ libraries you might have to deal
 with crashing processes.  The xdist-plugin provides the ``--boxed`` option
-to run each test in a controled subprocess.  Here is a basic example::
+to run each test in a controlled subprocess.  Here is a basic example::
 
     # content of test_module.py
 

setup.cfg

@@ -0,0 +1,5 @@
+[bdist_wheel]
+universal = 1
+
+[metadata]
+license_file = LICENSE

setup.py

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

testing/acceptance_test.py

@@ -10,9 +10,7 @@ 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("""
@@ -21,9 +19,7 @@ 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("""
@@ -57,9 +53,7 @@ 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("""
@@ -68,7 +62,7 @@ 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*",
         ])
@@ -84,8 +78,7 @@ class TestDistribution:
                     pass
                 def test_skip():
                     py.test.skip("hello")
-            """,
-        )
+            """, )
         result = testdir.runpytest(p1, "-v", '-d', '--tx=popen', '--tx=popen')
         result.stdout.fnmatch_lines([
             "*1*Python*",
@@ -113,11 +106,9 @@ class TestDistribution:
                 import py
                 assert tmpdir.relto(py.path.local(%r)), tmpdir
         """ % str(testdir.tmpdir))
-        result = testdir.runpytest(p1, "-n1")
+        result = testdir.runpytest_subprocess(p1, "-n1")
         assert result.ret == 0
-        result.stdout.fnmatch_lines([
-            "*1 passed*",
-        ])
+        result.stdout.fnmatch_lines(["*1 passed*", ])
 
     def test_dist_ini_specified(self, testdir):
         p1 = testdir.makepyfile("""
@@ -130,8 +121,7 @@ class TestDistribution:
                     pass
                 def test_skip():
                     py.test.skip("hello")
-            """,
-        )
+            """, )
         testdir.makeini("""
             [pytest]
             addopts = --tx=3*popen
@@ -163,13 +153,10 @@ class TestDistribution:
                     import os
                     time.sleep(0.5)
                     os.kill(os.getpid(), 15)
-            """
-        )
+            """)
         result = testdir.runpytest(p1, "-v", '-d', '-n1')
         result.stdout.fnmatch_lines([
-            "*Python*",
-            "*PASS**test_ok*",
-            "*node*down*",
+            "*Python*", "*PASS**test_ok*", "*node*down*",
             "*3 failed, 1 passed, 1 skipped*"
         ])
         assert result.ret == 1
@@ -182,13 +169,13 @@ class TestDistribution:
         p = subdir.join("test_one.py")
         p.write("def test_5():\n  assert not __file__.startswith(%r)" % str(p))
         result = testdir.runpytest("-v", "-d",
-            "--rsyncdir=%(subdir)s" % locals(),
-            "--tx=popen//chdir=%(dest)s" % locals(), p)
+                                   "--rsyncdir=%(subdir)s" % locals(),
+                                   "--tx=popen//chdir=%(dest)s" % locals(), p)
         assert result.ret == 0
         result.stdout.fnmatch_lines([
             "*0* *cwd*",
-            #"RSyncStart: [G1]",
-            #"RSyncFinished: [G1]",
+            # "RSyncStart: [G1]",
+            # "RSyncFinished: [G1]",
             "*1 passed*"
         ])
         assert dest.join(subdir.basename).check(dir=1)
@@ -221,14 +208,13 @@ 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*"
+            "*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()
         """)
@@ -243,7 +229,7 @@ class TestDistribution:
                 print ("s2call-finished")
         """)
         args = ["-n1", "--debug"]
-        result = testdir.runpytest(*args)
+        result = testdir.runpytest_subprocess(*args)
         s = result.stdout.str()
         assert result.ret == 2
         assert 's2call' in s
@@ -256,14 +242,14 @@ class TestDistribution:
                 import time
                 time.sleep(10)
         """)
-        child = testdir.spawn_pytest("-n1")
-        py.std.time.sleep(0.1)
-        child.expect(".*test session starts.*")
-        child.kill(2) # keyboard interrupt
+        child = testdir.spawn_pytest("-n1 -v")
+        child.expect(".*test_sleep.*")
+        child.kill(2)  # keyboard interrupt
         child.expect(".*KeyboardInterrupt.*")
-        #child.expect(".*seconds.*")
+        # child.expect(".*seconds.*")
         child.close()
-        #assert ret == 2
+        # assert ret == 2
+
 
 class TestDistEach:
     def test_simple(self, testdir):
@@ -271,11 +257,13 @@ class TestDistEach:
             def test_hello():
                 pass
         """)
-        result = testdir.runpytest("--debug", "--dist=each", "--tx=2*popen")
+        result = testdir.runpytest_subprocess("--debug", "--dist=each",
+                                              "--tx=2*popen")
         assert not result.ret
         result.stdout.fnmatch_lines(["*2 pass*"])
 
-    @py.test.mark.xfail(run=False,
+    @py.test.mark.xfail(
+        run=False,
         reason="other python versions might not have py.test installed")
     def test_simple_diffoutput(self, testdir):
         interpreters = []
@@ -285,7 +273,8 @@ 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])
@@ -299,6 +288,7 @@ class TestDistEach:
         assert "2...5" in s
         assert "2...6" in s
 
+
 class TestTerminalReporting:
     def test_pass_skip_fail(self, testdir):
         testdir.makepyfile("""
@@ -336,6 +326,7 @@ class TestTerminalReporting:
             "E       assert 0",
         ])
 
+
 def test_teardownfails_one_function(testdir):
     p = testdir.makepyfile("""
         def test_func():
@@ -345,10 +336,10 @@ def test_teardownfails_one_function(testdir):
     """)
     result = testdir.runpytest(p, '-n1', '--tx=popen')
     result.stdout.fnmatch_lines([
-        "*def teardown_function(function):*",
-        "*1 passed*1 error*"
+        "*def teardown_function(function):*", "*1 passed*1 error*"
     ])
 
+
 @py.test.mark.xfail
 def test_terminate_on_hangingnode(testdir):
     p = testdir.makeconftest("""
@@ -359,9 +350,7 @@ 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)
@@ -388,19 +377,43 @@ 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
     assert testdir.tmpdir.join("slave").check()
     assert testdir.tmpdir.join("master").check()
 
+
+def test_session_testscollected(testdir):
+    """
+    Make sure master node is updating the session object with the number
+    of tests collected from the slaves.
+    """
+    testdir.makepyfile(test_foo="""
+        import pytest
+        @pytest.mark.parametrize('i', range(3))
+        def test_ok(i):
+            pass
+    """)
+    testdir.makeconftest("""
+        def pytest_sessionfinish(session):
+            collected = getattr(session, 'testscollected', None)
+            with open('testscollected', 'w') as f:
+                f.write('collected = %s' % collected)
+    """)
+    result = testdir.inline_run("-n1")
+    result.assertoutcome(passed=3)
+    collected_file = testdir.tmpdir.join('testscollected')
+    assert collected_file.isfile()
+    assert collected_file.read() == 'collected = 3'
+
+
 def test_funcarg_teardown_failure(testdir):
     p = testdir.makepyfile("""
-        def pytest_funcarg__myarg(request):
+        import pytest
+        @pytest.fixture
+        def myarg(request):
             def teardown(val):
                 raise ValueError(val)
             return request.cached_setup(setup=lambda: 42, teardown=teardown,
@@ -408,13 +421,11 @@ def test_funcarg_teardown_failure(testdir):
         def test_hello(myarg):
             pass
     """)
-    result = testdir.runpytest("--debug", p) # , "-n1")
-    result.stdout.fnmatch_lines([
-        "*ValueError*42*",
-        "*1 passed*1 error*",
-    ])
+    result = testdir.runpytest_subprocess("--debug", p)  # , "-n1")
+    result.stdout.fnmatch_lines(["*ValueError*42*", "*1 passed*1 error*", ])
     assert result.ret
 
+
 def test_crashing_item(testdir):
     p = testdir.makepyfile("""
         import py
@@ -426,12 +437,10 @@ def test_crashing_item(testdir):
     """)
     result = testdir.runpytest("-n2", p)
     result.stdout.fnmatch_lines([
-        "*crashed*test_crash*",
-        "*1 failed*1 passed*"
+        "*crashed*test_crash*", "*1 failed*1 passed*"
     ])
 
 
-
 def test_skipping(testdir):
     p = testdir.makepyfile("""
         import pytest
@@ -440,10 +449,8 @@ def test_skipping(testdir):
     """)
     result = testdir.runpytest("-n1", '-rs', p)
     assert result.ret == 0
-    result.stdout.fnmatch_lines([
-        "*hello*",
-        "*1 skipped*"
-    ])
+    result.stdout.fnmatch_lines(["*hello*", "*1 skipped*"])
+
 
 def test_issue34_pluginloading_in_subprocess(testdir):
     testdir.tmpdir.join("plugin123.py").write(py.code.Source("""
@@ -455,11 +462,9 @@ def test_issue34_pluginloading_in_subprocess(testdir):
         def test_hello():
             assert pytest.sample_variable == "testing"
     """)
-    result = testdir.runpytest("-n1", "-p", "plugin123")
+    result = testdir.runpytest_subprocess("-n1", "-p", "plugin123")
     assert result.ret == 0
-    result.stdout.fnmatch_lines([
-        "*1 passed*",
-    ])
+    result.stdout.fnmatch_lines(["*1 passed*", ])
 
 
 def test_fixture_scope_caching_issue503(testdir):
@@ -468,7 +473,8 @@ def test_fixture_scope_caching_issue503(testdir):
 
             @pytest.fixture(scope='session')
             def fix():
-                assert fix.counter == 0, 'session fixture was invoked multiple times'
+                assert fix.counter == 0, \
+                    'session fixture was invoked multiple times'
                 fix.counter += 1
             fix.counter = 0
 
@@ -480,13 +486,57 @@ def test_fixture_scope_caching_issue503(testdir):
     """)
     result = testdir.runpytest(p1, '-v', '-n1')
     assert result.ret == 0
+    result.stdout.fnmatch_lines(["*2 passed*", ])
+
+
+def test_issue_594_random_parametrize(testdir):
+    """
+    Make sure that tests that are randomly parametrized display an appropriate
+    error message, instead of silently skipping the entire test run.
+    """
+    p1 = testdir.makepyfile("""
+        import pytest
+        import random
+
+        xs = list(range(10))
+        random.shuffle(xs)
+        @pytest.mark.parametrize('x', xs)
+        def test_foo(x):
+            assert 1
+    """)
+    result = testdir.runpytest(p1, '-v', '-n4')
+    assert result.ret == 1
     result.stdout.fnmatch_lines([
-    "*2 passed*",
+        "Different tests were collected between gw* and gw*",
     ])
 
 
-class TestNodeFailure:
+def test_tmpdir_disabled(testdir):
+    """Test xdist doesn't break if internal tmpdir plugin is disabled (#22).
+    """
+    p1 = testdir.makepyfile("""
+        def test_ok():
+            pass
+    """)
+    result = testdir.runpytest(p1, "-n1", '-p', 'no:tmpdir')
+    assert result.ret == 0
+    result.stdout.fnmatch_lines("*1 passed*")
 
+
+@pytest.mark.parametrize('plugin', ['xdist.looponfail', 'xdist.boxed'])
+def test_sub_plugins_disabled(testdir, plugin):
+    """Test that xdist doesn't break if we disable any of its sub-plugins. (#32)
+    """
+    p1 = testdir.makepyfile("""
+        def test_ok():
+            pass
+    """)
+    result = testdir.runpytest(p1, "-n1", '-p', 'no:%s' % plugin)
+    assert result.ret == 0
+    result.stdout.fnmatch_lines("*1 passed*")
+
+
+class TestNodeFailure:
     def test_load_single(self, testdir):
         f = testdir.makepyfile("""
             import os
@@ -495,7 +545,7 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '-n1')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*1 passed*",
         ])
@@ -510,7 +560,7 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '-n2')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*3 passed*",
         ])
@@ -523,11 +573,12 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '--dist=each', '--tx=popen')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*1 failed*1 passed*",
         ])
 
+    @pytest.mark.xfail(reason='#20: xdist race condition on node restart')
     def test_each_multiple(self, testdir):
         f = testdir.makepyfile("""
             import os
@@ -536,7 +587,103 @@ class TestNodeFailure:
         """)
         res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
         res.stdout.fnmatch_lines([
-            "*Replacing failed node*",
+            "*Replacing crashed slave*",
             "*Slave*crashed while running*",
             "*2 failed*2 passed*",
         ])
+
+    def test_max_slave_restart(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): os._exit(1)
+            def test_d(): pass
+        """)
+        res = testdir.runpytest(f, '-n4', '--max-slave-restart=1')
+        res.stdout.fnmatch_lines([
+            "*Replacing crashed slave*",
+            "*Maximum crashed slaves reached: 1*",
+            "*Slave*crashed while running*",
+            "*Slave*crashed while running*",
+            "*2 failed*2 passed*",
+        ])
+
+    def test_disable_restart(self, testdir):
+        f = testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): pass
+        """)
+        res = testdir.runpytest(f, '-n4', '--max-slave-restart=0')
+        res.stdout.fnmatch_lines([
+            "*Slave restarting disabled*",
+            "*Slave*crashed while running*",
+            "*1 failed*2 passed*",
+        ])
+
+
+@pytest.mark.parametrize('n', [0, 2])
+def test_worker_id_fixture(testdir, n):
+    import glob
+    f = testdir.makepyfile("""
+        import pytest
+        @pytest.mark.parametrize("run_num", range(2))
+        def test_worker_id1(worker_id, run_num):
+            with open("worker_id%s.txt" % run_num, "w") as f:
+                f.write(worker_id)
+    """)
+    result = testdir.runpytest(f, "-n%d" % n)
+    result.stdout.fnmatch_lines('* 2 passed in *')
+    worker_ids = set()
+    for fname in glob.glob(str(testdir.tmpdir.join("*.txt"))):
+        with open(fname) as f:
+            worker_ids.add(f.read().strip())
+    if n == 0:
+        assert worker_ids == set(['master'])
+    else:
+        assert worker_ids == set(['gw0', 'gw1'])
+
+
+def test_color_yes_collection_on_non_atty(testdir, request):
+    """skip collect progress report when working on non-terminals.
+
+    Similar to pytest-dev/pytest#1397
+    """
+    tr = request.config.pluginmanager.getplugin("terminalreporter")
+    if not hasattr(tr, 'isatty'):
+        pytest.skip('only valid for newer pytest versions')
+    testdir.makepyfile("""
+        import pytest
+        @pytest.mark.parametrize('i', range(10))
+        def test_this(i):
+            assert 1
+    """)
+    args = ['--color=yes', '-n2']
+    result = testdir.runpytest(*args)
+    assert 'test session starts' in result.stdout.str()
+    assert '\x1b[1m' in result.stdout.str()
+    assert 'gw0 [10] / gw1 [10]' in result.stdout.str()
+    assert 'gw0 C / gw1 C' not in result.stdout.str()
+
+
+def test_internal_error_with_maxfail(testdir):
+    """
+    Internal error when using --maxfail option (#62, #65).
+    """
+    testdir.makepyfile("""
+        import pytest
+
+        @pytest.fixture(params=['1', '2'])
+        def crasher():
+            raise RuntimeError
+
+        def test_aaa0(crasher):
+            pass
+        def test_aaa1(crasher):
+            pass
+    """)
+    result = testdir.runpytest_subprocess('--maxfail=1', '-n1')
+    result.stdout.fnmatch_lines(['* 1 error in *'])
+    assert 'INTERNALERROR' not in result.stderr.str()

testing/conftest.py

@@ -1,22 +1,59 @@
 import py
+import pytest
 import execnet
 
+
+@pytest.fixture(scope="session", autouse=True)
+def _ensure_imports():
+    # we import some modules because pytest-2.8's testdir fixture
+    # will unload all modules after each test and this cause
+    # (unknown) problems with execnet.Group()
+    execnet.Group
+    execnet.makegateway
+
+
 pytest_plugins = "pytester"
 
-#rsyncdirs = ['.', '../xdist', py.path.local(execnet.__file__).dirpath()]
+# rsyncdirs = ['.', '../xdist', py.path.local(execnet.__file__).dirpath()]
+
+
+@pytest.fixture(autouse=True)
+def _divert_atexit(request, monkeypatch):
+    import atexit
+    l = []
+
+    def finish():
+        while l:
+            l.pop()()
+
+    monkeypatch.setattr(atexit, "register", l.append)
+    request.addfinalizer(finish)
+
 
 def pytest_addoption(parser):
     parser.addoption('--gx',
-       action="append", dest="gspecs",
-       help=("add a global test environment, XSpec-syntax. "))
+                     action="append",
+                     dest="gspecs",
+                     help=("add a global test environment, XSpec-syntax. "))
 
-def pytest_funcarg__specssh(request):
+
+@pytest.fixture
+def specssh(request):
     return getspecssh(request.config)
 
+
+@pytest.fixture
+def testdir(testdir):
+    # pytest before 2.8 did not have a runpytest_subprocess
+    if not hasattr(testdir, "runpytest_subprocess"):
+        testdir.runpytest_subprocess = testdir.runpytest
+    return testdir
+
+
 # configuration information for tests
 def getgspecs(config):
-    return [execnet.XSpec(spec)
-                for spec in config.getvalueorskip("gspecs")]
+    return [execnet.XSpec(spec) for spec in config.getvalueorskip("gspecs")]
+
 
 def getspecssh(config):
     xspecs = getgspecs(config)
@@ -27,10 +64,10 @@ def getspecssh(config):
             return str(spec)
     py.test.skip("need '--gx ssh=...'")
 
+
 def getsocketspec(config):
     xspecs = getgspecs(config)
     for spec in xspecs:
         if spec.socket:
             return spec
     py.test.skip("need '--gx socket=...'")
-

testing/test_boxed.py

@@ -4,6 +4,7 @@ import os
 needsfork = pytest.mark.skipif(not hasattr(os, "fork"),
                                reason="os.fork required")
 
+
 @needsfork
 def test_functional_boxed(testdir):
     p1 = testdir.makepyfile("""
@@ -17,6 +18,7 @@ def test_functional_boxed(testdir):
         "*1 failed*"
     ])
 
+
 @needsfork
 @pytest.mark.parametrize("capmode", [
     "no",
@@ -41,6 +43,7 @@ def test_functional_boxed_capturing(testdir, capmode):
         *1 failed*
     """)
 
+
 class TestOptionEffects:
     def test_boxed_option_default(self, testdir):
         tmpdir = testdir.tmpdir.ensure("subdir", dir=1)
@@ -53,4 +56,3 @@ class TestOptionEffects:
     def test_is_not_boxed_by_default(self, testdir):
         config = testdir.parseconfig(testdir.tmpdir)
         assert not config.option.boxed
-

testing/test_dsession.py

@@ -1,8 +1,5 @@
 from xdist.dsession import (
-    DSession,
-    LoadScheduling,
-    EachScheduling,
-    report_collection_diff,
+    DSession, LoadScheduling, EachScheduling, report_collection_diff,
 )
 import py
 import pytest
@@ -10,23 +7,27 @@ import execnet
 
 XSpec = execnet.XSpec
 
+
 def run(item, node, excinfo=None):
     runner = item.config.pluginmanager.getplugin("runner")
-    rep = runner.ItemTestReport(item=item,
-        excinfo=excinfo, when="call")
+    rep = runner.ItemTestReport(item=item, excinfo=excinfo, when="call")
     rep.node = node
     return rep
 
+
 class MockGateway:
     _count = 0
+
     def __init__(self):
         self.id = str(self._count)
         self._count += 1
 
+
 class MockNode:
     def __init__(self):
         self.sent = []
         self.gateway = MockGateway()
+        self._shutdown = False
 
     def send_runtest_some(self, indices):
         self.sent.extend(indices)
@@ -35,137 +36,200 @@ class MockNode:
         self.sent.append("ALL")
 
     def shutdown(self):
-        self._shutdown=True
+        self._shutdown = True
+
+    @property
+    def shutting_down(self):
+        return self._shutdown
+
 
 def dumpqueue(queue):
     while queue.qsize():
         print(queue.get())
 
+
 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 not sched.tests_finished()
-        assert len(node1.sent) == 2
-        assert len(node2.sent) == 0
-        assert node1.sent == [0, 1]
-        sched.remove_item(node1, node1.sent[0])
-        assert sched.tests_finished()
-        sched.remove_item(node1, node1.sent[1])
-        assert sched.tests_finished()
+        assert sched.tests_finished
+        assert len(node1.sent) == 1
+        assert len(node2.sent) == 1
+        assert node1.sent == [0]
+        assert node2.sent == [1]
+        sched.mark_test_complete(node1, node1.sent[0])
+        assert sched.tests_finished
 
-    def test_init_distribute_chunksize(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, 1]
-        assert sent2 == [2, 3]
+        assert sent1 == [0, 2]
+        assert sent2 == [1, 3]
         assert sched.pending == [4, 5]
         assert sched.node2pending[node1] == sent1
         assert sched.node2pending[node2] == sent2
         assert len(sched.pending) == 2
-        sched.remove_item(node1, 0)
-        assert node1.sent == [0, 1, 4]
+        sched.mark_test_complete(node1, 0)
+        assert node1.sent == [0, 2, 4]
         assert sched.pending == [5]
-        assert node2.sent == [2, 3]
-        sched.remove_item(node1, 1)
-        assert node1.sent == [0, 1, 4, 5]
+        assert node2.sent == [1, 3]
+        sched.mark_test_complete(node1, 2)
+        assert node1.sent == [0, 2, 4, 5]
         assert not sched.pending
 
-    def test_add_remove_node(self):
+    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.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
+        assert sent1 == [0]
+        assert sent2 == [1]
+        assert sent3 == []
+        assert not sched.pending
+
+    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.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
+        assert sent1 == [0, 3]
+        assert sent2 == [1, 4]
+        assert sent3 == [2]
+        assert not sched.pending
+
+    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]
 
-    def test_schedule_different_tests_collected(self):
+    def test_different_tests_collected(self, testdir):
         """
-        Test that LoadScheduling is logging different tests were
-        collected by slaves when that happens.
+        Test that LoadScheduling is reporting collection errors when
+        different test ids are collected by slaves.
         """
+
+        class CollectHook(object):
+            """
+            Dummy hook that stores collection reports.
+            """
+
+            def __init__(self):
+                self.reports = []
+
+            def pytest_collectreport(self, report):
+                self.reports.append(report)
+
+        collect_hook = CollectHook()
+        config = testdir.parseconfig("--tx=2*popen")
+        config.pluginmanager.register(collect_hook, "collect_hook")
         node1 = MockNode()
         node2 = MockNode()
-        sched = LoadScheduling(2)
-        logged_messages = []
-        py.log.setconsumer('loadsched', logged_messages.append)
-        sched.addnode(node1)
-        sched.addnode(node2)
-        sched.addnode_collection(node1, ["a.py::test_1"])
-        sched.addnode_collection(node2, ["a.py::test_2"])
-        sched.init_distribute()
-        logged_content = ''.join(x.content() for x in logged_messages)
-        assert 'Different tests were collected between' in logged_content
-        assert 'Different tests collected, aborting run' in logged_content
+        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
 
 
 class TestDistReporter:
-
     @py.test.mark.xfail
     def test_rsync_printing(self, testdir, linecomp):
         config = testdir.parseconfig()
@@ -173,26 +237,26 @@ class TestDistReporter:
         rep = TerminalReporter(config, file=linecomp.stringio)
         config.pluginmanager.register(rep, "terminalreporter")
         dsession = DSession(config)
+
         class gw1:
             id = "X1"
             spec = execnet.XSpec("popen")
+
         class gw2:
             id = "X2"
             spec = execnet.XSpec("popen")
-        #class rinfo:
+        # class rinfo:
         #    version_info = (2, 5, 1, 'final', 0)
         #    executable = "hello"
         #    platform = "xyz"
         #    cwd = "qwe"
 
-        #dsession.pytest_xdist_newgateway(gw1, rinfo)
-        #linecomp.assert_contains_lines([
-        #    "*X1*popen*xyz*2.5*"
-        #])
+        # dsession.pytest_xdist_newgateway(gw1, rinfo)
+        # linecomp.assert_contains_lines([
+        #     "*X1*popen*xyz*2.5*"
+        # ])
         dsession.pytest_xdist_rsyncstart(source="hello", gateways=[gw1, gw2])
-        linecomp.assert_contains_lines([
-            "[X1,X2] rsyncing: hello",
-        ])
+        linecomp.assert_contains_lines(["[X1,X2] rsyncing: hello", ])
 
 
 def test_report_collection_diff_equal():
@@ -218,12 +282,12 @@ def test_report_collection_diff_different():
         ' bbb\n'
         '+XXX\n'
         ' ccc\n'
-        '-YYY'
-    )
+        '-YYY')
 
-    msg = report_collection_diff(from_collection, to_collection, 1, 2)
+    msg = report_collection_diff(from_collection, to_collection, '1', '2')
     assert msg == error_message
 
+
 @pytest.mark.xfail(reason="duplicate test ids not supported yet")
 def test_pytest_issue419(testdir):
     testdir.makepyfile("""

testing/test_looponfail.py

@@ -2,6 +2,7 @@ import py
 from xdist.looponfail import RemoteControl
 from xdist.looponfail import StatRecorder
 
+
 class TestStatRecorder:
     def test_filechange(self, tmpdir):
         tmp = tmpdir
@@ -87,6 +88,7 @@ class TestStatRecorder:
         sd.waitonchange(checkinterval=0.2)
         assert not l
 
+
 class TestRemoteControl:
     def test_nofailures(self, testdir):
         item = testdir.getitem("def test_func(): pass\n")
@@ -136,13 +138,14 @@ class TestRemoteControl:
         parent = modcol.fspath.dirpath().dirpath()
         parent.chdir()
         modcol.config.args = [py.path.local(x).relto(parent)
-                                for x in modcol.config.args]
+                              for x in modcol.config.args]
         control = RemoteControl(modcol.config)
         control.loop_once()
         assert control.failures
         control.loop_once()
         assert control.failures
 
+
 class TestLooponFailing:
     def test_looponfail_from_fail_to_ok(self, testdir):
         modcol = testdir.getmodulecol("""
@@ -225,7 +228,7 @@ class TestLooponFailing:
         def runsession_dups():
             # twisted.trial test cases may report multiple errors.
             failures, reports, collection_failed = orig_runsession()
-            print (failures)
+            print(failures)
             return failures * 2, reports, collection_failed
 
         monkeypatch.setattr(remotecontrol, 'runsession', runsession_dups)
@@ -240,8 +243,8 @@ class TestFunctional:
                 x = 0
                 assert x == 1
         """)
-        #p = testdir.mkdir("sub").join(p1.basename)
-        #p1.move(p)
+        # p = testdir.mkdir("sub").join(p1.basename)
+        # p1.move(p)
         child = testdir.spawn_pytest("-f %s --traceconfig" % p)
         child.expect("def test_one")
         child.expect("x == 1")
@@ -269,6 +272,7 @@ class TestFunctional:
         child.expect("waiting for changes")
         child.kill(15)
 
+
 def removepyc(path):
     # XXX damn those pyc files
     pyc = path + "c"
@@ -277,4 +281,3 @@ def removepyc(path):
     c = path.dirpath("__pycache__")
     if c.check():
         c.remove()
-

testing/test_newhooks.py

@@ -0,0 +1,55 @@
+import pytest
+
+
+class TestHooks:
+
+    @pytest.fixture(autouse=True)
+    def create_test_file(self, testdir):
+        testdir.makepyfile("""
+            import os
+            def test_a(): pass
+            def test_b(): pass
+            def test_c(): pass
+        """)
+
+    def test_runtest_logreport(self, testdir):
+        """Test that log reports from pytest_runtest_logreport when running
+        with xdist contain "node", "nodeid" and "worker_id" attributes. (#8)
+        """
+        testdir.makeconftest("""
+            def pytest_runtest_logreport(report):
+                if hasattr(report, 'node'):
+                    if report.when == "call":
+                        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([
+            '*HOOK: test_runtest_logreport.py::test_a gw0*',
+            '*HOOK: test_runtest_logreport.py::test_b gw0*',
+            '*HOOK: test_runtest_logreport.py::test_c gw0*',
+            '*3 passed*',
+        ])
+
+    def test_node_collection_finished(self, testdir):
+        """Test pytest_xdist_node_collection_finished hook (#8).
+        """
+        testdir.makeconftest("""
+            def pytest_xdist_node_collection_finished(node, ids):
+                slaveid = node.slaveinput['slaveid']
+                stripped_ids = [x.split('::')[1] for x in ids]
+                print("HOOK: %s %s" % (slaveid, ', '.join(stripped_ids)))
+        """)
+        res = testdir.runpytest('-n2', '-s')
+        res.stdout.fnmatch_lines_random([
+            '*HOOK: gw0 test_a, test_b, test_c',
+            '*HOOK: gw1 test_a, test_b, test_c',
+        ])
+        res.stdout.fnmatch_lines([
+            '*3 passed*',
+        ])

testing/test_plugin.py

@@ -2,6 +2,7 @@ import py
 import execnet
 from xdist.slavemanage import NodeManager
 
+
 def test_dist_incompatibility_messages(testdir):
     result = testdir.runpytest("--pdb", "--looponfail")
     assert result.ret != 0
@@ -12,8 +13,9 @@ def test_dist_incompatibility_messages(testdir):
     assert result.ret != 0
     assert "incompatible" in result.stderr.str()
 
+
 def test_dist_options(testdir):
-    from xdist.plugin import check_options
+    from xdist.plugin import pytest_cmdline_main as check_options
     config = testdir.parseconfigure("-n 2")
     check_options(config)
     assert config.option.dist == "load"
@@ -22,6 +24,22 @@ def test_dist_options(testdir):
     check_options(config)
     assert config.option.dist == "load"
 
+
+def test_auto_detect_cpus(testdir, monkeypatch):
+    import os
+    if hasattr(os, 'cpu_count'):
+        monkeypatch.setattr(os, 'cpu_count', lambda: 99)
+    else:
+        import multiprocessing
+        monkeypatch.setattr(multiprocessing, 'cpu_count', lambda: 99)
+
+    config = testdir.parseconfigure("-n2")
+    assert config.getoption('numprocesses') == 2
+
+    config = testdir.parseconfigure("-nauto")
+    assert config.getoption('numprocesses') == 99
+
+
 class TestDistOptions:
     def test_getxspecs(self, testdir):
         config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
@@ -33,7 +51,7 @@ class TestDistOptions:
         assert xspecs[1].ssh == "xyz"
 
     def test_xspecs_multiplied(self, testdir):
-        config = testdir.parseconfigure("--tx=3*popen",)
+        config = testdir.parseconfigure("--tx=3*popen", )
         xspecs = NodeManager(config)._getxspecs()
         assert len(xspecs) == 3
         assert xspecs[1].popen
@@ -60,11 +78,10 @@ class TestDistOptions:
             rsyncdirs= x
         """)
         config = testdir.parseconfigure(
-              testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
+            testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
         nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=xyz")])
         roots = nm._getrsyncdirs()
-        #assert len(roots) == 3 + 1 # pylib
+        # assert len(roots) == 3 + 1 # pylib
         assert py.path.local('y') in roots
         assert py.path.local('z') in roots
         assert testdir.tmpdir.join('x') in roots
-

testing/test_remote.py

@@ -1,12 +1,15 @@
 import py
+import pytest
 from xdist.slavemanage import SlaveController, unserialize_report
 from xdist.remote import serialize_report
 import execnet
-queue = py.builtin._tryimport("queue", "Queue")
 import marshal
 
+queue = py.builtin._tryimport("queue", "Queue")
+
 WAIT_TIMEOUT = 10.0
 
+
 def check_marshallable(d):
     try:
         marshal.dumps(d)
@@ -14,12 +17,14 @@ def check_marshallable(d):
         py.std.pprint.pprint(d)
         raise ValueError("not marshallable")
 
+
 class EventCall:
     def __init__(self, eventcall):
         self.name, self.kwargs = eventcall
 
     def __str__(self):
-        return "<EventCall %s(**%s)>" %(self.name, self.kwargs)
+        return "<EventCall %s(**%s)>" % (self.name, self.kwargs)
+
 
 class SlaveSetup:
     use_callback = False
@@ -31,11 +36,16 @@ class SlaveSetup:
 
     def setup(self, ):
         self.testdir.chdir()
-        #import os ; os.environ['EXECNET_DEBUG'] = "2"
+        # import os ; os.environ['EXECNET_DEBUG'] = "2"
         self.gateway = execnet.makegateway()
         self.config = config = self.testdir.parseconfigure()
         putevent = self.use_callback and self.events.put or None
-        self.slp = SlaveController(None, self.gateway, config, putevent)
+
+        class DummyMananger:
+            specs = [0, 1]
+
+        self.slp = SlaveController(DummyMananger, self.gateway, config,
+                                   putevent)
         self.request.addfinalizer(self.slp.ensure_teardown)
         self.slp.setup()
 
@@ -48,14 +58,18 @@ class SlaveSetup:
             ev = EventCall(data)
             if name is None or ev.name == name:
                 return ev
-            print("skipping %s" % (ev,))
+            print("skipping %s" % (ev, ))
 
     def sendcommand(self, name, **kwargs):
         self.slp.sendcommand(name, **kwargs)
 
-def pytest_funcarg__slave(request):
+
+@pytest.fixture
+def slave(request):
     return SlaveSetup(request)
 
+
+@pytest.mark.xfail(reason='#59')
 def test_remoteinitconfig(testdir):
     from xdist.remote import remote_initconfig
     config1 = testdir.parseconfig()
@@ -63,6 +77,7 @@ def test_remoteinitconfig(testdir):
     assert config2.option.__dict__ == config1.option.__dict__
     assert config2.pluginmanager.getplugin("terminal") in (-1, None)
 
+
 class TestReportSerialization:
     def test_itemreport_outcomes(self, testdir):
         reprec = testdir.inline_runsource("""
@@ -79,7 +94,7 @@ class TestReportSerialization:
                 py.test.xfail("hello")
         """)
         reports = reprec.getreports("pytest_runtest_logreport")
-        assert len(reports) == 17 # with setup/teardown "passed" reports
+        assert len(reports) == 17  # with setup/teardown "passed" reports
         for rep in reports:
             d = serialize_report(rep)
             check_marshallable(d)
@@ -158,7 +173,7 @@ class TestSlaveInteractor:
         ev = slave.popevent("logstart")
         assert ev.kwargs["nodeid"].endswith("test_func")
         assert len(ev.kwargs["location"]) == 3
-        ev = slave.popevent("testreport") # setup
+        ev = slave.popevent("testreport")  # setup
         ev = slave.popevent("testreport")
         assert ev.name == "testreport"
         rep = unserialize_report(ev.name, ev.kwargs['data'])
@@ -168,6 +183,8 @@ class TestSlaveInteractor:
         ev = slave.popevent("slavefinished")
         assert 'slaveoutput' in ev.kwargs
 
+    @pytest.mark.skipif(pytest.__version__ >= '3.0',
+                        reason='skip at module level illegal in pytest 3.0')
     def test_remote_collect_skip(self, slave):
         slave.testdir.makepyfile("""
             import py
@@ -246,3 +263,13 @@ class TestSlaveInteractor:
             ("pytest_collectreport", "report.collector.fspath == bbb"),
         ])
 
+
+def test_remote_env_vars(testdir):
+    testdir.makepyfile('''
+        import os
+        def test():
+            assert os.environ['PYTEST_XDIST_WORKER'] in ('gw0', 'gw1')
+            assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2'
+    ''')
+    result = testdir.runpytest('-n2', '--max-slave-restart=0')
+    assert result.ret == 0

testing/test_slavemanage.py

@@ -1,38 +1,49 @@
 import py
 import pytest
 import execnet
-from xdist import slavemanage
+from _pytest.pytester import HookRecorder
+from xdist import slavemanage, newhooks
 from xdist.slavemanage import HostRSync, NodeManager
 
-pytest_plugins = "pytester",
+pytest_plugins = "pytester"
 
-def pytest_funcarg__hookrecorder(request):
-    _pytest = request.getfuncargvalue('_pytest')
-    config = request.getfuncargvalue('config')
-    return _pytest.gethookrecorder(config.hook)
 
-def pytest_funcarg__config(request):
-    testdir = request.getfuncargvalue("testdir")
-    config = testdir.parseconfig()
-    return config
+@pytest.fixture
+def hookrecorder(request, config):
+    hookrecorder = HookRecorder(config.pluginmanager)
+    if hasattr(hookrecorder, "start_recording"):
+        hookrecorder.start_recording(newhooks)
+    request.addfinalizer(hookrecorder.finish_recording)
+    return hookrecorder
 
-def pytest_funcarg__mysetup(request):
+
+@pytest.fixture
+def config(testdir):
+    return testdir.parseconfig()
+
+
+@pytest.fixture
+def mysetup(tmpdir):
     class mysetup:
-        def __init__(self, request):
-            temp = request.getfuncargvalue("tmpdir")
-            self.source = temp.mkdir("source")
-            self.dest = temp.mkdir("dest")
-            request.getfuncargvalue("_pytest")
-    return mysetup(request)
+        source = tmpdir.mkdir("source")
+        dest = tmpdir.mkdir("dest")
+
+    return mysetup()
+
 
 @pytest.fixture
 def slavecontroller(monkeypatch):
     class MockController(object):
-        def __init__(self, *args): pass
-        def setup(self): pass
+        def __init__(self, *args):
+            pass
+
+        def setup(self):
+            pass
+
     monkeypatch.setattr(slavemanage, 'SlaveController', MockController)
     return MockController
 
+
 class TestNodeManagerPopen:
     def test_popen_no_default_chdir(self, config):
         gm = NodeManager(config, ["popen"])
@@ -45,8 +56,8 @@ class TestNodeManagerPopen:
         for spec in NodeManager(config, l, defaultchdir="abc").specs:
             assert spec.chdir == "abc"
 
-    def test_popen_makegateway_events(self, config,
-                                      hookrecorder, _pytest, slavecontroller):
+    def test_popen_makegateway_events(self, config, hookrecorder,
+                                      slavecontroller):
         hm = NodeManager(config, ["popen"] * 2)
         hm.setup_nodes(None)
         call = hookrecorder.popcall("pytest_xdist_setupnodes")
@@ -67,12 +78,16 @@ class TestNodeManagerPopen:
         hm.setup_nodes(None)
         assert len(hm.group) == 2
         for gw in hm.group:
+
             class pseudoexec:
                 args = []
+
                 def __init__(self, *args):
                     self.args.extend(args)
+
                 def waitclose(self):
                     pass
+
             gw.remote_exec = pseudoexec
         l = []
         for gw in hm.group:
@@ -98,8 +113,8 @@ class TestNodeManagerPopen:
         assert dest.join("dir1", "dir2").check()
         assert dest.join("dir1", "dir2", 'hello').check()
 
-    def test_rsync_same_popen_twice(self, config, mysetup,
-                                    hookrecorder, slavecontroller):
+    def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
+                                    slavecontroller):
         source, dest = mysetup.source, mysetup.dest
         hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
         hm.roots = []
@@ -113,15 +128,8 @@ class TestNodeManagerPopen:
         assert call.gateways[0] in hm.group
         call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
 
-class TestHRSync:
-    def pytest_funcarg__mysetup(self, request):
-        class mysetup:
-            def __init__(self, request):
-                tmp = request.getfuncargvalue('tmpdir')
-                self.source = tmp.mkdir("source")
-                self.dest = tmp.mkdir("dest")
-        return mysetup(request)
 
+class TestHRSync:
     def test_hrsync_filter(self, mysetup):
         source, _ = mysetup.source, mysetup.dest  # noqa
         source.ensure("dir", "file.txt")
@@ -129,8 +137,7 @@ class TestHRSync:
         source.ensure(".somedotfile", "moreentries")
         source.ensure("somedir", "editfile~")
         syncer = HostRSync(source, ignores=NodeManager.DEFAULT_IGNORES)
-        l = list(source.visit(rec=syncer.filter,
-                                   fil=syncer.filter))
+        l = list(source.visit(rec=syncer.filter, fil=syncer.filter))
         assert len(l) == 3
         basenames = [x.basename for x in l]
         assert 'dir' in basenames
@@ -151,12 +158,12 @@ class TestHRSync:
 
 
 class TestNodeManager:
-    @py.test.mark.xfail
+    @py.test.mark.xfail(run=False)
     def test_rsync_roots_no_roots(self, testdir, mysetup):
         mysetup.source.ensure("dir1", "file1").write("hello")
         config = testdir.parseconfig(mysetup.source)
         nodemanager = NodeManager(config, ["popen//chdir=%s" % mysetup.dest])
-        #assert nodemanager.config.topdir == source == config.topdir
+        # assert nodemanager.config.topdir == source == config.topdir
         nodemanager.makegateways()
         nodemanager.rsync_roots()
         p, = nodemanager.gwmanager.multi_exec(
@@ -175,10 +182,8 @@ class TestNodeManager:
         for rsyncroot in (dir1, source):
             dest.remove()
             nodemanager = NodeManager(testdir.parseconfig(
-                "--tx", "popen//chdir=%s" % dest,
-                "--rsyncdir", rsyncroot,
-                source,
-            ))
+                "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot,
+                source, ))
             nodemanager.setup_nodes(None)  # calls .rsync_roots()
             if rsyncroot == source:
                 dest = dest.join("source")
@@ -241,7 +246,8 @@ class TestNodeManager:
             assert not gwspec.chdir
 
     def test_ssh_setup_nodes(self, specssh, testdir):
-        testdir.makepyfile(__init__="", test_x="""
+        testdir.makepyfile(__init__="",
+                           test_x="""
             def test_one():
                 pass
         """)

tox.ini

@@ -1,42 +1,42 @@
 [tox]
-envlist=py26,py33,py34,py27,py27-pexpect,py33-pexpect,py26,py26-old,py33-old,flakes
+# if you change the envlist, please update .travis.yml file as well
+envlist=
+  py{26,27,34,35}-pytest{27,28,29,30}
+  py{27,35}-pytest28-pexpect
+  flakes
+  readme  
+
 
 [testenv]
 changedir=testing
-deps=pytest>=2.5.1
-commands= py.test {posargs}
-
-[testenv:py27-pexpect]
-deps={[testenv]deps}
-    pexpect
-[testenv:py33-pexpect]
-deps={[testenv]deps}
-    pexpect
+passenv = USER USERNAME
+deps =
+  pycmd
+  # to avoid .eggs
+  setuptools_scm
+  pytest27: pytest~=2.7.2
+  pytest28: pytest~=2.8.7
+  pytest29: pytest~=2.9.2
+  pytest30: pytest~=3.0.5
+  pexpect: pexpect
+platform=
+  pexpect: linux|darwin    
+commands=
+  # always clean to avoid code unmarshal mismatch on old python/pytest
+  py.cleanup -aq
+  py.test {posargs}
 
 [testenv:flakes]
 changedir=
-deps = pytest-flakes>=0.2
-commands = py.test --flakes -m flakes testing xdist
+deps = flake8
+commands = flake8 setup.py testing xdist
 
-[testenv:py26-old]
-basepython = python2.6
-deps=
-    pytest==2.5.2
-    pycmd
+[testenv:readme]
+changedir =
+deps = readme
+skip_install = true
+commands = python setup.py check -r -s
 
-commands= 
-    py.cleanup -a
-    py.test {posargs}
-
-[testenv:py33-old]
-basepython = python3.3
-deps=
-    pytest==2.5.2
-    pycmd
-
-commands= 
-    py.cleanup -a
-    py.test {posargs}
 
 [pytest]
 addopts = -rsfxX

xdist/__init__.py

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

xdist/boxed.py

@@ -0,0 +1,60 @@
+
+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,8 +1,10 @@
 import difflib
+import itertools
+from _pytest.runner import CollectReport
 
 import pytest
 import py
-from xdist.slavemanage import NodeManager
+from xdist.slavemanage import NodeManager, parse_spec_config
 
 
 queue = py.builtin._tryimport('queue', 'Queue')
@@ -13,17 +15,18 @@ class EachScheduling:
 
     If a node gets added after the test run is started then it is
     assumed to replace a node which got removed before it finished
-    it's collection.  In this case it will only be used if a a node
+    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 matching spec was removed before it
-    finished all it's pending items.  The new node will then be
+    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
+    def __init__(self, config, log=None):
+        self.config = config
+        self.numnodes = len(parse_spec_config(config))
         self.node2collection = {}
         self.node2pending = {}
         self._started = []
@@ -39,25 +42,7 @@ class EachScheduling:
         """A list of all nodes in the scheduler."""
         return list(self.node2pending.keys())
 
-    def hasnodes(self):
-        return bool(self.node2pending)
-
-    def haspending(self):
-        """Return True if there are pending test items
-
-        This indicates that collection has finished and nodes are
-        still processing test items, so can be thought of as "the
-        scheduler is active".
-        """
-        for pending in self.node2pending.values():
-            if pending:
-                return True
-        return False
-
-    def addnode(self, node):
-        assert node not in self.node2pending
-        self.node2pending[node] = []
-
+    @property
     def tests_finished(self):
         if not self.collection_is_completed:
             return False
@@ -68,15 +53,32 @@ class EachScheduling:
                 return False
         return True
 
-    def addnode_collection(self, node, collection):
+    @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 it's peding list is set to an empty
+        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 assing the pending items
-        here.  In either case ``.init_distribute()`` will instruct the
+        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
@@ -88,8 +90,9 @@ class EachScheduling:
         elif self._removed2pending:
             for deadnode in self._removed2pending:
                 if deadnode.gateway.spec == node.gateway.spec:
-                    if collection != self.node2collection[deadnode]:
-                        msg = report_collection_diff(self.collection,
+                    dead_collection = self.node2collection[deadnode]
+                    if collection != dead_collection:
+                        msg = report_collection_diff(dead_collection,
                                                      collection,
                                                      deadnode.gateway.id,
                                                      node.gateway.id)
@@ -99,11 +102,11 @@ class EachScheduling:
                     self.node2pending[node] = pending
                     break
 
-    def remove_item(self, node, item_index, duration=0):
+    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 addnode() yet
+        # KeyError if we didn't get an add_node() yet
         pending = self.node2pending.pop(node)
         if not pending:
             return
@@ -112,13 +115,13 @@ class EachScheduling:
             self._removed2pending[node] = pending
         return crashitem
 
-    def init_distribute(self):
+    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 ``.addnode_collection()``) then it replaces a
-        died node and we only need to run those tests.
+        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():
@@ -133,19 +136,19 @@ class EachScheduling:
 
 
 class LoadScheduling:
-    """Implement load scheduling accross nodes.
+    """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 suit and
+    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 devided up in
-    chunks and chunks get submitted to nodes.  Whenver a node finishes
-    an item they call ``.remove_item()`` which will trigger the
+    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
+    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:
@@ -153,7 +156,7 @@ class LoadScheduling:
     :numnodes: The expected number of nodes taking part.  The actual
        number of nodes will vary during the scheduler's lifetime as
        nodes are added by the DSession as they are brought up and
-       removed either because of a died node or normal shutdown.  This
+       removed either because of a dead node or normal shutdown.  This
        number is primarily used to know when the initial collection is
        completed.
 
@@ -167,7 +170,7 @@ class LoadScheduling:
 
     :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.
+       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
@@ -175,10 +178,11 @@ class LoadScheduling:
 
     :log: A py.log.Producer instance.
 
+    :config: Config object, used for handling hooks.
     """
 
-    def __init__(self, numnodes, log=None):
-        self.numnodes = numnodes
+    def __init__(self, config, log=None):
+        self.numnodes = len(parse_spec_config(config))
         self.node2collection = {}
         self.node2pending = {}
         self.pending = []
@@ -187,6 +191,7 @@ class LoadScheduling:
             self.log = py.log.Producer("loadsched")
         else:
             self.log = log.loadsched
+        self.config = config
 
     @property
     def nodes(self):
@@ -203,36 +208,7 @@ class LoadScheduling:
         """
         return len(self.node2collection) >= self.numnodes
 
-    def haspending(self):
-        """Return True if there are pending test items
-
-        This indicates that collection has finished and nodes are
-        still processing test items, so can be thought of as "the
-        scheduler is active".
-        """
-        if self.pending:
-            return True
-        for pending in self.node2pending.values():
-            if pending:
-                return True
-        return False
-
-    def hasnodes(self):
-        """Return True if nodes exist in the scheduler."""
-        return bool(self.node2pending)
-
-    def addnode(self, node):
-        """Add a new node in the scheduler.
-
-        From now on the node will be allocated chunks of tests to
-        execute.
-
-        Called by the ``DSession.slave_slaveready`` hook when it
-        sucessfully bootstrapped a new node.
-        """
-        assert node not in self.node2pending
-        self.node2pending[node] = []
-
+    @property
     def tests_finished(self):
         """Return True if all tests have been executed by the nodes."""
         if not self.collection_is_completed:
@@ -244,7 +220,34 @@ class LoadScheduling:
                 return False
         return True
 
-    def addnode_collection(self, node, collection):
+    @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.
@@ -253,8 +256,9 @@ class LoadScheduling:
         assert node in self.node2pending
         if self.collection_is_completed:
             # A new node has been added later, perhaps an original one died.
-            assert self.collection  # .init_distribute() should have
-                                    # been called by now
+            # .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,
@@ -265,7 +269,7 @@ class LoadScheduling:
                 return
         self.node2collection[node] = list(collection)
 
-    def remove_item(self, node, item_index, duration=0):
+    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
@@ -284,6 +288,9 @@ class LoadScheduling:
         ``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)
@@ -303,7 +310,7 @@ class LoadScheduling:
         self.log("num items waiting for node:", len(self.pending))
 
     def remove_node(self, node):
-        """Remove an node from the scheduler
+        """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
@@ -326,7 +333,7 @@ class LoadScheduling:
             self.check_schedule(node)
         return crashitem
 
-    def init_distribute(self):
+    def schedule(self):
         """Initiate distribution of the test collection
 
         Initiate scheduling of the items across the nodes.  If this
@@ -336,12 +343,10 @@ class LoadScheduling:
 
         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 happend, reschedule on all nodes
+        # Initial distribution already happened, reschedule on all nodes
         if self.collection is not None:
             for node in self.nodes:
                 self.check_schedule(node)
@@ -358,13 +363,22 @@ class LoadScheduling:
         if not self.collection:
             return
 
-        # how many items per node do we have about?
-        items_per_node = len(self.collection) // len(self.node2pending)
-        # take a fraction of tests for initial distribution
-        node_chunksize = max(items_per_node // 4, 2)
-        # and initialize each node with a chunk of tests
-        for node in self.nodes:
-            self._send_tests(node, node_chunksize)
+        # 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]
@@ -376,8 +390,9 @@ class LoadScheduling:
     def _check_nodes_have_same_collection(self):
         """Return True if all nodes have collected the same items.
 
-        If collections differ this returns False and logs the
-        collection differences as they are found.
+        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]
@@ -390,8 +405,13 @@ class LoadScheduling:
                 node.gateway.id,
             )
             if msg:
-                self.log(msg)
                 same_collection = False
+                self.log(msg)
+                if self.config is not None:
+                    rep = CollectReport(
+                        node.gateway.id, 'failed',
+                        longrepr=msg, result=[])
+                    self.config.hook.pytest_collectreport(report=rep)
 
         return same_collection
 
@@ -442,12 +462,19 @@ class DSession:
         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")
         self.queue = queue.Queue()
+        self._session = None
         self._failed_collection_errors = {}
         self._active_nodes = set()
+        self._failed_nodes_count = 0
+        self._max_slave_restart = self.config.getoption('max_slave_restart')
+        if self._max_slave_restart is not None:
+            self._max_slave_restart = int(self._max_slave_restart)
         try:
             self.terminal = config.pluginmanager.getplugin("terminalreporter")
         except KeyError:
@@ -461,7 +488,7 @@ class DSession:
         """Return True if the distributed session has finished
 
         This means all nodes have executed all test items.  This is
-        used to by pytest_runtestloop to break out of it's loop.
+        used by pytest_runtestloop to break out of its loop.
         """
         return bool(self.shuttingdown and not self._active_nodes)
 
@@ -479,30 +506,39 @@ class DSession:
         self.nodemanager = NodeManager(self.config)
         nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
         self._active_nodes.update(nodes)
+        self._session = session
 
     def pytest_sessionfinish(self, session):
         """Shutdown all nodes."""
         nm = getattr(self, 'nodemanager', None)  # if not fully initialized
         if nm is not None:
             nm.teardown_nodes()
+        self._session = None
 
     def pytest_collection(self):
         # prohibit collection of test items in master process
         return True
 
-    def pytest_runtestloop(self):
-        numnodes = len(self.nodemanager.specs)
-        dist = self.config.getvalue("dist")
+    @pytest.mark.trylast
+    def pytest_xdist_make_scheduler(self, config, log):
+        dist = config.getvalue("dist")
         if dist == "load":
-            self.sched = LoadScheduling(numnodes, log=self.log)
+            return LoadScheduling(config, log)
         elif dist == "each":
-            self.sched = EachScheduling(numnodes, log=self.log)
-        else:
-            assert 0, dist
+            return EachScheduling(config, log)
+
+    def pytest_runtestloop(self):
+        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()
             if self.shouldstop:
+                self.triggershutdown()
                 raise Interrupted(str(self.shouldstop))
         return True
 
@@ -520,7 +556,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()
 
     #
@@ -540,14 +576,14 @@ 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.
 
         Removes the node from the scheduler.
 
-        The node might not be the scheduler if it had not emitted
+        The node might not be in the scheduler if it had not emitted
         slaveready before shutdown was triggered.
         """
         self.config.hook.pytest_testnodedown(node=node, error=None)
@@ -570,8 +606,20 @@ class DSession:
         else:
             if crashitem:
                 self.handle_crashitem(crashitem, node)
-        self.report_line("Replacing failed node %s" % node.gateway.id)
-        self._clone_node(node)
+
+        self._failed_nodes_count += 1
+        maximum_reached = (self._max_slave_restart is not None and
+                           self._failed_nodes_count > self._max_slave_restart)
+        if maximum_reached:
+            if self._max_slave_restart == 0:
+                msg = 'Slave restarting disabled'
+            else:
+                msg = "Maximum crashed slaves reached: %d" % \
+                      self._max_slave_restart
+            self.report_line(msg)
+        else:
+            self.report_line("Replacing crashed slave %s" % node.gateway.id)
+            self._clone_node(node)
         self._active_nodes.remove(node)
 
     def slave_collectionfinish(self, node, ids):
@@ -579,22 +627,27 @@ class DSession:
 
         This adds the collection for this node to the scheduler.  If
         the scheduler indicates collection is finished (i.e. all
-        initial nodes have submitted their collection), then tells the
+        initial nodes have submitted their collections), then tells the
         scheduler to schedule the collected items.  When initiating
         scheduling the first time it logs which scheduler is in use.
         """
         if self.shuttingdown:
             return
-        self.sched.addnode_collection(node, ids)
+        self.config.hook.pytest_xdist_node_collection_finished(node=node,
+                                                               ids=ids)
+        # tell session which items were effectively collected otherwise
+        # the master node will finish the session with EXIT_NOTESTSCOLLECTED
+        self._session.testscollected = len(ids)
+        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."""
@@ -604,12 +657,12 @@ class DSession:
     def slave_testreport(self, node, rep):
         """Emitted when a node calls the pytest_runtest_logreport hook.
 
-        If the node indicates it is finished with a test item remove
+        If the node indicates it is finished with a test item, remove
         the item from the pending list in the scheduler.
         """
         if rep.when == "call" or (rep.when == "setup" and not rep.passed):
-            self.sched.remove_item(node, rep.item_index, rep.duration)
-        #self.report_line("testreport %s: %s" %(rep.id, rep.status))
+            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)
         self._handlefailures(rep)
@@ -622,9 +675,9 @@ class DSession:
     def _clone_node(self, node):
         """Return new node based on an existing one.
 
-        This is normally for when a node died, this will copy the spec
+        This is normally for when a node dies, this will copy the spec
         of the existing node and create a new one with a new id.  The
-        new node will have been setup so will start calling the
+        new node will have been setup so it will start calling the
         "slave_*" hooks and do work soon.
         """
         spec = node.gateway.spec
@@ -673,22 +726,23 @@ class TerminalDistReporter:
         self.tr = config.pluginmanager.getplugin("terminalreporter")
         self._status = {}
         self._lastlen = 0
+        self._isatty = getattr(self.tr, 'isatty', self.tr.hasmarkup)
 
     def write_line(self, msg):
         self.tr.write_line(msg)
 
     def ensure_show_status(self):
-        if not self.tr.hasmarkup:
+        if not self._isatty:
             self.write_line(self.getstatus())
 
     def setstatus(self, spec, status, show=True):
         self._status[spec.id] = status
-        if show and self.tr.hasmarkup:
+        if show and self._isatty:
             self.rewrite(self.getstatus())
 
     def getstatus(self):
-        parts = ["%s %s" %(spec.id, self._status[spec.id])
-                   for spec in self._specs]
+        parts = ["%s %s" % (spec.id, self._status[spec.id])
+                 for spec in self._specs]
         return " / ".join(parts)
 
     def rewrite(self, line, newline=False):
@@ -719,7 +773,7 @@ class TerminalDistReporter:
     def pytest_testnodeready(self, node):
         if self.config.option.verbose > 0:
             d = node.slaveinfo
-            infoline = "[%s] Python %s" %(
+            infoline = "[%s] Python %s" % (
                 d['id'],
                 d['version'].replace('\n', ' -- '),)
             self.rewrite(infoline, newline=True)
@@ -728,13 +782,12 @@ class TerminalDistReporter:
     def pytest_testnodedown(self, node, error):
         if not error:
             return
-        self.write_line("[%s] node down: %s" %(node.gateway.id, error))
+        self.write_line("[%s] node down: %s" % (node.gateway.id, error))
 
-    #def pytest_xdist_rsyncstart(self, source, gateways):
+    # def pytest_xdist_rsyncstart(self, source, gateways):
     #    targets = ",".join([gw.id for gw in gateways])
     #    msg = "[%s] rsyncing: %s" %(targets, source)
     #    self.write_line(msg)
-    #def pytest_xdist_rsyncfinish(self, source, gateways):
+    # def pytest_xdist_rsyncfinish(self, source, gateways):
     #    targets = ", ".join(["[%s]" % gw.id for gw in gateways])
     #    self.write_line("rsyncfinish: %s -> %s" %(source, targets))
-

xdist/looponfail.py

@@ -7,10 +7,32 @@
     the controlling process which should best never happen.
 """
 
-import py, pytest
+import py
+import pytest
 import sys
 import execnet
 
+
+def pytest_addoption(parser):
+    group = parser.getgroup("xdist", "distributed and subprocess testing")
+    group._addoption(
+        '-f', '--looponfail',
+        action="store_true", dest="looponfail", default=False,
+        help="run tests in subprocess, wait for modified files "
+             "and re-run failing test set until all pass.")
+
+
+def pytest_cmdline_main(config):
+
+    if config.getoption("looponfail"):
+        usepdb = config.getoption('usepdb')  # a core option
+        if usepdb:
+            raise pytest.UsageError(
+                "--pdb incompatible with --looponfail.")
+        looponfail_main(config)
+        return 2  # looponfail only can get stop with ctrl-C anyway
+
+
 def looponfail_main(config):
     remotecontrol = RemoteControl(config)
     rootdirs = config.getini("looponfailroots")
@@ -19,7 +41,8 @@ def looponfail_main(config):
         while 1:
             remotecontrol.loop_once()
             if not remotecontrol.failures and remotecontrol.wasfailing:
-                continue # the last failures passed, let's immediately rerun all
+                # the last failures passed, let's immediately rerun all
+                continue
             repr_pytest_looponfailinfo(
                 failreports=remotecontrol.failures,
                 rootdirs=rootdirs)
@@ -27,6 +50,7 @@ def looponfail_main(config):
     except KeyboardInterrupt:
         print()
 
+
 class RemoteControl(object):
     def __init__(self, config):
         self.config = config
@@ -47,11 +71,13 @@ class RemoteControl(object):
             raise ValueError("already have gateway %r" % self.gateway)
         self.trace("setting up slave session")
         self.gateway = self.initgateway()
-        self.channel = channel = self.gateway.remote_exec(init_slave_session,
+        self.channel = channel = self.gateway.remote_exec(
+            init_slave_session,
             args=self.config.args,
             option_dict=vars(self.config.option),
         )
         remote_outchannel = channel.receive()
+
         def write(s):
             out._file.write(s)
             out._file.flush()
@@ -87,7 +113,7 @@ class RemoteControl(object):
         result = self.runsession()
         failures, reports, collection_failed = result
         if collection_failed:
-            pass # "Collection failed, keeping previous failure set"
+            pass  # "Collection failed, keeping previous failure set"
         else:
             uniq_failures = []
             for failure in failures:
@@ -95,6 +121,7 @@ class RemoteControl(object):
                     uniq_failures.append(failure)
             self.failures = uniq_failures
 
+
 def repr_pytest_looponfailinfo(failreports, rootdirs):
     tr = py.io.TerminalWriter()
     if failreports:
@@ -104,11 +131,12 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
                 tr.line(report, red=True)
     tr.sep("#", "waiting for changes", bold=True)
     for rootdir in rootdirs:
-        tr.line("### Watching:   %s" %(rootdir,), bold=True)
+        tr.line("### Watching:   %s" % (rootdir,), bold=True)
 
 
 def init_slave_session(channel, args, option_dict):
-    import os, sys
+    import os
+    import sys
     outchannel = channel.gateway.newchannel()
     sys.stdout = sys.stderr = outchannel.makefile('w')
     channel.send(outchannel)
@@ -121,13 +149,14 @@ def init_slave_session(channel, args, option_dict):
             newpaths.append(p)
     sys.path[:] = newpaths
 
-    #fullwidth, hasmarkup = channel.receive()
+    # fullwidth, hasmarkup = channel.receive()
     from _pytest.config import Config
     config = Config.fromdictargs(option_dict, list(args))
     config.args = args
     from xdist.looponfail import SlaveFailSession
     SlaveFailSession(config, channel).main()
 
+
 class SlaveFailSession:
     def __init__(self, config, channel):
         self.config = config
@@ -150,7 +179,8 @@ class SlaveFailSession:
             items = session.perform_collect(self.trails or None)
         except pytest.UsageError:
             items = session.perform_collect(None)
-        hook.pytest_collection_modifyitems(session=session, config=session.config, items=items)
+        hook.pytest_collection_modifyitems(
+            session=session, config=session.config, items=items)
         hook.pytest_collection_finish(session=session)
         return True
 
@@ -168,7 +198,7 @@ class SlaveFailSession:
         try:
             command = self.channel.receive()
         except KeyboardInterrupt:
-            return # in the slave we can't do much about this
+            return  # in the slave we can't do much about this
         self.DEBUG("received", command)
         self.current_command = command
         self.config.hook.pytest_cmdline_main(config=self.config)
@@ -180,14 +210,16 @@ class SlaveFailSession:
             failreports.append(loc)
         self.channel.send((trails, failreports, self.collection_failed))
 
+
 class StatRecorder:
     def __init__(self, rootdirlist):
         self.rootdirlist = rootdirlist
         self.statcache = {}
-        self.check() # snapshot state
+        self.check()  # snapshot state
 
     def fil(self, p):
         return p.check(file=1, dotfile=0) and p.ext != ".pyc"
+
     def rec(self, p):
         return p.check(dotfile=0)
 
@@ -198,7 +230,7 @@ class StatRecorder:
                 return
             py.std.time.sleep(checkinterval)
 
-    def check(self, removepycfiles=True):
+    def check(self, removepycfiles=True):  # noqa, too complex
         changed = False
         statcache = self.statcache
         newstat = {}
@@ -212,8 +244,8 @@ class StatRecorder:
                         changed = True
                 else:
                     if oldstat:
-                       if oldstat.mtime != curstat.mtime or \
-                          oldstat.size != curstat.size:
+                        if oldstat.mtime != curstat.mtime or \
+                           oldstat.size != curstat.size:
                             changed = True
                             py.builtin.print_("# MODIFIED", path)
                             if removepycfiles and path.ext == ".py":
@@ -227,4 +259,3 @@ class StatRecorder:
             changed = True
         self.statcache = newstat
         return changed
-

xdist/newhooks.py

@@ -1,21 +1,52 @@
+"""
+xdist hooks.
+
+Additionally, pytest-xdist will also decorate a few other hooks
+with the worker instance that executed the hook originally:
+
+``pytest_runtest_logreport``: ``rep`` parameter has a ``node`` attribute.
+
+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
+"""
+import pytest
+
 
 def pytest_xdist_setupnodes(config, specs):
     """ called before any remote node is set up. """
 
+
 def pytest_xdist_newgateway(gateway):
     """ called on new raw gateway creation. """
 
+
 def pytest_xdist_rsyncstart(source, gateways):
     """ called before rsyncing a directory to remote gateways takes place. """
 
+
 def pytest_xdist_rsyncfinish(source, gateways):
     """ called after rsyncing a directory to remote gateways takes place. """
 
+
 def pytest_configure_node(node):
     """ configure node information before it gets instantiated. """
 
+
 def pytest_testnodeready(node):
     """ Test Node is ready to operate. """
 
+
 def pytest_testnodedown(node, error):
     """ Test Node is down. """
+
+
+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

@@ -1,65 +1,89 @@
 import py
 import pytest
 
+
+def parse_numprocesses(s):
+    if s == 'auto':
+        try:
+            from os import cpu_count
+        except ImportError:
+            from multiprocessing import cpu_count
+        try:
+            n = cpu_count()
+        except NotImplementedError:
+            return 1
+        return n if n else 1
+    else:
+        return int(s)
+
+
 def pytest_addoption(parser):
     group = parser.getgroup("xdist", "distributed and subprocess testing")
-    group._addoption('-f', '--looponfail',
-           action="store_true", dest="looponfail", default=False,
-           help="run tests in subprocess, wait for modified files "
-                "and re-run failing test set until all pass.")
-    group._addoption('-n', dest="numprocesses", metavar="numprocesses",
-           action="store", type="int",
-           help="shortcut for '--dist=load --tx=NUM*popen'")
-    group.addoption('--boxed',
-           action="store_true", dest="boxed", default=False,
-           help="box each test run in a separate process (unix)")
-    group._addoption('--dist', metavar="distmode",
-           action="store", choices=['load', 'each', 'no'],
-           type="choice", 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"
-                 "(default) no: run tests inprocess, don't distribute."))
-    group._addoption('--tx', dest="tx", action="append", default=[],
-           metavar="xspec",
-           help=("add a test execution environment. some examples: "
-                 "--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
-                 "--tx ssh=user@codespeak.net//chdir=testcache"))
-    group._addoption('-d',
-           action="store_true", dest="distload", default=False,
-           help="load-balance tests.  shortcut for '--dist=load'")
-    group.addoption('--rsyncdir', action="append", default=[], metavar="DIR",
-           help="add directory for rsyncing to remote tx nodes.")
-    group.addoption('--rsyncignore', action="append", default=[], metavar="GLOB",
-           help="add expression for ignores when rsyncing to remote tx nodes.")
+    group._addoption(
+        '-n', 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(
+        '--dist', metavar="distmode",
+        action="store", choices=['load', 'each', '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"
+              "(default) no: run tests inprocess, don't distribute."))
+    group._addoption(
+        '--tx', dest="tx", action="append", default=[],
+        metavar="xspec",
+        help=("add a test execution environment. some examples: "
+              "--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
+              "--tx ssh=user@codespeak.net//chdir=testcache"))
+    group._addoption(
+        '-d',
+        action="store_true", dest="distload", default=False,
+        help="load-balance tests.  shortcut for '--dist=load'")
+    group.addoption(
+        '--rsyncdir', action="append", default=[], metavar="DIR",
+        help="add directory for rsyncing to remote tx nodes.")
+    group.addoption(
+        '--rsyncignore', action="append", default=[], metavar="GLOB",
+        help="add expression for ignores when rsyncing to remote tx nodes.")
 
-    parser.addini('rsyncdirs', 'list of (relative) paths to be rsynced for'
-         ' remote distributed testing.', type="pathlist")
-    parser.addini('rsyncignore', 'list of (relative) glob-style paths to be ignored '
-         'for rsyncing.', type="pathlist")
-    parser.addini("looponfailroots", type="pathlist",
+    parser.addini(
+        'rsyncdirs', 'list of (relative) paths to be rsynced for'
+        ' remote distributed testing.', type="pathlist")
+    parser.addini(
+        'rsyncignore', 'list of (relative) glob-style paths to be ignored '
+        'for rsyncing.', type="pathlist")
+    parser.addini(
+        "looponfailroots", type="pathlist",
         help="directories to check for changes", default=[py.path.local()])
 
 # -------------------------------------------------------------------------
 # distributed testing hooks
 # -------------------------------------------------------------------------
+
+
 def pytest_addhooks(pluginmanager):
     from xdist import newhooks
-    pluginmanager.addhooks(newhooks)
+    # avoid warnings with pytest-2.8
+    method = getattr(pluginmanager, "add_hookspecs", None)
+    if method is None:
+        method = pluginmanager.addhooks
+    method(newhooks)
 
 # -------------------------------------------------------------------------
 # distributed testing initialization
 # -------------------------------------------------------------------------
 
-def pytest_cmdline_main(config):
-    check_options(config)
-    if config.getoption("looponfail"):
-        from xdist.looponfail import looponfail_main
-        looponfail_main(config)
-        return 2 # looponfail only can get stop with ctrl-C anyway
 
-def pytest_configure(config, __multicall__):
-    __multicall__.execute()
+@pytest.mark.trylast
+def pytest_configure(config):
     if config.getoption("dist") != "no":
         from xdist.dsession import DSession
         session = DSession(config)
@@ -67,65 +91,30 @@ def pytest_configure(config, __multicall__):
         tr = config.pluginmanager.getplugin("terminalreporter")
         tr.showfspath = False
 
-def check_options(config):
+
+@pytest.mark.tryfirst
+def pytest_cmdline_main(config):
     if config.option.numprocesses:
         config.option.dist = "load"
-        config.option.tx = ['popen'] * int(config.option.numprocesses)
+        config.option.tx = ['popen'] * config.option.numprocesses
     if config.option.distload:
         config.option.dist = "load"
     val = config.getvalue
     if not val("collectonly"):
-        usepdb = config.option.usepdb  # a core option
-        if val("looponfail"):
+        usepdb = config.getoption('usepdb')  # a core option
+        if val("dist") != "no":
             if usepdb:
-                raise pytest.UsageError("--pdb incompatible with --looponfail.")
-        elif val("dist") != "no":
-            if usepdb:
-                raise pytest.UsageError("--pdb incompatible with distributing tests.")
+                raise pytest.UsageError(
+                    "--pdb incompatible with distributing tests.")
+
+# -------------------------------------------------------------------------
+# fixtures
+# -------------------------------------------------------------------------
 
 
-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]
+@pytest.fixture(scope="session")
+def worker_id(request):
+    if hasattr(request.config, 'slaveinput'):
+        return request.config.slaveinput['slaveid']
     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
+        return 'master'

xdist/remote.py

@@ -6,7 +6,9 @@
     needs not to be installed in remote environments.
 """
 
-import sys, os
+import sys
+import os
+
 
 class SlaveInteractor:
     def __init__(self, config, channel):
@@ -44,7 +46,10 @@ class SlaveInteractor:
         self.log("entering main loop")
         torun = []
         while 1:
-            name, kwargs = self.channel.receive()
+            try:
+                name, kwargs = self.channel.receive()
+            except EOFError:
+                return True
             self.log("received command", name, kwargs)
             if name == "runtests":
                 torun.extend(kwargs['indices'])
@@ -72,7 +77,8 @@ class SlaveInteractor:
             nextitem=nextitem)
 
     def pytest_collection_finish(self, session):
-        self.sendevent("collectionfinish",
+        self.sendevent(
+            "collectionfinish",
             topdir=str(session.fspath),
             ids=[item.nodeid for item in session.items])
 
@@ -82,6 +88,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)
 
@@ -89,6 +96,7 @@ class SlaveInteractor:
         data = serialize_report(report)
         self.sendevent("collectreport", data=data)
 
+
 def serialize_report(rep):
     import py
     d = rep.__dict__.copy()
@@ -100,20 +108,22 @@ def serialize_report(rep):
         if isinstance(d[name], py.path.local):
             d[name] = str(d[name])
         elif name == "result":
-            d[name] = None # for now
+            d[name] = None  # for now
     return d
 
+
 def getinfodict():
     import platform
     return dict(
-        version = sys.version,
-        version_info = tuple(sys.version_info),
-        sysplatform = sys.platform,
-        platform = platform.platform(),
-        executable = sys.executable,
-        cwd = os.getcwd(),
+        version=sys.version,
+        version_info=tuple(sys.version_info),
+        sysplatform=sys.platform,
+        platform=platform.platform(),
+        executable=sys.executable,
+        cwd=os.getcwd(),
     )
 
+
 def remote_initconfig(option_dict, args):
     from _pytest.config import Config
     option_dict['plugins'].append("no:terminal")
@@ -131,14 +141,17 @@ if __name__ == '__channelexec__':
     channel = channel  # noqa
     # python3.2 is not concurrent import safe, so let's play it safe
     # https://bitbucket.org/hpk42/pytest/issue/347/pytest-xdist-and-python-32
-    if sys.version_info[:2] == (3,2):
+    if sys.version_info[:2] == (3, 2):
         os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
-    slaveinput,args,option_dict = channel.receive()
+    slaveinput, args, option_dict = channel.receive()
     importpath = os.getcwd()
-    sys.path.insert(0, importpath) # XXX only for remote situations
-    os.environ['PYTHONPATH'] = (importpath + os.pathsep +
+    sys.path.insert(0, importpath)  # XXX only for remote situations
+    os.environ['PYTHONPATH'] = (
+        importpath + os.pathsep +
         os.environ.get('PYTHONPATH', ''))
-    #os.environ['PYTHONPATH'] = importpath
+    os.environ['PYTEST_XDIST_WORKER'] = slaveinput['slaveid']
+    os.environ['PYTEST_XDIST_WORKER_COUNT'] = str(slaveinput['slavecount'])
+    # os.environ['PYTHONPATH'] = importpath
     import py
     config = remote_initconfig(option_dict, args)
     config.slaveinput = slaveinput

xdist/slavemanage.py

@@ -1,16 +1,35 @@
 import fnmatch
 import os
+import re
 
 import py
 import pytest
 import execnet
 import xdist.remote
 
-from _pytest import runner # XXX load dynamically
+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', '*~']
+
     def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
         self.config = config
         self._nodesready = py.std.threading.Event()
@@ -59,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:
@@ -79,11 +86,12 @@ class NodeManager(object):
                 break
         else:
             return []
-        import pytest, _pytest
+        import pytest
+        import _pytest
         pytestpath = pytest.__file__.rstrip("co")
         pytestdir = py.path.local(_pytest.__file__).dirpath()
         config = self.config
-        candidates = [py._pydir,pytestpath,pytestdir]
+        candidates = [py._pydir, pytestpath, pytestdir]
         candidates += config.option.rsyncdir
         rsyncroots = config.getini("rsyncdirs")
         if rsyncroots:
@@ -92,7 +100,7 @@ class NodeManager(object):
         for root in candidates:
             root = py.path.local(root).realpath()
             if not root.check():
-                raise pytest.UsageError("rsyncdir doesn't exist: %r" %(root,))
+                raise pytest.UsageError("rsyncdir doesn't exist: %r" % (root,))
             if root not in roots:
                 roots.append(root)
         return roots
@@ -124,6 +132,7 @@ class NodeManager(object):
             return
         if (spec, source) in self._rsynced_specs:
             return
+
         def finished():
             if notify:
                 notify("rsyncrootready", spec, source)
@@ -139,22 +148,23 @@ class NodeManager(object):
             gateways=[gateway],
         )
 
+
 class HostRSync(execnet.RSync):
     """ RSyncer that filters out common files
     """
     def __init__(self, sourcedir, *args, **kwargs):
         self._synced = {}
-        ignores= None
-        if 'ignores' in kwargs:
-            ignores = kwargs.pop('ignores')
-        self._ignores = ignores or []
+        self._ignores = []
+        ignores = kwargs.pop('ignores', None) or []
+        for x in ignores:
+            x = getattr(x, 'strpath', x)
+            self._ignores.append(re.compile(fnmatch.translate(x)))
         super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
 
     def filter(self, path):
         path = py.path.local(path)
-        for x in self._ignores:
-            x = getattr(x, 'strpath', x)
-            if fnmatch.fnmatch(path.basename, x) or fnmatch.fnmatch(path.strpath, x):
+        for cre in self._ignores:
+            if cre.match(path.basename) or cre.match(path.strpath):
                 return False
         else:
             return True
@@ -190,6 +200,7 @@ def make_reltoroot(roots, args):
         l.append(splitcode.join(parts))
     return l
 
+
 class SlaveController(object):
     ENDMARK = -1
 
@@ -198,14 +209,20 @@ class SlaveController(object):
         self.putevent = putevent
         self.gateway = gateway
         self.config = config
-        self.slaveinput = {'slaveid': gateway.id}
+        self.slaveinput = {'slaveid': gateway.id,
+                           'slavecount': len(nodemanager.specs)}
         self._down = False
+        self._shutdown_sent = False
         self.log = py.log.Producer("slavectl-%s" % gateway.id)
         if not self.config.option.debug:
             py.log.setconsumer(self.log._keywords, None)
 
     def __repr__(self):
-        return "<%s %s>" %(self.__class__.__name__, self.gateway.id,)
+        return "<%s %s>" % (self.__class__.__name__, self.gateway.id,)
+
+    @property
+    def shutting_down(self):
+        return self._down or self._shutdown_sent
 
     def setup(self):
         self.log("setting up slave session")
@@ -216,13 +233,15 @@ class SlaveController(object):
         option_dict = vars(self.config.option)
         if spec.popen:
             name = "popen-%s" % self.gateway.id
-            basetemp = self.config._tmpdirhandler.getbasetemp()
-            option_dict['basetemp'] = str(basetemp.join(name))
+            if hasattr(self.config, '_tmpdirhandler'):
+                basetemp = self.config._tmpdirhandler.getbasetemp()
+                option_dict['basetemp'] = str(basetemp.join(name))
         self.config.hook.pytest_configure_node(node=self)
         self.channel = self.gateway.remote_exec(xdist.remote)
         self.channel.send((self.slaveinput, args, option_dict))
         if self.putevent:
-            self.channel.setcallback(self.process_from_remote,
+            self.channel.setcallback(
+                self.process_from_remote,
                 endmarker=self.ENDMARK)
 
     def ensure_teardown(self):
@@ -230,11 +249,11 @@ class SlaveController(object):
             if not self.channel.isclosed():
                 self.log("closing", self.channel)
                 self.channel.close()
-            #del self.channel
+            # del self.channel
         if hasattr(self, 'gateway'):
             self.log("exiting", self.gateway)
             self.gateway.exit()
-            #del self.gateway
+            # del self.gateway
 
     def send_runtest_some(self, indices):
         self.sendcommand("runtests", indices=indices)
@@ -248,6 +267,7 @@ class SlaveController(object):
                 self.sendcommand("shutdown")
             except IOError:
                 pass
+            self._shutdown_sent = True
 
     def sendcommand(self, name, **kwargs):
         """ send a named parametrized command to the other side. """
@@ -258,7 +278,7 @@ class SlaveController(object):
         self.log("queuing %s(**%s)" % (eventname, kwargs))
         self.putevent((eventname, kwargs))
 
-    def process_from_remote(self, eventcall):
+    def process_from_remote(self, eventcall):  # noqa too complex
         """ this gets called for each object we receive from
             the other side and if the channel closes.
 
@@ -271,13 +291,13 @@ class SlaveController(object):
                 err = self.channel._getremoteerror()
                 if not self._down:
                     if not err or isinstance(err, EOFError):
-                        err = "Not properly terminated" # lost connection?
+                        err = "Not properly terminated"  # lost connection?
                     self.notify_inproc("errordown", node=self, error=err)
                     self._down = True
                 return
             eventname, kwargs = eventcall
-            if eventname in ("collectionstart"):
-                self.log("ignoring %s(%s)" %(eventname, kwargs))
+            if eventname in ("collectionstart",):
+                self.log("ignoring %s(%s)" % (eventname, kwargs))
             elif eventname == "slaveready":
                 self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "slavefinished":
@@ -286,7 +306,8 @@ class SlaveController(object):
                 self.notify_inproc("slavefinished", node=self)
             elif eventname == "logstart":
                 self.notify_inproc(eventname, node=self, **kwargs)
-            elif eventname in ("testreport", "collectreport", "teardownreport"):
+            elif eventname in (
+                    "testreport", "collectreport", "teardownreport"):
                 item_index = kwargs.pop("item_index", None)
                 rep = unserialize_report(eventname, kwargs['data'])
                 if item_index is not None:
@@ -295,7 +316,7 @@ class SlaveController(object):
             elif eventname == "collectionfinish":
                 self.notify_inproc(eventname, node=self, ids=kwargs['ids'])
             else:
-                raise ValueError("unknown event: %s" %(eventname,))
+                raise ValueError("unknown event: %s" % (eventname,))
         except KeyboardInterrupt:
             # should not land in receiver-thread
             raise
@@ -304,6 +325,7 @@ class SlaveController(object):
             py.builtin.print_("!" * 20, excinfo)
             self.config.pluginmanager.notify_exception(excinfo)
 
+
 def unserialize_report(name, reportdict):
     if name == "testreport":
         return runner.TestReport(**reportdict)