make looponfailing work with the new py-trunk developments, refactor internal sesssion handling
This commit is contained in:
holger krekel
192
xdist/looponfail.py
Normal file
192
xdist/looponfail.py
Normal file
@@ -0,0 +1,192 @@
|
||||
"""
|
||||
Implement -f aka looponfailing for py.test.
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
import py
|
||||
import sys
|
||||
import execnet
|
||||
from py._test.session import gettopdir
|
||||
from xdist import util
|
||||
|
||||
def looponfail_main(config):
|
||||
remotecontrol = RemoteControl(config)
|
||||
# XXX better configure rootdir
|
||||
rootdirs = [gettopdir(config.args)]
|
||||
statrecorder = util.StatRecorder(rootdirs)
|
||||
try:
|
||||
while 1:
|
||||
remotecontrol.loop_once()
|
||||
if not remotecontrol.failures and remotecontrol.wasfailing:
|
||||
continue # the last failures passed, let's immediately rerun all
|
||||
statrecorder.waitonchange(checkinterval=2.0)
|
||||
except KeyboardInterrupt:
|
||||
print()
|
||||
|
||||
class RemoteControl(object):
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.remote_topdir = None
|
||||
self.failures = []
|
||||
|
||||
def trace(self, *args):
|
||||
if self.config.option.debug:
|
||||
msg = " ".join([str(x) for x in args])
|
||||
py.builtin.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 slave session")
|
||||
self.gateway = self.initgateway()
|
||||
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()
|
||||
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.remote_topdir, self.failures))
|
||||
self.channel.send((self.remote_topdir, 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()
|
||||
topdir, failures, reports, collection_failed = result
|
||||
if collection_failed:
|
||||
reports = ["Collection failed, keeping previous failure set"]
|
||||
else:
|
||||
self.remote_topdir, self.failures = topdir, failures
|
||||
|
||||
repr_pytest_looponfailinfo(
|
||||
failreports=reports,
|
||||
rootdirs=[self.remote_topdir],)
|
||||
|
||||
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_slave_session(channel, args, option_dict):
|
||||
import os, sys
|
||||
import py
|
||||
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()
|
||||
config = py.test.config
|
||||
config.option.__dict__.update(option_dict)
|
||||
config._preparse(args)
|
||||
config.args = args
|
||||
from xdist.looponfail import SlaveFailSession
|
||||
SlaveFailSession(config, channel).main()
|
||||
|
||||
class SlaveFailSession:
|
||||
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_perform_collection(self, session):
|
||||
self.session = session
|
||||
self.collection = session.collection
|
||||
self.topdir, self.trails = self.current_command
|
||||
if self.topdir and self.trails:
|
||||
self.topdir = py.path.local(self.topdir)
|
||||
self.collection.topdir = self.topdir
|
||||
nodes = []
|
||||
for trail in self.trails:
|
||||
names = self.collection._parsearg(trail, base=self.topdir)
|
||||
try:
|
||||
self.collection.genitems(
|
||||
[self.collection._topcollector], names, nodes)
|
||||
except self.config.Error:
|
||||
pass # ignore collect errors / vanished tests
|
||||
self.collection.items = nodes
|
||||
return True
|
||||
self.topdir = session.collection.topdir
|
||||
|
||||
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("SLAVE: received configuration, waiting for command trails")
|
||||
try:
|
||||
command = self.channel.receive()
|
||||
except KeyboardInterrupt:
|
||||
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)
|
||||
trails, failreports = [], []
|
||||
for rep in self.recorded_failures:
|
||||
trails.append(self.collection.getid(rep.getnode()))
|
||||
loc = rep._getcrashline()
|
||||
failreports.append(loc)
|
||||
topdir = str(self.topdir)
|
||||
self.channel.send((topdir, trails, failreports, self.collection_failed))
|
||||
|
||||
@@ -195,16 +195,27 @@ def pytest_configure(config):
|
||||
if val("looponfail"):
|
||||
if usepdb:
|
||||
raise config.Error("--pdb incompatible with --looponfail.")
|
||||
from xdist.remote import LooponfailingSession
|
||||
config.setsessionclass(LooponfailingSession)
|
||||
config._isdistsession = True
|
||||
elif val("dist") != "no":
|
||||
if usepdb:
|
||||
raise config.Error("--pdb incompatible with distributing tests.")
|
||||
from xdist.dsession import DSession
|
||||
config.setsessionclass(DSession)
|
||||
config._isdistsession = True
|
||||
|
||||
def pytest_cmdline_main(config):
|
||||
if config.getvalue("looponfail"):
|
||||
from xdist.looponfail import looponfail_main
|
||||
looponfail_main(config)
|
||||
return 2 # looponfail only can get stop with ctrl-C anyway
|
||||
elif config.getvalue("dist"):
|
||||
pass
|
||||
return
|
||||
from py._test.session import Session, Collection
|
||||
collection = Collection(config)
|
||||
# instantiate session already because it
|
||||
# records failures and implements maxfail handling
|
||||
session = Session(config, collection)
|
||||
exitstatus = collection.do_collection()
|
||||
if not exitstatus:
|
||||
exitstatus = session.main()
|
||||
return exitstatus
|
||||
|
||||
def pytest_sessionstart(session):
|
||||
config = session.config
|
||||
|
||||
189
xdist/remote.py
189
xdist/remote.py
@@ -1,189 +0,0 @@
|
||||
"""
|
||||
LooponfailingSession and Helpers.
|
||||
|
||||
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.
|
||||
"""
|
||||
import py
|
||||
import sys
|
||||
import execnet
|
||||
from py._test.session import Session
|
||||
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.remotecontrol = RemoteControl(self.config)
|
||||
self.out = py.io.TerminalWriter()
|
||||
|
||||
def main(self, initialitems):
|
||||
try:
|
||||
self.loopstate = loopstate = LoopState([])
|
||||
while 1:
|
||||
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)
|
||||
except KeyboardInterrupt:
|
||||
print
|
||||
|
||||
def loop_once(self, loopstate):
|
||||
self.remotecontrol.setup()
|
||||
colitems = loopstate.colitems
|
||||
loopstate.wasfailing = colitems and len(colitems)
|
||||
loopstate.colitems = self.remotecontrol.runsession(colitems or ())
|
||||
|
||||
class LoopState:
|
||||
def __init__(self, colitems=None):
|
||||
self.colitems = colitems
|
||||
|
||||
class RemoteControl(object):
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
|
||||
def trace(self, *args):
|
||||
if self.config.option.debug:
|
||||
msg = " ".join([str(x) for x in args])
|
||||
py.builtin.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 slave session")
|
||||
self.gateway = self.initgateway()
|
||||
self.channel = channel = self.gateway.remote_exec("""
|
||||
import os, sys
|
||||
import py
|
||||
chdir = channel.receive()
|
||||
outchannel = channel.gateway.newchannel()
|
||||
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
|
||||
os.chdir(chdir) # unpickling config uses cwd as topdir
|
||||
|
||||
config_state = channel.receive()
|
||||
fullwidth, hasmarkup = channel.receive()
|
||||
py.test.config.__setstate__(config_state)
|
||||
|
||||
import sys
|
||||
sys.stdout = sys.stderr = outchannel.makefile('w')
|
||||
|
||||
from xdist.remote import slave_runsession
|
||||
slave_runsession(channel, py.test.config, fullwidth, hasmarkup)
|
||||
""")
|
||||
channel.send(str(self.config.topdir))
|
||||
remote_outchannel = channel.receive()
|
||||
def write(s):
|
||||
out._file.write(s)
|
||||
out._file.flush()
|
||||
remote_outchannel.setcallback(write)
|
||||
channel.send(self.config.__getstate__())
|
||||
channel.send((out.fullwidth, out.hasmarkup))
|
||||
self.trace("set up of slave session complete")
|
||||
|
||||
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, colitems=()):
|
||||
try:
|
||||
self.trace("sending", colitems)
|
||||
trails = colitems
|
||||
self.channel.send(trails)
|
||||
try:
|
||||
return self.channel.receive()
|
||||
except self.channel.RemoteError:
|
||||
e = sys.exc_info()[1]
|
||||
self.trace("ERROR", e)
|
||||
raise
|
||||
finally:
|
||||
self.ensure_teardown()
|
||||
|
||||
def slave_runsession(channel, config, fullwidth, hasmarkup):
|
||||
""" we run this on the other side. """
|
||||
if config.option.debug:
|
||||
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
|
||||
try:
|
||||
trails = channel.receive()
|
||||
except KeyboardInterrupt:
|
||||
return # in the slave we can't do much about this
|
||||
config.pluginmanager.do_configure(config)
|
||||
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
|
||||
config.pytest_terminal_hasmarkup = hasmarkup
|
||||
config.pytest_terminal_fullwidth = fullwidth
|
||||
if trails:
|
||||
colitems = []
|
||||
for trail in trails:
|
||||
try:
|
||||
colitem = config._rootcol.fromtrail(trail)
|
||||
except ValueError:
|
||||
#XXX send info for "test disappeared" or so
|
||||
continue
|
||||
colitems.append(colitem)
|
||||
else:
|
||||
colitems = config.getinitialnodes()
|
||||
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),
|
||||
rootdirs=[config.topdir])
|
||||
rootcol = session.config._rootcol
|
||||
channel.send([rootcol.totrail(rep.getnode()) for rep in failreports])
|
||||
|
||||
|
||||
def repr_pytest_looponfailinfo(failreports, rootdirs):
|
||||
tr = py.io.TerminalWriter()
|
||||
if failreports:
|
||||
tr.sep("#", "LOOPONFAILING", red=True)
|
||||
for report in failreports:
|
||||
loc = report._getcrashline()
|
||||
if loc:
|
||||
tr.line(loc, red=True)
|
||||
tr.sep("#", "waiting for changes")
|
||||
for rootdir in rootdirs:
|
||||
tr.line("### Watching: %s" %(rootdir,), bold=True)
|
||||
Reference in New Issue
Block a user