IQ.Lvbs/pytest-xdist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #268 from feuillemorte/234-master-worker-terminology

Browse Source 
#234 use master/worker terminology
This commit is contained in:
Bruno Oliveira
2018-02-06 07:27:31 -02:00
committed by GitHub
parent 3098f27731 6fb01801ac
commit 23ce4502f1
17 changed files with 249 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

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

6
README.rst
View File

@@ -77,9 +77,9 @@ 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``.
that worker and report the failure as usual. You can use the
``--max-worker-restart`` option to limit the number of workers that can
be restarted, or disable restarting altogether using ``--max-worker-restart=0``.
Running tests in a Python subprocess

1
changelog/234.trivial Normal file
View File

@@ -0,0 +1 @@
Change terminology used by ``pytest-xdist`` to *master* and *worker* in arguments and messages (for example ``--max-worker-reset``).

92
testing/acceptance_test.py
View File

@@ -192,27 +192,41 @@ class TestDistribution:
])
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):
testdir.makeconftest("""
# This hook only called on master.
def pytest_configure_node(node):
node.slaveinput['a'] = 42
node.slaveinput['b'] = 7
node.workerinput['a'] = 42
node.workerinput['b'] = 7
def pytest_configure(config):
# this attribute is only set on slaves
if hasattr(config, 'slaveinput'):
a = config.slaveinput['a']
b = config.slaveinput['b']
# this attribute is only set on workers
if hasattr(config, 'workerinput'):
a = config.workerinput['a']
b = config.workerinput['b']
r = a + b
config.slaveoutput['r'] = r
config.workeroutput['r'] = r
# This hook only called on master.
def pytest_testnodedown(node, error):
node.config.calc_result = node.slaveoutput['r']
node.config.calc_result = node.workeroutput['r']
def pytest_terminal_summary(terminalreporter):
if not hasattr(terminalreporter.config, 'slaveinput'):
if not hasattr(terminalreporter.config, 'workerinput'):
calc_result = terminalreporter.config.calc_result
terminalreporter._tw.sep('-',
'calculated result is %s' % calc_result)
@@ -232,12 +246,12 @@ class TestDistribution:
""")
testdir.makeconftest("""
def pytest_sessionfinish(session):
# on the slave
if hasattr(session.config, 'slaveoutput'):
session.config.slaveoutput['s2'] = 42
# on the worker
if hasattr(session.config, 'workeroutput'):
session.config.workeroutput['s2'] = 42
# on the master
def pytest_testnodedown(node, error):
assert node.slaveoutput['s2'] == 42
assert node.workeroutput['s2'] == 42
print ("s2call-finished")
""")
args = ["-n1", "--debug"]
@@ -411,7 +425,7 @@ def test_teardownfails_one_function(testdir):
def test_terminate_on_hangingnode(testdir):
p = testdir.makeconftest("""
def pytest_sessionfinish(session):
if session.nodeid == "my": # running on slave
if session.nodeid == "my": # running on worker
import time
time.sleep(3)
""")
@@ -429,15 +443,15 @@ def test_session_hooks(testdir):
def pytest_sessionstart(session):
sys.pytestsessionhooks = session
def pytest_sessionfinish(session):
if hasattr(session.config, 'slaveinput'):
name = "slave"
if hasattr(session.config, 'workerinput'):
name = "worker"
else:
name = "master"
f = open(name, "w")
f.write("xy")
f.close()
# let's fail on the slave
if name == "slave":
# let's fail on the worker
if name == "worker":
raise ValueError(42)
""")
p = testdir.makepyfile("""
@@ -453,14 +467,14 @@ def test_session_hooks(testdir):
assert not result.ret
d = result.parseoutcomes()
assert d['passed'] == 1
assert testdir.tmpdir.join("slave").check()
assert testdir.tmpdir.join("worker").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.
of tests collected from the workers.
"""
testdir.makepyfile(test_foo="""
import pytest
@@ -667,8 +681,8 @@ class TestNodeFailure:
""")
res = testdir.runpytest(f, '-n1')
res.stdout.fnmatch_lines([
"*Replacing crashed slave*",
"*Slave*crashed while running*",
"*Replacing crashed worker*",
"*Worker*crashed while running*",
"*1 failed*1 passed*",
])
@@ -682,8 +696,8 @@ class TestNodeFailure:
""")
res = testdir.runpytest(f, '-n2')
res.stdout.fnmatch_lines([
"*Replacing crashed slave*",
"*Slave*crashed while running*",
"*Replacing crashed worker*",
"*Worker*crashed while running*",
"*1 failed*3 passed*",
])
@@ -695,8 +709,8 @@ class TestNodeFailure:
""")
res = testdir.runpytest(f, '--dist=each', '--tx=popen')
res.stdout.fnmatch_lines([
"*Replacing crashed slave*",
"*Slave*crashed while running*",
"*Replacing crashed worker*",
"*Worker*crashed while running*",
"*1 failed*1 passed*",
])
@@ -709,12 +723,12 @@ class TestNodeFailure:
""")
res = testdir.runpytest(f, '--dist=each', '--tx=2*popen')
res.stdout.fnmatch_lines([
"*Replacing crashed slave*",
"*Slave*crashed while running*",
"*Replacing crashed worker*",
"*Worker*crashed while running*",
"*2 failed*2 passed*",
])
def test_max_slave_restart(self, testdir):
def test_max_worker_restart(self, testdir):
f = testdir.makepyfile("""
import os
def test_a(): pass
@@ -722,21 +736,21 @@ class TestNodeFailure:
def test_c(): os._exit(1)
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([
"*Replacing crashed slave*",
"*Maximum crashed slaves reached: 1*",
"*Slave*crashed while running*",
"*Slave*crashed while running*",
"*Replacing crashed worker*",
"*Maximum crashed workers reached: 1*",
"*Worker*crashed while running*",
"*Worker*crashed while running*",
"*2 failed*2 passed*",
])
def test_max_slave_restart_die(self, testdir):
def test_max_worker_restart_die(self, testdir):
f = testdir.makepyfile("""
import os
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([
"*Unexpectedly no active workers*",
"*INTERNALERROR*"
@@ -749,10 +763,10 @@ class TestNodeFailure:
def test_b(): os._exit(1)
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([
"*Slave restarting disabled*",
"*Slave*crashed while running*",
"*Worker restarting disabled*",
"*Worker*crashed while running*",
"*1 failed*2 passed*",
])

2
testing/test_dsession.py
View File

@@ -203,7 +203,7 @@ class TestLoadScheduling:
def test_different_tests_collected(self, testdir):
"""
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):

10
testing/test_newhooks.py
View File

@@ -20,10 +20,10 @@ class TestHooks:
def pytest_runtest_logreport(report):
if hasattr(report, 'node'):
if report.when == "call":
slaveid = report.node.slaveinput['slaveid']
if slaveid != report.worker_id:
workerid = report.node.workerinput['workerid']
if workerid != report.worker_id:
print("HOOK: Worker id mismatch: %s %s"
% (slaveid, report.worker_id))
% (workerid, report.worker_id))
else:
print("HOOK: %s %s"
% (report.nodeid, report.worker_id))
@@ -41,9 +41,9 @@ class TestHooks:
"""
testdir.makeconftest("""
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]
print("HOOK: %s %s" % (slaveid, ', '.join(stripped_ids)))
print("HOOK: %s %s" % (workerid, ', '.join(stripped_ids)))
""")
res = testdir.runpytest('-n2', '-s')
res.stdout.fnmatch_lines_random([

2
testing/test_plugin.py
View File

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

124
testing/test_remote.py
View File

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

24
testing/test_slavemanage.py
View File

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

88
xdist/dsession.py
View File

@@ -1,7 +1,7 @@
import py
import pytest
from xdist.slavemanage import NodeManager
from xdist.workermanage import NodeManager
from xdist.scheduler import (
EachScheduling,
LoadScheduling,
@@ -22,7 +22,7 @@ class DSession:
At the beginning of the test session this creates a NodeManager
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.
Once a node is started it will automatically start running the
@@ -46,9 +46,9 @@ class DSession:
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)
self._max_worker_restart = self.config.option.maxworkerrestart
if self._max_worker_restart is not None:
self._max_worker_restart = int(self._max_worker_restart)
try:
self.terminal = config.pluginmanager.getplugin("terminalreporter")
except KeyError:
@@ -75,7 +75,7 @@ class DSession:
"""Creates and starts the nodes.
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)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
@@ -120,7 +120,7 @@ class DSession:
return True
def loop_once(self):
"""Process one callback from one of the slaves."""
"""Process one callback from one of the workers."""
while 1:
if not self._active_nodes:
# If everything has died stop looping
@@ -133,7 +133,7 @@ class DSession:
continue
callname, kwargs = eventcall
assert callname, kwargs
method = "slave_" + callname
method = "worker_" + callname
call = getattr(self, method)
self.log("calling method", method, kwargs)
call(**kwargs)
@@ -141,44 +141,48 @@ class DSession:
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.
This adds the node to the scheduler, nodes continue with
collection without any further input.
"""
node.slaveinfo = slaveinfo
node.slaveinfo['id'] = node.gateway.id
node.slaveinfo['spec'] = node.gateway.spec
node.workerinfo = workerinfo
node.workerinfo['id'] = node.gateway.id
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)
if self.shuttingdown:
node.shutdown()
else:
self.sched.add_node(node)
def slave_slavefinished(self, node):
def worker_workerfinished(self, node):
"""Emitted when node executes its pytest_sessionfinish hook.
Removes the node from the scheduler.
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)
if node.slaveoutput['exitstatus'] == 2: # keyboard-interrupt
if node.workeroutput['exitstatus'] == 2: # keyboard-interrupt
self.shouldstop = "%s received keyboard-interrupt" % (node,)
self.slave_errordown(node, "keyboard-interrupt")
self.worker_errordown(node, "keyboard-interrupt")
return
if node in self.sched.nodes:
crashitem = self.sched.remove_node(node)
assert not crashitem, (crashitem, node)
self._active_nodes.remove(node)
def slave_errordown(self, node, error):
"""Emitted by the SlaveController when a node dies."""
def worker_errordown(self, node, error):
"""Emitted by the WorkerController when a node dies."""
self.config.hook.pytest_testnodedown(node=node, error=error)
try:
crashitem = self.sched.remove_node(node)
@@ -189,22 +193,22 @@ class DSession:
self.handle_crashitem(crashitem, node)
self._failed_nodes_count += 1
maximum_reached = (self._max_slave_restart is not None and
self._failed_nodes_count > self._max_slave_restart)
maximum_reached = (self._max_worker_restart is not None and
self._failed_nodes_count > self._max_worker_restart)
if maximum_reached:
if self._max_slave_restart == 0:
msg = 'Slave restarting disabled'
if self._max_worker_restart == 0:
msg = 'Worker restarting disabled'
else:
msg = "Maximum crashed slaves reached: %d" % \
self._max_slave_restart
msg = "Maximum crashed workers reached: %d" % \
self._max_worker_restart
self.report_line(msg)
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._active_nodes.remove(node)
def slave_collectionfinish(self, node, ids):
"""Slave has finished test collection.
def worker_collectionfinish(self, node, ids):
"""worker has finished test collection.
This adds the collection for this node to the scheduler. If
the scheduler indicates collection is finished (i.e. all
@@ -230,23 +234,23 @@ class DSession:
self.sched.__class__.__name__))
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."""
self.config.hook.pytest_runtest_logstart(
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."""
self.config.hook.pytest_runtest_logfinish(
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."""
rep.node = node
self.config.hook.pytest_runtest_logreport(report=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,
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)
def slave_collectreport(self, node, rep):
def worker_collectreport(self, node, rep):
"""Emitted when a node calls the pytest_collectreport hook."""
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."""
kwargs = dict(message=message, code=code, nodeid=nodeid, fslocation=fslocation)
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
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
"slave_*" hooks and do work soon.
"worker_*" hooks and do work soon.
"""
spec = node.gateway.spec
spec.id = None
@@ -279,9 +283,9 @@ class DSession:
self._active_nodes.add(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
# another slave).
# another worker).
if rep.longrepr not in self._failed_collection_errors:
self._failed_collection_errors[rep.longrepr] = True
self.config.hook.pytest_collectreport(report=rep)
@@ -300,15 +304,15 @@ class DSession:
for node in self.sched.nodes:
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 count no of failures and retry N times
runner = self.config.pluginmanager.getplugin("runner")
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),
(), "failed", msg, "???")
rep.node = slave
rep.node = worker
self.config.hook.pytest_runtest_logreport(report=rep)
@@ -364,7 +368,7 @@ class TerminalDistReporter:
def pytest_testnodeready(self, node):
if self.config.option.verbose > 0:
d = node.slaveinfo
d = node.workerinfo
infoline = "[%s] Python %s" % (
d['id'],
d['version'].replace('\n', ' -- '),)

16
xdist/looponfail.py
View File

@@ -69,10 +69,10 @@ class RemoteControl(object):
out = py.io.TerminalWriter()
if hasattr(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.channel = channel = self.gateway.remote_exec(
init_slave_session,
init_worker_session,
args=self.config.args,
option_dict=vars(self.config.option),
)
@@ -134,7 +134,7 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
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 sys
outchannel = channel.gateway.newchannel()
@@ -153,11 +153,11 @@ def init_slave_session(channel, args, option_dict):
from _pytest.config import Config
config = Config.fromdictargs(option_dict, list(args))
config.args = args
from xdist.looponfail import SlaveFailSession
SlaveFailSession(config, channel).main()
from xdist.looponfail import WorkerFailSession
WorkerFailSession(config, channel).main()
class SlaveFailSession:
class WorkerFailSession:
def __init__(self, config, channel):
self.config = config
self.channel = channel
@@ -194,11 +194,11 @@ class SlaveFailSession:
self.collection_failed = True
def main(self):
self.DEBUG("SLAVE: received configuration, waiting for command trails")
self.DEBUG("WORKER: received configuration, waiting for command trails")
try:
command = self.channel.receive()
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.current_command = command
self.config.hook.pytest_cmdline_main(config=self.config)

13
xdist/plugin.py
View File

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

31
xdist/remote.py
View File

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

2
xdist/scheduler/each.py
View File

@@ -1,6 +1,6 @@
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

14
xdist/scheduler/load.py
View File

@@ -3,7 +3,7 @@ from itertools import cycle
from py.log import Producer
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
@@ -113,7 +113,7 @@ class LoadScheduling:
From now on the node will be allocated chunks of tests to
execute.
Called by the ``DSession.slave_slaveready`` hook when it
Called by the ``DSession.worker_workerready`` hook when it
successfully bootstraps a new node.
"""
assert node not in self.node2pending
@@ -123,7 +123,7 @@ class LoadScheduling:
"""Add the collected test items from a node
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
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 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.check_schedule(node, duration=duration)
@@ -187,8 +187,8 @@ class LoadScheduling:
This should be called either when the node crashed or at
shutdown time. In the former case any pending items assigned
to the node will be re-scheduled. Called by the
``DSession.slave_slavefinished`` and
``DSession.slave_errordown`` hooks.
``DSession.worker_workerfinished`` and
``DSession.worker_errordown`` hooks.
Return the item which was being executing while the node
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
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.
"""
assert self.collection_is_completed

14
xdist/scheduler/loadscope.py
View File

@@ -3,7 +3,7 @@ from collections import OrderedDict
from _pytest.runner import CollectReport
from py.log import Producer
from xdist.report import report_collection_diff
from xdist.slavemanage import parse_spec_config
from xdist.workermanage import parse_spec_config
class LoadScopeScheduling:
@@ -151,7 +151,7 @@ class LoadScopeScheduling:
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.
"""
assert node not in self.assigned_work
@@ -166,8 +166,8 @@ class LoadScopeScheduling:
Called by the hooks:
- ``DSession.slave_slavefinished``.
- ``DSession.slave_errordown``.
- ``DSession.worker_workerfinished``.
- ``DSession.worker_errordown``.
Return the item being executed while the node crashed or None if the
node has no more pending items.
@@ -206,7 +206,7 @@ class LoadScopeScheduling:
Called by the hook:
- ``DSession.slave_collectionfinish``.
- ``DSession.worker_collectionfinish``.
"""
# Check that add_node() was called on the node before
@@ -239,7 +239,7 @@ class LoadScopeScheduling:
Called by the hook:
- ``DSession.slave_testreport``.
- ``DSession.worker_testreport``.
"""
nodeid = self.registered_collections[node][item_index]
scope = self._split_scope(nodeid)
@@ -336,7 +336,7 @@ class LoadScopeScheduling:
If ``.collection_is_completed`` is True, this is called by the hook:
- ``DSession.slave_collectionfinish``.
- ``DSession.worker_collectionfinish``.
"""
assert self.collection_is_completed

27
xdist/slavemanage.py → xdist/workermanage.py
View File

@@ -68,7 +68,7 @@ class NodeManager(object):
gw = self.group.makegateway(spec)
self.config.hook.pytest_xdist_newgateway(gateway=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
node.setup()
self.trace("started node %r" % node)
@@ -201,7 +201,7 @@ def make_reltoroot(roots, args):
return result
class SlaveController(object):
class WorkerController(object):
ENDMARK = -1
def __init__(self, nodemanager, gateway, config, putevent):
@@ -209,11 +209,16 @@ class SlaveController(object):
self.putevent = putevent
self.gateway = gateway
self.config = config
self.slaveinput = {'slaveid': gateway.id,
'slavecount': len(nodemanager.specs)}
self.workerinput = {'workerid': gateway.id,
'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._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:
py.log.setconsumer(self.log._keywords, None)
@@ -225,7 +230,7 @@ class SlaveController(object):
return self._down or self._shutdown_sent
def setup(self):
self.log("setting up slave session")
self.log("setting up worker session")
spec = self.gateway.spec
args = self.config.args
if not spec.popen or spec.chdir:
@@ -238,7 +243,7 @@ class SlaveController(object):
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))
self.channel.send((self.workerinput, args, option_dict))
if self.putevent:
self.channel.setcallback(
self.process_from_remote,
@@ -298,12 +303,12 @@ class SlaveController(object):
eventname, kwargs = eventcall
if eventname in ("collectionstart",):
self.log("ignoring %s(%s)" % (eventname, kwargs))
elif eventname == "slaveready":
elif eventname == "workerready":
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname == "slavefinished":
elif eventname == "workerfinished":
self._down = True
self.slaveoutput = kwargs['slaveoutput']
self.notify_inproc("slavefinished", node=self)
self.workeroutput = kwargs['workeroutput']
self.notify_inproc("workerfinished", node=self)
elif eventname in ("logstart", "logfinish"):
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname in (