Compare commits

..

26 Commits

Author SHA1 Message Date
Bruno Oliveira
c92bbb021a Preparing 1.22.2 release 2018-02-26 18:46:11 -03:00
Bruno Oliveira
710b324902 Merge pull request #286 from nicoddemus/fix-slaveoutput
Add backward compatibility for ``slaveoutput`` attribute to ``WorkerController`` instances
2018-02-26 18:44:24 -03:00
Bruno Oliveira
76e7b8d0c0 Add backward compatibility for `slaveoutput attribute to WorkerController` instances
Fix #285
Fix pytest-dev/pytest-cov#191
2018-02-26 14:52:55 -03:00
Bruno Oliveira
63b329c251 Merge pull request #281 from nicoddemus/release-1.22.1
Release 1.22.1
2018-02-19 19:10:07 -03:00
Bruno Oliveira
de993e1ee3 Prepare changelog for 1.22.1 2018-02-19 18:10:50 -03:00
Bruno Oliveira
af4409753a Show content for trivial changelog entries 2018-02-19 18:10:10 -03:00
Bruno Oliveira
c39cd10ae5 Small improvements to HOWTORELEASE 2018-02-19 18:07:13 -03:00
Ronny Pfannschmidt
11e3599785 Merge pull request #277 from lmtierney/single-loadscope
Ensure loadscope and loadfile work when the first scope has one work unit
2018-02-19 20:02:22 +01:00
Bruno Oliveira
1be8a463ff Add docstring to tests for #277 2018-02-16 20:14:47 -02:00
lmtierney
bd3c7dba8b Ensure loadscope and loadfile work when the first scope has one work unit 2018-02-08 19:51:42 -06:00
Ronny Pfannschmidt
d91ec5503e Merge pull request #262 from nicoddemus/titles-readme
Improve README to be more newcomer friendly
2018-02-08 07:49:08 +01:00
Bruno Oliveira
23ce4502f1 Merge pull request #268 from feuillemorte/234-master-worker-terminology
#234 use master/worker terminology
2018-02-06 07:27:31 -02:00
feuillemorte
6fb01801ac #234 Fix comments 2018-02-06 11:53:45 +03:00
Bruno Oliveira
ee2a1b4f4d Add an explicit test for backward compatibility with "config.slaveinput"
Related to #234
2018-01-30 19:13:14 -02:00
feuillemorte
b04703b6ba #234 Fix flake 8 and fix comments 2018-01-17 23:34:39 +03:00
feuillemorte
9a5a81962c Merge branch '234-master-worker-terminology' of github.com:feuillemorte/pytest-xdist into 234-master-worker-terminology 2018-01-17 23:07:45 +03:00
feuillemorte
40491279ef #234 Added master/worker terminology 2018-01-17 23:07:29 +03:00
Bruno Oliveira
9efe14946a Improve changelog formatting 2018-01-16 18:26:50 -02:00
feuillemorte
708256228b Add changelog entry 2018-01-16 21:49:54 +03:00
feuillemorte
f1e6dc4344 use master/worker terminology 2018-01-16 21:46:43 +03:00
Bruno Oliveira
3098f27731 Merge pull request #267 from nicoddemus/release-1.22.0
Prepare release 1.22.0
2018-01-11 18:13:41 -02:00
Bruno Oliveira
e72eadb413 Fix linting 2017-12-22 17:30:47 -02:00
Bruno Oliveira
6cd1eb2f43 Use more modern python and reword part about pytest ini configuration 2017-12-22 16:53:18 -02:00
Bruno Oliveira
7c3a86a549 Mention pytest-forked instead of the help for --boxed 2017-12-22 16:49:14 -02:00
Bruno Oliveira
b2c63a1ba5 Reworded some sections to be more newbie friendly 2017-12-22 16:42:35 -02:00
Bruno Oliveira
3f545ee621 Change section titles so they are more prominent in the README 2017-12-22 16:29:10 -02:00
19 changed files with 391 additions and 357 deletions

1
.gitignore vendored
View File

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

View File

@@ -1,3 +1,33 @@
pytest-xdist 1.22.2 (2018-02-26)
================================
Bug Fixes
---------
- Add backward compatibility for ``slaveoutput`` attribute to
``WorkerController`` instances. (`#285
<https://github.com/pytest-dev/pytest-xdist/issues/285>`_)
pytest-xdist 1.22.1 (2018-02-19)
================================
Bug Fixes
---------
- Fix issue when using ``loadscope`` or ``loadfile`` where tests would fail to
start if the first scope had only one test. (`#257
<https://github.com/pytest-dev/pytest-xdist/issues/257>`_)
Trivial Changes
---------------
- Change terminology used by ``pytest-xdist`` to *master* and *worker* in
arguments and messages (for example ``--max-worker-reset``). (`#234
<https://github.com/pytest-dev/pytest-xdist/issues/234>`_)
pytest-xdist 1.22.0 (2018-01-11) pytest-xdist 1.22.0 (2018-01-11)
================================ ================================

View File

@@ -26,13 +26,11 @@ To publish a new release ``X.Y.Z``, the steps are as follows:
#. Install ``pytest-xdist`` and dev requirements in a virtualenv:: #. Install ``pytest-xdist`` and dev requirements in a virtualenv::
$ pip install -e . -r dev-requirements.txt $ pip install -e . -U -r dev-requirements.txt
#. Update ``CHANGELOG.rst`` file by running:: #. Update ``CHANGELOG.rst`` file by running::
$ towncrier --version X.Y.Z $ towncrier --version X.Y.Z --yes
It might ask for confirmation to remove news fragments; answer yes.
#. Commit and push the branch for review. #. Commit and push the branch for review.

View File

@@ -21,7 +21,7 @@
:target: https://ci.appveyor.com/project/pytestbot/pytest-xdist :target: https://ci.appveyor.com/project/pytestbot/pytest-xdist
xdist: pytest distributed testing plugin xdist: pytest distributed testing plugin
========================================= ========================================
The `pytest-xdist`_ plugin extends py.test with some unique The `pytest-xdist`_ plugin extends py.test with some unique
test execution modes: test execution modes:
@@ -49,7 +49,7 @@ If you would like to know how pytest-xdist works under the covers, checkout
Installation Installation
----------------------- ------------
Install the plugin with:: Install the plugin with::
@@ -58,43 +58,54 @@ Install the plugin with::
or use the package in develop/in-place mode with or use the package in develop/in-place mode with
a checkout of the `pytest-xdist repository`_ :: a checkout of the `pytest-xdist repository`_ ::
python setup.py develop pip install --editable .
Usage examples
---------------------
.. _parallelization: .. _parallelization:
Speed up test runs by sending tests to multiple CPUs Speed up test runs by sending tests to multiple CPUs
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ----------------------------------------------------
To send tests to multiple CPUs, type:: To send tests to multiple CPUs, type::
py.test -n NUM py.test -n NUM
Especially for longer running tests or tests requiring Especially for longer running tests or tests requiring
a lot of IO this can lead to considerable speed ups. This option can a lot of I/O this can lead to considerable speed ups. This option can
also be set to ``auto`` for automatic detection of the number of CPUs. also be set to ``auto`` for automatic detection of the number of CPUs.
If a test crashes the interpreter, pytest-xdist will automatically restart If a test crashes the interpreter, pytest-xdist will automatically restart
that slave and report the failure as usual. You can use the that worker and report the failure as usual. You can use the
``--max-slave-restart`` option to limit the number of slaves that can ``--max-worker-restart`` option to limit the number of workers that can
be restarted, or disable restarting altogether using ``--max-slave-restart=0``. be restarted, or disable restarting altogether using ``--max-worker-restart=0``.
By default, the ``-n`` option will send pending tests to any worker that is available, without
any guaranteed order, but you can control this with these options:
* ``--dist=loadscope``: tests will be grouped by **module** for *test functions* and
by **class** for *test methods*, then each group will be sent to an available worker,
guaranteeing that all tests in a group run in the same process. This can be useful if you have
expensive module-level or class-level fixtures. Currently the groupings can't be customized,
with grouping by class takes priority over grouping by module.
This feature was added in version ``1.19``.
* ``--dist=loadfile``: tests will be grouped by file name, and then will be sent to an available
worker, guaranteeing that all tests in a group run in the same worker. This feature was added
in version ``1.21``.
Running tests in a Python subprocess Running tests in a Python subprocess
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ------------------------------------
To instantiate a python2.5 sub process and send tests to it, you may type:: To instantiate a python3.5 subprocess and send tests to it, you may type::
py.test -d --tx popen//python=python2.5 py.test -d --tx popen//python=python3.5
This will start a subprocess which is run with the "python2.5" This will start a subprocess which is run with the ``python3.5``
Python interpreter, found in your system binary lookup path. Python interpreter, found in your system binary lookup path.
If you prefix the --tx option value like this:: If you prefix the --tx option value like this::
--tx 3*popen//python=python2.5 --tx 3*popen//python=python3.5
then three subprocesses would be created and tests then three subprocesses would be created and tests
will be load-balanced across these three processes. will be load-balanced across these three processes.
@@ -102,28 +113,16 @@ will be load-balanced across these three processes.
.. _boxed: .. _boxed:
Running tests in a boxed subprocess 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.
This functionality has been moved to the
`pytest-forked <https://github.com/pytest-dev/pytest-forked>`_ plugin, but the ``--boxed`` option
is still kept for backward compatibility.
.. _`remote machines`: .. _`remote machines`:
Sending tests to remote SSH accounts Sending tests to remote SSH accounts
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ------------------------------------
Suppose you have a package ``mypkg`` which contains some Suppose you have a package ``mypkg`` which contains some
tests that you can successfully run locally. And you tests that you can successfully run locally. And you
@@ -158,7 +157,7 @@ ini-file option(s).
Sending tests to remote Socket Servers Sending tests to remote Socket Servers
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ --------------------------------------
Download the single-module `socketserver.py`_ Python program Download the single-module `socketserver.py`_ Python program
and run it like this:: and run it like this::
@@ -177,7 +176,7 @@ new socket host with something like this::
Running tests on many platforms at once Running tests on many platforms at once
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ---------------------------------------
The basic command to run tests on multiple platforms is:: The basic command to run tests on multiple platforms is::
@@ -195,7 +194,7 @@ at once. The specifications strings use the `xspec syntax`_.
.. _`execnet`: http://codespeak.net/execnet .. _`execnet`: http://codespeak.net/execnet
Identifying the worker process during a test Identifying the worker process during a test
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ --------------------------------------------
*New in version 1.15.* *New in version 1.15.*
@@ -219,16 +218,15 @@ defined:
* ``PYTEST_XDIST_WORKER_COUNT``: the total number of workers in this session, * ``PYTEST_XDIST_WORKER_COUNT``: the total number of workers in this session,
e.g., ``"4"`` when ``-n 4`` is given in the command-line. 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 The information about the worker_id in a test is stored in the ``TestReport`` as
well, under worker_id attribute. well, under the ``worker_id`` attribute.
Specifying test exec environments in an ini file Specifying test exec environments in an ini file
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ------------------------------------------------
pytest (since version 2.0) supports ini-style cofiguration. You can use pytest's ini file configuration to avoid typing common options.
You can for example make running with three subprocesses You can for example make running with three subprocesses your default like this:
your default like this:
.. code-block:: ini .. code-block:: ini
@@ -240,7 +238,7 @@ You can also add default environments like this:
.. code-block:: ini .. code-block:: ini
[pytest] [pytest]
addopts = --tx ssh=myhost//python=python2.5 --tx ssh=myhost//python=python3.6 addopts = --tx ssh=myhost//python=python3.5 --tx ssh=myhost//python=python3.6
and then just type:: and then just type::
@@ -249,100 +247,8 @@ and then just type::
to run tests in each of the environments. to run tests in each of the environments.
Sending groups of related tests to the same worker
++++++++++++++++++++++++++++++++++++++++++++++++++
*New in version 1.19.*
.. note::
This is an **experimental** feature: the actual functionality will
likely stay the same, but the CLI might change slightly in future versions.
You can send groups of related tests to the same worker by using the
``--dist=loadscope`` option. Tests will be grouped by **module**
for *test functions* and by **class** for *test methods*.
For example, consider this two test files:
.. code-block:: python
# content of test_container.py
import pytest
def test_container_startup():
pass
def test_container_logging():
pass
@pytest.mark.parametrize('methods', ['ssh', 'http'])
def test_container_communication(methods):
pass
# content of test_io.py
class TestHDF:
def test_listing(self):
pass
def test_search(self):
pass
class TestXML:
def test_listing(self):
pass
def test_search(self):
pass
By executing ``pytest -v --dist=loadscope -n4`` you might get this output
(sorted by worker for readability)::
============================= test session starts =============================
<skip header>
gw0 [8] / gw1 [8] / gw2 [8] / gw3 [8]
scheduling tests via LoadScopeScheduling
[gw0] PASSED test_container.py::test_container_communication[http]
[gw0] PASSED test_container.py::test_container_communication[ssh]
[gw0] PASSED test_container.py::test_container_logging
[gw0] PASSED test_container.py::test_container_startup
[gw1] PASSED test_io.py::TestHDF::test_listing
[gw1] PASSED test_io.py::TestHDF::test_search
[gw2] PASSED test_io.py::TestXML::test_listing
[gw2] PASSED test_io.py::TestXML::test_search
========================== 8 passed in 0.56 seconds ===========================
As you can see, all test functions from ``test_container.py`` executed on
the same worker ``gw0``, while the test methods from classes ``TestHDF`` and
``TestXML`` executed in workers ``gw1`` and ``gw2`` respectively.
Currently the groupings can't be customized, with grouping by class takes
priority over grouping by module.
Sending tests to the same worker based on their file
++++++++++++++++++++++++++++++++++++++++++++++++++++
*New in version 1.21.*
.. note::
This is an **experimental** feature: the actual functionality will
likely stay the same, but the CLI might change slightly in future versions.
You can send tests to the same worker grouped by their filename by using the
``--dist=loadfile`` option, so tests of the same file are guaranteed to run
in the same worker.
Using the example in the previous section, all tests from ``test_container.py`` will
run in the same worker, as well as the tests in ``test_io.py``.
Specifying "rsync" dirs in an ini-file Specifying "rsync" dirs in an ini-file
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ --------------------------------------
In a ``tox.ini`` or ``setup.cfg`` file in your root project directory In a ``tox.ini`` or ``setup.cfg`` file in your root project directory
you may specify directories to include or to exclude in synchronisation: you may specify directories to include or to exclude in synchronisation:

View File

@@ -33,4 +33,4 @@ template = "changelog/_template.rst"
[[tool.towncrier.type]] [[tool.towncrier.type]]
directory = "trivial" directory = "trivial"
name = "Trivial Changes" name = "Trivial Changes"
showcontent = false showcontent = true

View File

@@ -192,27 +192,41 @@ class TestDistribution:
]) ])
assert dest.join(subdir.basename).check(dir=1) assert dest.join(subdir.basename).check(dir=1)
def test_backward_compatibility_worker_terminology(self, testdir):
"""Ensure that we still support "config.slaveinput" for backward compatibility (#234).
Keep in mind that removing this compatibility will break a ton of plugins and user code.
"""
testdir.makepyfile("""
def test(pytestconfig):
assert hasattr(pytestconfig, 'slaveinput')
assert hasattr(pytestconfig, 'workerinput')
""")
result = testdir.runpytest("-n1")
result.stdout.fnmatch_lines("*1 passed*")
assert result.ret == 0
def test_data_exchange(self, testdir): def test_data_exchange(self, testdir):
testdir.makeconftest(""" testdir.makeconftest("""
# This hook only called on master. # This hook only called on master.
def pytest_configure_node(node): def pytest_configure_node(node):
node.slaveinput['a'] = 42 node.workerinput['a'] = 42
node.slaveinput['b'] = 7 node.workerinput['b'] = 7
def pytest_configure(config): def pytest_configure(config):
# this attribute is only set on slaves # this attribute is only set on workers
if hasattr(config, 'slaveinput'): if hasattr(config, 'workerinput'):
a = config.slaveinput['a'] a = config.workerinput['a']
b = config.slaveinput['b'] b = config.workerinput['b']
r = a + b r = a + b
config.slaveoutput['r'] = r config.workeroutput['r'] = r
# This hook only called on master. # This hook only called on master.
def pytest_testnodedown(node, error): def pytest_testnodedown(node, error):
node.config.calc_result = node.slaveoutput['r'] node.config.calc_result = node.workeroutput['r']
def pytest_terminal_summary(terminalreporter): def pytest_terminal_summary(terminalreporter):
if not hasattr(terminalreporter.config, 'slaveinput'): if not hasattr(terminalreporter.config, 'workerinput'):
calc_result = terminalreporter.config.calc_result calc_result = terminalreporter.config.calc_result
terminalreporter._tw.sep('-', terminalreporter._tw.sep('-',
'calculated result is %s' % calc_result) 'calculated result is %s' % calc_result)
@@ -232,12 +246,12 @@ class TestDistribution:
""") """)
testdir.makeconftest(""" testdir.makeconftest("""
def pytest_sessionfinish(session): def pytest_sessionfinish(session):
# on the slave # on the worker
if hasattr(session.config, 'slaveoutput'): if hasattr(session.config, 'workeroutput'):
session.config.slaveoutput['s2'] = 42 session.config.workeroutput['s2'] = 42
# on the master # on the master
def pytest_testnodedown(node, error): def pytest_testnodedown(node, error):
assert node.slaveoutput['s2'] == 42 assert node.workeroutput['s2'] == 42
print ("s2call-finished") print ("s2call-finished")
""") """)
args = ["-n1", "--debug"] args = ["-n1", "--debug"]
@@ -411,7 +425,7 @@ def test_teardownfails_one_function(testdir):
def test_terminate_on_hangingnode(testdir): def test_terminate_on_hangingnode(testdir):
p = testdir.makeconftest(""" p = testdir.makeconftest("""
def pytest_sessionfinish(session): def pytest_sessionfinish(session):
if session.nodeid == "my": # running on slave if session.nodeid == "my": # running on worker
import time import time
time.sleep(3) time.sleep(3)
""") """)
@@ -429,15 +443,15 @@ def test_session_hooks(testdir):
def pytest_sessionstart(session): def pytest_sessionstart(session):
sys.pytestsessionhooks = session sys.pytestsessionhooks = session
def pytest_sessionfinish(session): def pytest_sessionfinish(session):
if hasattr(session.config, 'slaveinput'): if hasattr(session.config, 'workerinput'):
name = "slave" name = "worker"
else: else:
name = "master" name = "master"
f = open(name, "w") f = open(name, "w")
f.write("xy") f.write("xy")
f.close() f.close()
# let's fail on the slave # let's fail on the worker
if name == "slave": if name == "worker":
raise ValueError(42) raise ValueError(42)
""") """)
p = testdir.makepyfile(""" p = testdir.makepyfile("""
@@ -453,14 +467,14 @@ def test_session_hooks(testdir):
assert not result.ret assert not result.ret
d = result.parseoutcomes() d = result.parseoutcomes()
assert d['passed'] == 1 assert d['passed'] == 1
assert testdir.tmpdir.join("slave").check() assert testdir.tmpdir.join("worker").check()
assert testdir.tmpdir.join("master").check() assert testdir.tmpdir.join("master").check()
def test_session_testscollected(testdir): def test_session_testscollected(testdir):
""" """
Make sure master node is updating the session object with the number Make sure master node is updating the session object with the number
of tests collected from the slaves. of tests collected from the workers.
""" """
testdir.makepyfile(test_foo=""" testdir.makepyfile(test_foo="""
import pytest import pytest
@@ -667,8 +681,8 @@ class TestNodeFailure:
""") """)
res = testdir.runpytest(f, '-n1') res = testdir.runpytest(f, '-n1')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Replacing crashed slave*", "*Replacing crashed worker*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*1 failed*1 passed*", "*1 failed*1 passed*",
]) ])
@@ -682,8 +696,8 @@ class TestNodeFailure:
""") """)
res = testdir.runpytest(f, '-n2') res = testdir.runpytest(f, '-n2')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Replacing crashed slave*", "*Replacing crashed worker*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*1 failed*3 passed*", "*1 failed*3 passed*",
]) ])
@@ -695,8 +709,8 @@ class TestNodeFailure:
""") """)
res = testdir.runpytest(f, '--dist=each', '--tx=popen') res = testdir.runpytest(f, '--dist=each', '--tx=popen')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Replacing crashed slave*", "*Replacing crashed worker*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*1 failed*1 passed*", "*1 failed*1 passed*",
]) ])
@@ -709,12 +723,12 @@ class TestNodeFailure:
""") """)
res = testdir.runpytest(f, '--dist=each', '--tx=2*popen') res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Replacing crashed slave*", "*Replacing crashed worker*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*2 failed*2 passed*", "*2 failed*2 passed*",
]) ])
def test_max_slave_restart(self, testdir): def test_max_worker_restart(self, testdir):
f = testdir.makepyfile(""" f = testdir.makepyfile("""
import os import os
def test_a(): pass def test_a(): pass
@@ -722,21 +736,21 @@ class TestNodeFailure:
def test_c(): os._exit(1) def test_c(): os._exit(1)
def test_d(): pass def test_d(): pass
""") """)
res = testdir.runpytest(f, '-n4', '--max-slave-restart=1') res = testdir.runpytest(f, '-n4', '--max-worker-restart=1')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Replacing crashed slave*", "*Replacing crashed worker*",
"*Maximum crashed slaves reached: 1*", "*Maximum crashed workers reached: 1*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*2 failed*2 passed*", "*2 failed*2 passed*",
]) ])
def test_max_slave_restart_die(self, testdir): def test_max_worker_restart_die(self, testdir):
f = testdir.makepyfile(""" f = testdir.makepyfile("""
import os import os
os._exit(1) os._exit(1)
""") """)
res = testdir.runpytest(f, '-n4', '--max-slave-restart=0') res = testdir.runpytest(f, '-n4', '--max-worker-restart=0')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Unexpectedly no active workers*", "*Unexpectedly no active workers*",
"*INTERNALERROR*" "*INTERNALERROR*"
@@ -749,10 +763,10 @@ class TestNodeFailure:
def test_b(): os._exit(1) def test_b(): os._exit(1)
def test_c(): pass def test_c(): pass
""") """)
res = testdir.runpytest(f, '-n4', '--max-slave-restart=0') res = testdir.runpytest(f, '-n4', '--max-worker-restart=0')
res.stdout.fnmatch_lines([ res.stdout.fnmatch_lines([
"*Slave restarting disabled*", "*Worker restarting disabled*",
"*Slave*crashed while running*", "*Worker*crashed while running*",
"*1 failed*2 passed*", "*1 failed*2 passed*",
]) ])
@@ -874,6 +888,39 @@ class TestLoadScope:
assert get_workers_and_test_count_by_prefix( assert get_workers_and_test_count_by_prefix(
'test_a.py::TestB', result.outlines) in ({'gw0': 10}, {'gw1': 10}) 'test_a.py::TestB', result.outlines) in ({'gw0': 10}, {'gw1': 10})
def test_module_single_start(self, testdir):
"""Fix test suite never finishing in case all workers start with a single test (#277)."""
test_file1 = """
import pytest
def test():
pass
"""
test_file2 = """
import pytest
def test_1():
pass
def test_2():
pass
"""
testdir.makepyfile(
test_a=test_file1,
test_b=test_file1,
test_c=test_file2
)
result = testdir.runpytest('-n2', '--dist=loadscope', '-v')
a = get_workers_and_test_count_by_prefix('test_a.py::test',
result.outlines)
b = get_workers_and_test_count_by_prefix('test_b.py::test',
result.outlines)
c1 = get_workers_and_test_count_by_prefix('test_c.py::test_1',
result.outlines)
c2 = get_workers_and_test_count_by_prefix('test_c.py::test_2',
result.outlines)
assert a in ({'gw0': 1}, {'gw1': 1})
assert b in ({'gw0': 1}, {'gw1': 1})
assert a.items() != b.items()
assert c1 == c2
class TestFileScope: class TestFileScope:
@@ -929,6 +976,39 @@ class TestFileScope:
assert test_b_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \ assert test_b_workers_and_test_count in ({'gw0': 10}, {'gw1': 0}) or \
test_b_workers_and_test_count in ({'gw0': 0}, {'gw1': 10}) test_b_workers_and_test_count in ({'gw0': 0}, {'gw1': 10})
def test_module_single_start(self, testdir):
"""Fix test suite never finishing in case all workers start with a single test (#277)."""
test_file1 = """
import pytest
def test():
pass
"""
test_file2 = """
import pytest
def test_1():
pass
def test_2():
pass
"""
testdir.makepyfile(
test_a=test_file1,
test_b=test_file1,
test_c=test_file2
)
result = testdir.runpytest('-n2', '--dist=loadfile', '-v')
a = get_workers_and_test_count_by_prefix('test_a.py::test',
result.outlines)
b = get_workers_and_test_count_by_prefix('test_b.py::test',
result.outlines)
c1 = get_workers_and_test_count_by_prefix('test_c.py::test_1',
result.outlines)
c2 = get_workers_and_test_count_by_prefix('test_c.py::test_2',
result.outlines)
assert a in ({'gw0': 1}, {'gw1': 1})
assert b in ({'gw0': 1}, {'gw1': 1})
assert a.items() != b.items()
assert c1 == c2
def parse_tests_and_workers_from_output(lines): def parse_tests_and_workers_from_output(lines):
result = [] result = []

View File

@@ -203,7 +203,7 @@ class TestLoadScheduling:
def test_different_tests_collected(self, testdir): def test_different_tests_collected(self, testdir):
""" """
Test that LoadScheduling is reporting collection errors when Test that LoadScheduling is reporting collection errors when
different test ids are collected by slaves. different test ids are collected by workers.
""" """
class CollectHook(object): class CollectHook(object):

View File

@@ -20,10 +20,10 @@ class TestHooks:
def pytest_runtest_logreport(report): def pytest_runtest_logreport(report):
if hasattr(report, 'node'): if hasattr(report, 'node'):
if report.when == "call": if report.when == "call":
slaveid = report.node.slaveinput['slaveid'] workerid = report.node.workerinput['workerid']
if slaveid != report.worker_id: if workerid != report.worker_id:
print("HOOK: Worker id mismatch: %s %s" print("HOOK: Worker id mismatch: %s %s"
% (slaveid, report.worker_id)) % (workerid, report.worker_id))
else: else:
print("HOOK: %s %s" print("HOOK: %s %s"
% (report.nodeid, report.worker_id)) % (report.nodeid, report.worker_id))
@@ -41,9 +41,9 @@ class TestHooks:
""" """
testdir.makeconftest(""" testdir.makeconftest("""
def pytest_xdist_node_collection_finished(node, ids): def pytest_xdist_node_collection_finished(node, ids):
slaveid = node.slaveinput['slaveid'] workerid = node.workerinput['workerid']
stripped_ids = [x.split('::')[1] for x in ids] stripped_ids = [x.split('::')[1] for x in ids]
print("HOOK: %s %s" % (slaveid, ', '.join(stripped_ids))) print("HOOK: %s %s" % (workerid, ', '.join(stripped_ids)))
""") """)
res = testdir.runpytest('-n2', '-s') res = testdir.runpytest('-n2', '-s')
res.stdout.fnmatch_lines_random([ res.stdout.fnmatch_lines_random([

View File

@@ -1,6 +1,6 @@
import py import py
import execnet import execnet
from xdist.slavemanage import NodeManager from xdist.workermanage import NodeManager
def test_dist_incompatibility_messages(testdir): def test_dist_incompatibility_messages(testdir):

View File

@@ -1,6 +1,6 @@
import py import py
import pytest import pytest
from xdist.slavemanage import SlaveController, unserialize_report from xdist.workermanage import WorkerController, unserialize_report
from xdist.remote import serialize_report from xdist.remote import serialize_report
import execnet import execnet
import marshal import marshal
@@ -26,7 +26,7 @@ class EventCall:
return "<EventCall %s(**%s)>" % (self.name, self.kwargs) return "<EventCall %s(**%s)>" % (self.name, self.kwargs)
class SlaveSetup: class WorkerSetup:
use_callback = False use_callback = False
def __init__(self, request, testdir): def __init__(self, request, testdir):
@@ -44,7 +44,7 @@ class SlaveSetup:
class DummyMananger: class DummyMananger:
specs = [0, 1] specs = [0, 1]
self.slp = SlaveController(DummyMananger, self.gateway, config, self.slp = WorkerController(DummyMananger, self.gateway, config,
putevent) putevent)
self.request.addfinalizer(self.slp.ensure_teardown) self.request.addfinalizer(self.slp.ensure_teardown)
self.slp.setup() self.slp.setup()
@@ -65,8 +65,8 @@ class SlaveSetup:
@pytest.fixture @pytest.fixture
def slave(request, testdir): def worker(request, testdir):
return SlaveSetup(request, testdir) return WorkerSetup(request, testdir)
@pytest.mark.xfail(reason='#59') @pytest.mark.xfail(reason='#59')
@@ -243,107 +243,107 @@ class TestReportSerialization:
assert newrep.longrepr == str(rep.longrepr) assert newrep.longrepr == str(rep.longrepr)
class TestSlaveInteractor: class TestWorkerInteractor:
def test_basic_collect_and_runtests(self, slave): def test_basic_collect_and_runtests(self, worker):
slave.testdir.makepyfile(""" worker.testdir.makepyfile("""
def test_func(): def test_func():
pass pass
""") """)
slave.setup() worker.setup()
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "slaveready" assert ev.name == "workerready"
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectionstart" assert ev.name == "collectionstart"
assert not ev.kwargs assert not ev.kwargs
ev = slave.popevent("collectionfinish") ev = worker.popevent("collectionfinish")
assert ev.kwargs['topdir'] == slave.testdir.tmpdir assert ev.kwargs['topdir'] == worker.testdir.tmpdir
ids = ev.kwargs['ids'] ids = ev.kwargs['ids']
assert len(ids) == 1 assert len(ids) == 1
slave.sendcommand("runtests", indices=list(range(len(ids)))) worker.sendcommand("runtests", indices=list(range(len(ids))))
slave.sendcommand("shutdown") worker.sendcommand("shutdown")
ev = slave.popevent("logstart") ev = worker.popevent("logstart")
assert ev.kwargs["nodeid"].endswith("test_func") assert ev.kwargs["nodeid"].endswith("test_func")
assert len(ev.kwargs["location"]) == 3 assert len(ev.kwargs["location"]) == 3
ev = slave.popevent("testreport") # setup ev = worker.popevent("testreport") # setup
ev = slave.popevent("testreport") ev = worker.popevent("testreport")
assert ev.name == "testreport" assert ev.name == "testreport"
rep = unserialize_report(ev.name, ev.kwargs['data']) rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.nodeid.endswith("::test_func") assert rep.nodeid.endswith("::test_func")
assert rep.passed assert rep.passed
assert rep.when == "call" assert rep.when == "call"
ev = slave.popevent("slavefinished") ev = worker.popevent("workerfinished")
assert 'slaveoutput' in ev.kwargs assert 'workeroutput' in ev.kwargs
@pytest.mark.skipif(pytest.__version__ >= '3.0', @pytest.mark.skipif(pytest.__version__ >= '3.0',
reason='skip at module level illegal in pytest 3.0') reason='skip at module level illegal in pytest 3.0')
def test_remote_collect_skip(self, slave): def test_remote_collect_skip(self, worker):
slave.testdir.makepyfile(""" worker.testdir.makepyfile("""
import py import py
py.test.skip("hello") py.test.skip("hello")
""") """)
slave.setup() worker.setup()
ev = slave.popevent("collectionstart") ev = worker.popevent("collectionstart")
assert not ev.kwargs assert not ev.kwargs
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectreport" assert ev.name == "collectreport"
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectreport" assert ev.name == "collectreport"
rep = unserialize_report(ev.name, ev.kwargs['data']) rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.skipped assert rep.skipped
ev = slave.popevent("collectionfinish") ev = worker.popevent("collectionfinish")
assert not ev.kwargs['ids'] assert not ev.kwargs['ids']
def test_remote_collect_fail(self, slave): def test_remote_collect_fail(self, worker):
slave.testdir.makepyfile("""aasd qwe""") worker.testdir.makepyfile("""aasd qwe""")
slave.setup() worker.setup()
ev = slave.popevent("collectionstart") ev = worker.popevent("collectionstart")
assert not ev.kwargs assert not ev.kwargs
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectreport" assert ev.name == "collectreport"
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectreport" assert ev.name == "collectreport"
rep = unserialize_report(ev.name, ev.kwargs['data']) rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.failed assert rep.failed
ev = slave.popevent("collectionfinish") ev = worker.popevent("collectionfinish")
assert not ev.kwargs['ids'] assert not ev.kwargs['ids']
def test_runtests_all(self, slave): def test_runtests_all(self, worker):
slave.testdir.makepyfile(""" worker.testdir.makepyfile("""
def test_func(): pass def test_func(): pass
def test_func2(): pass def test_func2(): pass
""") """)
slave.setup() worker.setup()
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "slaveready" assert ev.name == "workerready"
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "collectionstart" assert ev.name == "collectionstart"
assert not ev.kwargs assert not ev.kwargs
ev = slave.popevent("collectionfinish") ev = worker.popevent("collectionfinish")
ids = ev.kwargs['ids'] ids = ev.kwargs['ids']
assert len(ids) == 2 assert len(ids) == 2
slave.sendcommand("runtests_all", ) worker.sendcommand("runtests_all", )
slave.sendcommand("shutdown", ) worker.sendcommand("shutdown", )
for func in "::test_func", "::test_func2": for func in "::test_func", "::test_func2":
for i in range(3): # setup/call/teardown for i in range(3): # setup/call/teardown
ev = slave.popevent("testreport") ev = worker.popevent("testreport")
assert ev.name == "testreport" assert ev.name == "testreport"
rep = unserialize_report(ev.name, ev.kwargs['data']) rep = unserialize_report(ev.name, ev.kwargs['data'])
assert rep.nodeid.endswith(func) assert rep.nodeid.endswith(func)
ev = slave.popevent("slavefinished") ev = worker.popevent("workerfinished")
assert 'slaveoutput' in ev.kwargs assert 'workeroutput' in ev.kwargs
def test_happy_run_events_converted(self, testdir, slave): def test_happy_run_events_converted(self, testdir, worker):
py.test.xfail("implement a simple test for event production") py.test.xfail("implement a simple test for event production")
assert not slave.use_callback assert not worker.use_callback
slave.testdir.makepyfile(""" worker.testdir.makepyfile("""
def test_func(): def test_func():
pass pass
""") """)
slave.setup() worker.setup()
hookrec = testdir.getreportrecorder(slave.config) hookrec = testdir.getreportrecorder(worker.config)
for data in slave.slp.channel: for data in worker.slp.channel:
slave.slp.process_from_remote(data) worker.slp.process_from_remote(data)
slave.slp.process_from_remote(slave.slp.ENDMARK) worker.slp.process_from_remote(worker.slp.ENDMARK)
py.std.pprint.pprint(hookrec.hookrecorder.calls) py.std.pprint.pprint(hookrec.hookrecorder.calls)
hookrec.hookrecorder.contains([ hookrec.hookrecorder.contains([
("pytest_collectstart", "collector.fspath == aaa"), ("pytest_collectstart", "collector.fspath == aaa"),
@@ -354,13 +354,13 @@ class TestSlaveInteractor:
("pytest_collectreport", "report.collector.fspath == bbb"), ("pytest_collectreport", "report.collector.fspath == bbb"),
]) ])
def test_process_from_remote_error_handling(self, slave, capsys): def test_process_from_remote_error_handling(self, worker, capsys):
slave.use_callback = True worker.use_callback = True
slave.setup() worker.setup()
slave.slp.process_from_remote(('<nonono>', ())) worker.slp.process_from_remote(('<nonono>', ()))
out, err = capsys.readouterr() out, err = capsys.readouterr()
assert 'INTERNALERROR> ValueError: unknown event: <nonono>' in out assert 'INTERNALERROR> ValueError: unknown event: <nonono>' in out
ev = slave.popevent() ev = worker.popevent()
assert ev.name == "errordown" assert ev.name == "errordown"
@@ -371,5 +371,5 @@ def test_remote_env_vars(testdir):
assert os.environ['PYTEST_XDIST_WORKER'] in ('gw0', 'gw1') assert os.environ['PYTEST_XDIST_WORKER'] in ('gw0', 'gw1')
assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2' assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2'
''') ''')
result = testdir.runpytest('-n2', '--max-slave-restart=0') result = testdir.runpytest('-n2', '--max-worker-restart=0')
assert result.ret == 0 assert result.ret == 0

View File

@@ -2,8 +2,8 @@ import py
import pytest import pytest
import execnet import execnet
from _pytest.pytester import HookRecorder from _pytest.pytester import HookRecorder
from xdist import slavemanage, newhooks from xdist import workermanage, newhooks
from xdist.slavemanage import HostRSync, NodeManager from xdist.workermanage import HostRSync, NodeManager
pytest_plugins = "pytester" pytest_plugins = "pytester"
@@ -32,7 +32,7 @@ def mysetup(tmpdir):
@pytest.fixture @pytest.fixture
def slavecontroller(monkeypatch): def workercontroller(monkeypatch):
class MockController(object): class MockController(object):
def __init__(self, *args): def __init__(self, *args):
pass pass
@@ -40,7 +40,7 @@ def slavecontroller(monkeypatch):
def setup(self): def setup(self):
pass pass
monkeypatch.setattr(slavemanage, 'SlaveController', MockController) monkeypatch.setattr(workermanage, 'WorkerController', MockController)
return MockController return MockController
@@ -57,7 +57,7 @@ class TestNodeManagerPopen:
assert spec.chdir == "abc" assert spec.chdir == "abc"
def test_popen_makegateway_events(self, config, hookrecorder, def test_popen_makegateway_events(self, config, hookrecorder,
slavecontroller): workercontroller):
hm = NodeManager(config, ["popen"] * 2) hm = NodeManager(config, ["popen"] * 2)
hm.setup_nodes(None) hm.setup_nodes(None)
call = hookrecorder.popcall("pytest_xdist_setupnodes") call = hookrecorder.popcall("pytest_xdist_setupnodes")
@@ -72,7 +72,7 @@ class TestNodeManagerPopen:
hm.teardown_nodes() hm.teardown_nodes()
assert not len(hm.group) assert not len(hm.group)
def test_popens_rsync(self, config, mysetup, slavecontroller): def test_popens_rsync(self, config, mysetup, workercontroller):
source = mysetup.source source = mysetup.source
hm = NodeManager(config, ["popen"] * 2) hm = NodeManager(config, ["popen"] * 2)
hm.setup_nodes(None) hm.setup_nodes(None)
@@ -97,7 +97,7 @@ class TestNodeManagerPopen:
assert not len(hm.group) assert not len(hm.group)
assert "sys.path.insert" in gw.remote_exec.args[0] assert "sys.path.insert" in gw.remote_exec.args[0]
def test_rsync_popen_with_path(self, config, mysetup, slavecontroller): def test_rsync_popen_with_path(self, config, mysetup, workercontroller):
source, dest = mysetup.source, mysetup.dest source, dest = mysetup.source, mysetup.dest
hm = NodeManager(config, ["popen//chdir=%s" % dest] * 1) hm = NodeManager(config, ["popen//chdir=%s" % dest] * 1)
hm.setup_nodes(None) hm.setup_nodes(None)
@@ -114,7 +114,7 @@ class TestNodeManagerPopen:
assert dest.join("dir1", "dir2", 'hello').check() assert dest.join("dir1", "dir2", 'hello').check()
def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder, def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
slavecontroller): workercontroller):
source, dest = mysetup.source, mysetup.dest source, dest = mysetup.source, mysetup.dest
hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2) hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
hm.roots = [] hm.roots = []
@@ -174,7 +174,7 @@ class TestNodeManager:
assert p.join("dir1").check() assert p.join("dir1").check()
assert p.join("dir1", "file1").check() assert p.join("dir1", "file1").check()
def test_popen_rsync_subdir(self, testdir, mysetup, slavecontroller): def test_popen_rsync_subdir(self, testdir, mysetup, workercontroller):
source, dest = mysetup.source, mysetup.dest source, dest = mysetup.source, mysetup.dest
dir1 = mysetup.source.mkdir("dir1") dir1 = mysetup.source.mkdir("dir1")
dir2 = dir1.mkdir("dir2") dir2 = dir1.mkdir("dir2")
@@ -192,7 +192,7 @@ class TestNodeManager:
assert dest.join("dir1", "dir2", 'hello').check() assert dest.join("dir1", "dir2", 'hello').check()
nodemanager.teardown_nodes() nodemanager.teardown_nodes()
def test_init_rsync_roots(self, testdir, mysetup, slavecontroller): def test_init_rsync_roots(self, testdir, mysetup, workercontroller):
source, dest = mysetup.source, mysetup.dest source, dest = mysetup.source, mysetup.dest
dir2 = source.ensure("dir1", "dir2", dir=1) dir2 = source.ensure("dir1", "dir2", dir=1)
source.ensure("dir1", "somefile", dir=1) source.ensure("dir1", "somefile", dir=1)
@@ -209,7 +209,7 @@ class TestNodeManager:
assert not dest.join("dir1").check() assert not dest.join("dir1").check()
assert not dest.join("bogus").check() assert not dest.join("bogus").check()
def test_rsyncignore(self, testdir, mysetup, slavecontroller): def test_rsyncignore(self, testdir, mysetup, workercontroller):
source, dest = mysetup.source, mysetup.dest source, dest = mysetup.source, mysetup.dest
dir2 = source.ensure("dir1", "dir2", dir=1) dir2 = source.ensure("dir1", "dir2", dir=1)
source.ensure("dir5", "dir6", "bogus") source.ensure("dir5", "dir6", "bogus")
@@ -233,7 +233,7 @@ class TestNodeManager:
assert not dest.join('foo').check() assert not dest.join('foo').check()
assert not dest.join('bar').check() assert not dest.join('bar').check()
def test_optimise_popen(self, testdir, mysetup, slavecontroller): def test_optimise_popen(self, testdir, mysetup, workercontroller):
source = mysetup.source source = mysetup.source
specs = ["popen"] * 3 specs = ["popen"] * 3
source.join("conftest.py").write("rsyncdirs = ['a']") source.join("conftest.py").write("rsyncdirs = ['a']")

View File

@@ -1,7 +1,7 @@
import py import py
import pytest import pytest
from xdist.slavemanage import NodeManager from xdist.workermanage import NodeManager
from xdist.scheduler import ( from xdist.scheduler import (
EachScheduling, EachScheduling,
LoadScheduling, LoadScheduling,
@@ -22,7 +22,7 @@ class DSession:
At the beginning of the test session this creates a NodeManager At the beginning of the test session this creates a NodeManager
instance which creates and starts all nodes. Nodes then emit instance which creates and starts all nodes. Nodes then emit
events processed in the pytest_runtestloop hook using the slave_* events processed in the pytest_runtestloop hook using the worker_*
methods. methods.
Once a node is started it will automatically start running the Once a node is started it will automatically start running the
@@ -46,9 +46,9 @@ class DSession:
self._failed_collection_errors = {} self._failed_collection_errors = {}
self._active_nodes = set() self._active_nodes = set()
self._failed_nodes_count = 0 self._failed_nodes_count = 0
self._max_slave_restart = self.config.getoption('max_slave_restart') self._max_worker_restart = self.config.option.maxworkerrestart
if self._max_slave_restart is not None: if self._max_worker_restart is not None:
self._max_slave_restart = int(self._max_slave_restart) self._max_worker_restart = int(self._max_worker_restart)
try: try:
self.terminal = config.pluginmanager.getplugin("terminalreporter") self.terminal = config.pluginmanager.getplugin("terminalreporter")
except KeyError: except KeyError:
@@ -75,7 +75,7 @@ class DSession:
"""Creates and starts the nodes. """Creates and starts the nodes.
The nodes are setup to put their events onto self.queue. As The nodes are setup to put their events onto self.queue. As
soon as nodes start they will emit the slave_slaveready event. soon as nodes start they will emit the worker_workerready event.
""" """
self.nodemanager = NodeManager(self.config) self.nodemanager = NodeManager(self.config)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put) nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
@@ -120,7 +120,7 @@ class DSession:
return True return True
def loop_once(self): def loop_once(self):
"""Process one callback from one of the slaves.""" """Process one callback from one of the workers."""
while 1: while 1:
if not self._active_nodes: if not self._active_nodes:
# If everything has died stop looping # If everything has died stop looping
@@ -133,7 +133,7 @@ class DSession:
continue continue
callname, kwargs = eventcall callname, kwargs = eventcall
assert callname, kwargs assert callname, kwargs
method = "slave_" + callname method = "worker_" + callname
call = getattr(self, method) call = getattr(self, method)
self.log("calling method", method, kwargs) self.log("calling method", method, kwargs)
call(**kwargs) call(**kwargs)
@@ -141,44 +141,48 @@ class DSession:
self.triggershutdown() self.triggershutdown()
# #
# callbacks for processing events from slaves # callbacks for processing events from workers
# #
def slave_slaveready(self, node, slaveinfo): def worker_workerready(self, node, workerinfo):
"""Emitted when a node first starts up. """Emitted when a node first starts up.
This adds the node to the scheduler, nodes continue with This adds the node to the scheduler, nodes continue with
collection without any further input. collection without any further input.
""" """
node.slaveinfo = slaveinfo node.workerinfo = workerinfo
node.slaveinfo['id'] = node.gateway.id node.workerinfo['id'] = node.gateway.id
node.slaveinfo['spec'] = node.gateway.spec node.workerinfo['spec'] = node.gateway.spec
# TODO: (#234 task) needs this for pytest. Remove when refactor in pytest repo
node.slaveinfo = node.workerinfo
self.config.hook.pytest_testnodeready(node=node) self.config.hook.pytest_testnodeready(node=node)
if self.shuttingdown: if self.shuttingdown:
node.shutdown() node.shutdown()
else: else:
self.sched.add_node(node) self.sched.add_node(node)
def slave_slavefinished(self, node): def worker_workerfinished(self, node):
"""Emitted when node executes its pytest_sessionfinish hook. """Emitted when node executes its pytest_sessionfinish hook.
Removes the node from the scheduler. Removes the node from the scheduler.
The node might not be in 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. workerready before shutdown was triggered.
""" """
self.config.hook.pytest_testnodedown(node=node, error=None) self.config.hook.pytest_testnodedown(node=node, error=None)
if node.slaveoutput['exitstatus'] == 2: # keyboard-interrupt if node.workeroutput['exitstatus'] == 2: # keyboard-interrupt
self.shouldstop = "%s received keyboard-interrupt" % (node,) self.shouldstop = "%s received keyboard-interrupt" % (node,)
self.slave_errordown(node, "keyboard-interrupt") self.worker_errordown(node, "keyboard-interrupt")
return return
if node in self.sched.nodes: if node in self.sched.nodes:
crashitem = self.sched.remove_node(node) crashitem = self.sched.remove_node(node)
assert not crashitem, (crashitem, node) assert not crashitem, (crashitem, node)
self._active_nodes.remove(node) self._active_nodes.remove(node)
def slave_errordown(self, node, error): def worker_errordown(self, node, error):
"""Emitted by the SlaveController when a node dies.""" """Emitted by the WorkerController when a node dies."""
self.config.hook.pytest_testnodedown(node=node, error=error) self.config.hook.pytest_testnodedown(node=node, error=error)
try: try:
crashitem = self.sched.remove_node(node) crashitem = self.sched.remove_node(node)
@@ -189,22 +193,22 @@ class DSession:
self.handle_crashitem(crashitem, node) self.handle_crashitem(crashitem, node)
self._failed_nodes_count += 1 self._failed_nodes_count += 1
maximum_reached = (self._max_slave_restart is not None and maximum_reached = (self._max_worker_restart is not None and
self._failed_nodes_count > self._max_slave_restart) self._failed_nodes_count > self._max_worker_restart)
if maximum_reached: if maximum_reached:
if self._max_slave_restart == 0: if self._max_worker_restart == 0:
msg = 'Slave restarting disabled' msg = 'Worker restarting disabled'
else: else:
msg = "Maximum crashed slaves reached: %d" % \ msg = "Maximum crashed workers reached: %d" % \
self._max_slave_restart self._max_worker_restart
self.report_line(msg) self.report_line(msg)
else: else:
self.report_line("Replacing crashed slave %s" % node.gateway.id) self.report_line("Replacing crashed worker %s" % node.gateway.id)
self._clone_node(node) self._clone_node(node)
self._active_nodes.remove(node) self._active_nodes.remove(node)
def slave_collectionfinish(self, node, ids): def worker_collectionfinish(self, node, ids):
"""Slave has finished test collection. """worker has finished test collection.
This adds the collection for this node to the scheduler. If This adds the collection for this node to the scheduler. If
the scheduler indicates collection is finished (i.e. all the scheduler indicates collection is finished (i.e. all
@@ -230,23 +234,23 @@ class DSession:
self.sched.__class__.__name__)) self.sched.__class__.__name__))
self.sched.schedule() self.sched.schedule()
def slave_logstart(self, node, nodeid, location): def worker_logstart(self, node, nodeid, location):
"""Emitted when a node calls the pytest_runtest_logstart hook.""" """Emitted when a node calls the pytest_runtest_logstart hook."""
self.config.hook.pytest_runtest_logstart( self.config.hook.pytest_runtest_logstart(
nodeid=nodeid, location=location) nodeid=nodeid, location=location)
def slave_logfinish(self, node, nodeid, location): def worker_logfinish(self, node, nodeid, location):
"""Emitted when a node calls the pytest_runtest_logfinish hook.""" """Emitted when a node calls the pytest_runtest_logfinish hook."""
self.config.hook.pytest_runtest_logfinish( self.config.hook.pytest_runtest_logfinish(
nodeid=nodeid, location=location) nodeid=nodeid, location=location)
def slave_testreport(self, node, rep): def worker_testreport(self, node, rep):
"""Emitted when a node calls the pytest_runtest_logreport hook.""" """Emitted when a node calls the pytest_runtest_logreport hook."""
rep.node = node rep.node = node
self.config.hook.pytest_runtest_logreport(report=rep) self.config.hook.pytest_runtest_logreport(report=rep)
self._handlefailures(rep) self._handlefailures(rep)
def slave_runtest_protocol_complete(self, node, item_index, duration): def worker_runtest_protocol_complete(self, node, item_index, duration):
""" """
Emitted when a node fires the 'runtest_protocol_complete' event, Emitted when a node fires the 'runtest_protocol_complete' event,
signalling that a test has completed the runtestprotocol and should be signalling that a test has completed the runtestprotocol and should be
@@ -254,12 +258,12 @@ class DSession:
""" """
self.sched.mark_test_complete(node, item_index, duration) self.sched.mark_test_complete(node, item_index, duration)
def slave_collectreport(self, node, rep): def worker_collectreport(self, node, rep):
"""Emitted when a node calls the pytest_collectreport hook.""" """Emitted when a node calls the pytest_collectreport hook."""
if rep.failed: if rep.failed:
self._failed_slave_collectreport(node, rep) self._failed_worker_collectreport(node, rep)
def slave_logwarning(self, message, code, nodeid, fslocation): def worker_logwarning(self, message, code, nodeid, fslocation):
"""Emitted when a node calls the pytest_logwarning hook.""" """Emitted when a node calls the pytest_logwarning hook."""
kwargs = dict(message=message, code=code, nodeid=nodeid, fslocation=fslocation) kwargs = dict(message=message, code=code, nodeid=nodeid, fslocation=fslocation)
self.config.hook.pytest_logwarning.call_historic(kwargs=kwargs) self.config.hook.pytest_logwarning.call_historic(kwargs=kwargs)
@@ -270,7 +274,7 @@ class DSession:
This is normally for when a node dies, 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 of the existing node and create a new one with a new id. The
new node will have been setup so it will start calling the new node will have been setup so it will start calling the
"slave_*" hooks and do work soon. "worker_*" hooks and do work soon.
""" """
spec = node.gateway.spec spec = node.gateway.spec
spec.id = None spec.id = None
@@ -279,9 +283,9 @@ class DSession:
self._active_nodes.add(node) self._active_nodes.add(node)
return node return node
def _failed_slave_collectreport(self, node, rep): def _failed_worker_collectreport(self, node, rep):
# Check we haven't already seen this report (from # Check we haven't already seen this report (from
# another slave). # another worker).
if rep.longrepr not in self._failed_collection_errors: if rep.longrepr not in self._failed_collection_errors:
self._failed_collection_errors[rep.longrepr] = True self._failed_collection_errors[rep.longrepr] = True
self.config.hook.pytest_collectreport(report=rep) self.config.hook.pytest_collectreport(report=rep)
@@ -300,15 +304,15 @@ class DSession:
for node in self.sched.nodes: for node in self.sched.nodes:
node.shutdown() node.shutdown()
def handle_crashitem(self, nodeid, slave): def handle_crashitem(self, nodeid, worker):
# XXX get more reporting info by recording pytest_runtest_logstart? # XXX get more reporting info by recording pytest_runtest_logstart?
# XXX count no of failures and retry N times # XXX count no of failures and retry N times
runner = self.config.pluginmanager.getplugin("runner") runner = self.config.pluginmanager.getplugin("runner")
fspath = nodeid.split("::")[0] fspath = nodeid.split("::")[0]
msg = "Slave %r crashed while running %r" % (slave.gateway.id, nodeid) msg = "Worker %r crashed while running %r" % (worker.gateway.id, nodeid)
rep = runner.TestReport(nodeid, (fspath, None, fspath), rep = runner.TestReport(nodeid, (fspath, None, fspath),
(), "failed", msg, "???") (), "failed", msg, "???")
rep.node = slave rep.node = worker
self.config.hook.pytest_runtest_logreport(report=rep) self.config.hook.pytest_runtest_logreport(report=rep)
@@ -364,7 +368,7 @@ class TerminalDistReporter:
def pytest_testnodeready(self, node): def pytest_testnodeready(self, node):
if self.config.option.verbose > 0: if self.config.option.verbose > 0:
d = node.slaveinfo d = node.workerinfo
infoline = "[%s] Python %s" % ( infoline = "[%s] Python %s" % (
d['id'], d['id'],
d['version'].replace('\n', ' -- '),) d['version'].replace('\n', ' -- '),)

View File

@@ -69,10 +69,10 @@ class RemoteControl(object):
out = py.io.TerminalWriter() out = py.io.TerminalWriter()
if hasattr(self, 'gateway'): if hasattr(self, 'gateway'):
raise ValueError("already have gateway %r" % self.gateway) raise ValueError("already have gateway %r" % self.gateway)
self.trace("setting up slave session") self.trace("setting up worker session")
self.gateway = self.initgateway() self.gateway = self.initgateway()
self.channel = channel = self.gateway.remote_exec( self.channel = channel = self.gateway.remote_exec(
init_slave_session, init_worker_session,
args=self.config.args, args=self.config.args,
option_dict=vars(self.config.option), option_dict=vars(self.config.option),
) )
@@ -134,7 +134,7 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
tr.line("### Watching: %s" % (rootdir,), bold=True) tr.line("### Watching: %s" % (rootdir,), bold=True)
def init_slave_session(channel, args, option_dict): def init_worker_session(channel, args, option_dict):
import os import os
import sys import sys
outchannel = channel.gateway.newchannel() outchannel = channel.gateway.newchannel()
@@ -153,11 +153,11 @@ def init_slave_session(channel, args, option_dict):
from _pytest.config import Config from _pytest.config import Config
config = Config.fromdictargs(option_dict, list(args)) config = Config.fromdictargs(option_dict, list(args))
config.args = args config.args = args
from xdist.looponfail import SlaveFailSession from xdist.looponfail import WorkerFailSession
SlaveFailSession(config, channel).main() WorkerFailSession(config, channel).main()
class SlaveFailSession: class WorkerFailSession:
def __init__(self, config, channel): def __init__(self, config, channel):
self.config = config self.config = config
self.channel = channel self.channel = channel
@@ -194,11 +194,11 @@ class SlaveFailSession:
self.collection_failed = True self.collection_failed = True
def main(self): def main(self):
self.DEBUG("SLAVE: received configuration, waiting for command trails") self.DEBUG("WORKER: received configuration, waiting for command trails")
try: try:
command = self.channel.receive() command = self.channel.receive()
except KeyboardInterrupt: except KeyboardInterrupt:
return # in the slave we can't do much about this return # in the worker we can't do much about this
self.DEBUG("received", command) self.DEBUG("received", command)
self.current_command = command self.current_command = command
self.config.hook.pytest_cmdline_main(config=self.config) self.config.hook.pytest_cmdline_main(config=self.config)

View File

@@ -26,9 +26,12 @@ def pytest_addoption(parser):
help="shortcut for '--dist=load --tx=NUM*popen', " help="shortcut for '--dist=load --tx=NUM*popen', "
"you can use 'auto' here for auto detection CPUs number on " "you can use 'auto' here for auto detection CPUs number on "
"host system") "host system")
group.addoption('--max-slave-restart', action="store", default=None, group.addoption('--max-worker-restart', '--max-slave-restart', action="store", default=None,
help="maximum number of slaves that can be restarted " dest="maxworkerrestart",
"when crashed (set to zero to disable this feature)") help="maximum number of workers that can be restarted "
"when crashed (set to zero to disable this feature)\n"
"'--max-slave-restart' option is deprecated and will be removed in "
"a future release")
group.addoption( group.addoption(
'--dist', metavar="distmode", '--dist', metavar="distmode",
action="store", choices=['each', 'load', 'loadscope', 'loadfile', 'no'], action="store", choices=['each', 'load', 'loadscope', 'loadfile', 'no'],
@@ -129,7 +132,7 @@ def worker_id(request):
"""Return the id of the current worker ('gw0', 'gw1', etc) or 'master' """Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
if running on the master node. if running on the master node.
""" """
if hasattr(request.config, 'slaveinput'): if hasattr(request.config, 'workerinput'):
return request.config.slaveinput['slaveid'] return request.config.workerinput['workerid']
else: else:
return 'master' return 'master'

View File

@@ -14,11 +14,11 @@ import _pytest.hookspec
import pytest import pytest
class SlaveInteractor: class WorkerInteractor:
def __init__(self, config, channel): def __init__(self, config, channel):
self.config = config self.config = config
self.slaveid = config.slaveinput.get('slaveid', "?") self.workerid = config.workerinput.get('workerid', "?")
self.log = py.log.Producer("slave-%s" % self.slaveid) self.log = py.log.Producer("worker-%s" % self.workerid)
if not config.option.debug: if not config.option.debug:
py.log.setconsumer(self.log._keywords, None) py.log.setconsumer(self.log._keywords, None)
self.channel = channel self.channel = channel
@@ -34,14 +34,14 @@ class SlaveInteractor:
def pytest_sessionstart(self, session): def pytest_sessionstart(self, session):
self.session = session self.session = session
slaveinfo = getinfodict() workerinfo = getinfodict()
self.sendevent("slaveready", slaveinfo=slaveinfo) self.sendevent("workerready", workerinfo=workerinfo)
@pytest.hookimpl(hookwrapper=True) @pytest.hookimpl(hookwrapper=True)
def pytest_sessionfinish(self, exitstatus): def pytest_sessionfinish(self, exitstatus):
self.config.slaveoutput['exitstatus'] = exitstatus self.config.workeroutput['exitstatus'] = exitstatus
yield yield
self.sendevent("slavefinished", slaveoutput=self.config.slaveoutput) self.sendevent("workerfinished", workeroutput=self.config.workeroutput)
def pytest_collection(self, session): def pytest_collection(self, session):
self.sendevent("collectionstart") self.sendevent("collectionstart")
@@ -103,7 +103,7 @@ class SlaveInteractor:
def pytest_runtest_logreport(self, report): def pytest_runtest_logreport(self, report):
data = serialize_report(report) data = serialize_report(report)
data["item_index"] = self.item_index data["item_index"] = self.item_index
data["worker_id"] = self.slaveid data["worker_id"] = self.workerid
assert self.session.items[self.item_index].nodeid == report.nodeid assert self.session.items[self.item_index].nodeid == report.nodeid
self.sendevent("testreport", data=data) self.sendevent("testreport", data=data)
@@ -185,18 +185,21 @@ def remote_initconfig(option_dict, args):
if __name__ == '__channelexec__': if __name__ == '__channelexec__':
channel = channel # noqa channel = channel # noqa
slaveinput, args, option_dict = channel.receive() workerinput, args, option_dict = channel.receive()
importpath = os.getcwd() importpath = os.getcwd()
sys.path.insert(0, importpath) # XXX only for remote situations sys.path.insert(0, importpath) # XXX only for remote situations
os.environ['PYTHONPATH'] = ( os.environ['PYTHONPATH'] = (
importpath + os.pathsep + importpath + os.pathsep +
os.environ.get('PYTHONPATH', '')) os.environ.get('PYTHONPATH', ''))
os.environ['PYTEST_XDIST_WORKER'] = slaveinput['slaveid'] os.environ['PYTEST_XDIST_WORKER'] = workerinput['workerid']
os.environ['PYTEST_XDIST_WORKER_COUNT'] = str(slaveinput['slavecount']) os.environ['PYTEST_XDIST_WORKER_COUNT'] = str(workerinput['workercount'])
# os.environ['PYTHONPATH'] = importpath # os.environ['PYTHONPATH'] = importpath
import py import py
config = remote_initconfig(option_dict, args) config = remote_initconfig(option_dict, args)
config.slaveinput = slaveinput config.workerinput = workerinput
config.slaveoutput = {} config.workeroutput = {}
interactor = SlaveInteractor(config, channel) # TODO: deprecated name, backward compatibility only. Remove it in future
config.slaveinput = config.workerinput
config.slaveoutput = config.workeroutput
interactor = WorkerInteractor(config, channel)
config.hook.pytest_cmdline_main(config=config) config.hook.pytest_cmdline_main(config=config)

View File

@@ -1,6 +1,6 @@
from py.log import Producer from py.log import Producer
from xdist.slavemanage import parse_spec_config from xdist.workermanage import parse_spec_config
from xdist.report import report_collection_diff from xdist.report import report_collection_diff

View File

@@ -3,7 +3,7 @@ from itertools import cycle
from py.log import Producer from py.log import Producer
from _pytest.runner import CollectReport from _pytest.runner import CollectReport
from xdist.slavemanage import parse_spec_config from xdist.workermanage import parse_spec_config
from xdist.report import report_collection_diff from xdist.report import report_collection_diff
@@ -113,7 +113,7 @@ class LoadScheduling:
From now on the node will be allocated chunks of tests to From now on the node will be allocated chunks of tests to
execute. execute.
Called by the ``DSession.slave_slaveready`` hook when it Called by the ``DSession.worker_workerready`` hook when it
successfully bootstraps a new node. successfully bootstraps a new node.
""" """
assert node not in self.node2pending assert node not in self.node2pending
@@ -123,7 +123,7 @@ class LoadScheduling:
"""Add the collected test items from a node """Add the collected test items from a node
The collection is stored in the ``.node2collection`` map. The collection is stored in the ``.node2collection`` map.
Called by the ``DSession.slave_collectionfinish`` hook. Called by the ``DSession.worker_collectionfinish`` hook.
""" """
assert node in self.node2pending assert node in self.node2pending
if self.collection_is_completed: if self.collection_is_completed:
@@ -147,7 +147,7 @@ class LoadScheduling:
The duration it took to execute the item is used as a hint to The duration it took to execute the item is used as a hint to
the scheduler. the scheduler.
This is called by the ``DSession.slave_testreport`` hook. This is called by the ``DSession.worker_testreport`` hook.
""" """
self.node2pending[node].remove(item_index) self.node2pending[node].remove(item_index)
self.check_schedule(node, duration=duration) self.check_schedule(node, duration=duration)
@@ -187,8 +187,8 @@ class LoadScheduling:
This should be called either when the node crashed or at This should be called either when the node crashed or at
shutdown time. In the former case any pending items assigned shutdown time. In the former case any pending items assigned
to the node will be re-scheduled. Called by the to the node will be re-scheduled. Called by the
``DSession.slave_slavefinished`` and ``DSession.worker_workerfinished`` and
``DSession.slave_errordown`` hooks. ``DSession.worker_errordown`` hooks.
Return the item which was being executing while the node Return the item which was being executing while the node
crashed or None if the node has no more pending items. crashed or None if the node has no more pending items.
@@ -213,7 +213,7 @@ class LoadScheduling:
``.check_schedule()`` on all nodes so that newly added nodes ``.check_schedule()`` on all nodes so that newly added nodes
will start to be used. will start to be used.
This is called by the ``DSession.slave_collectionfinish`` hook This is called by the ``DSession.worker_collectionfinish`` hook
if ``.collection_is_completed`` is True. if ``.collection_is_completed`` is True.
""" """
assert self.collection_is_completed assert self.collection_is_completed

View File

@@ -3,7 +3,7 @@ from collections import OrderedDict
from _pytest.runner import CollectReport from _pytest.runner import CollectReport
from py.log import Producer from py.log import Producer
from xdist.report import report_collection_diff from xdist.report import report_collection_diff
from xdist.slavemanage import parse_spec_config from xdist.workermanage import parse_spec_config
class LoadScopeScheduling: class LoadScopeScheduling:
@@ -151,7 +151,7 @@ class LoadScopeScheduling:
From now on the node will be assigned work units to be executed. From now on the node will be assigned work units to be executed.
Called by the ``DSession.slave_slaveready`` hook when it successfully Called by the ``DSession.worker_workerready`` hook when it successfully
bootstraps a new node. bootstraps a new node.
""" """
assert node not in self.assigned_work assert node not in self.assigned_work
@@ -166,8 +166,8 @@ class LoadScopeScheduling:
Called by the hooks: Called by the hooks:
- ``DSession.slave_slavefinished``. - ``DSession.worker_workerfinished``.
- ``DSession.slave_errordown``. - ``DSession.worker_errordown``.
Return the item being executed while the node crashed or None if the Return the item being executed while the node crashed or None if the
node has no more pending items. node has no more pending items.
@@ -206,7 +206,7 @@ class LoadScopeScheduling:
Called by the hook: Called by the hook:
- ``DSession.slave_collectionfinish``. - ``DSession.worker_collectionfinish``.
""" """
# Check that add_node() was called on the node before # Check that add_node() was called on the node before
@@ -239,7 +239,7 @@ class LoadScopeScheduling:
Called by the hook: Called by the hook:
- ``DSession.slave_testreport``. - ``DSession.worker_testreport``.
""" """
nodeid = self.registered_collections[node][item_index] nodeid = self.registered_collections[node][item_index]
scope = self._split_scope(nodeid) scope = self._split_scope(nodeid)
@@ -336,7 +336,7 @@ class LoadScopeScheduling:
If ``.collection_is_completed`` is True, this is called by the hook: If ``.collection_is_completed`` is True, this is called by the hook:
- ``DSession.slave_collectionfinish``. - ``DSession.worker_collectionfinish``.
""" """
assert self.collection_is_completed assert self.collection_is_completed
@@ -380,6 +380,10 @@ class LoadScopeScheduling:
for node in self.nodes: for node in self.nodes:
self._assign_work_unit(node) self._assign_work_unit(node)
# Ensure nodes start with at least two work units if possible (#277)
for node in self.nodes:
self._reschedule(node)
# Initial distribution sent all tests, start node shutdown # Initial distribution sent all tests, start node shutdown
if not self.workqueue: if not self.workqueue:
for node in self.nodes: for node in self.nodes:

View File

@@ -68,7 +68,7 @@ class NodeManager(object):
gw = self.group.makegateway(spec) gw = self.group.makegateway(spec)
self.config.hook.pytest_xdist_newgateway(gateway=gw) self.config.hook.pytest_xdist_newgateway(gateway=gw)
self.rsync_roots(gw) self.rsync_roots(gw)
node = SlaveController(self, gw, self.config, putevent) node = WorkerController(self, gw, self.config, putevent)
gw.node = node # keep the node alive gw.node = node # keep the node alive
node.setup() node.setup()
self.trace("started node %r" % node) self.trace("started node %r" % node)
@@ -201,7 +201,7 @@ def make_reltoroot(roots, args):
return result return result
class SlaveController(object): class WorkerController(object):
ENDMARK = -1 ENDMARK = -1
def __init__(self, nodemanager, gateway, config, putevent): def __init__(self, nodemanager, gateway, config, putevent):
@@ -209,11 +209,16 @@ class SlaveController(object):
self.putevent = putevent self.putevent = putevent
self.gateway = gateway self.gateway = gateway
self.config = config self.config = config
self.slaveinput = {'slaveid': gateway.id, self.workerinput = {'workerid': gateway.id,
'slavecount': len(nodemanager.specs)} 'workercount': len(nodemanager.specs),
'slaveid': gateway.id,
'slavecount': len(nodemanager.specs)
}
# TODO: deprecated name, backward compatibility only. Remove it in future
self.slaveinput = self.workerinput
self._down = False self._down = False
self._shutdown_sent = False self._shutdown_sent = False
self.log = py.log.Producer("slavectl-%s" % gateway.id) self.log = py.log.Producer("workerctl-%s" % gateway.id)
if not self.config.option.debug: if not self.config.option.debug:
py.log.setconsumer(self.log._keywords, None) py.log.setconsumer(self.log._keywords, None)
@@ -225,7 +230,7 @@ class SlaveController(object):
return self._down or self._shutdown_sent return self._down or self._shutdown_sent
def setup(self): def setup(self):
self.log("setting up slave session") self.log("setting up worker session")
spec = self.gateway.spec spec = self.gateway.spec
args = self.config.args args = self.config.args
if not spec.popen or spec.chdir: if not spec.popen or spec.chdir:
@@ -238,7 +243,7 @@ class SlaveController(object):
option_dict['basetemp'] = str(basetemp.join(name)) option_dict['basetemp'] = str(basetemp.join(name))
self.config.hook.pytest_configure_node(node=self) self.config.hook.pytest_configure_node(node=self)
self.channel = self.gateway.remote_exec(xdist.remote) self.channel = self.gateway.remote_exec(xdist.remote)
self.channel.send((self.slaveinput, args, option_dict)) self.channel.send((self.workerinput, args, option_dict))
if self.putevent: if self.putevent:
self.channel.setcallback( self.channel.setcallback(
self.process_from_remote, self.process_from_remote,
@@ -298,12 +303,12 @@ class SlaveController(object):
eventname, kwargs = eventcall eventname, kwargs = eventcall
if eventname in ("collectionstart",): if eventname in ("collectionstart",):
self.log("ignoring %s(%s)" % (eventname, kwargs)) self.log("ignoring %s(%s)" % (eventname, kwargs))
elif eventname == "slaveready": elif eventname == "workerready":
self.notify_inproc(eventname, node=self, **kwargs) self.notify_inproc(eventname, node=self, **kwargs)
elif eventname == "slavefinished": elif eventname == "workerfinished":
self._down = True self._down = True
self.slaveoutput = kwargs['slaveoutput'] self.workeroutput = self.slaveoutput = kwargs['workeroutput']
self.notify_inproc("slavefinished", node=self) self.notify_inproc("workerfinished", node=self)
elif eventname in ("logstart", "logfinish"): elif eventname in ("logstart", "logfinish"):
self.notify_inproc(eventname, node=self, **kwargs) self.notify_inproc(eventname, node=self, **kwargs)
elif eventname in ( elif eventname in (