test_ok2/_pytest/core.py

476 lines
16 KiB
Python

"""
pytest PluginManager, basic initialization and tracing.
(c) Holger Krekel 2004-2010
"""
import sys, os
import inspect
import py
from _pytest import hookspec # the extension point definitions
assert py.__version__.split(".")[:2] >= ['1', '4'], ("installation problem: "
"%s is too old, remove or upgrade 'py'" % (py.__version__))
default_plugins = (
"config mark main terminal runner python pdb unittest capture skipping "
"tmpdir monkeypatch recwarn pastebin helpconfig nose assertion genscript "
"junitxml resultlog doctest").split()
class TagTracer:
def __init__(self):
self._tag2proc = {}
self.writer = None
self.indent = 0
def get(self, name):
return TagTracerSub(self, (name,))
def processmessage(self, tags, args):
if self.writer is not None:
if args:
indent = " " * self.indent
content = " ".join(map(str, args))
self.writer("%s%s [%s]\n" %(indent, content, ":".join(tags)))
try:
self._tag2proc[tags](tags, args)
except KeyError:
pass
def setwriter(self, writer):
self.writer = writer
def setprocessor(self, tags, processor):
if isinstance(tags, str):
tags = tuple(tags.split(":"))
else:
assert isinstance(tags, tuple)
self._tag2proc[tags] = processor
class TagTracerSub:
def __init__(self, root, tags):
self.root = root
self.tags = tags
def __call__(self, *args):
self.root.processmessage(self.tags, args)
def setmyprocessor(self, processor):
self.root.setprocessor(self.tags, processor)
def get(self, name):
return self.__class__(self.root, self.tags + (name,))
class PluginManager(object):
def __init__(self, load=False):
self._name2plugin = {}
self._listattrcache = {}
self._plugins = []
self._hints = []
self.trace = TagTracer().get("pluginmanage")
self._plugin_distinfo = []
if os.environ.get('PYTEST_DEBUG'):
err = sys.stderr
encoding = getattr(err, 'encoding', 'utf8')
try:
err = py.io.dupfile(err, encoding=encoding)
except Exception:
pass
self.trace.root.setwriter(err.write)
self.hook = HookRelay([hookspec], pm=self)
self.register(self)
if load:
for spec in default_plugins:
self.import_plugin(spec)
def register(self, plugin, name=None, prepend=False):
if self._name2plugin.get(name, None) == -1:
return
name = name or getattr(plugin, '__name__', str(id(plugin)))
if self.isregistered(plugin, name):
raise ValueError("Plugin already registered: %s=%s" %(name, plugin))
#self.trace("registering", name, plugin)
self._name2plugin[name] = plugin
self.call_plugin(plugin, "pytest_addhooks", {'pluginmanager': self})
self.hook.pytest_plugin_registered(manager=self, plugin=plugin)
if not prepend:
self._plugins.append(plugin)
else:
self._plugins.insert(0, plugin)
return True
def unregister(self, plugin=None, name=None):
if plugin is None:
plugin = self.getplugin(name=name)
self._plugins.remove(plugin)
self.hook.pytest_plugin_unregistered(plugin=plugin)
for name, value in list(self._name2plugin.items()):
if value == plugin:
del self._name2plugin[name]
def isregistered(self, plugin, name=None):
if self.getplugin(name) is not None:
return True
for val in self._name2plugin.values():
if plugin == val:
return True
def addhooks(self, spec):
self.hook._addhooks(spec, prefix="pytest_")
def getplugins(self):
return list(self._plugins)
def skipifmissing(self, name):
if not self.hasplugin(name):
py.test.skip("plugin %r is missing" % name)
def hasplugin(self, name):
return bool(self.getplugin(name))
def getplugin(self, name):
if name is None:
return None
try:
return self._name2plugin[name]
except KeyError:
return self._name2plugin.get("_pytest." + name, None)
# API for bootstrapping
#
def _envlist(self, varname):
val = py.std.os.environ.get(varname, None)
if val is not None:
return val.split(',')
return ()
def consider_env(self):
for spec in self._envlist("PYTEST_PLUGINS"):
self.import_plugin(spec)
def consider_setuptools_entrypoints(self):
try:
from pkg_resources import iter_entry_points, DistributionNotFound
except ImportError:
return # XXX issue a warning
for ep in iter_entry_points('pytest11'):
name = ep.name
if name.startswith("pytest_"):
name = name[7:]
if ep.name in self._name2plugin or name in self._name2plugin:
continue
try:
plugin = ep.load()
except DistributionNotFound:
continue
self._plugin_distinfo.append((ep.dist, plugin))
self.register(plugin, name=name)
def consider_preparse(self, args):
for opt1,opt2 in zip(args, args[1:]):
if opt1 == "-p":
self.consider_pluginarg(opt2)
def consider_pluginarg(self, arg):
if arg.startswith("no:"):
name = arg[3:]
if self.getplugin(name) is not None:
self.unregister(None, name=name)
self._name2plugin[name] = -1
else:
if self.getplugin(arg) is None:
self.import_plugin(arg)
def consider_conftest(self, conftestmodule):
if self.register(conftestmodule, name=conftestmodule.__file__):
self.consider_module(conftestmodule)
def consider_module(self, mod):
attr = getattr(mod, "pytest_plugins", ())
if attr:
if not isinstance(attr, (list, tuple)):
attr = (attr,)
for spec in attr:
self.import_plugin(spec)
def import_plugin(self, modname):
assert isinstance(modname, str)
if self.getplugin(modname) is not None:
return
try:
#self.trace("importing", modname)
mod = importplugin(modname)
except KeyboardInterrupt:
raise
except ImportError:
if modname.startswith("pytest_"):
return self.import_plugin(modname[7:])
raise
except:
e = py.std.sys.exc_info()[1]
if not hasattr(py.test, 'skip'):
raise
elif not isinstance(e, py.test.skip.Exception):
raise
self._hints.append("skipped plugin %r: %s" %((modname, e.msg)))
else:
self.register(mod, modname)
self.consider_module(mod)
def pytest_configure(self, config):
config.addinivalue_line("markers",
"tryfirst: mark a hook implementation function such that the "
"plugin machinery will try to call it first/as early as possible.")
config.addinivalue_line("markers",
"trylast: mark a hook implementation function such that the "
"plugin machinery will try to call it last/as late as possible.")
def pytest_plugin_registered(self, plugin):
import pytest
dic = self.call_plugin(plugin, "pytest_namespace", {}) or {}
if dic:
self._setns(pytest, dic)
if hasattr(self, '_config'):
self.call_plugin(plugin, "pytest_addoption",
{'parser': self._config._parser})
self.call_plugin(plugin, "pytest_configure",
{'config': self._config})
def _setns(self, obj, dic):
import pytest
for name, value in dic.items():
if isinstance(value, dict):
mod = getattr(obj, name, None)
if mod is None:
modname = "pytest.%s" % name
mod = py.std.types.ModuleType(modname)
sys.modules[modname] = mod
mod.__all__ = []
setattr(obj, name, mod)
obj.__all__.append(name)
self._setns(mod, value)
else:
setattr(obj, name, value)
obj.__all__.append(name)
#if obj != pytest:
# pytest.__all__.append(name)
setattr(pytest, name, value)
def pytest_terminal_summary(self, terminalreporter):
tw = terminalreporter._tw
if terminalreporter.config.option.traceconfig:
for hint in self._hints:
tw.line("hint: %s" % hint)
def do_addoption(self, parser):
mname = "pytest_addoption"
methods = reversed(self.listattr(mname))
MultiCall(methods, {'parser': parser}).execute()
def do_configure(self, config):
assert not hasattr(self, '_config')
self._config = config
config.hook.pytest_configure(config=self._config)
def do_unconfigure(self, config):
config = self._config
del self._config
config.hook.pytest_unconfigure(config=config)
config.pluginmanager.unregister(self)
def notify_exception(self, excinfo, option=None):
if option and option.fulltrace:
style = "long"
else:
style = "native"
excrepr = excinfo.getrepr(funcargs=True,
showlocals=getattr(option, 'showlocals', False),
style=style,
)
res = self.hook.pytest_internalerror(excrepr=excrepr)
if not py.builtin.any(res):
for line in str(excrepr).split("\n"):
sys.stderr.write("INTERNALERROR> %s\n" %line)
sys.stderr.flush()
def listattr(self, attrname, plugins=None):
if plugins is None:
plugins = self._plugins
key = (attrname,) + tuple(plugins)
try:
return list(self._listattrcache[key])
except KeyError:
pass
l = []
last = []
for plugin in plugins:
try:
meth = getattr(plugin, attrname)
if hasattr(meth, 'tryfirst'):
last.append(meth)
elif hasattr(meth, 'trylast'):
l.insert(0, meth)
else:
l.append(meth)
except AttributeError:
continue
l.extend(last)
self._listattrcache[key] = list(l)
return l
def call_plugin(self, plugin, methname, kwargs):
return MultiCall(methods=self.listattr(methname, plugins=[plugin]),
kwargs=kwargs, firstresult=True).execute()
def importplugin(importspec):
name = importspec
try:
mod = "_pytest." + name
__import__(mod)
return sys.modules[mod]
except ImportError:
#e = py.std.sys.exc_info()[1]
#if str(e).find(name) == -1:
# raise
pass #
__import__(importspec)
return sys.modules[importspec]
class MultiCall:
""" execute a call into multiple python functions/methods. """
def __init__(self, methods, kwargs, firstresult=False):
self.methods = list(methods)
self.kwargs = kwargs
self.results = []
self.firstresult = firstresult
def __repr__(self):
status = "%d results, %d meths" % (len(self.results), len(self.methods))
return "<MultiCall %s, kwargs=%r>" %(status, self.kwargs)
def execute(self):
while self.methods:
method = self.methods.pop()
kwargs = self.getkwargs(method)
res = method(**kwargs)
if res is not None:
self.results.append(res)
if self.firstresult:
return res
if not self.firstresult:
return self.results
def getkwargs(self, method):
kwargs = {}
for argname in varnames(method):
try:
kwargs[argname] = self.kwargs[argname]
except KeyError:
if argname == "__multicall__":
kwargs[argname] = self
return kwargs
def varnames(func):
try:
return func._varnames
except AttributeError:
pass
if not inspect.isfunction(func) and not inspect.ismethod(func):
func = getattr(func, '__call__', func)
ismethod = inspect.ismethod(func)
rawcode = py.code.getrawcode(func)
try:
x = rawcode.co_varnames[ismethod:rawcode.co_argcount]
except AttributeError:
x = ()
py.builtin._getfuncdict(func)['_varnames'] = x
return x
class HookRelay:
def __init__(self, hookspecs, pm, prefix="pytest_"):
if not isinstance(hookspecs, list):
hookspecs = [hookspecs]
self._hookspecs = []
self._pm = pm
self.trace = pm.trace.root.get("hook")
for hookspec in hookspecs:
self._addhooks(hookspec, prefix)
def _addhooks(self, hookspecs, prefix):
self._hookspecs.append(hookspecs)
added = False
for name, method in vars(hookspecs).items():
if name.startswith(prefix):
firstresult = getattr(method, 'firstresult', False)
hc = HookCaller(self, name, firstresult=firstresult)
setattr(self, name, hc)
added = True
#print ("setting new hook", name)
if not added:
raise ValueError("did not find new %r hooks in %r" %(
prefix, hookspecs,))
class HookCaller:
def __init__(self, hookrelay, name, firstresult):
self.hookrelay = hookrelay
self.name = name
self.firstresult = firstresult
self.trace = self.hookrelay.trace
def __repr__(self):
return "<HookCaller %r>" %(self.name,)
def __call__(self, **kwargs):
methods = self.hookrelay._pm.listattr(self.name)
return self._docall(methods, kwargs)
def pcall(self, plugins, **kwargs):
methods = self.hookrelay._pm.listattr(self.name, plugins=plugins)
return self._docall(methods, kwargs)
def _docall(self, methods, kwargs):
self.trace(self.name, kwargs)
self.trace.root.indent += 1
mc = MultiCall(methods, kwargs, firstresult=self.firstresult)
try:
res = mc.execute()
if res:
self.trace("finish", self.name, "-->", res)
finally:
self.trace.root.indent -= 1
return res
_preinit = []
def _preloadplugins():
_preinit.append(PluginManager(load=True))
def _prepareconfig(args=None, plugins=None):
if args is None:
args = sys.argv[1:]
elif isinstance(args, py.path.local):
args = [str(args)]
elif not isinstance(args, (tuple, list)):
if not isinstance(args, str):
raise ValueError("not a string or argument list: %r" % (args,))
args = py.std.shlex.split(args)
if _preinit:
_pluginmanager = _preinit.pop(0)
else: # subsequent calls to main will create a fresh instance
_pluginmanager = PluginManager(load=True)
hook = _pluginmanager.hook
if plugins:
for plugin in plugins:
_pluginmanager.register(plugin)
return hook.pytest_cmdline_parse(
pluginmanager=_pluginmanager, args=args)
def main(args=None, plugins=None):
""" returned exit code integer, after an in-process testing run
with the given command line arguments, preloading an optional list
of passed in plugin objects. """
config = _prepareconfig(args, plugins)
exitstatus = config.hook.pytest_cmdline_main(config=config)
return exitstatus
class UsageError(Exception):
""" error in py.test usage or invocation"""