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

flake8 cleanup

Browse Source
This commit is contained in:
Ronny Pfannschmidt
2015-09-01 22:37:51 +02:00
parent fa1ec6742a
commit 0f5ef95be8
18 changed files with 298 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
xdist/boxed.py
View File

@@ -4,9 +4,10 @@ import py
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group.addoption('--boxed',
action="store_true", dest="boxed", default=False,
help="box each test run in a separate process (unix)")
group.addoption(
'--boxed',
action="store_true", dest="boxed", default=False,
help="box each test run in a separate process (unix)")
def pytest_runtest_protocol(item):
@@ -16,6 +17,7 @@ def pytest_runtest_protocol(item):
item.ihook.pytest_runtest_logreport(report=rep)
return True
def forked_run_report(item):
# for now, we run setup/teardown in the subprocess
# XXX optionally allow sharing of setup/teardown
@@ -24,6 +26,7 @@ def forked_run_report(item):
import marshal
from xdist.remote import serialize_report
from xdist.slavemanage import unserialize_report
def runforked():
try:
reports = runtestprotocol(item, log=False)
@@ -38,9 +41,10 @@ def forked_run_report(item):
return [unserialize_report("testreport", x) for x in report_dumps]
else:
if result.exitstatus == EXITSTATUS_TESTEXIT:
py.test.exit("forked test item %s raised Exit" %(item,))
py.test.exit("forked test item %s raised Exit" % (item,))
return [report_process_crash(item, result)]
def report_process_crash(item, result):
path, lineno = item._getfslineno()
info = ("%s:%s: running the test CRASHED with signal %d" %

25
xdist/dsession.py
View File

@@ -257,8 +257,9 @@ class LoadScheduling:
assert node in self.node2pending
if self.collection_is_completed:
# A new node has been added later, perhaps an original one died.
assert self.collection # .init_distribute() should have
# been called by now
# .init_distribute() should have
# been called by now
assert self.collection
if collection != self.collection:
other_node = next(iter(self.node2collection.keys()))
msg = report_collection_diff(self.collection,
@@ -398,8 +399,9 @@ class LoadScheduling:
same_collection = False
self.log(msg)
if self.config is not None:
rep = CollectReport(node.gateway.id, 'failed', longrepr=msg,
result=[])
rep = CollectReport(
node.gateway.id, 'failed',
longrepr=msg, result=[])
self.config.hook.pytest_collectreport(report=rep)
return same_collection
@@ -641,7 +643,7 @@ class DSession:
"""
if rep.when == "call" or (rep.when == "setup" and not rep.passed):
self.sched.remove_item(node, rep.item_index, rep.duration)
#self.report_line("testreport %s: %s" %(rep.id, rep.status))
# self.report_line("testreport %s: %s" %(rep.id, rep.status))
rep.node = node
self.config.hook.pytest_runtest_logreport(report=rep)
self._handlefailures(rep)
@@ -719,8 +721,8 @@ class TerminalDistReporter:
self.rewrite(self.getstatus())
def getstatus(self):
parts = ["%s %s" %(spec.id, self._status[spec.id])
for spec in self._specs]
parts = ["%s %s" % (spec.id, self._status[spec.id])
for spec in self._specs]
return " / ".join(parts)
def rewrite(self, line, newline=False):
@@ -751,7 +753,7 @@ class TerminalDistReporter:
def pytest_testnodeready(self, node):
if self.config.option.verbose > 0:
d = node.slaveinfo
infoline = "[%s] Python %s" %(
infoline = "[%s] Python %s" % (
d['id'],
d['version'].replace('\n', ' -- '),)
self.rewrite(infoline, newline=True)
@@ -760,13 +762,12 @@ class TerminalDistReporter:
def pytest_testnodedown(self, node, error):
if not error:
return
self.write_line("[%s] node down: %s" %(node.gateway.id, error))
self.write_line("[%s] node down: %s" % (node.gateway.id, error))
#def pytest_xdist_rsyncstart(self, source, gateways):
# def pytest_xdist_rsyncstart(self, source, gateways):
# targets = ",".join([gw.id for gw in gateways])
# msg = "[%s] rsyncing: %s" %(targets, source)
# self.write_line(msg)
#def pytest_xdist_rsyncfinish(self, source, gateways):
# def pytest_xdist_rsyncfinish(self, source, gateways):
# targets = ", ".join(["[%s]" % gw.id for gw in gateways])
# self.write_line("rsyncfinish: %s -> %s" %(source, targets))

53
xdist/looponfail.py
View File

@@ -7,23 +7,26 @@
the controlling process which should best never happen.
"""
import py, pytest
import py
import pytest
import sys
import execnet
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption('-f', '--looponfail',
action="store_true", dest="looponfail", default=False,
help="run tests in subprocess, wait for modified files "
"and re-run failing test set until all pass.")
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"):
looponfail_main(config)
return 2 # looponfail only can get stop with ctrl-C anyway
return 2 # looponfail only can get stop with ctrl-C anyway
def looponfail_main(config):
@@ -34,7 +37,8 @@ def looponfail_main(config):
while 1:
remotecontrol.loop_once()
if not remotecontrol.failures and remotecontrol.wasfailing:
continue # the last failures passed, let's immediately rerun all
# the last failures passed, let's immediately rerun all
continue
repr_pytest_looponfailinfo(
failreports=remotecontrol.failures,
rootdirs=rootdirs)
@@ -42,6 +46,7 @@ def looponfail_main(config):
except KeyboardInterrupt:
print()
class RemoteControl(object):
def __init__(self, config):
self.config = config
@@ -62,11 +67,13 @@ class RemoteControl(object):
raise ValueError("already have gateway %r" % self.gateway)
self.trace("setting up slave session")
self.gateway = self.initgateway()
self.channel = channel = self.gateway.remote_exec(init_slave_session,
self.channel = channel = self.gateway.remote_exec(
init_slave_session,
args=self.config.args,
option_dict=vars(self.config.option),
)
remote_outchannel = channel.receive()
def write(s):
out._file.write(s)
out._file.flush()
@@ -102,7 +109,7 @@ class RemoteControl(object):
result = self.runsession()
failures, reports, collection_failed = result
if collection_failed:
pass # "Collection failed, keeping previous failure set"
pass # "Collection failed, keeping previous failure set"
else:
uniq_failures = []
for failure in failures:
@@ -110,6 +117,7 @@ class RemoteControl(object):
uniq_failures.append(failure)
self.failures = uniq_failures
def repr_pytest_looponfailinfo(failreports, rootdirs):
tr = py.io.TerminalWriter()
if failreports:
@@ -119,11 +127,12 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
tr.line(report, red=True)
tr.sep("#", "waiting for changes", bold=True)
for rootdir in rootdirs:
tr.line("### Watching: %s" %(rootdir,), bold=True)
tr.line("### Watching: %s" % (rootdir,), bold=True)
def init_slave_session(channel, args, option_dict):
import os, sys
import os
import sys
outchannel = channel.gateway.newchannel()
sys.stdout = sys.stderr = outchannel.makefile('w')
channel.send(outchannel)
@@ -136,13 +145,14 @@ def init_slave_session(channel, args, option_dict):
newpaths.append(p)
sys.path[:] = newpaths
#fullwidth, hasmarkup = channel.receive()
# fullwidth, hasmarkup = channel.receive()
from _pytest.config import Config
config = Config.fromdictargs(option_dict, list(args))
config.args = args
from xdist.looponfail import SlaveFailSession
SlaveFailSession(config, channel).main()
class SlaveFailSession:
def __init__(self, config, channel):
self.config = config
@@ -165,7 +175,8 @@ class SlaveFailSession:
items = session.perform_collect(self.trails or None)
except pytest.UsageError:
items = session.perform_collect(None)
hook.pytest_collection_modifyitems(session=session, config=session.config, items=items)
hook.pytest_collection_modifyitems(
session=session, config=session.config, items=items)
hook.pytest_collection_finish(session=session)
return True
@@ -183,7 +194,7 @@ class SlaveFailSession:
try:
command = self.channel.receive()
except KeyboardInterrupt:
return # in the slave we can't do much about this
return # in the slave we can't do much about this
self.DEBUG("received", command)
self.current_command = command
self.config.hook.pytest_cmdline_main(config=self.config)
@@ -195,14 +206,16 @@ class SlaveFailSession:
failreports.append(loc)
self.channel.send((trails, failreports, self.collection_failed))
class StatRecorder:
def __init__(self, rootdirlist):
self.rootdirlist = rootdirlist
self.statcache = {}
self.check() # snapshot state
self.check() # snapshot state
def fil(self, p):
return p.check(file=1, dotfile=0) and p.ext != ".pyc"
def rec(self, p):
return p.check(dotfile=0)
@@ -213,7 +226,7 @@ class StatRecorder:
return
py.std.time.sleep(checkinterval)
def check(self, removepycfiles=True):
def check(self, removepycfiles=True): # noqa, too complex
changed = False
statcache = self.statcache
newstat = {}
@@ -227,8 +240,8 @@ class StatRecorder:
changed = True
else:
if oldstat:
if oldstat.mtime != curstat.mtime or \
oldstat.size != curstat.size:
if oldstat.mtime != curstat.mtime or \
oldstat.size != curstat.size:
changed = True
py.builtin.print_("# MODIFIED", path)
if removepycfiles and path.ext == ".py":

6
xdist/newhooks.py
View File

@@ -2,20 +2,26 @@
def pytest_xdist_setupnodes(config, specs):
""" called before any remote node is set up. """
def pytest_xdist_newgateway(gateway):
""" called on new raw gateway creation. """
def pytest_xdist_rsyncstart(source, gateways):
""" called before rsyncing a directory to remote gateways takes place. """
def pytest_xdist_rsyncfinish(source, gateways):
""" called after rsyncing a directory to remote gateways takes place. """
def pytest_configure_node(node):
""" configure node information before it gets instantiated. """
def pytest_testnodeready(node):
""" Test Node is ready to operate. """
def pytest_testnodedown(node, error):
""" Test Node is down. """

78
xdist/plugin.py
View File

@@ -12,45 +12,56 @@ def parse_numprocesses(s):
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption('-n', dest="numprocesses", metavar="numprocesses",
action="store",
type=parse_numprocesses,
help="shortcut for '--dist=load --tx=NUM*popen', "
"you can use 'auto' here for auto detection CPUs number on "
"host system")
group._addoption(
'-n', dest="numprocesses", metavar="numprocesses",
action="store",
type=parse_numprocesses,
help="shortcut for '--dist=load --tx=NUM*popen', "
"you can use 'auto' here for auto detection CPUs number on "
"host system")
group._addoption('--max-slave-restart', action="store", default=None,
help="maximum number of slaves that can be restarted "
"when crashed (set to zero to disable this feature)")
group._addoption('--dist', metavar="distmode",
action="store", choices=['load', 'each', 'no'],
type="choice", dest="dist", default="no",
help=("set mode for distributing tests to exec environments.\n\n"
"each: send each test to each available environment.\n\n"
"load: send each test to available environment.\n\n"
"(default) no: run tests inprocess, don't distribute."))
group._addoption('--tx', dest="tx", action="append", default=[],
metavar="xspec",
help=("add a test execution environment. some examples: "
"--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
"--tx ssh=user@codespeak.net//chdir=testcache"))
group._addoption('-d',
action="store_true", dest="distload", default=False,
help="load-balance tests. shortcut for '--dist=load'")
group.addoption('--rsyncdir', action="append", default=[], metavar="DIR",
help="add directory for rsyncing to remote tx nodes.")
group.addoption('--rsyncignore', action="append", default=[], metavar="GLOB",
help="add expression for ignores when rsyncing to remote tx nodes.")
group._addoption(
'--dist', metavar="distmode",
action="store", choices=['load', 'each', 'no'],
type="choice", dest="dist", default="no",
help=("set mode for distributing tests to exec environments.\n\n"
"each: send each test to each available environment.\n\n"
"load: send each test to available environment.\n\n"
"(default) no: run tests inprocess, don't distribute."))
group._addoption(
'--tx', dest="tx", action="append", default=[],
metavar="xspec",
help=("add a test execution environment. some examples: "
"--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
"--tx ssh=user@codespeak.net//chdir=testcache"))
group._addoption(
'-d',
action="store_true", dest="distload", default=False,
help="load-balance tests. shortcut for '--dist=load'")
group.addoption(
'--rsyncdir', action="append", default=[], metavar="DIR",
help="add directory for rsyncing to remote tx nodes.")
group.addoption(
'--rsyncignore', action="append", default=[], metavar="GLOB",
help="add expression for ignores when rsyncing to remote tx nodes.")
parser.addini('rsyncdirs', 'list of (relative) paths to be rsynced for'
' remote distributed testing.', type="pathlist")
parser.addini('rsyncignore', 'list of (relative) glob-style paths to be ignored '
'for rsyncing.', type="pathlist")
parser.addini("looponfailroots", type="pathlist",
parser.addini(
'rsyncdirs', 'list of (relative) paths to be rsynced for'
' remote distributed testing.', type="pathlist")
parser.addini(
'rsyncignore', 'list of (relative) glob-style paths to be ignored '
'for rsyncing.', type="pathlist")
parser.addini(
"looponfailroots", type="pathlist",
help="directories to check for changes", default=[py.path.local()])
# -------------------------------------------------------------------------
# distributed testing hooks
# -------------------------------------------------------------------------
def pytest_addhooks(pluginmanager):
from xdist import newhooks
# avoid warnings with pytest-2.8
@@ -73,6 +84,7 @@ def pytest_configure(config):
tr = config.pluginmanager.getplugin("terminalreporter")
tr.showfspath = False
@pytest.mark.tryfirst
def pytest_cmdline_main(config):
if config.option.numprocesses:
@@ -85,7 +97,9 @@ def pytest_cmdline_main(config):
usepdb = config.option.usepdb # a core option
if val("looponfail"):
if usepdb:
raise pytest.UsageError("--pdb incompatible with --looponfail.")
raise pytest.UsageError(
"--pdb incompatible with --looponfail.")
elif val("dist") != "no":
if usepdb:
raise pytest.UsageError("--pdb incompatible with distributing tests.")
raise pytest.UsageError(
"--pdb incompatible with distributing tests.")

35
xdist/remote.py
View File

@@ -6,7 +6,9 @@
needs not to be installed in remote environments.
"""
import sys, os
import sys
import os
class SlaveInteractor:
def __init__(self, config, channel):
@@ -72,7 +74,8 @@ class SlaveInteractor:
nextitem=nextitem)
def pytest_collection_finish(self, session):
self.sendevent("collectionfinish",
self.sendevent(
"collectionfinish",
topdir=str(session.fspath),
ids=[item.nodeid for item in session.items])
@@ -89,6 +92,7 @@ class SlaveInteractor:
data = serialize_report(report)
self.sendevent("collectreport", data=data)
def serialize_report(rep):
import py
d = rep.__dict__.copy()
@@ -100,20 +104,22 @@ def serialize_report(rep):
if isinstance(d[name], py.path.local):
d[name] = str(d[name])
elif name == "result":
d[name] = None # for now
d[name] = None # for now
return d
def getinfodict():
import platform
return dict(
version = sys.version,
version_info = tuple(sys.version_info),
sysplatform = sys.platform,
platform = platform.platform(),
executable = sys.executable,
cwd = os.getcwd(),
version=sys.version,
version_info=tuple(sys.version_info),
sysplatform=sys.platform,
platform=platform.platform(),
executable=sys.executable,
cwd=os.getcwd(),
)
def remote_initconfig(option_dict, args):
from _pytest.config import Config
option_dict['plugins'].append("no:terminal")
@@ -131,14 +137,15 @@ if __name__ == '__channelexec__':
channel = channel # noqa
# python3.2 is not concurrent import safe, so let's play it safe
# https://bitbucket.org/hpk42/pytest/issue/347/pytest-xdist-and-python-32
if sys.version_info[:2] == (3,2):
if sys.version_info[:2] == (3, 2):
os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
slaveinput,args,option_dict = channel.receive()
slaveinput, args, option_dict = channel.receive()
importpath = os.getcwd()
sys.path.insert(0, importpath) # XXX only for remote situations
os.environ['PYTHONPATH'] = (importpath + os.pathsep +
sys.path.insert(0, importpath) # XXX only for remote situations
os.environ['PYTHONPATH'] = (
importpath + os.pathsep +
os.environ.get('PYTHONPATH', ''))
#os.environ['PYTHONPATH'] = importpath
# os.environ['PYTHONPATH'] = importpath
import py
config = remote_initconfig(option_dict, args)
config.slaveinput = slaveinput

47
xdist/slavemanage.py
View File

@@ -1,16 +1,19 @@
import fnmatch
import os
import re
import py
import pytest
import execnet
import xdist.remote
from _pytest import runner # XXX load dynamically
from _pytest import runner # XXX load dynamically
class NodeManager(object):
EXIT_TIMEOUT = 10
DEFAULT_IGNORES = ['.*', '*.pyc', '*.pyo', '*~']
def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
self.config = config
self._nodesready = py.std.threading.Event()
@@ -79,11 +82,12 @@ class NodeManager(object):
break
else:
return []
import pytest, _pytest
import pytest
import _pytest
pytestpath = pytest.__file__.rstrip("co")
pytestdir = py.path.local(_pytest.__file__).dirpath()
config = self.config
candidates = [py._pydir,pytestpath,pytestdir]
candidates = [py._pydir, pytestpath, pytestdir]
candidates += config.option.rsyncdir
rsyncroots = config.getini("rsyncdirs")
if rsyncroots:
@@ -92,7 +96,7 @@ class NodeManager(object):
for root in candidates:
root = py.path.local(root).realpath()
if not root.check():
raise pytest.UsageError("rsyncdir doesn't exist: %r" %(root,))
raise pytest.UsageError("rsyncdir doesn't exist: %r" % (root,))
if root not in roots:
roots.append(root)
return roots
@@ -124,6 +128,7 @@ class NodeManager(object):
return
if (spec, source) in self._rsynced_specs:
return
def finished():
if notify:
notify("rsyncrootready", spec, source)
@@ -139,19 +144,23 @@ class NodeManager(object):
gateways=[gateway],
)
class HostRSync(execnet.RSync):
""" RSyncer that filters out common files
"""
def __init__(self, sourcedir, *args, **kwargs):
self._synced = {}
self._ignores = kwargs.pop('ignores', None) or []
self._ignores = []
ignores = kwargs.pop('ignores', None) or []
for x in ignores:
x = getattr(x, 'strpath', x)
self.ignores.append(re.compile(fnmatch.translate(x)))
super(HostRSync, self).__init__(sourcedir=sourcedir, **kwargs)
def filter(self, path):
path = py.path.local(path)
for x in self._ignores:
x = getattr(x, 'strpath', x)
if fnmatch.fnmatch(path.basename, x) or fnmatch.fnmatch(path.strpath, x):
for check in self._ignores:
if check(path.basename) or check(path.strpath):
return False
else:
return True
@@ -187,6 +196,7 @@ def make_reltoroot(roots, args):
l.append(splitcode.join(parts))
return l
class SlaveController(object):
ENDMARK = -1
@@ -202,7 +212,7 @@ class SlaveController(object):
py.log.setconsumer(self.log._keywords, None)
def __repr__(self):
return "<%s %s>" %(self.__class__.__name__, self.gateway.id,)
return "<%s %s>" % (self.__class__.__name__, self.gateway.id,)
def setup(self):
self.log("setting up slave session")
@@ -219,7 +229,8 @@ class SlaveController(object):
self.channel = self.gateway.remote_exec(xdist.remote)
self.channel.send((self.slaveinput, args, option_dict))
if self.putevent:
self.channel.setcallback(self.process_from_remote,
self.channel.setcallback(
self.process_from_remote,
endmarker=self.ENDMARK)
def ensure_teardown(self):
@@ -227,11 +238,11 @@ class SlaveController(object):
if not self.channel.isclosed():
self.log("closing", self.channel)
self.channel.close()
#del self.channel
# del self.channel
if hasattr(self, 'gateway'):
self.log("exiting", self.gateway)
self.gateway.exit()
#del self.gateway
# del self.gateway
def send_runtest_some(self, indices):
self.sendcommand("runtests", indices=indices)
@@ -255,7 +266,7 @@ class SlaveController(object):
self.log("queuing %s(**%s)" % (eventname, kwargs))
self.putevent((eventname, kwargs))
def process_from_remote(self, eventcall):
def process_from_remote(self, eventcall): # noqa too complex
""" this gets called for each object we receive from
the other side and if the channel closes.
@@ -268,13 +279,13 @@ class SlaveController(object):
err = self.channel._getremoteerror()
if not self._down:
if not err or isinstance(err, EOFError):
err = "Not properly terminated" # lost connection?
err = "Not properly terminated" # lost connection?
self.notify_inproc("errordown", node=self, error=err)
self._down = True
return
eventname, kwargs = eventcall
if eventname in ("collectionstart"):
self.log("ignoring %s(%s)" %(eventname, kwargs))
self.log("ignoring %s(%s)" % (eventname, kwargs))
elif eventname == "slaveready":
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname == "slavefinished":
@@ -283,7 +294,8 @@ class SlaveController(object):
self.notify_inproc("slavefinished", node=self)
elif eventname == "logstart":
self.notify_inproc(eventname, node=self, **kwargs)
elif eventname in ("testreport", "collectreport", "teardownreport"):
elif eventname in (
"testreport", "collectreport", "teardownreport"):
item_index = kwargs.pop("item_index", None)
rep = unserialize_report(eventname, kwargs['data'])
if item_index is not None:
@@ -292,7 +304,7 @@ class SlaveController(object):
elif eventname == "collectionfinish":
self.notify_inproc(eventname, node=self, ids=kwargs['ids'])
else:
raise ValueError("unknown event: %s" %(eventname,))
raise ValueError("unknown event: %s" % (eventname,))
except KeyboardInterrupt:
# should not land in receiver-thread
raise
@@ -301,6 +313,7 @@ class SlaveController(object):
py.builtin.print_("!" * 20, excinfo)
self.config.pluginmanager.notify_exception(excinfo)
def unserialize_report(name, reportdict):
if name == "testreport":
return runner.TestReport(**reportdict)