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

Adopt 'src' layout and add 'testing' extras

Browse Source
This commit is contained in:
Bruno Oliveira
2019-06-06 19:35:13 -03:00
parent cdf422dec8
commit 7d41db3685
16 changed files with 6 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/xdist/__init__.py Normal file
View File

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

399
src/xdist/dsession.py Normal file
View File

@@ -0,0 +1,399 @@
import py
import pytest
from xdist.workermanage import NodeManager
from xdist.scheduler import (
EachScheduling,
LoadScheduling,
LoadScopeScheduling,
LoadFileScheduling,
)
from six.moves.queue import Empty, Queue
class Interrupted(KeyboardInterrupt):
""" signals an immediate interruption. """
class DSession(object):
"""A pytest plugin which runs a distributed test session
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 worker_*
methods.
Once a node is started it will automatically start running the
pytest mainloop with some custom hooks. This means a node
automatically starts collecting tests. Once tests are collected
it will wait for instructions.
"""
def __init__(self, config):
self.config = config
self.log = py.log.Producer("dsession")
if not config.option.debug:
py.log.setconsumer(self.log._keywords, None)
self.nodemanager = None
self.sched = None
self.shuttingdown = False
self.countfailures = 0
self.maxfail = config.getvalue("maxfail")
self.queue = Queue()
self._session = None
self._failed_collection_errors = {}
self._active_nodes = set()
self._failed_nodes_count = 0
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:
self.terminal = None
else:
self.trdist = TerminalDistReporter(config)
config.pluginmanager.register(self.trdist, "terminaldistreporter")
@property
def session_finished(self):
"""Return True if the distributed session has finished
This means all nodes have executed all test items. This is
used by pytest_runtestloop to break out of its loop.
"""
return bool(self.shuttingdown and not self._active_nodes)
def report_line(self, line):
if self.terminal and self.config.option.verbose >= 0:
self.terminal.write_line(line)
@pytest.mark.trylast
def pytest_sessionstart(self, session):
"""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 worker_workerready event.
"""
self.nodemanager = NodeManager(self.config)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
self._active_nodes.update(nodes)
self._session = session
def pytest_sessionfinish(self, session):
"""Shutdown all nodes."""
nm = getattr(self, "nodemanager", None) # if not fully initialized
if nm is not None:
nm.teardown_nodes()
self._session = None
def pytest_collection(self):
# prohibit collection of test items in master process
return True
@pytest.mark.trylast
def pytest_xdist_make_scheduler(self, config, log):
dist = config.getvalue("dist")
schedulers = {
"each": EachScheduling,
"load": LoadScheduling,
"loadscope": LoadScopeScheduling,
"loadfile": LoadFileScheduling,
}
return schedulers[dist](config, log)
def pytest_runtestloop(self):
self.sched = self.config.hook.pytest_xdist_make_scheduler(
config=self.config, log=self.log
)
assert self.sched is not None
self.shouldstop = False
while not self.session_finished:
self.loop_once()
if self.shouldstop:
self.triggershutdown()
raise Interrupted(str(self.shouldstop))
return True
def loop_once(self):
"""Process one callback from one of the workers."""
while 1:
if not self._active_nodes:
# If everything has died stop looping
self.triggershutdown()
raise RuntimeError("Unexpectedly no active workers available")
try:
eventcall = self.queue.get(timeout=2.0)
break
except Empty:
continue
callname, kwargs = eventcall
assert callname, kwargs
method = "worker_" + callname
call = getattr(self, method)
self.log("calling method", method, kwargs)
call(**kwargs)
if self.sched.tests_finished:
self.triggershutdown()
#
# callbacks for processing events from workers
#
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.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 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
workerready before shutdown was triggered.
"""
self.config.hook.pytest_testnodedown(node=node, error=None)
if node.workeroutput["exitstatus"] == 2: # keyboard-interrupt
self.shouldstop = "%s received keyboard-interrupt" % (node,)
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 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)
except KeyError:
pass
else:
if crashitem:
self.handle_crashitem(crashitem, node)
self._failed_nodes_count += 1
maximum_reached = (
self._max_worker_restart is not None
and self._failed_nodes_count > self._max_worker_restart
)
if maximum_reached:
if self._max_worker_restart == 0:
msg = "Worker restarting disabled"
else:
msg = "Maximum crashed workers reached: %d" % self._max_worker_restart
self.report_line(msg)
else:
self.report_line("Replacing crashed worker %s" % node.gateway.id)
self._clone_node(node)
self._active_nodes.remove(node)
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
initial nodes have submitted their collections), then tells the
scheduler to schedule the collected items. When initiating
scheduling the first time it logs which scheduler is in use.
"""
if self.shuttingdown:
return
self.config.hook.pytest_xdist_node_collection_finished(node=node, ids=ids)
# tell session which items were effectively collected otherwise
# the master node will finish the session with EXIT_NOTESTSCOLLECTED
self._session.testscollected = len(ids)
self.sched.add_node_collection(node, ids)
if self.terminal:
self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
if self.sched.collection_is_completed:
if self.terminal and not self.sched.has_pending:
self.trdist.ensure_show_status()
self.terminal.write_line("")
if self.config.option.verbose > 0:
self.terminal.write_line(
"scheduling tests via %s" % (self.sched.__class__.__name__)
)
self.sched.schedule()
def worker_logstart(self, node, nodeid, location):
"""Emitted when a node calls the pytest_runtest_logstart hook."""
self.config.hook.pytest_runtest_logstart(nodeid=nodeid, location=location)
def worker_logfinish(self, node, nodeid, location):
"""Emitted when a node calls the pytest_runtest_logfinish hook."""
self.config.hook.pytest_runtest_logfinish(nodeid=nodeid, location=location)
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 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
removed from the pending list in the scheduler.
"""
self.sched.mark_test_complete(node, item_index, duration)
def worker_collectreport(self, node, rep):
"""Emitted when a node calls the pytest_collectreport hook.
Because we only need the report when there's a failure/skip, as optimization
we only expect to receive failed/skipped reports from workers (#330).
"""
assert not rep.passed
self._failed_worker_collectreport(node, rep)
def worker_logwarning(self, message, code, nodeid, fslocation):
"""Emitted when a node calls the pytest_logwarning hook."""
kwargs = dict(message=message, code=code, nodeid=nodeid, fslocation=fslocation)
self.config.hook.pytest_logwarning.call_historic(kwargs=kwargs)
def worker_warning_captured(self, warning_message, when, item):
"""Emitted when a node calls the pytest_logwarning hook."""
kwargs = dict(warning_message=warning_message, when=when, item=item)
self.config.hook.pytest_warning_captured.call_historic(kwargs=kwargs)
def _clone_node(self, node):
"""Return new node based on an existing one.
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
"worker_*" hooks and do work soon.
"""
spec = node.gateway.spec
spec.id = None
self.nodemanager.group.allocate_id(spec)
node = self.nodemanager.setup_node(spec, self.queue.put)
self._active_nodes.add(node)
return node
def _failed_worker_collectreport(self, node, rep):
# Check we haven't already seen this report (from
# 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)
self._handlefailures(rep)
def _handlefailures(self, rep):
if rep.failed:
self.countfailures += 1
if self.maxfail and self.countfailures >= self.maxfail:
self.shouldstop = "stopping after %d failures" % (self.countfailures)
def triggershutdown(self):
self.log("triggering shutdown")
self.shuttingdown = True
for node in self.sched.nodes:
node.shutdown()
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 = "Worker %r crashed while running %r" % (worker.gateway.id, nodeid)
rep = runner.TestReport(
nodeid, (fspath, None, fspath), (), "failed", msg, "???"
)
rep.node = worker
self.config.hook.pytest_runtest_logreport(report=rep)
class TerminalDistReporter(object):
def __init__(self, config):
self.config = config
self.tr = config.pluginmanager.getplugin("terminalreporter")
self._status = {}
self._lastlen = 0
self._isatty = getattr(self.tr, "isatty", self.tr.hasmarkup)
def write_line(self, msg):
self.tr.write_line(msg)
def ensure_show_status(self):
if not self._isatty:
self.write_line(self.getstatus())
def setstatus(self, spec, status, show=True):
self._status[spec.id] = status
if show and self._isatty:
self.rewrite(self.getstatus())
def getstatus(self):
if self.config.option.verbose >= 0:
parts = ["%s %s" % (spec.id, self._status[spec.id]) for spec in self._specs]
return " / ".join(parts)
else:
return "bringing up nodes..."
def rewrite(self, line, newline=False):
pline = line + " " * max(self._lastlen - len(line), 0)
if newline:
self._lastlen = 0
pline += "\n"
else:
self._lastlen = len(line)
self.tr.rewrite(pline, bold=True)
def pytest_xdist_setupnodes(self, specs):
self._specs = specs
for spec in specs:
self.setstatus(spec, "I", show=False)
self.setstatus(spec, "I", show=True)
self.ensure_show_status()
def pytest_xdist_newgateway(self, gateway):
if self.config.option.verbose > 0:
rinfo = gateway._rinfo()
version = "%s.%s.%s" % rinfo.version_info[:3]
self.rewrite(
"[%s] %s Python %s cwd: %s"
% (gateway.id, rinfo.platform, version, rinfo.cwd),
newline=True,
)
self.setstatus(gateway.spec, "C")
def pytest_testnodeready(self, node):
if self.config.option.verbose > 0:
d = node.workerinfo
infoline = "[%s] Python %s" % (d["id"], d["version"].replace("\n", " -- "))
self.rewrite(infoline, newline=True)
self.setstatus(node.gateway.spec, "ok")
def pytest_testnodedown(self, node, error):
if not error:
return
self.write_line("[%s] node down: %s" % (node.gateway.id, error))
# def pytest_xdist_rsyncstart(self, source, gateways):
# targets = ",".join([gw.id for gw in gateways])
# msg = "[%s] rsyncing: %s" %(targets, source)
# self.write_line(msg)
# def pytest_xdist_rsyncfinish(self, source, gateways):
# targets = ", ".join(["[%s]" % gw.id for gw in gateways])
# self.write_line("rsyncfinish: %s -> %s" %(source, targets))

272
src/xdist/looponfail.py Normal file
View File

@@ -0,0 +1,272 @@
"""
Implement -f aka looponfailing for pytest.
NOTE that we try to avoid loading and depending on application modules
within the controlling process (the one that starts repeatedly test
processes) otherwise changes to source code can crash
the controlling process which should best never happen.
"""
from __future__ import print_function
import py
import pytest
import sys
import time
import execnet
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption(
"-f",
"--looponfail",
action="store_true",
dest="looponfail",
default=False,
help="run tests in subprocess, wait for modified files "
"and re-run failing test set until all pass.",
)
def pytest_cmdline_main(config):
if config.getoption("looponfail"):
usepdb = config.getoption("usepdb", False) # a core option
if usepdb:
raise pytest.UsageError("--pdb is incompatible with --looponfail.")
looponfail_main(config)
return 2 # looponfail only can get stop with ctrl-C anyway
def looponfail_main(config):
remotecontrol = RemoteControl(config)
rootdirs = config.getini("looponfailroots")
statrecorder = StatRecorder(rootdirs)
try:
while 1:
remotecontrol.loop_once()
if not remotecontrol.failures and remotecontrol.wasfailing:
# the last failures passed, let's immediately rerun all
continue
repr_pytest_looponfailinfo(
failreports=remotecontrol.failures, rootdirs=rootdirs
)
statrecorder.waitonchange(checkinterval=2.0)
except KeyboardInterrupt:
print()
class RemoteControl(object):
def __init__(self, config):
self.config = config
self.failures = []
def trace(self, *args):
if self.config.option.debug:
msg = " ".join([str(x) for x in args])
print("RemoteControl:", msg)
def initgateway(self):
return execnet.makegateway("popen")
def setup(self, out=None):
if out is None:
out = py.io.TerminalWriter()
if hasattr(self, "gateway"):
raise ValueError("already have gateway %r" % self.gateway)
self.trace("setting up worker session")
self.gateway = self.initgateway()
self.channel = channel = self.gateway.remote_exec(
init_worker_session,
args=self.config.args,
option_dict=vars(self.config.option),
)
remote_outchannel = channel.receive()
def write(s):
out._file.write(s)
out._file.flush()
remote_outchannel.setcallback(write)
def ensure_teardown(self):
if hasattr(self, "channel"):
if not self.channel.isclosed():
self.trace("closing", self.channel)
self.channel.close()
del self.channel
if hasattr(self, "gateway"):
self.trace("exiting", self.gateway)
self.gateway.exit()
del self.gateway
def runsession(self):
try:
self.trace("sending", self.failures)
self.channel.send(self.failures)
try:
return self.channel.receive()
except self.channel.RemoteError:
e = sys.exc_info()[1]
self.trace("ERROR", e)
raise
finally:
self.ensure_teardown()
def loop_once(self):
self.setup()
self.wasfailing = self.failures and len(self.failures)
result = self.runsession()
failures, reports, collection_failed = result
if collection_failed:
pass # "Collection failed, keeping previous failure set"
else:
uniq_failures = []
for failure in failures:
if failure not in uniq_failures:
uniq_failures.append(failure)
self.failures = uniq_failures
def repr_pytest_looponfailinfo(failreports, rootdirs):
tr = py.io.TerminalWriter()
if failreports:
tr.sep("#", "LOOPONFAILING", bold=True)
for report in failreports:
if report:
tr.line(report, red=True)
tr.sep("#", "waiting for changes", bold=True)
for rootdir in rootdirs:
tr.line("### Watching: %s" % (rootdir,), bold=True)
def init_worker_session(channel, args, option_dict):
import os
import sys
outchannel = channel.gateway.newchannel()
sys.stdout = sys.stderr = outchannel.makefile("w")
channel.send(outchannel)
# prune sys.path to not contain relative paths
newpaths = []
for p in sys.path:
if p:
if not os.path.isabs(p):
p = os.path.abspath(p)
newpaths.append(p)
sys.path[:] = newpaths
# fullwidth, hasmarkup = channel.receive()
from _pytest.config import Config
config = Config.fromdictargs(option_dict, list(args))
config.args = args
from xdist.looponfail import WorkerFailSession
WorkerFailSession(config, channel).main()
class WorkerFailSession(object):
def __init__(self, config, channel):
self.config = config
self.channel = channel
self.recorded_failures = []
self.collection_failed = False
config.pluginmanager.register(self)
config.option.looponfail = False
config.option.usepdb = False
def DEBUG(self, *args):
if self.config.option.debug:
print(" ".join(map(str, args)))
def pytest_collection(self, session):
self.session = session
self.trails = self.current_command
hook = self.session.ihook
try:
items = session.perform_collect(self.trails or None)
except pytest.UsageError:
items = session.perform_collect(None)
hook.pytest_collection_modifyitems(
session=session, config=session.config, items=items
)
hook.pytest_collection_finish(session=session)
return True
def pytest_runtest_logreport(self, report):
if report.failed:
self.recorded_failures.append(report)
def pytest_collectreport(self, report):
if report.failed:
self.recorded_failures.append(report)
self.collection_failed = True
def main(self):
self.DEBUG("WORKER: received configuration, waiting for command trails")
try:
command = self.channel.receive()
except KeyboardInterrupt:
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)
trails, failreports = [], []
for rep in self.recorded_failures:
trails.append(rep.nodeid)
loc = rep.longrepr
loc = str(getattr(loc, "reprcrash", loc))
failreports.append(loc)
self.channel.send((trails, failreports, self.collection_failed))
class StatRecorder(object):
def __init__(self, rootdirlist):
self.rootdirlist = rootdirlist
self.statcache = {}
self.check() # snapshot state
def fil(self, p):
return p.check(file=1, dotfile=0) and p.ext != ".pyc"
def rec(self, p):
return p.check(dotfile=0)
def waitonchange(self, checkinterval=1.0):
while 1:
changed = self.check()
if changed:
return
time.sleep(checkinterval)
def check(self, removepycfiles=True): # noqa, too complex
changed = False
statcache = self.statcache
newstat = {}
for rootdir in self.rootdirlist:
for path in rootdir.visit(self.fil, self.rec):
oldstat = statcache.pop(path, None)
try:
newstat[path] = curstat = path.stat()
except py.error.ENOENT:
if oldstat:
changed = True
else:
if oldstat:
if (
oldstat.mtime != curstat.mtime
or oldstat.size != curstat.size
):
changed = True
print("# MODIFIED", path)
if removepycfiles and path.ext == ".py":
pycfile = path + "c"
if pycfile.check():
pycfile.remove()
else:
changed = True
if statcache:
changed = True
self.statcache = newstat
return changed

57
src/xdist/newhooks.py Normal file
View File

@@ -0,0 +1,57 @@
"""
xdist hooks.
Additionally, pytest-xdist will also decorate a few other hooks
with the worker instance that executed the hook originally:
``pytest_runtest_logreport``: ``rep`` parameter has a ``node`` attribute.
You can use this hooks just as you would use normal pytest hooks, but some care
must be taken in plugins in case ``xdist`` is not installed. Please see:
http://pytest.org/en/latest/writing_plugins.html#optionally-using-hooks-from-3rd-party-plugins
"""
import pytest
def pytest_xdist_setupnodes(config, specs):
""" called before any remote node is set up. """
def pytest_xdist_newgateway(gateway):
""" called on new raw gateway creation. """
def pytest_xdist_rsyncstart(source, gateways):
""" called before rsyncing a directory to remote gateways takes place. """
def pytest_xdist_rsyncfinish(source, gateways):
""" called after rsyncing a directory to remote gateways takes place. """
@pytest.mark.firstresult
def pytest_xdist_getremotemodule():
""" called when creating remote node"""
def pytest_configure_node(node):
""" configure node information before it gets instantiated. """
def pytest_testnodeready(node):
""" Test Node is ready to operate. """
def pytest_testnodedown(node, error):
""" Test Node is down. """
def pytest_xdist_node_collection_finished(node, ids):
"""called by the master node when a node finishes collecting.
"""
@pytest.mark.firstresult
def pytest_xdist_make_scheduler(config, log):
""" return a node scheduler implementation """

219
src/xdist/plugin.py Normal file
View File

@@ -0,0 +1,219 @@
import os
import py
import pytest
def auto_detect_cpus():
try:
from os import sched_getaffinity
except ImportError:
if os.environ.get("TRAVIS") == "true":
# workaround https://bitbucket.org/pypy/pypy/issues/2375
return 2
try:
from os import cpu_count
except ImportError:
from multiprocessing import cpu_count
else:
def cpu_count():
return len(sched_getaffinity(0))
try:
n = cpu_count()
except NotImplementedError:
return 1
return n if n else 1
class AutoInt(int):
"""Mark value as auto-detected."""
def parse_numprocesses(s):
if s == "auto":
return AutoInt(auto_detect_cpus())
elif s is not None:
return int(s)
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption(
"-n",
"--numprocesses",
dest="numprocesses",
metavar="numprocesses",
action="store",
type=parse_numprocesses,
help="shortcut for '--dist=load --tx=NUM*popen', "
"you can use 'auto' here for auto detection CPUs number on "
"host system and it will be 0 when used with --pdb",
)
group.addoption(
"--maxprocesses",
dest="maxprocesses",
metavar="maxprocesses",
action="store",
type=int,
help="limit the maximum number of workers to process the tests when using --numprocesses=auto",
)
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"],
dest="dist",
default="no",
help=(
"set mode for distributing tests to exec environments.\n\n"
"each: send each test to all available environments.\n\n"
"load: load balance by sending any pending test to any"
" available environment.\n\n"
"loadscope: load balance by sending pending groups of tests in"
" the same scope to any available environment.\n\n"
"loadfile: load balance by sending test grouped by file"
" to any available environment.\n\n"
"(default) no: run tests inprocess, don't distribute."
),
)
group.addoption(
"--tx",
dest="tx",
action="append",
default=[],
metavar="xspec",
help=(
"add a test execution environment. some examples: "
"--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
"--tx ssh=user@codespeak.net//chdir=testcache"
),
)
group._addoption(
"-d",
action="store_true",
dest="distload",
default=False,
help="load-balance tests. shortcut for '--dist=load'",
)
group.addoption(
"--rsyncdir",
action="append",
default=[],
metavar="DIR",
help="add directory for rsyncing to remote tx nodes.",
)
group.addoption(
"--rsyncignore",
action="append",
default=[],
metavar="GLOB",
help="add expression for ignores when rsyncing to remote tx nodes.",
)
group.addoption(
"--boxed",
action="store_true",
help="backward compatibility alias for pytest-forked --forked",
)
parser.addini(
"rsyncdirs",
"list of (relative) paths to be rsynced for remote distributed testing.",
type="pathlist",
)
parser.addini(
"rsyncignore",
"list of (relative) glob-style paths to be ignored for rsyncing.",
type="pathlist",
)
parser.addini(
"looponfailroots",
type="pathlist",
help="directories to check for changes",
default=[py.path.local()],
)
# -------------------------------------------------------------------------
# distributed testing hooks
# -------------------------------------------------------------------------
def pytest_addhooks(pluginmanager):
from xdist import newhooks
# avoid warnings with pytest-2.8
method = getattr(pluginmanager, "add_hookspecs", None)
if method is None:
method = pluginmanager.addhooks
method(newhooks)
# -------------------------------------------------------------------------
# distributed testing initialization
# -------------------------------------------------------------------------
@pytest.mark.trylast
def pytest_configure(config):
if config.getoption("dist") != "no" and not config.getvalue("collectonly"):
from xdist.dsession import DSession
session = DSession(config)
config.pluginmanager.register(session, "dsession")
tr = config.pluginmanager.getplugin("terminalreporter")
tr.showfspath = False
if config.getoption("boxed"):
config.option.forked = True
@pytest.mark.tryfirst
def pytest_cmdline_main(config):
usepdb = config.getoption("usepdb", False) # a core option
if isinstance(config.option.numprocesses, AutoInt):
config.option.numprocesses = 0 if usepdb else int(config.option.numprocesses)
if config.option.numprocesses:
if config.option.dist == "no":
config.option.dist = "load"
numprocesses = config.option.numprocesses
if config.option.maxprocesses:
numprocesses = min(numprocesses, config.option.maxprocesses)
config.option.tx = ["popen"] * numprocesses
if config.option.distload:
config.option.dist = "load"
val = config.getvalue
if not val("collectonly"):
if val("dist") != "no":
if usepdb:
raise pytest.UsageError(
"--pdb is incompatible with distributing tests; try using -n0 or -nauto."
) # noqa: E501
# -------------------------------------------------------------------------
# fixtures
# -------------------------------------------------------------------------
@pytest.fixture(scope="session")
def worker_id(request):
"""Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
if running on the master node.
"""
if hasattr(request.config, "workerinput"):
return request.config.workerinput["workerid"]
else:
return "master"

249
src/xdist/remote.py Normal file
View File

@@ -0,0 +1,249 @@
"""
This module is executed in remote subprocesses and helps to
control a remote testing session and relay back information.
It assumes that 'py' is importable and does not have dependencies
on the rest of the xdist code. This means that the xdist-plugin
needs not to be installed in remote environments.
"""
import sys
import os
import time
import py
import _pytest.hookspec
import pytest
from execnet.gateway_base import dumps, DumpError
class WorkerInteractor(object):
def __init__(self, config, channel):
self.config = config
self.workerid = config.workerinput.get("workerid", "?")
self.log = py.log.Producer("worker-%s" % self.workerid)
if not config.option.debug:
py.log.setconsumer(self.log._keywords, None)
self.channel = channel
config.pluginmanager.register(self)
def sendevent(self, name, **kwargs):
self.log("sending", name, kwargs)
self.channel.send((name, kwargs))
def pytest_internalerror(self, excrepr):
for line in str(excrepr).split("\n"):
self.log("IERROR>", line)
def pytest_sessionstart(self, session):
self.session = session
workerinfo = getinfodict()
self.sendevent("workerready", workerinfo=workerinfo)
@pytest.hookimpl(hookwrapper=True)
def pytest_sessionfinish(self, exitstatus):
self.config.workeroutput["exitstatus"] = exitstatus
yield
self.sendevent("workerfinished", workeroutput=self.config.workeroutput)
def pytest_collection(self, session):
self.sendevent("collectionstart")
def pytest_runtestloop(self, session):
self.log("entering main loop")
torun = []
while 1:
try:
name, kwargs = self.channel.receive()
except EOFError:
return True
self.log("received command", name, kwargs)
if name == "runtests":
torun.extend(kwargs["indices"])
elif name == "runtests_all":
torun.extend(range(len(session.items)))
self.log("items to run:", torun)
# only run if we have an item and a next item
while len(torun) >= 2:
self.run_one_test(torun)
if name == "shutdown":
if torun:
self.run_one_test(torun)
break
return True
def run_one_test(self, torun):
items = self.session.items
self.item_index = torun.pop(0)
item = items[self.item_index]
if torun:
nextitem = items[torun[0]]
else:
nextitem = None
start = time.time()
self.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
duration = time.time() - start
self.sendevent(
"runtest_protocol_complete", item_index=self.item_index, duration=duration
)
def pytest_collection_finish(self, session):
self.sendevent(
"collectionfinish",
topdir=str(session.fspath),
ids=[item.nodeid for item in session.items],
)
def pytest_runtest_logstart(self, nodeid, location):
self.sendevent("logstart", nodeid=nodeid, location=location)
# the pytest_runtest_logfinish hook was introduced in pytest 3.4
if hasattr(_pytest.hookspec, "pytest_runtest_logfinish"):
def pytest_runtest_logfinish(self, nodeid, location):
self.sendevent("logfinish", nodeid=nodeid, location=location)
def pytest_runtest_logreport(self, report):
data = self.config.hook.pytest_report_to_serializable(
config=self.config, report=report
)
data["item_index"] = self.item_index
data["worker_id"] = self.workerid
assert self.session.items[self.item_index].nodeid == report.nodeid
self.sendevent("testreport", data=data)
def pytest_collectreport(self, report):
# send only reports that have not passed to master as optimization (#330)
if not report.passed:
data = self.config.hook.pytest_report_to_serializable(
config=self.config, report=report
)
self.sendevent("collectreport", data=data)
# the pytest_logwarning hook was deprecated since pytest 4.0
if hasattr(
_pytest.hookspec, "pytest_logwarning"
) and not _pytest.hookspec.pytest_logwarning.pytest_spec.get("warn_on_impl"):
def pytest_logwarning(self, message, code, nodeid, fslocation):
self.sendevent(
"logwarning",
message=message,
code=code,
nodeid=nodeid,
fslocation=str(fslocation),
)
# the pytest_warning_captured hook was introduced in pytest 3.8
if hasattr(_pytest.hookspec, "pytest_warning_captured"):
def pytest_warning_captured(self, warning_message, when, item):
self.sendevent(
"warning_captured",
warning_message_data=serialize_warning_message(warning_message),
when=when,
# item cannot be serialized and will always be None when used with xdist
item=None,
)
def serialize_warning_message(warning_message):
if isinstance(warning_message.message, Warning):
message_module = type(warning_message.message).__module__
message_class_name = type(warning_message.message).__name__
message_str = str(warning_message.message)
# check now if we can serialize the warning arguments (#349)
# if not, we will just use the exception message on the master node
try:
dumps(warning_message.message.args)
except DumpError:
message_args = None
else:
message_args = warning_message.message.args
else:
message_str = warning_message.message
message_module = None
message_class_name = None
message_args = None
if warning_message.category:
category_module = warning_message.category.__module__
category_class_name = warning_message.category.__name__
else:
category_module = None
category_class_name = None
result = {
"message_str": message_str,
"message_module": message_module,
"message_class_name": message_class_name,
"message_args": message_args,
"category_module": category_module,
"category_class_name": category_class_name,
}
# access private _WARNING_DETAILS because the attributes vary between Python versions
for attr_name in warning_message._WARNING_DETAILS:
if attr_name in ("message", "category"):
continue
attr = getattr(warning_message, attr_name)
# Check if we can serialize the warning detail, marking `None` otherwise
# Note that we need to define the attr (even as `None`) to allow deserializing
try:
dumps(attr)
except DumpError:
result[attr_name] = repr(attr)
else:
result[attr_name] = attr
return result
def getinfodict():
import platform
return dict(
version=sys.version,
version_info=tuple(sys.version_info),
sysplatform=sys.platform,
platform=platform.platform(),
executable=sys.executable,
cwd=os.getcwd(),
)
def remote_initconfig(option_dict, args):
from _pytest.config import Config
option_dict["plugins"].append("no:terminal")
config = Config.fromdictargs(option_dict, args)
config.option.looponfail = False
config.option.usepdb = False
config.option.dist = "no"
config.option.distload = False
config.option.numprocesses = None
config.option.maxprocesses = None
config.args = args
return config
if __name__ == "__channelexec__":
channel = channel # noqa
workerinput, args, option_dict, change_sys_path = channel.receive()
if change_sys_path:
importpath = os.getcwd()
sys.path.insert(0, importpath)
os.environ["PYTHONPATH"] = (
importpath + os.pathsep + os.environ.get("PYTHONPATH", "")
)
os.environ["PYTEST_XDIST_WORKER"] = workerinput["workerid"]
os.environ["PYTEST_XDIST_WORKER_COUNT"] = str(workerinput["workercount"])
config = remote_initconfig(option_dict, args)
config._parser.prog = os.path.basename(workerinput["mainargv"][0])
config.workerinput = workerinput
config.workeroutput = {}
# TODO: deprecated name, backward compatibility only. Remove it in future
config.slaveinput = config.workerinput
config.slaveoutput = config.workeroutput
interactor = WorkerInteractor(config, channel)
config.hook.pytest_cmdline_main(config=config)

21
src/xdist/report.py Normal file
View File

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

4
src/xdist/scheduler/__init__.py Normal file
View File

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

132
src/xdist/scheduler/each.py Normal file
View File

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

286
src/xdist/scheduler/load.py Normal file
View File

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

52
src/xdist/scheduler/loadfile.py Normal file
View File

@@ -0,0 +1,52 @@
from .loadscope import LoadScopeScheduling
from py.log import Producer
class LoadFileScheduling(LoadScopeScheduling):
"""Implement load scheduling across nodes, but grouping test test file.
This distributes the tests collected across all nodes so each test is run
just once. All nodes collect and submit the list of tests and when all
collections are received it is verified they are identical collections.
Then the collection gets divided up in work units, grouped by test file,
and those work units get submitted to nodes. Whenever a node finishes an
item, it calls ``.mark_test_complete()`` which will trigger the scheduler
to assign more work units if the number of pending tests for the node falls
below a low-watermark.
When created, ``numnodes`` defines how many nodes are expected to submit a
collection. This is used to know when all nodes have finished collection.
This class behaves very much like LoadScopeScheduling, but with a file-level scope.
"""
def __init__(self, config, log=None):
super(LoadFileScheduling, self).__init__(config, log)
if log is None:
self.log = Producer("loadfilesched")
else:
self.log = log.loadfilesched
def _split_scope(self, nodeid):
"""Determine the scope (grouping) of a nodeid.
There are usually 3 cases for a nodeid::
example/loadsuite/test/test_beta.py::test_beta0
example/loadsuite/test/test_delta.py::Delta1::test_delta0
example/loadsuite/epsilon/__init__.py::epsilon.epsilon
#. Function in a test module.
#. Method of a class in a test module.
#. Doctest in a function in a package.
This function will group tests with the scope determined by splitting
the first ``::`` from the left. That is, test will be grouped in a
single work unit when they reside in the same file.
In the above example, scopes will be::
example/loadsuite/test/test_beta.py
example/loadsuite/test/test_delta.py
example/loadsuite/epsilon/__init__.py
"""
return nodeid.split("::", 1)[0]

409
src/xdist/scheduler/loadscope.py Normal file
View File

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

409
src/xdist/workermanage.py Normal file
View File

@@ -0,0 +1,409 @@
from __future__ import print_function
import fnmatch
import os
import re
import sys
import py
import pytest
import execnet
import xdist.remote
def parse_spec_config(config):
xspeclist = []
for xspec in config.getvalue("tx"):
i = xspec.find("*")
try:
num = int(xspec[:i])
except ValueError:
xspeclist.append(xspec)
else:
xspeclist.extend([xspec[i + 1 :]] * num)
if not xspeclist:
raise pytest.UsageError(
"MISSING test execution (tx) nodes: please specify --tx"
)
return xspeclist
class NodeManager(object):
EXIT_TIMEOUT = 10
DEFAULT_IGNORES = [".*", "*.pyc", "*.pyo", "*~"]
def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
self.config = config
self.trace = self.config.trace.get("nodemanager")
self.group = execnet.Group()
if specs is None:
specs = self._getxspecs()
self.specs = []
for spec in specs:
if not isinstance(spec, execnet.XSpec):
spec = execnet.XSpec(spec)
if not spec.chdir and not spec.popen:
spec.chdir = defaultchdir
self.group.allocate_id(spec)
self.specs.append(spec)
self.roots = self._getrsyncdirs()
self.rsyncoptions = self._getrsyncoptions()
self._rsynced_specs = set()
def rsync_roots(self, gateway):
"""Rsync the set of roots to the node's gateway cwd."""
if self.roots:
for root in self.roots:
self.rsync(gateway, root, **self.rsyncoptions)
def setup_nodes(self, putevent):
self.config.hook.pytest_xdist_setupnodes(config=self.config, specs=self.specs)
self.trace("setting up nodes")
nodes = []
for spec in self.specs:
nodes.append(self.setup_node(spec, putevent))
return nodes
def setup_node(self, spec, putevent):
gw = self.group.makegateway(spec)
self.config.hook.pytest_xdist_newgateway(gateway=gw)
self.rsync_roots(gw)
node = WorkerController(self, gw, self.config, putevent)
gw.node = node # keep the node alive
node.setup()
self.trace("started node %r" % node)
return node
def teardown_nodes(self):
self.group.terminate(self.EXIT_TIMEOUT)
def _getxspecs(self):
return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
def _getrsyncdirs(self):
for spec in self.specs:
if not spec.popen or spec.chdir:
break
else:
return []
import pytest
import _pytest
pytestpath = pytest.__file__.rstrip("co")
pytestdir = py.path.local(_pytest.__file__).dirpath()
config = self.config
candidates = [py._pydir, pytestpath, pytestdir]
candidates += config.option.rsyncdir
rsyncroots = config.getini("rsyncdirs")
if rsyncroots:
candidates.extend(rsyncroots)
roots = []
for root in candidates:
root = py.path.local(root).realpath()
if not root.check():
raise pytest.UsageError("rsyncdir doesn't exist: %r" % (root,))
if root not in roots:
roots.append(root)
return roots
def _getrsyncoptions(self):
"""Get options to be passed for rsync."""
ignores = list(self.DEFAULT_IGNORES)
ignores += self.config.option.rsyncignore
ignores += self.config.getini("rsyncignore")
return {"ignores": ignores, "verbose": self.config.option.verbose}
def rsync(self, gateway, source, notify=None, verbose=False, ignores=None):
"""Perform rsync to remote hosts for node."""
# XXX This changes the calling behaviour of
# pytest_xdist_rsyncstart and pytest_xdist_rsyncfinish to
# be called once per rsync target.
rsync = HostRSync(source, verbose=verbose, ignores=ignores)
spec = gateway.spec
if spec.popen and not spec.chdir:
# XXX This assumes that sources are python-packages
# and that adding the basedir does not hurt.
gateway.remote_exec(
"""
import sys ; sys.path.insert(0, %r)
"""
% os.path.dirname(str(source))
).waitclose()
return
if (spec, source) in self._rsynced_specs:
return
def finished():
if notify:
notify("rsyncrootready", spec, source)
rsync.add_target_host(gateway, finished=finished)
self._rsynced_specs.add((spec, source))
self.config.hook.pytest_xdist_rsyncstart(source=source, gateways=[gateway])
rsync.send()
self.config.hook.pytest_xdist_rsyncfinish(source=source, gateways=[gateway])
class HostRSync(execnet.RSync):
""" RSyncer that filters out common files
"""
def __init__(self, sourcedir, *args, **kwargs):
self._synced = {}
self._ignores = []
ignores = kwargs.pop("ignores", None) or []
for x in ignores:
x = getattr(x, "strpath", x)
self._ignores.append(re.compile(fnmatch.translate(x)))
super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
def filter(self, path):
path = py.path.local(path)
for cre in self._ignores:
if cre.match(path.basename) or cre.match(path.strpath):
return False
else:
return True
def add_target_host(self, gateway, finished=None):
remotepath = os.path.basename(self._sourcedir)
super(HostRSync, self).add_target(
gateway, remotepath, finishedcallback=finished, delete=True
)
def _report_send_file(self, gateway, modified_rel_path):
if self._verbose:
path = os.path.basename(self._sourcedir) + "/" + modified_rel_path
remotepath = gateway.spec.chdir
print("%s:%s <= %s" % (gateway.spec, remotepath, path))
def make_reltoroot(roots, args):
# XXX introduce/use public API for splitting pytest args
splitcode = "::"
result = []
for arg in args:
parts = arg.split(splitcode)
fspath = py.path.local(parts[0])
for root in roots:
x = fspath.relto(root)
if x or fspath == root:
parts[0] = root.basename + "/" + x
break
else:
raise ValueError("arg %s not relative to an rsync root" % (arg,))
result.append(splitcode.join(parts))
return result
class WorkerController(object):
ENDMARK = -1
class RemoteHook:
@pytest.mark.trylast
def pytest_xdist_getremotemodule(self):
return xdist.remote
def __init__(self, nodemanager, gateway, config, putevent):
config.pluginmanager.register(self.RemoteHook())
self.nodemanager = nodemanager
self.putevent = putevent
self.gateway = gateway
self.config = config
self.workerinput = {
"workerid": gateway.id,
"workercount": len(nodemanager.specs),
"slaveid": gateway.id,
"slavecount": len(nodemanager.specs),
"mainargv": sys.argv,
}
# TODO: deprecated name, backward compatibility only. Remove it in future
self.slaveinput = self.workerinput
self._down = False
self._shutdown_sent = False
self.log = py.log.Producer("workerctl-%s" % gateway.id)
if not self.config.option.debug:
py.log.setconsumer(self.log._keywords, None)
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.gateway.id)
@property
def shutting_down(self):
return self._down or self._shutdown_sent
def setup(self):
self.log("setting up worker session")
spec = self.gateway.spec
args = self.config.args
if not spec.popen or spec.chdir:
args = make_reltoroot(self.nodemanager.roots, args)
option_dict = vars(self.config.option)
if spec.popen:
name = "popen-%s" % self.gateway.id
if hasattr(self.config, "_tmpdirhandler"):
basetemp = self.config._tmpdirhandler.getbasetemp()
option_dict["basetemp"] = str(basetemp.join(name))
self.config.hook.pytest_configure_node(node=self)
remote_module = self.config.hook.pytest_xdist_getremotemodule()
self.channel = self.gateway.remote_exec(remote_module)
# change sys.path only for remote workers
change_sys_path = not self.gateway.spec.popen
self.channel.send((self.workerinput, args, option_dict, change_sys_path))
if self.putevent:
self.channel.setcallback(self.process_from_remote, endmarker=self.ENDMARK)
def ensure_teardown(self):
if hasattr(self, "channel"):
if not self.channel.isclosed():
self.log("closing", self.channel)
self.channel.close()
# del self.channel
if hasattr(self, "gateway"):
self.log("exiting", self.gateway)
self.gateway.exit()
# del self.gateway
def send_runtest_some(self, indices):
self.sendcommand("runtests", indices=indices)
def send_runtest_all(self):
self.sendcommand("runtests_all")
def shutdown(self):
if not self._down:
try:
self.sendcommand("shutdown")
except (IOError, OSError):
pass
self._shutdown_sent = True
def sendcommand(self, name, **kwargs):
""" send a named parametrized command to the other side. """
self.log("sending command %s(**%s)" % (name, kwargs))
self.channel.send((name, kwargs))
def notify_inproc(self, eventname, **kwargs):
self.log("queuing %s(**%s)" % (eventname, kwargs))
self.putevent((eventname, kwargs))
def process_from_remote(self, eventcall): # noqa too complex
""" this gets called for each object we receive from
the other side and if the channel closes.
Note that channel callbacks run in the receiver
thread of execnet gateways - we need to
avoid raising exceptions or doing heavy work.
"""
try:
if eventcall == self.ENDMARK:
err = self.channel._getremoteerror()
if not self._down:
if not err or isinstance(err, EOFError):
err = "Not properly terminated" # lost connection?
self.notify_inproc("errordown", node=self, error=err)
self._down = True
return
eventname, kwargs = eventcall
if eventname in ("collectionstart",):
self.log("ignoring %s(%s)" % (eventname, kwargs))
elif eventname == "workerready":
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname == "workerfinished":
self._down = True
self.workeroutput = self.slaveoutput = kwargs["workeroutput"]
self.notify_inproc("workerfinished", node=self)
elif eventname in ("logstart", "logfinish"):
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname in ("testreport", "collectreport", "teardownreport"):
item_index = kwargs.pop("item_index", None)
rep = self.config.hook.pytest_report_from_serializable(
config=self.config, data=kwargs["data"]
)
if item_index is not None:
rep.item_index = item_index
self.notify_inproc(eventname, node=self, rep=rep)
elif eventname == "collectionfinish":
self.notify_inproc(eventname, node=self, ids=kwargs["ids"])
elif eventname == "runtest_protocol_complete":
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname == "logwarning":
self.notify_inproc(
eventname,
message=kwargs["message"],
code=kwargs["code"],
nodeid=kwargs["nodeid"],
fslocation=kwargs["nodeid"],
)
elif eventname == "warning_captured":
warning_message = unserialize_warning_message(
kwargs["warning_message_data"]
)
self.notify_inproc(
eventname,
warning_message=warning_message,
when=kwargs["when"],
item=kwargs["item"],
)
else:
raise ValueError("unknown event: %s" % (eventname,))
except KeyboardInterrupt:
# should not land in receiver-thread
raise
except: # noqa
from _pytest._code import ExceptionInfo
# ExceptionInfo API changed in pytest 4.1
if hasattr(ExceptionInfo, "from_current"):
excinfo = ExceptionInfo.from_current()
else:
excinfo = ExceptionInfo()
print("!" * 20, excinfo)
self.config.notify_exception(excinfo)
self.shutdown()
self.notify_inproc("errordown", node=self, error=excinfo)
def unserialize_warning_message(data):
import warnings
import importlib
if data["message_module"]:
mod = importlib.import_module(data["message_module"])
cls = getattr(mod, data["message_class_name"])
message = None
if data["message_args"] is not None:
try:
message = cls(*data["message_args"])
except TypeError:
pass
if message is None:
# could not recreate the original warning instance;
# create a generic Warning instance with the original
# message at least
message_text = "{mod}.{cls}: {msg}".format(
mod=data["message_module"],
cls=data["message_class_name"],
msg=data["message_str"],
)
message = Warning(message_text)
else:
message = data["message_str"]
if data["category_module"]:
mod = importlib.import_module(data["category_module"])
category = getattr(mod, data["category_class_name"])
else:
category = None
kwargs = {"message": message, "category": category}
# access private _WARNING_DETAILS because the attributes vary between Python versions
for attr_name in warnings.WarningMessage._WARNING_DETAILS:
if attr_name in ("message", "category"):
continue
kwargs[attr_name] = data[attr_name]
return warnings.WarningMessage(**kwargs)