jiuyuan
/
test_ok1
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

clean up and simplify startup test protocols and objects 

introduce some new experimental hooks pytest_runtest_mainloop
to better integrate distributed testing

--HG--
branch : trunk
This commit is contained in:
holger krekel 2010-09-26 16:23:43 +02:00
parent 2cf22e3124
commit 7d1585215d
12 changed files with 141 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
py/_plugin/hookspec.py
View File

@ -24,6 +24,10 @@ def pytest_cmdline_main(config):
""" called for performing the main (cmdline) action. """
pytest_cmdline_main.firstresult = True
def pytest_runtest_mainloop(session):
""" called for performing the main runtest loop (after collection. """
pytest_runtest_mainloop.firstresult = True
def pytest_unconfigure(config):
""" called before test process is exited. """
@ -31,10 +35,11 @@ def pytest_unconfigure(config):
# collection hooks
# -------------------------------------------------------------------------
def pytest_log_startcollection(collection):
""" called before collection.perform_collection() is called. """
def pytest_perform_collection(session):
""" perform the collection protocol for the given session. """
pytest_perform_collection.firstresult = True
def pytest_collection_modifyitems(collection):
def pytest_collection_modifyitems(config, items):
""" called to allow filtering and selecting of test items (inplace). """
def pytest_log_finishcollection(collection):
@ -139,6 +144,7 @@ def pytest_sessionstart(session):
def pytest_sessionfinish(session, exitstatus):
""" whole test run finishes. """
# -------------------------------------------------------------------------
# hooks for influencing reporting (invoked from pytest_terminal)
# -------------------------------------------------------------------------

34
py/_plugin/pytest_default.py
View File

@ -4,15 +4,26 @@ import sys
import py
def pytest_cmdline_main(config):
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
from py._test.session import Session
return Session(config).main()
def pytest_perform_collection(session):
collection = session.collection
assert not hasattr(collection, 'items')
hook = session.config.hook
collection.items = items = collection.perform_collect()
hook.pytest_collection_modifyitems(config=session.config, items=items)
hook.pytest_log_finishcollection(collection=collection)
return True
def pytest_runtest_mainloop(session):
if session.config.option.collectonly:
return True
for item in session.collection.items:
item.config.hook.pytest_runtest_protocol(item=item)
if session.shouldstop:
raise session.Interrupted(session.shouldstop)
return True
def pytest_ignore_collect(path, config):
ignore_paths = config.getconftest_pathlist("collect_ignore", path=path)
@ -21,11 +32,6 @@ def pytest_ignore_collect(path, config):
if excludeopt:
ignore_paths.extend([py.path.local(x) for x in excludeopt])
return path in ignore_paths
# XXX more refined would be:
if ignore_paths:
for p in ignore_paths:
if path == p or path.relto(p):
return True
def pytest_collect_directory(path, parent):
# XXX reconsider the following comment

4
py/_plugin/pytest_genscript.py
View File

@ -11,7 +11,7 @@ def pytest_addoption(parser):
dest="genscript", metavar="path",
help="create standalone py.test script at given target path.")
def pytest_configure(config):
def pytest_cmdline_main(config):
genscript = config.getvalue("genscript")
if genscript:
import py
@ -20,7 +20,7 @@ def pytest_configure(config):
pybasedir = py.path.local(py.__file__).dirpath().dirpath()
genscript = py.path.local(genscript)
main(pybasedir, outfile=genscript, infile=infile)
raise SystemExit(0)
return 0
def main(pybasedir, outfile, infile):
import base64

15
py/_plugin/pytest_helpconfig.py
View File

@ -23,15 +23,18 @@ def pytest_addoption(parser):
help="show available conftest.py and ENV-variable names.")
def pytest_configure(__multicall__, config):
def pytest_cmdline_main(config):
if config.option.version:
p = py.path.local(py.__file__).dirpath()
sys.stderr.write("This is py.test version %s, imported from %s\n" %
(py.__version__, p))
sys.exit(0)
if not config.option.helpconfig:
return
__multicall__.execute()
return 0
elif config.option.helpconfig:
config.pluginmanager.do_configure(config)
showpluginhelp(config)
return 0
def showpluginhelp(config):
options = []
for group in config._parser._groups:
options.extend(group.options)
@ -65,9 +68,7 @@ def pytest_configure(__multicall__, config):
help,
)
tw.line(line[:tw.fullwidth])
tw.sep("-")
sys.exit(0)
conftest_options = (
('pytest_plugins', 'list of plugin names to load'),

7
py/_plugin/pytest_keyword.py
View File

@ -8,8 +8,7 @@ def pytest_addoption(parser):
"Terminate the expression with ':' to treat a match as a signal "
"to run all subsequent tests. ")
def pytest_collection_modifyitems(collection):
config = collection.config
def pytest_collection_modifyitems(items, config):
keywordexpr = config.option.keyword
if not keywordexpr:
return
@ -20,7 +19,7 @@ def pytest_collection_modifyitems(collection):
remaining = []
deselected = []
for colitem in collection.items:
for colitem in items:
if keywordexpr and skipbykeyword(colitem, keywordexpr):
deselected.append(colitem)
else:
@ -30,7 +29,7 @@ def pytest_collection_modifyitems(collection):
if deselected:
config.hook.pytest_deselected(items=deselected)
collection.items[:] = remaining
items[:] = remaining
def skipbykeyword(colitem, keywordexpr):
""" return True if they given keyword expression means to

4
py/_plugin/pytest_pytester.py
View File

@ -193,9 +193,9 @@ class TmpTestdir:
args = ("-s", ) + args # otherwise FD leakage
config = self.parseconfig(*args)
reprec = self.getreportrecorder(config)
config.pluginmanager.do_configure(config)
#config.pluginmanager.do_configure(config)
config.hook.pytest_cmdline_main(config=config)
config.pluginmanager.do_unconfigure(config)
#config.pluginmanager.do_unconfigure(config)
return reprec
def config_preparse(self):

25
py/_plugin/pytest_terminal.py
View File

@ -29,21 +29,13 @@ def pytest_addoption(parser):
help="don't cut any tracebacks (default is to cut).")
def pytest_configure(config):
if config.option.showfuncargs:
return
if config.option.collectonly:
reporter = CollectonlyReporter(config)
elif config.option.showfuncargs:
reporter = None
else:
reporter = TerminalReporter(config)
if reporter:
# XXX see remote.py's XXX
for attr in 'pytest_terminal_hasmarkup', 'pytest_terminal_fullwidth':
if hasattr(config, attr):
#print "SETTING TERMINAL OPTIONS", attr, getattr(config, attr)
name = attr.split("_")[-1]
assert hasattr(self.reporter._tw, name), name
setattr(reporter._tw, name, getattr(config, attr))
config.pluginmanager.register(reporter, 'terminalreporter')
config.pluginmanager.register(reporter, 'terminalreporter')
def getreportopt(config):
reportopts = ""
@ -192,7 +184,10 @@ class TerminalReporter:
markup = {}
self.stats.setdefault(cat, []).append(rep)
if not self.config.option.verbose:
self.write_fspath_result(self._getfspath(rep.item), letter)
fspath = getattr(rep, 'fspath', None)
if not fspath:
fspath = self._getfspath(rep.item)
self.write_fspath_result(fspath, letter)
else:
line = self._reportinfoline(rep.item)
if not hasattr(rep, 'node'):
@ -217,17 +212,19 @@ class TerminalReporter:
def pytest_sessionstart(self, session):
self.write_sep("=", "test session starts", bold=True)
self._sessionstarttime = py.std.time.time()
verinfo = ".".join(map(str, sys.version_info[:3]))
msg = "platform %s -- Python %s" % (sys.platform, verinfo)
msg += " -- pytest-%s" % (py.__version__)
if self.config.option.verbose or self.config.option.debug or getattr(self.config.option, 'pastebin', None):
if self.config.option.verbose or self.config.option.debug or \
getattr(self.config.option, 'pastebin', None):
msg += " -- " + str(sys.executable)
self.write_line(msg)
lines = self.config.hook.pytest_report_header(config=self.config)
lines.reverse()
for line in flatten(lines):
self.write_line(line)
def pytest_log_finishcollection(self):
for i, testarg in enumerate(self.config.args):
self.write_line("test path %d: %s" %(i+1, testarg))

11
py/_test/pluginmanager.py
View File

@ -32,7 +32,7 @@ class PluginManager(object):
name = id(plugin)
return name
def register(self, plugin, name=None):
def register(self, plugin, name=None, prepend=False):
assert not self.isregistered(plugin), plugin
assert not self.registry.isregistered(plugin), plugin
name = self._getpluginname(plugin, name)
@ -41,7 +41,7 @@ class PluginManager(object):
self._name2plugin[name] = plugin
self.call_plugin(plugin, "pytest_addhooks", {'pluginmanager': self})
self.hook.pytest_plugin_registered(manager=self, plugin=plugin)
self.registry.register(plugin)
self.registry.register(plugin, prepend=prepend)
return True
def unregister(self, plugin):
@ -277,10 +277,13 @@ class Registry:
plugins = []
self._plugins = plugins
def register(self, plugin):
def register(self, plugin, prepend=False):
assert not isinstance(plugin, str)
assert not plugin in self._plugins
self._plugins.append(plugin)
if not prepend:
self._plugins.append(plugin)
else:
self._plugins.insert(0, plugin)
def unregister(self, plugin):
self._plugins.remove(plugin)

149
py/_test/session.py
View File

@ -6,7 +6,7 @@
"""
import py
import sys
import os, sys
#
# main entry point
@ -16,11 +16,9 @@ def main(args=None):
if args is None:
args = sys.argv[1:]
config = py.test.config
config.parse(args)
try:
config.parse(args)
config.pluginmanager.do_configure(config)
exitstatus = config.hook.pytest_cmdline_main(config=config)
config.pluginmanager.do_unconfigure(config)
except config.Error:
e = sys.exc_info()[1]
sys.stderr.write("ERROR: %s\n" %(e.args[0],))
@ -28,7 +26,6 @@ def main(args=None):
py.test.config = py.test.config.__class__()
return exitstatus
# exitcodes for the command line
EXIT_OK = 0
EXIT_TESTSFAILED = 1
@ -36,27 +33,25 @@ EXIT_INTERRUPTED = 2
EXIT_INTERNALERROR = 3
EXIT_NOHOSTS = 4
# imports used for genitems()
Item = py.test.collect.Item
Collector = py.test.collect.Collector
class Session(object):
nodeid = ""
class Interrupted(KeyboardInterrupt):
""" signals an interrupted test run. """
__module__ = 'builtins' # for py3
def __init__(self, config, collection):
def __init__(self, config):
self.config = config
self.pluginmanager = config.pluginmanager # shortcut
self.pluginmanager.register(self)
self.config.pluginmanager.register(self, name="session", prepend=True)
self._testsfailed = 0
self.shouldstop = False
self.collection = collection
self.collection = Collection(config)
def sessionstarts(self):
""" setup any neccessary resources ahead of the test run. """
self.config.hook.pytest_sessionstart(session=self)
def sessionfinishes(self, exitstatus):
# XXX move to main loop / refactor mainloop
self.config.hook.pytest_sessionfinish(
session=self,
exitstatus=exitstatus,
)
def pytest_runtest_logreport(self, report):
if report.failed:
@ -68,24 +63,22 @@ class Session(object):
self.collection.shouldstop = self.shouldstop
pytest_collectreport = pytest_runtest_logreport
def sessionfinishes(self, exitstatus):
""" teardown any resources after a test run. """
self.config.hook.pytest_sessionfinish(
session=self,
exitstatus=exitstatus,
)
def main(self):
""" main loop for running tests. """
self.shouldstop = False
self.sessionstarts()
exitstatus = EXIT_OK
config = self.config
try:
self._mainloop()
config.pluginmanager.do_configure(config)
config.hook.pytest_sessionstart(session=self)
config.hook.pytest_perform_collection(session=self)
config.hook.pytest_runtest_mainloop(session=self)
if self._testsfailed:
exitstatus = EXIT_TESTSFAILED
self.sessionfinishes(exitstatus=exitstatus)
config.pluginmanager.do_unconfigure(config)
except self.config.Error:
raise
except KeyboardInterrupt:
excinfo = py.code.ExceptionInfo()
self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo)
@ -94,19 +87,12 @@ class Session(object):
excinfo = py.code.ExceptionInfo()
self.config.pluginmanager.notify_exception(excinfo)
exitstatus = EXIT_INTERNALERROR
if excinfo.errisinstance(SystemExit):
sys.stderr.write("mainloop: caught Spurious SystemExit!\n")
if exitstatus in (EXIT_INTERNALERROR, EXIT_INTERRUPTED):
self.sessionfinishes(exitstatus=exitstatus)
return exitstatus
def _mainloop(self):
if self.config.option.collectonly:
return
for item in self.collection.items:
item.config.hook.pytest_runtest_protocol(item=item)
if self.shouldstop:
raise self.Interrupted(self.shouldstop)
class Collection:
def __init__(self, config):
self.config = config
@ -121,13 +107,15 @@ class Collection:
def _normalizearg(self, arg):
return "::".join(self._parsearg(arg))
def _parsearg(self, arg):
def _parsearg(self, arg, base=None):
""" return normalized name list for a command line specified id
which might be of the form x/y/z::name1::name2
and should result into the form x::y::z::name1::name2
"""
if base is None:
base = py.path.local()
parts = str(arg).split("::")
path = py.path.local(parts[0])
path = base.join(parts[0], abs=True)
if not path.check():
raise self.config.Error("file not found: %s" %(path,))
topdir = self.topdir
@ -137,17 +125,21 @@ class Collection:
topparts = path.relto(topdir).split(path.sep)
return topparts + parts[1:]
def getid(self, node, relative=True):
def getid(self, node):
""" return id for node, relative to topdir. """
path = node.fspath
chain = [x for x in node.listchain() if x.fspath == path]
chain = chain[1:]
names = [x.name for x in chain if x.name != "()"]
if relative:
relpath = path.relto(self.topdir)
if relpath:
path = relpath
names = relpath.split(node.fspath.sep) + names
relpath = path.relto(self.topdir)
if not relpath:
assert path == self.topdir
path = ''
else:
path = relpath
if os.sep != "/":
path = str(path).replace(os.sep, "/")
names.insert(0, path)
return "::".join(names)
def getbyid(self, id):
@ -158,6 +150,9 @@ class Collection:
names = id.split("::")
while names:
name = names.pop(0)
newnames = name.split("/")
name = newnames[0]
names[:0] = newnames[1:]
l = []
for current in matching:
for x in current._memocollect():
@ -172,22 +167,6 @@ class Collection:
matching = l
return matching
def do_collection(self):
assert not hasattr(self, 'items')
hook = self.config.hook
hook.pytest_log_startcollection(collection=self)
try:
self.items = self.perform_collect()
except self.config.Error:
raise
except Exception:
self.config.pluginmanager.notify_exception()
return EXIT_INTERNALERROR
else:
hook.pytest_collection_modifyitems(collection=self)
res = hook.pytest_log_finishcollection(collection=self)
return res and max(res) or 0 # returncode
def getinitialnodes(self):
idlist = [self._normalizearg(arg) for arg in self.config.args]
nodes = []
@ -206,36 +185,36 @@ class Collection:
names = list(names)
name = names and names.pop(0) or None
for node in matching:
if isinstance(node, Item):
if isinstance(node, py.test.collect.Item):
if name is None:
self.config.hook.pytest_log_itemcollect(item=node)
result.append(node)
else:
assert isinstance(node, Collector)
node.ihook.pytest_collectstart(collector=node)
rep = node.ihook.pytest_make_collect_report(collector=node)
#print "matching", rep.result, "against name", name
if rep.passed:
if name:
matched = False
for subcol in rep.result:
if subcol.name != name and subcol.name == "()":
names.insert(0, name)
name = "()"
# see doctests/custom naming XXX
if subcol.name == name or subcol.fspath.basename == name:
self.genitems([subcol], names, result)
matched = True
if not matched:
raise self.config.Error(
"can't collect: %s" % (name,))
continue
assert isinstance(node, py.test.collect.Collector)
node.ihook.pytest_collectstart(collector=node)
rep = node.ihook.pytest_make_collect_report(collector=node)
#print "matching", rep.result, "against name", name
if rep.passed:
if name:
matched = False
for subcol in rep.result:
if subcol.name != name and subcol.name == "()":
names.insert(0, name)
name = "()"
# see doctests/custom naming XXX
if subcol.name == name or subcol.fspath.basename == name:
self.genitems([subcol], names, result)
matched = True
if not matched:
raise self.config.Error(
"can't collect: %s" % (name,))
else:
self.genitems(rep.result, [], result)
node.ihook.pytest_collectreport(report=rep)
x = getattr(self, 'shouldstop', None)
if x:
raise self.Interrupted(x)
else:
self.genitems(rep.result, [], result)
node.ihook.pytest_collectreport(report=rep)
x = getattr(self, 'shouldstop', None)
if x:
raise Session.Interrupted(x)
def gettopdir(args):
""" return the top directory for the given paths.

2
testing/plugin/test_pytest_genscript.py
View File

@ -26,7 +26,7 @@ def test_gen(testdir, anypython, standalone):
"*imported from*mypytest"
])
@py.test.mark.xfail(reason="fix-dist")
@py.test.mark.xfail(reason="fix-dist", run=False)
def test_rundist(testdir, pytestconfig, standalone):
pytestconfig.pluginmanager.skipifmissing("xdist")
testdir.makepyfile("""

2
testing/test_collection.py
View File

@ -145,7 +145,7 @@ class TestCollection:
bbb = testdir.mkpydir("bbb")
p.copy(aaa.join("test_aaa.py"))
p.move(bbb.join("test_bbb.py"))
id = "."
config = testdir.parseconfig(id)
rcol = Collection(config)

13
testing/test_pluginmanager.py
View File

@ -160,6 +160,19 @@ class TestBootstrapping:
pp.unregister(a2)
assert not pp.isregistered(a2)
def test_registry_ordering(self):
pp = PluginManager()
class A: pass
a1, a2 = A(), A()
pp.register(a1)
pp.register(a2, "hello")
l = pp.getplugins()
assert l.index(a1) < l.index(a2)
a3 = A()
pp.register(a3, prepend=True)
l = pp.getplugins()
assert l.index(a3) == 0
def test_register_imported_modules(self):
pp = PluginManager()
mod = py.std.types.ModuleType("x.y.pytest_hello")