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

remove trailing whitespace from sources

Browse Source
This commit is contained in:
holger krekel
2010-09-07 10:46:58 +02:00
parent 198a2c2d38
commit 4d0422a548
27 changed files with 365 additions and 360 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
xdist/__init__.py
View File

@@ -1,3 +1,3 @@
#
__version__ = "1.4"
__version__ = "1.5a1"

62
xdist/dsession.py
View File

@@ -1,5 +1,5 @@
import py
from py._test import session
from py._test import session
from xdist.nodemanage import NodeManager
queue = py.builtin._tryimport('queue', 'Queue')
@@ -14,10 +14,10 @@ class LoopState(object):
def __init__(self, dsession, colitems):
self.dsession = dsession
self.colitems = colitems
self.exitstatus = None
# loopstate.dowork is False after reschedule events
# because otherwise we might very busily loop
# waiting for a host to become ready.
self.exitstatus = None
# loopstate.dowork is False after reschedule events
# because otherwise we might very busily loop
# waiting for a host to become ready.
self.dowork = True
self.shuttingdown = False
self.testsfailed = 0
@@ -47,8 +47,8 @@ class LoopState(object):
crashitem = pending[0]
debug("determined crashitem", crashitem)
self.dsession.handle_crashitem(crashitem, node)
# XXX recovery handling for "each"?
# currently pending items are not retried
# XXX recovery handling for "each"?
# currently pending items are not retried
if self.dsession.config.option.dist == "load":
self.colitems.extend(pending[1:])
@@ -59,10 +59,10 @@ class LoopState(object):
self.dowork = False # avoid busywait, nodes still have work
class DSession(session.Session):
"""
"""
Session drives the collection and running of tests
and generates test events for reporters.
"""
and generates test events for reporters.
"""
LOAD_THRESHOLD_NEWITEMS = 5
ITEM_CHUNKSIZE = 10
@@ -96,7 +96,7 @@ class DSession(session.Session):
allitems = self.collect_all_items(colitems)
exitstatus = self.loop(allitems)
self.teardown()
self.sessionfinishes(exitstatus=exitstatus)
self.sessionfinishes(exitstatus=exitstatus)
return exitstatus
def collect_all_items(self, colitems):
@@ -111,19 +111,19 @@ class DSession(session.Session):
def loop_once(self, loopstate):
if loopstate.shuttingdown:
return self.loop_once_shutdown(loopstate)
colitems = loopstate.colitems
colitems = loopstate.colitems
if self._nodesready.isSet() and loopstate.dowork and colitems:
self.triggertesting(loopstate.colitems)
self.triggertesting(loopstate.colitems)
colitems[:] = []
# we use a timeout here so that control-C gets through
# we use a timeout here so that control-C gets through
while 1:
try:
eventcall = self.queue.get(timeout=2.0)
break
except queue.Empty:
continue
loopstate.dowork = True
loopstate.dowork = True
callname, args, kwargs = eventcall
if callname is not None:
call = getattr(self.config.hook, callname)
@@ -132,9 +132,9 @@ class DSession(session.Session):
# termination conditions
maxfail = self.config.getvalue("maxfail")
if (not self.node2pending or
(loopstate.testsfailed and maxfail and
loopstate.testsfailed >= maxfail) or
if (not self.node2pending or
(loopstate.testsfailed and maxfail and
loopstate.testsfailed >= maxfail) or
(not self.item2nodes and not colitems and not self.queue.qsize())):
if maxfail and loopstate.testsfailed >= maxfail:
raise self.Interrupted("stopping after %d failures" % (
@@ -143,10 +143,10 @@ class DSession(session.Session):
loopstate.shuttingdown = True
if not self.node2pending:
loopstate.exitstatus = session.EXIT_NOHOSTS
def loop_once_shutdown(self, loopstate):
# once we are in shutdown mode we dont send
# events other than HostDown upstream
# once we are in shutdown mode we dont send
# events other than HostDown upstream
eventname, args, kwargs = self.queue.get()
if eventname == "pytest_testnodedown":
self.config.hook.pytest_testnodedown(**kwargs)
@@ -181,7 +181,7 @@ class DSession(session.Session):
self.loop_once(loopstate)
if loopstate.exitstatus is not None:
exitstatus = loopstate.exitstatus
break
break
except KeyboardInterrupt:
excinfo = py.code.ExceptionInfo()
self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo)
@@ -201,7 +201,7 @@ class DSession(session.Session):
def addnode(self, node):
assert node not in self.node2pending
self.node2pending[node] = []
if (not hasattr(self, 'nodemanager') or
if (not hasattr(self, 'nodemanager') or
len(self.node2pending) == len(self.nodemanager.gwmanager.group)):
self._nodesready.set()
@@ -219,7 +219,7 @@ class DSession(session.Session):
return pending
def triggertesting(self, colitems):
# for now we don't allow sending collectors
# for now we don't allow sending collectors
for next in colitems:
assert isinstance(next, py.test.collect.Item), next
senditems = list(colitems)
@@ -230,11 +230,11 @@ class DSession(session.Session):
self.senditems_load(senditems)
def queueevent(self, eventname, **kwargs):
self.queue.put((eventname, (), kwargs))
self.queue.put((eventname, (), kwargs))
def senditems_each(self, tosend):
if not tosend:
return
return
for node, pending in self.node2pending.items():
node.sendlist(tosend)
pending.extend(tosend)
@@ -247,7 +247,7 @@ class DSession(session.Session):
def senditems_load(self, tosend):
if not tosend:
return
return
available = []
for node, pending in self.node2pending.items():
if len(pending) < self.LOAD_THRESHOLD_NEWITEMS:
@@ -281,7 +281,7 @@ class DSession(session.Session):
pending.remove(item)
def handle_crashitem(self, item, node):
runner = item.config.pluginmanager.getplugin("runner")
runner = item.config.pluginmanager.getplugin("runner")
info = "!!! Node %r crashed during running of test %r" %(node, item)
rep = runner.ItemTestReport(item=item, excinfo=info, when="???")
rep.node = node
@@ -290,11 +290,11 @@ class DSession(session.Session):
def setup(self):
""" setup any neccessary resources ahead of the test run. """
if not self.config.getvalue("verbose"):
self.report_line("instantiating gateways (use -v for details): %s" %
self.report_line("instantiating gateways (use -v for details): %s" %
",".join(self.config.option.tx))
self.nodemanager = NodeManager(self.config)
self.nodemanager.setup_nodes(putevent=self.queue.put)
def teardown(self):
""" teardown any resources after a test run. """
""" teardown any resources after a test run. """
self.nodemanager.teardown_nodes()

20
xdist/gwmanage.py
View File

@@ -29,8 +29,8 @@ class GatewayManager:
gateway=gw, platinfo=gw._rinfo())
def rsync(self, source, notify=None, verbose=False, ignores=None):
""" perform rsync to all remote hosts.
"""
""" perform rsync to all remote hosts.
"""
rsync = HostRSync(source, verbose=verbose, ignores=ignores)
seen = py.builtin.set()
gateways = []
@@ -38,7 +38,7 @@ class GatewayManager:
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
# and that adding the basedir does not hurt
gateway.remote_exec("""
import sys ; sys.path.insert(0, %r)
""" % os.path.dirname(str(source))).waitclose()
@@ -52,20 +52,20 @@ class GatewayManager:
gateways.append(gateway)
if seen:
self.hook.pytest_gwmanage_rsyncstart(
source=source,
gateways=gateways,
source=source,
gateways=gateways,
)
rsync.send()
self.hook.pytest_gwmanage_rsyncfinish(
source=source,
gateways=gateways,
source=source,
gateways=gateways,
)
def exit(self):
self.group.terminate(self.EXIT_TIMEOUT)
class HostRSync(execnet.RSync):
""" RSyncer that filters out common files
""" RSyncer that filters out common files
"""
def __init__(self, sourcedir, *args, **kwargs):
self._synced = {}
@@ -78,7 +78,7 @@ class HostRSync(execnet.RSync):
def filter(self, path):
path = py.path.local(path)
if not path.ext in ('.pyc', '.pyo'):
if not path.basename.endswith('~'):
if not path.basename.endswith('~'):
if path.check(dotfile=0):
for x in self._ignores:
if path == x:
@@ -88,7 +88,7 @@ class HostRSync(execnet.RSync):
def add_target_host(self, gateway, finished=None):
remotepath = os.path.basename(self._sourcedir)
super(HostRSync, self).add_target(gateway, remotepath,
super(HostRSync, self).add_target(gateway, remotepath,
finishedcallback=finished,
delete=True,)

48
xdist/mypickle.py
View File

@@ -1,14 +1,14 @@
"""
Pickling support for two processes that want to exchange
*immutable* object instances. Immutable in the sense
that the receiving side of an object can modify its
copy but when it sends it back the original sending
Pickling support for two processes that want to exchange
*immutable* object instances. Immutable in the sense
that the receiving side of an object can modify its
copy but when it sends it back the original sending
side will continue to see its unmodified version
(and no actual state will go over the wire).
This module also implements an experimental
execnet pickling channel using this idea.
This module also implements an experimental
execnet pickling channel using this idea.
"""
@@ -18,7 +18,7 @@ import sys, os, struct
if sys.version_info >= (3,0):
makekey = lambda x: x
fromkey = lambda x: x
fromkey = lambda x: x
from pickle import _Pickler as Pickler
from pickle import _Unpickler as Unpickler
else:
@@ -29,7 +29,7 @@ else:
class MyPickler(Pickler):
""" Pickler with a custom memoize()
to take care of unique ID creation.
to take care of unique ID creation.
See the usage in ImmutablePickler
"""
def __init__(self, immo, file, protocol, uneven):
@@ -37,7 +37,7 @@ class MyPickler(Pickler):
self.uneven = uneven
self._unpicklememo = immo._unpicklememo
self.memo = immo._picklememo
def memoize(self, obj):
if self.fast:
return
@@ -55,23 +55,23 @@ class MyPickler(Pickler):
# def save_string(self, obj, pack=struct.pack):
# obj = unicode(obj)
# self.save_unicode(obj, pack=pack)
# Pickler.dispatch[str] = save_string
# Pickler.dispatch[str] = save_string
class UnpicklingDict(dict):
def __init__(self, picklememo):
super(UnpicklingDict, self).__init__()
self._picklememo = picklememo
def __setitem__(self, key, obj):
super(UnpicklingDict, self).__setitem__(key, obj)
self._picklememo[id(obj)] = (fromkey(key), obj)
class ImmutablePickler:
def __init__(self, uneven, protocol=0):
""" ImmutablePicklers are instantiated in Pairs.
""" ImmutablePicklers are instantiated in Pairs.
The two sides need to create unique IDs
while pickling their objects. This is
done by using either even or uneven
done by using either even or uneven
numbers, depending on the instantiation
parameter.
"""
@@ -82,8 +82,8 @@ class ImmutablePickler:
def selfmemoize(self, obj):
# this is for feeding objects to ourselfes
# which be the case e.g. if you want to pickle
# from a forked process back to the original
# which be the case e.g. if you want to pickle
# from a forked process back to the original
f = py.io.BytesIO()
pickler = MyPickler(self, f, self._protocol, uneven=self.uneven)
pickler.memoize(obj)
@@ -92,7 +92,7 @@ class ImmutablePickler:
f = py.io.BytesIO()
pickler = MyPickler(self, f, self._protocol, uneven=self.uneven)
pickler.dump(obj)
#print >>debug, "dumped", obj
#print >>debug, "dumped", obj
#print >>debug, "picklememo", self._picklememo
return f.getvalue()
@@ -117,20 +117,20 @@ class UnpickleError(Exception):
return self.formatted
class PickleChannel(object):
""" PickleChannels wrap execnet channels
""" PickleChannels wrap execnet channels
and allow to send/receive by using
"immutable pickling".
"immutable pickling".
"""
_unpicklingerror = None
def __init__(self, channel):
self._channel = channel
# we use the fact that each side of a
# we use the fact that each side of a
# gateway connection counts with uneven
# or even numbers depending on which
# or even numbers depending on which
# side it is (for the purpose of creating
# unique ids - which is what we need it here for)
uneven = channel.gateway._channelfactory.count % 2
self._ipickle = ImmutablePickler(uneven=uneven)
uneven = channel.gateway._channelfactory.count % 2
self._ipickle = ImmutablePickler(uneven=uneven)
self.RemoteError = channel.RemoteError
def send(self, obj):

2
xdist/newhooks.py
View File

@@ -1,6 +1,6 @@
def pytest_gwmanage_newgateway(gateway, platinfo):
""" called on new raw gateway creation. """
""" called on new raw gateway creation. """
def pytest_gwmanage_rsyncstart(source, gateways):
""" called before rsyncing a directory to remote gateways takes place. """

28
xdist/nodemanage.py
View File

@@ -4,10 +4,10 @@ import xdist
from xdist.txnode import TXNode
from xdist.gwmanage import GatewayManager
import execnet
class NodeManager(object):
def __init__(self, config, specs=None):
self.config = config
self.config = config
if specs is None:
specs = self._getxspecs()
self.roots = self._getrsyncdirs()
@@ -23,32 +23,32 @@ class NodeManager(object):
def rsync_roots(self):
""" make sure that all remote gateways
have the same set of roots in their
current directory.
current directory.
"""
self.makegateways()
options = {
'ignores': self.config_getignores(),
'ignores': self.config_getignores(),
'verbose': self.config.option.verbose,
}
if self.roots:
# send each rsync root
for root in self.roots:
self.gwmanager.rsync(root, **options)
else:
XXX # do we want to care for situations without explicit rsyncdirs?
else:
XXX # do we want to care for situations without explicit rsyncdirs?
# we transfer our topdir as the root
self.gwmanager.rsync(self.config.topdir, **options)
# and cd into it
# and cd into it
self.gwmanager.multi_chdir(self.config.topdir.basename, inplacelocal=False)
def makegateways(self):
# we change to the topdir sot that
# PopenGateways will have their cwd
# such that unpickling configs will
# pick it up as the right topdir
# we change to the topdir sot that
# PopenGateways will have their cwd
# such that unpickling configs will
# pick it up as the right topdir
# (for other gateways this chdir is irrelevant)
self.trace("making gateways")
old = self.config.topdir.chdir()
old = self.config.topdir.chdir()
try:
self.gwmanager.makegateways()
finally:
@@ -59,7 +59,7 @@ class NodeManager(object):
self.trace("setting up nodes")
for gateway in self.gwmanager.group:
node = TXNode(self, gateway, self.config, putevent)
gateway.node = node # to keep node alive
gateway.node = node # to keep node alive
self.trace("started node %r" % node)
def teardown_nodes(self):
@@ -83,7 +83,7 @@ class NodeManager(object):
def _getrsyncdirs(self):
config = self.config
candidates = [py._pydir]
candidates = [py._pydir]
candidates += [py.path.local(xdist.__file__).dirpath()]
candidates += config.option.rsyncdir
conftestroots = config.getconftest_pathlist("rsyncdirs")

96
xdist/plugin.py
View File

@@ -1,23 +1,23 @@
"""loop on failing tests, distribute test runs to CPUs and hosts.
The `pytest-xdist`_ plugin extends py.test with some unique
The `pytest-xdist`_ plugin extends py.test with some unique
test execution modes:
* Looponfail: run your tests repeatedly in a subprocess. After each run py.test
waits until a file in your project changes and then re-runs the previously
failing tests. This is repeated until all tests pass after which again
a full run is performed.
a full run is performed.
* Load-balancing: if you have multiple CPUs or hosts you can use
those for a combined test run. This allows to speed up
development or to use special resources of remote machines.
those for a combined test run. This allows to speed up
development or to use special resources of remote machines.
* Multi-Platform coverage: you can specify different Python interpreters
or different platforms and run tests in parallel on all of them.
or different platforms and run tests in parallel on all of them.
Before running tests remotely, ``py.test`` efficiently synchronizes your
program source code to the remote place. All test results
are reported back and displayed to your local test session.
Before running tests remotely, ``py.test`` efficiently synchronizes your
program source code to the remote place. All test results
are reported back and displayed to your local test session.
You may specify different Python versions and interpreters.
.. _`pytest-xdist`: http://pypi.python.org/pypi/pytest-xdist
@@ -32,11 +32,11 @@ To send tests to multiple CPUs, type::
py.test -n NUM
Especially for longer running tests or tests requiring
a lot of IO this can lead to considerable speed ups.
Especially for longer running tests or tests requiring
a lot of IO this can lead to considerable speed ups.
Running tests in a Python subprocess
Running tests in a Python subprocess
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
To instantiate a python2.4 sub process and send tests to it, you may type::
@@ -44,51 +44,51 @@ To instantiate a python2.4 sub process and send tests to it, you may type::
py.test -d --tx popen//python=python2.4
This will start a subprocess which is run with the "python2.4"
Python interpreter, found in your system binary lookup path.
Python interpreter, found in your system binary lookup path.
If you prefix the --tx option value like this::
--tx 3*popen//python=python2.4
then three subprocesses would be created and tests
will be load-balanced across these three processes.
will be load-balanced across these three processes.
Sending tests to remote SSH accounts
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Suppose you have a package ``mypkg`` which contains some
Suppose you have a package ``mypkg`` which contains some
tests that you can successfully run locally. And you
have a ssh-reachable machine ``myhost``. Then
have a ssh-reachable machine ``myhost``. Then
you can ad-hoc distribute your tests by typing::
py.test -d --tx ssh=myhostpopen --rsyncdir mypkg mypkg
This will synchronize your ``mypkg`` package directory
to an remote ssh account and then locally collect tests
and send them to remote places for execution.
This will synchronize your ``mypkg`` package directory
to an remote ssh account and then locally collect tests
and send them to remote places for execution.
You can specify multiple ``--rsyncdir`` directories
to be sent to the remote side.
You can specify multiple ``--rsyncdir`` directories
to be sent to the remote side.
**NOTE:** For py.test to collect and send tests correctly
you not only need to make sure all code and tests
directories are rsynced, but that any test (sub) directory
also has an ``__init__.py`` file because internally
py.test references tests as a fully qualified python
module path. **You will otherwise get strange errors**
module path. **You will otherwise get strange errors**
during setup of the remote side.
Sending tests to remote Socket Servers
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Download the single-module `socketserver.py`_ Python program
Download the single-module `socketserver.py`_ Python program
and run it like this::
python socketserver.py
It will tell you that it starts listening on the default
port. You can now on your home machine specify this
port. You can now on your home machine specify this
new socket host with something like this::
py.test -d --tx socket=192.168.1.102:8888 --rsyncdir mypkg mypkg
@@ -96,17 +96,17 @@ new socket host with something like this::
.. _`atonce`:
Running tests on many platforms at once
Running tests on many platforms at once
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
The basic command to run tests on multiple platforms is::
py.test --dist=each --tx=spec1 --tx=spec2
py.test --dist=each --tx=spec1 --tx=spec2
If you specify a windows host, an OSX host and a Linux
environment this command will send each tests to all
environment this command will send each tests to all
platforms - and report back failures from all platforms
at once. The specifications strings use the `xspec syntax`_.
at once. The specifications strings use the `xspec syntax`_.
.. _`xspec syntax`: http://codespeak.net/execnet/trunk/basics.html#xspec
@@ -117,14 +117,14 @@ at once. The specifications strings use the `xspec syntax`_.
Specifying test exec environments in a conftest.py
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Instead of specifying command line options, you can
Instead of specifying command line options, you can
put options values in a ``conftest.py`` file like this::
option_tx = ['ssh=myhost//python=python2.5', 'popen//python=python2.5']
option_dist = True
Any commandline ``--tx`` specifictions will add to the list of
available execution environments.
available execution environments.
Specifying "rsync" dirs in a conftest.py
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -144,33 +144,33 @@ import sys
import py
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption('-f', '--looponfail',
action="store_true", dest="looponfail", default=False,
help="run tests in subprocess, wait for modified files "
"and re-run failing test set until all pass.")
group._addoption('-n', dest="numprocesses", metavar="numprocesses",
action="store", type="int",
group._addoption('-n', dest="numprocesses", metavar="numprocesses",
action="store", type="int",
help="shortcut for '--dist=load --tx=NUM*popen'")
group.addoption('--boxed',
action="store_true", dest="boxed", default=False,
help="box each test run in a separate process (unix)")
group._addoption('--dist', metavar="distmode",
action="store", choices=['load', 'each', 'no'],
type="choice", dest="dist", default="no",
help="box each test run in a separate process (unix)")
group._addoption('--dist', metavar="distmode",
action="store", choices=['load', 'each', 'no'],
type="choice", dest="dist", default="no",
help=("set mode for distributing tests to exec environments.\n\n"
"each: send each test to each available environment.\n\n"
"load: send each test to available environment.\n\n"
"(default) no: run tests inprocess, don't distribute."))
group._addoption('--tx', dest="tx", action="append", default=[],
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',
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="dir1",
group.addoption('--rsyncdir', action="append", default=[], metavar="dir1",
help="add directory for rsyncing to remote tx nodes.")
# -------------------------------------------------------------------------
@@ -223,20 +223,20 @@ def pytest_runtest_protocol(item):
return True
def forked_run_report(item):
# for now, we run setup/teardown in the subprocess
# XXX optionally allow sharing of setup/teardown
# for now, we run setup/teardown in the subprocess
# XXX optionally allow sharing of setup/teardown
from py._plugin.pytest_runner import runtestprotocol
EXITSTATUS_TESTEXIT = 4
from xdist.mypickle import ImmutablePickler
ipickle = ImmutablePickler(uneven=0)
ipickle.selfmemoize(item.config)
# XXX workaround the issue that 2.6 cannot pickle
# XXX workaround the issue that 2.6 cannot pickle
# instances of classes defined in global conftest.py files
ipickle.selfmemoize(item)
ipickle.selfmemoize(item)
def runforked():
try:
reports = runtestprotocol(item, log=False)
except KeyboardInterrupt:
except KeyboardInterrupt:
py.std.os._exit(EXITSTATUS_TESTEXIT)
return ipickle.dumps(reports)
@@ -275,16 +275,16 @@ class TerminalDistReporter:
def pytest_runtest_logreport(self, report):
if hasattr(report, 'node'):
report.headerlines.append(self.gateway2info.get(
report.node.gateway,
report.node.gateway,
"node %r (platinfo not found? strange)"))
def pytest_gwmanage_newgateway(self, gateway, platinfo):
#self.write_line("%s instantiated gateway from spec %r" %(gateway.id, gateway.spec._spec))
d = {}
d['version'] = self.tplugin.repr_pythonversion(platinfo.version_info)
d['id'] = gateway.id
d['spec'] = gateway.spec._spec
d['platform'] = platinfo.platform
d['spec'] = gateway.spec._spec
d['platform'] = platinfo.platform
if self.config.option.verbose:
d['extra'] = "- " + platinfo.executable
else:

40
xdist/remote.py
View File

@@ -1,11 +1,11 @@
"""
LooponfailingSession and Helpers.
LooponfailingSession and Helpers.
NOTE that one really has to avoid loading and depending on
application modules within the controlling process
NOTE that one really has 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 never happen.
otherwise changes to source code can crash
the controlling process which should never happen.
"""
import py
import sys
@@ -16,8 +16,8 @@ from xdist import util
class LooponfailingSession(Session):
def __init__(self, config):
super(LooponfailingSession, self).__init__(config=config)
self.rootdirs = [self.config.topdir] # xxx dist_rsync_roots?
self.statrecorder = util.StatRecorder(self.rootdirs)
self.rootdirs = [self.config.topdir] # xxx dist_rsync_roots?
self.statrecorder = util.StatRecorder(self.rootdirs)
self.remotecontrol = RemoteControl(self.config)
self.out = py.io.TerminalWriter()
@@ -28,7 +28,7 @@ class LooponfailingSession(Session):
self.loop_once(loopstate)
if not loopstate.colitems and loopstate.wasfailing:
continue # the last failures passed, let's rerun all
self.statrecorder.waitonchange(checkinterval=2.0)
self.statrecorder.waitonchange(checkinterval=2.0)
except KeyboardInterrupt:
print
@@ -74,9 +74,9 @@ class RemoteControl(object):
if not os.path.isabs(p):
p = os.path.abspath(p)
newpaths.append(p)
sys.path[:] = newpaths
sys.path[:] = newpaths
os.chdir(chdir) # unpickling config uses cwd as topdir
config_state = channel.receive()
fullwidth, hasmarkup = channel.receive()
py.test.config.__setstate__(config_state)
@@ -85,7 +85,7 @@ class RemoteControl(object):
sys.stdout = sys.stderr = outchannel.makefile('w')
from xdist.remote import slave_runsession
slave_runsession(channel, py.test.config, fullwidth, hasmarkup)
slave_runsession(channel, py.test.config, fullwidth, hasmarkup)
""")
channel.send(str(self.config.topdir))
remote_outchannel = channel.receive()
@@ -125,15 +125,15 @@ class RemoteControl(object):
def slave_runsession(channel, config, fullwidth, hasmarkup):
""" we run this on the other side. """
if config.option.debug:
def DEBUG(*args):
def DEBUG(*args):
print(" ".join(map(str, args)))
else:
def DEBUG(*args): pass
DEBUG("SLAVE: received configuration, using topdir:", config.topdir)
#config.option.session = None
config.option.looponfail = False
config.option.usepdb = False
config.option.looponfail = False
config.option.usepdb = False
try:
trails = channel.receive()
except KeyboardInterrupt:
@@ -142,7 +142,7 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
DEBUG("SLAVE: initsession()")
session = config.initsession()
# XXX configure the reporter object's terminal writer more directly
# XXX and write a test for this remote-terminal setting logic
# XXX and write a test for this remote-terminal setting logic
config.pytest_terminal_hasmarkup = hasmarkup
config.pytest_terminal_fullwidth = fullwidth
if trails:
@@ -152,25 +152,25 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
colitem = config._rootcol.fromtrail(trail)
except ValueError:
#XXX send info for "test disappeared" or so
continue
continue
colitems.append(colitem)
else:
colitems = config.getinitialnodes()
session.shouldclose = channel.isclosed
session.shouldclose = channel.isclosed
class Failures(list):
def pytest_runtest_logreport(self, report):
if report.failed:
self.append(report)
pytest_collectreport = pytest_runtest_logreport
failreports = Failures()
session.pluginmanager.register(failreports)
DEBUG("SLAVE: starting session.main()")
session.main(colitems)
repr_pytest_looponfailinfo(
failreports=list(failreports),
failreports=list(failreports),
rootdirs=[config.topdir])
rootcol = session.config._rootcol
channel.send([rootcol.totrail(rep.getnode()) for rep in failreports])

52
xdist/txnode.py
View File

@@ -1,22 +1,22 @@
"""
Manage setup, running and local representation of remote nodes/processes.
Manage setup, running and local representation of remote nodes/processes.
"""
import py
from xdist.mypickle import PickleChannel
from py._test.session import Session
class TXNode(object):
""" Represents a Test Execution environment in the controlling process.
- sets up a slave node through an execnet gateway
""" Represents a Test Execution environment in the controlling process.
- sets up a slave node through an execnet gateway
- manages sending of test-items and receival of results and events
- creates events when the remote side crashes
- creates events when the remote side crashes
"""
ENDMARK = -1
def __init__(self, nodemanager, gateway, config, putevent):
self.nodemanager = nodemanager
self.config = config
self.putevent = putevent
self.config = config
self.putevent = putevent
self.gateway = gateway
self.slaveinput = {}
self.channel = install_slave(self)
@@ -31,13 +31,13 @@ class TXNode(object):
def notify(self, eventname, *args, **kwargs):
assert not args
self.putevent((eventname, args, kwargs))
def callback(self, eventcall):
""" this gets called for each object we receive from
the other side and if the channel closes.
""" 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
thread of execnet gateways - we need to
avoid raising exceptions or doing heavy work.
"""
try:
@@ -45,11 +45,11 @@ class TXNode(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("pytest_testnodedown", node=self, error=err)
self._down = True
return
eventname, args, kwargs = eventcall
eventname, args, kwargs = eventcall
if eventname == "slaveready":
self.notify("pytest_testnodeready", node=self)
elif eventname == "slavefinished":
@@ -57,15 +57,15 @@ class TXNode(object):
self.slaveoutput = kwargs['slaveoutput']
error = kwargs['error']
self.notify("pytest_testnodedown", error=error, node=self)
elif eventname in ("pytest_runtest_logreport",
elif eventname in ("pytest_runtest_logreport",
"pytest__teardown_final_logerror"):
kwargs['report'].node = self
self.notify(eventname, **kwargs)
else:
self.notify(eventname, **kwargs)
except KeyboardInterrupt:
except KeyboardInterrupt:
# should not land in receiver-thread
raise
raise
except:
excinfo = py.code.ExceptionInfo()
py.builtin.print_("!" * 20, excinfo)
@@ -84,11 +84,11 @@ class TXNode(object):
else:
self.channel.send(None)
# configuring and setting up slave node
# configuring and setting up slave node
def install_slave(node):
channel = node.gateway.remote_exec(source="""
import os, sys
sys.path.insert(0, os.getcwd())
import os, sys
sys.path.insert(0, os.getcwd())
from xdist.mypickle import PickleChannel
from xdist.txnode import SlaveSession
channel.send("basicimport")
@@ -107,11 +107,11 @@ def install_slave(node):
channel.receive()
channel = PickleChannel(channel)
basetemp = None
config = node.config
config = node.config
config.hook.pytest_configure_node(node=node)
if node.gateway.spec.popen:
popenbase = config.ensuretemp("popen")
basetemp = py.path.local.make_numbered_dir(prefix="slave-",
basetemp = py.path.local.make_numbered_dir(prefix="slave-",
keep=0, rootdir=popenbase)
basetemp = str(basetemp)
channel.send((config, node.slaveinput, basetemp, node.gateway.id))
@@ -147,13 +147,13 @@ class SlaveSession(Session):
self.sendevent("slaveready")
self.main(None)
error = getattr(self, '_slaveerror', None)
self.sendevent("slavefinished", error=error,
self.sendevent("slavefinished", error=error,
slaveoutput=self.config.slaveoutput)
def _mainloop(self, colitems):
while 1:
task = self.channel.receive()
if task is None:
if task is None:
break
if isinstance(task, list):
for item in task:
@@ -165,10 +165,10 @@ class SlaveSession(Session):
call = self.runner.CallInfo(item._reraiseunpicklingproblem, when='setup')
if call.excinfo:
# likely it is not collectable here because of
# platform/import-dependency induced skips
# platform/import-dependency induced skips
# we fake a setup-error report with the obtained exception
# and do not care about capturing or non-runner hooks
# and do not care about capturing or non-runner hooks
rep = self.runner.pytest_runtest_makereport(item=item, call=call)
self.pytest_runtest_logreport(rep)
return
item.config.hook.pytest_runtest_protocol(item=item)
item.config.hook.pytest_runtest_protocol(item=item)

4
xdist/util.py
View File

@@ -6,7 +6,7 @@ class StatRecorder:
self.statcache = {}
self.check() # snapshot state
def fil(self, p):
def fil(self, p):
return p.ext in ('.py', '.txt', '.c', '.h')
def rec(self, p):
return p.check(dotfile=0)
@@ -43,7 +43,7 @@ class StatRecorder:
pycfile = path + "c"
if pycfile.check():
pycfile.remove()
else:
changed = True
if statcache: