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

Merged in hpk42/pytest-patches/plugin_no_pytest (pull request #278) 

Refactor pluginmanagement
This commit is contained in:
holger krekel 2015-04-25 09:08:21 +02:00
parent 2b80caf1af 9020bf48b7
commit d8e91d9fee
20 changed files with 676 additions and 698 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGELOG
View File

@ -18,6 +18,15 @@
from ``inline_run()`` to allow temporary modules to be reloaded.
Thanks Eduardo Schettino.
- internally refactor pluginmanager API and code so that there
is a clear distinction between a pytest-agnostic rather simple
pluginmanager and the PytestPluginManager which adds a lot of
behaviour, among it handling of the local conftest files.
In terms of documented methods this is a backward compatible
change but it might still break 3rd party plugins which relied on
details like especially the pluginmanager.add_shutdown() API.
Thanks Holger Krekel.
2.7.1.dev (compared to 2.7.0)
-----------------------------

11
_pytest/assertion/__init__.py
View File

@ -70,12 +70,11 @@ def pytest_configure(config):
config._assertstate = AssertionState(config, mode)
config._assertstate.hook = hook
config._assertstate.trace("configured with mode set to %r" % (mode,))
def pytest_unconfigure(config):
hook = config._assertstate.hook
if hook is not None and hook in sys.meta_path:
sys.meta_path.remove(hook)
def undo():
hook = config._assertstate.hook
if hook is not None and hook in sys.meta_path:
sys.meta_path.remove(hook)
config.add_cleanup(undo)
def pytest_collection(session):

4
_pytest/capture.py
View File

@ -37,13 +37,13 @@ def pytest_load_initial_conftests(early_config, parser, args):
pluginmanager.register(capman, "capturemanager")
# make sure that capturemanager is properly reset at final shutdown
pluginmanager.add_shutdown(capman.reset_capturings)
early_config.add_cleanup(capman.reset_capturings)
# make sure logging does not raise exceptions at the end
def silence_logging_at_shutdown():
if "logging" in sys.modules:
sys.modules["logging"].raiseExceptions = False
pluginmanager.add_shutdown(silence_logging_at_shutdown)
early_config.add_cleanup(silence_logging_at_shutdown)
# finally trigger conftest loading but while capturing (issue93)
capman.init_capturings()

352
_pytest/config.py
View File

@ -53,6 +53,10 @@ default_plugins = (
"tmpdir monkeypatch recwarn pastebin helpconfig nose assertion genscript "
"junitxml resultlog doctest").split()
builtin_plugins = set(default_plugins)
builtin_plugins.add("pytester")
def _preloadplugins():
assert not _preinit
_preinit.append(get_plugin_manager())
@ -77,19 +81,31 @@ def _prepareconfig(args=None, plugins=None):
raise ValueError("not a string or argument list: %r" % (args,))
args = shlex.split(args)
pluginmanager = get_plugin_manager()
try:
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
except Exception:
pluginmanager.ensure_shutdown()
raise
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
def exclude_pytest_names(name):
return not name.startswith(name) or name == "pytest_plugins" or \
name.startswith("pytest_funcarg__")
class PytestPluginManager(PluginManager):
def __init__(self, hookspecs=[hookspec]):
super(PytestPluginManager, self).__init__(hookspecs=hookspecs)
def __init__(self):
super(PytestPluginManager, self).__init__(prefix="pytest_",
excludefunc=exclude_pytest_names)
self._warnings = []
self._plugin_distinfo = []
self._globalplugins = []
# state related to local conftest plugins
self._path2confmods = {}
self._conftestpath2mod = {}
self._confcutdir = None
self.addhooks(hookspec)
self.register(self)
if os.environ.get('PYTEST_DEBUG'):
err = sys.stderr
@ -100,6 +116,25 @@ class PytestPluginManager(PluginManager):
pass
self.set_tracing(err.write)
def register(self, plugin, name=None, conftest=False):
ret = super(PytestPluginManager, self).register(plugin, name)
if ret and not conftest:
self._globalplugins.append(plugin)
return ret
def _do_register(self, plugin, name):
# called from core PluginManager class
if hasattr(self, "config"):
self.config._register_plugin(plugin, name)
return super(PytestPluginManager, self)._do_register(plugin, name)
def unregister(self, plugin):
super(PytestPluginManager, self).unregister(plugin)
try:
self._globalplugins.remove(plugin)
except ValueError:
pass
def pytest_configure(self, config):
config.addinivalue_line("markers",
"tryfirst: mark a hook implementation function such that the "
@ -110,6 +145,172 @@ class PytestPluginManager(PluginManager):
for warning in self._warnings:
config.warn(code="I1", message=warning)
#
# internal API for local conftest plugin handling
#
def _set_initial_conftests(self, namespace):
""" load initial conftest files given a preparsed "namespace".
As conftest files may add their own command line options
which have arguments ('--my-opt somepath') we might get some
false positives. All builtin and 3rd party plugins will have
been loaded, however, so common options will not confuse our logic
here.
"""
current = py.path.local()
self._confcutdir = current.join(namespace.confcutdir, abs=True) \
if namespace.confcutdir else None
testpaths = namespace.file_or_dir
foundanchor = False
for path in testpaths:
path = str(path)
# remove node-id syntax
i = path.find("::")
if i != -1:
path = path[:i]
anchor = current.join(path, abs=1)
if exists(anchor): # we found some file object
self._try_load_conftest(anchor)
foundanchor = True
if not foundanchor:
self._try_load_conftest(current)
def _try_load_conftest(self, anchor):
self._getconftestmodules(anchor)
# let's also consider test* subdirs
if anchor.check(dir=1):
for x in anchor.listdir("test*"):
if x.check(dir=1):
self._getconftestmodules(x)
def _getconftestmodules(self, path):
try:
return self._path2confmods[path]
except KeyError:
clist = []
for parent in path.parts():
if self._confcutdir and self._confcutdir.relto(parent):
continue
conftestpath = parent.join("conftest.py")
if conftestpath.check(file=1):
mod = self._importconftest(conftestpath)
clist.append(mod)
self._path2confmods[path] = clist
return clist
def _rget_with_confmod(self, name, path):
modules = self._getconftestmodules(path)
for mod in reversed(modules):
try:
return mod, getattr(mod, name)
except AttributeError:
continue
raise KeyError(name)
def _importconftest(self, conftestpath):
try:
return self._conftestpath2mod[conftestpath]
except KeyError:
pkgpath = conftestpath.pypkgpath()
if pkgpath is None:
_ensure_removed_sysmodule(conftestpath.purebasename)
try:
mod = conftestpath.pyimport()
except Exception:
raise ConftestImportFailure(conftestpath, sys.exc_info())
self._conftestpath2mod[conftestpath] = mod
dirpath = conftestpath.dirpath()
if dirpath in self._path2confmods:
for path, mods in self._path2confmods.items():
if path and path.relto(dirpath) or path == dirpath:
assert mod not in mods
mods.append(mod)
self.trace("loaded conftestmodule %r" %(mod))
self.consider_conftest(mod)
return mod
#
# API for bootstrapping plugin loading
#
#
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:]
plugin = self.getplugin(name)
if plugin is not None:
self.unregister(plugin)
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__,
conftest=True):
self.consider_module(conftestmodule)
def consider_env(self):
self._import_plugin_specs(os.environ.get("PYTEST_PLUGINS"))
def consider_module(self, mod):
self._import_plugin_specs(getattr(mod, "pytest_plugins", None))
def _import_plugin_specs(self, spec):
if spec:
if isinstance(spec, str):
spec = spec.split(",")
for import_spec in spec:
self.import_plugin(import_spec)
def import_plugin(self, modname):
# most often modname refers to builtin modules, e.g. "pytester",
# "terminal" or "capture". Those plugins are registered under their
# basename for historic purposes but must be imported with the
# _pytest prefix.
assert isinstance(modname, str)
if self.getplugin(modname) is not None:
return
if modname in builtin_plugins:
importspec = "_pytest." + modname
else:
importspec = modname
try:
__import__(importspec)
except ImportError:
raise
except Exception as e:
import pytest
if not hasattr(pytest, 'skip') or not isinstance(e, pytest.skip.Exception):
raise
self._warnings.append("skipped plugin %r: %s" %((modname, e.msg)))
else:
mod = sys.modules[importspec]
self.register(mod, modname)
self.consider_module(mod)
class Parser:
""" Parser for command line arguments and ini-file values. """
@ -464,96 +665,6 @@ class DropShorterLongHelpFormatter(argparse.HelpFormatter):
return action._formatted_action_invocation
class Conftest(object):
""" the single place for accessing values and interacting
towards conftest modules from pytest objects.
"""
def __init__(self, onimport=None):
self._path2confmods = {}
self._onimport = onimport
self._conftestpath2mod = {}
self._confcutdir = None
def setinitial(self, namespace):
""" load initial conftest files given a preparsed "namespace".
As conftest files may add their own command line options
which have arguments ('--my-opt somepath') we might get some
false positives. All builtin and 3rd party plugins will have
been loaded, however, so common options will not confuse our logic
here.
"""
current = py.path.local()
self._confcutdir = current.join(namespace.confcutdir, abs=True) \
if namespace.confcutdir else None
testpaths = namespace.file_or_dir
foundanchor = False
for path in testpaths:
path = str(path)
# remove node-id syntax
i = path.find("::")
if i != -1:
path = path[:i]
anchor = current.join(path, abs=1)
if exists(anchor): # we found some file object
self._try_load_conftest(anchor)
foundanchor = True
if not foundanchor:
self._try_load_conftest(current)
def _try_load_conftest(self, anchor):
self.getconftestmodules(anchor)
# let's also consider test* subdirs
if anchor.check(dir=1):
for x in anchor.listdir("test*"):
if x.check(dir=1):
self.getconftestmodules(x)
def getconftestmodules(self, path):
try:
return self._path2confmods[path]
except KeyError:
clist = []
for parent in path.parts():
if self._confcutdir and self._confcutdir.relto(parent):
continue
conftestpath = parent.join("conftest.py")
if conftestpath.check(file=1):
mod = self.importconftest(conftestpath)
clist.append(mod)
self._path2confmods[path] = clist
return clist
def rget_with_confmod(self, name, path):
modules = self.getconftestmodules(path)
for mod in reversed(modules):
try:
return mod, getattr(mod, name)
except AttributeError:
continue
raise KeyError(name)
def importconftest(self, conftestpath):
try:
return self._conftestpath2mod[conftestpath]
except KeyError:
pkgpath = conftestpath.pypkgpath()
if pkgpath is None:
_ensure_removed_sysmodule(conftestpath.purebasename)
try:
mod = conftestpath.pyimport()
except Exception:
raise ConftestImportFailure(conftestpath, sys.exc_info())
self._conftestpath2mod[conftestpath] = mod
dirpath = conftestpath.dirpath()
if dirpath in self._path2confmods:
for path, mods in self._path2confmods.items():
if path and path.relto(dirpath) or path == dirpath:
assert mod not in mods
mods.append(mod)
if self._onimport:
self._onimport(mod)
return mod
def _ensure_removed_sysmodule(modname):
try:
@ -589,13 +700,11 @@ class Config(object):
#: a pluginmanager instance
self.pluginmanager = pluginmanager
self.trace = self.pluginmanager.trace.root.get("config")
self._conftest = Conftest(onimport=self._onimportconftest)
self.hook = self.pluginmanager.hook
self._inicache = {}
self._opt2dest = {}
self._cleanup = []
self.pluginmanager.register(self, "pytestconfig")
self.pluginmanager.set_register_callback(self._register_plugin)
self._configured = False
def _register_plugin(self, plugin, name):
@ -612,16 +721,23 @@ class Config(object):
if self._configured:
call_plugin(plugin, "pytest_configure", {'config': self})
def do_configure(self):
def add_cleanup(self, func):
""" Add a function to be called when the config object gets out of
use (usually coninciding with pytest_unconfigure)."""
self._cleanup.append(func)
def _do_configure(self):
assert not self._configured
self._configured = True
self.hook.pytest_configure(config=self)
def do_unconfigure(self):
assert self._configured
self._configured = False
self.hook.pytest_unconfigure(config=self)
self.pluginmanager.ensure_shutdown()
def _ensure_unconfigure(self):
if self._configured:
self._configured = False
self.hook.pytest_unconfigure(config=self)
while self._cleanup:
fin = self._cleanup.pop()
fin()
def warn(self, code, message):
""" generate a warning for this test session. """
@ -636,11 +752,6 @@ class Config(object):
self.parse(args)
return self
def pytest_unconfigure(config):
while config._cleanup:
fin = config._cleanup.pop()
fin()
def notify_exception(self, excinfo, option=None):
if option and option.fulltrace:
style = "long"
@ -675,10 +786,6 @@ class Config(object):
config.pluginmanager.consider_pluginarg(x)
return config
def _onimportconftest(self, conftestmodule):
self.trace("loaded conftestmodule %r" %(conftestmodule,))
self.pluginmanager.consider_conftest(conftestmodule)
def _processopt(self, opt):
for name in opt._short_opts + opt._long_opts:
self._opt2dest[name] = opt.dest
@ -688,11 +795,11 @@ class Config(object):
setattr(self.option, opt.dest, opt.default)
def _getmatchingplugins(self, fspath):
return self.pluginmanager._plugins + \
self._conftest.getconftestmodules(fspath)
return self.pluginmanager._globalplugins + \
self.pluginmanager._getconftestmodules(fspath)
def pytest_load_initial_conftests(self, early_config):
self._conftest.setinitial(early_config.known_args_namespace)
self.pluginmanager._set_initial_conftests(early_config.known_args_namespace)
pytest_load_initial_conftests.trylast = True
def _initini(self, args):
@ -799,7 +906,7 @@ class Config(object):
def _getconftest_pathlist(self, name, path):
try:
mod, relroots = self._conftest.rget_with_confmod(name, path)
mod, relroots = self.pluginmanager._rget_with_confmod(name, path)
except KeyError:
return None
modpath = py.path.local(mod.__file__).dirpath()
@ -933,3 +1040,4 @@ def setns(obj, dic):
#if obj != pytest:
# pytest.__all__.append(name)
setattr(pytest, name, value)

350
_pytest/core.py
View File

@ -1,14 +1,9 @@
"""
pytest PluginManager, basic initialization and tracing.
PluginManager, basic initialization and tracing.
"""
import os
import sys
import inspect
import py
# don't import pytest to avoid circular imports
assert py.__version__.split(".")[:2] >= ['1', '4'], ("installation problem: "
"%s is too old, remove or upgrade 'py'" % (py.__version__))
py3 = sys.version_info > (3,0)
@ -139,202 +134,133 @@ class CallOutcome:
class PluginManager(object):
def __init__(self, hookspecs=None, prefix="pytest_"):
""" Core Pluginmanager class which manages registration
of plugin objects and 1:N hook calling.
You can register new hooks by calling ``addhooks(module_or_class)``.
You can register plugin objects (which contain hooks) by calling
``register(plugin)``. The Pluginmanager is initialized with a
prefix that is searched for in the names of the dict of registered
plugin objects. An optional excludefunc allows to blacklist names which
are not considered as hooks despite a matching prefix.
For debugging purposes you can call ``set_tracing(writer)``
which will subsequently send debug information to the specified
write function.
"""
def __init__(self, prefix, excludefunc=None):
self._prefix = prefix
self._excludefunc = excludefunc
self._name2plugin = {}
self._plugins = []
self._conftestplugins = []
self._plugin2hookcallers = {}
self._warnings = []
self.trace = TagTracer().get("pluginmanage")
self._plugin_distinfo = []
self._shutdown = []
self.hook = HookRelay(hookspecs or [], pm=self, prefix=prefix)
self.hook = HookRelay(pm=self)
def set_tracing(self, writer):
""" turn on tracing to the given writer method and
return an undo function. """
self.trace.root.setwriter(writer)
# reconfigure HookCalling to perform tracing
assert not hasattr(self, "_wrapping")
self._wrapping = True
hooktrace = self.hook.trace
def _docall(self, methods, kwargs):
trace = self.hookrelay.trace
trace.root.indent += 1
trace(self.name, kwargs)
hooktrace.root.indent += 1
hooktrace(self.name, kwargs)
box = yield
if box.excinfo is None:
trace("finish", self.name, "-->", box.result)
trace.root.indent -= 1
hooktrace("finish", self.name, "-->", box.result)
hooktrace.root.indent -= 1
undo = add_method_wrapper(HookCaller, _docall)
self.add_shutdown(undo)
return add_method_wrapper(HookCaller, _docall)
def do_configure(self, config):
# backward compatibility
config.do_configure()
def make_hook_caller(self, name, plugins):
caller = getattr(self.hook, name)
methods = self.listattr(name, plugins=plugins)
return HookCaller(caller.name, caller.firstresult,
argnames=caller.argnames, methods=methods)
def set_register_callback(self, callback):
assert not hasattr(self, "_registercallback")
self._registercallback = callback
def register(self, plugin, name=None, prepend=False, conftest=False):
def register(self, plugin, name=None):
""" Register a plugin with the given name and ensure that all its
hook implementations are integrated. If the name is not specified
we use the ``__name__`` attribute of the plugin object or, if that
doesn't exist, the id of the plugin. This method will raise a
ValueError if the eventual name is already registered. """
name = name or self._get_canonical_name(plugin)
if self._name2plugin.get(name, None) == -1:
return
name = name or getattr(plugin, '__name__', str(id(plugin)))
if self.isregistered(plugin, name):
if self.hasplugin(name):
raise ValueError("Plugin already registered: %s=%s\n%s" %(
name, plugin, self._name2plugin))
#self.trace("registering", name, plugin)
reg = getattr(self, "_registercallback", None)
if reg is not None:
reg(plugin, name) # may call addhooks
hookcallers = list(self.hook._scan_plugin(plugin))
# allow subclasses to intercept here by calling a helper
return self._do_register(plugin, name)
def _do_register(self, plugin, name):
hookcallers = list(self._scan_plugin(plugin))
self._plugin2hookcallers[plugin] = hookcallers
self._name2plugin[name] = plugin
if conftest:
self._conftestplugins.append(plugin)
else:
if not prepend:
self._plugins.append(plugin)
else:
self._plugins.insert(0, plugin)
# finally make sure that the methods of the new plugin take part
self._plugins.append(plugin)
# rescan all methods for the hookcallers we found
for hookcaller in hookcallers:
hookcaller.scan_methods()
self._scan_methods(hookcaller)
return True
def unregister(self, plugin):
try:
self._plugins.remove(plugin)
except KeyError:
self._conftestplugins.remove(plugin)
""" unregister the plugin object and all its contained hook implementations
from internal data structures. """
self._plugins.remove(plugin)
for name, value in list(self._name2plugin.items()):
if value == plugin:
del self._name2plugin[name]
hookcallers = self._plugin2hookcallers.pop(plugin)
for hookcaller in hookcallers:
hookcaller.scan_methods()
self._scan_methods(hookcaller)
def add_shutdown(self, func):
self._shutdown.append(func)
def ensure_shutdown(self):
while self._shutdown:
func = self._shutdown.pop()
func()
self._plugins = self._conftestplugins = []
self._name2plugin.clear()
def isregistered(self, plugin, name=None):
if self.getplugin(name) is not None:
return True
return plugin in self._plugins or plugin in self._conftestplugins
def addhooks(self, spec, prefix="pytest_"):
self.hook._addhooks(spec, prefix=prefix)
def addhooks(self, module_or_class):
""" add new hook definitions from the given module_or_class using
the prefix/excludefunc with which the PluginManager was initialized. """
isclass = int(inspect.isclass(module_or_class))
names = []
for name in dir(module_or_class):
if name.startswith(self._prefix):
method = module_or_class.__dict__[name]
firstresult = getattr(method, 'firstresult', False)
hc = HookCaller(name, firstresult=firstresult,
argnames=varnames(method, startindex=isclass))
setattr(self.hook, name, hc)
names.append(name)
if not names:
raise ValueError("did not find new %r hooks in %r"
%(self._prefix, module_or_class))
def getplugins(self):
return self._plugins + self._conftestplugins
""" return the complete list of registered plugins. NOTE that
you will get the internal list and need to make a copy if you
modify the list."""
return self._plugins
def skipifmissing(self, name):
if not self.hasplugin(name):
import pytest
pytest.skip("plugin %r is missing" % name)
def isregistered(self, plugin):
""" Return True if the plugin is already registered under its
canonical name. """
return self.hasplugin(self._get_canonical_name(plugin)) or \
plugin in self._plugins
def hasplugin(self, name):
return bool(self.getplugin(name))
""" Return True if there is a registered with the given name. """
return name in self._name2plugin
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 = 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:]
plugin = self.getplugin(name)
if plugin is not None:
self.unregister(plugin)
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__,
conftest=True):
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:
mod = importplugin(modname)
except KeyboardInterrupt:
raise
except ImportError:
if modname.startswith("pytest_"):
return self.import_plugin(modname[7:])
raise
except:
e = sys.exc_info()[1]
import pytest
if not hasattr(pytest, 'skip') or not isinstance(e, pytest.skip.Exception):
raise
self._warnings.append("skipped plugin %r: %s" %((modname, e.msg)))
else:
self.register(mod, modname)
self.consider_module(mod)
""" Return a plugin or None for the given name. """
return self._name2plugin.get(name)
def listattr(self, attrname, plugins=None):
if plugins is None:
plugins = self._plugins + self._conftestplugins
plugins = self._plugins
l = []
last = []
wrappers = []
@ -355,20 +281,43 @@ class PluginManager(object):
l.extend(wrappers)
return l
def _scan_methods(self, hookcaller):
hookcaller.methods = self.listattr(hookcaller.name)
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:
__import__(importspec)
return sys.modules[importspec]
def _scan_plugin(self, plugin):
def fail(msg, *args):
name = getattr(plugin, '__name__', plugin)
raise PluginValidationError("plugin %r\n%s" %(name, msg % args))
for name in dir(plugin):
if name[0] == "_" or not name.startswith(self._prefix):
continue
hook = getattr(self.hook, name, None)
method = getattr(plugin, name)
if hook is None:
if self._excludefunc is not None and self._excludefunc(name):
continue
if getattr(method, 'optionalhook', False):
continue
fail("found unknown hook: %r", name)
for arg in varnames(method):
if arg not in hook.argnames:
fail("argument %r not available\n"
"actual definition: %s\n"
"available hookargs: %s",
arg, formatdef(method),
", ".join(hook.argnames))
yield hook
def _get_canonical_name(self, plugin):
return getattr(plugin, "__name__", None) or str(id(plugin))
class MultiCall:
""" execute a call into multiple python functions/methods. """
@ -441,65 +390,13 @@ def varnames(func, startindex=None):
class HookRelay:
def __init__(self, hookspecs, pm, prefix="pytest_"):
if not isinstance(hookspecs, list):
hookspecs = [hookspecs]
def __init__(self, pm):
self._pm = pm
self.trace = pm.trace.root.get("hook")
self.prefix = prefix
for hookspec in hookspecs:
self._addhooks(hookspec, prefix)
def _addhooks(self, hookspec, prefix):
added = False
isclass = int(inspect.isclass(hookspec))
for name, method in vars(hookspec).items():
if name.startswith(prefix):
firstresult = getattr(method, 'firstresult', False)
hc = HookCaller(self, name, firstresult=firstresult,
argnames=varnames(method, startindex=isclass))
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, hookspec,))
def _getcaller(self, name, plugins):
caller = getattr(self, name)
methods = self._pm.listattr(name, plugins=plugins)
if methods:
return caller.new_cached_caller(methods)
return caller
def _scan_plugin(self, plugin):
def fail(msg, *args):
name = getattr(plugin, '__name__', plugin)
raise PluginValidationError("plugin %r\n%s" %(name, msg % args))
for name in dir(plugin):
if not name.startswith(self.prefix):
continue
hook = getattr(self, name, None)
method = getattr(plugin, name)
if hook is None:
is_optional = getattr(method, 'optionalhook', False)
if not isgenerichook(name) and not is_optional:
fail("found unknown hook: %r", name)
continue
for arg in varnames(method):
if arg not in hook.argnames:
fail("argument %r not available\n"
"actual definition: %s\n"
"available hookargs: %s",
arg, formatdef(method),
", ".join(hook.argnames))
yield hook
class HookCaller:
def __init__(self, hookrelay, name, firstresult, argnames, methods=()):
self.hookrelay = hookrelay
def __init__(self, name, firstresult, argnames, methods=()):
self.name = name
self.firstresult = firstresult
self.argnames = ["__multicall__"]
@ -507,16 +404,9 @@ class HookCaller:
assert "self" not in argnames # sanity check
self.methods = methods
def new_cached_caller(self, methods):
return HookCaller(self.hookrelay, self.name, self.firstresult,
argnames=self.argnames, methods=methods)
def __repr__(self):
return "<HookCaller %r>" %(self.name,)
def scan_methods(self):
self.methods = self.hookrelay._pm.listattr(self.name)
def __call__(self, **kwargs):
return self._docall(self.methods, kwargs)
@ -531,13 +421,9 @@ class HookCaller:
class PluginValidationError(Exception):
""" plugin failed validation. """
def isgenerichook(name):
return name == "pytest_plugins" or \
name.startswith("pytest_funcarg__")
def formatdef(func):
return "%s%s" % (
func.__name__,
inspect.formatargspec(*inspect.getargspec(func))
)

2
_pytest/doctest.py
View File

@ -132,7 +132,7 @@ class DoctestModule(pytest.File):
def collect(self):
import doctest
if self.fspath.basename == "conftest.py":
module = self.config._conftest.importconftest(self.fspath)
module = self.config._conftest._importconftest(self.fspath)
else:
try:
module = self.fspath.pyimport()

26
_pytest/helpconfig.py
View File

@ -28,24 +28,20 @@ def pytest_cmdline_parse():
config = outcome.get_result()
if config.option.debug:
path = os.path.abspath("pytestdebug.log")
f = open(path, 'w')
config._debugfile = f
f.write("versions pytest-%s, py-%s, "
debugfile = open(path, 'w')
debugfile.write("versions pytest-%s, py-%s, "
"python-%s\ncwd=%s\nargs=%s\n\n" %(
pytest.__version__, py.__version__,
".".join(map(str, sys.version_info)),
os.getcwd(), config._origargs))
config.pluginmanager.set_tracing(f.write)
config.pluginmanager.set_tracing(debugfile.write)
sys.stderr.write("writing pytestdebug information to %s\n" % path)
@pytest.mark.trylast
def pytest_unconfigure(config):
if hasattr(config, '_debugfile'):
config._debugfile.close()
sys.stderr.write("wrote pytestdebug information to %s\n" %
config._debugfile.name)
config.trace.root.setwriter(None)
def unset_tracing():
debugfile.close()
sys.stderr.write("wrote pytestdebug information to %s\n" %
debugfile.name)
config.trace.root.setwriter(None)
config.add_cleanup(unset_tracing)
def pytest_cmdline_main(config):
if config.option.version:
@ -58,9 +54,9 @@ def pytest_cmdline_main(config):
sys.stderr.write(line + "\n")
return 0
elif config.option.help:
config.do_configure()
config._do_configure()
showhelp(config)
config.do_unconfigure()
config._ensure_unconfigure()
return 0
def showhelp(config):

2
_pytest/hookspec.py
View File

@ -6,7 +6,7 @@
def pytest_addhooks(pluginmanager):
"""called at plugin load time to allow adding new hooks via a call to
pluginmanager.registerhooks(module)."""
pluginmanager.addhooks(module_or_class, prefix)."""
def pytest_namespace():

12
_pytest/main.py
View File

@ -77,7 +77,7 @@ def wrap_session(config, doit):
initstate = 0
try:
try:
config.do_configure()
config._do_configure()
initstate = 1
config.hook.pytest_sessionstart(session=session)
initstate = 2
@ -107,9 +107,7 @@ def wrap_session(config, doit):
config.hook.pytest_sessionfinish(
session=session,
exitstatus=session.exitstatus)
if initstate >= 1:
config.do_unconfigure()
config.pluginmanager.ensure_shutdown()
config._ensure_unconfigure()
return session.exitstatus
def pytest_cmdline_main(config):
@ -160,7 +158,7 @@ class FSHookProxy(object):
def __getattr__(self, name):
plugins = self.config._getmatchingplugins(self.fspath)
x = self.config.hook._getcaller(name, plugins)
x = self.config.pluginmanager.make_hook_caller(name, plugins)
self.__dict__[name] = x
return x
@ -510,7 +508,7 @@ class Session(FSCollector):
def __init__(self, config):
FSCollector.__init__(self, config.rootdir, parent=None,
config=config, session=self)
self.config.pluginmanager.register(self, name="session", prepend=True)
self.config.pluginmanager.register(self, name="session")
self._testsfailed = 0
self.shouldstop = False
self.trace = config.trace.root.get("collection")
@ -521,10 +519,12 @@ class Session(FSCollector):
def _makeid(self):
return ""
@pytest.mark.tryfirst
def pytest_collectstart(self):
if self.shouldstop:
raise self.Interrupted(self.shouldstop)
@pytest.mark.tryfirst
def pytest_runtest_logreport(self, report):
if report.failed and not hasattr(report, 'wasxfail'):
self._testsfailed += 1

4
_pytest/mark.py
View File

@ -44,14 +44,14 @@ def pytest_addoption(parser):
def pytest_cmdline_main(config):
if config.option.markers:
config.do_configure()
config._do_configure()
tw = py.io.TerminalWriter()
for line in config.getini("markers"):
name, rest = line.split(":", 1)
tw.write("@pytest.mark.%s:" % name, bold=True)
tw.line(rest)
tw.line()
config.do_unconfigure()
config._ensure_unconfigure()
return 0
pytest_cmdline_main.tryfirst = True

14
_pytest/pytester.py
View File

@ -83,7 +83,8 @@ class HookRecorder:
self.calls.append(ParsedCall(hookcaller.name, kwargs))
yield
self._undo_wrapping = add_method_wrapper(HookCaller, _docall)
pluginmanager.add_shutdown(self._undo_wrapping)
#if hasattr(pluginmanager, "config"):
# pluginmanager.add_shutdown(self._undo_wrapping)
def finish_recording(self):
self._undo_wrapping()
@ -589,12 +590,7 @@ class TmpTestdir:
# we don't know what the test will do with this half-setup config
# object and thus we make sure it gets unconfigured properly in any
# case (otherwise capturing could still be active, for example)
def ensure_unconfigure():
if hasattr(config.pluginmanager, "_config"):
config.pluginmanager.do_unconfigure(config)
config.pluginmanager.ensure_shutdown()
self.request.addfinalizer(ensure_unconfigure)
self.request.addfinalizer(config._ensure_unconfigure)
return config
def parseconfigure(self, *args):
@ -606,8 +602,8 @@ class TmpTestdir:
"""
config = self.parseconfig(*args)
config.do_configure()
self.request.addfinalizer(config.do_unconfigure)
config._do_configure()
self.request.addfinalizer(config._ensure_unconfigure)
return config
def getitem(self, source, funcname="test_func"):

1
testing/conftest.py
View File

@ -66,6 +66,7 @@ def check_open_files(config):
error.append(error[0])
raise AssertionError("\n".join(error))
@pytest.mark.trylast
def pytest_runtest_teardown(item, __multicall__):
item.config._basedir.chdir()
if hasattr(item.config, '_openfiles'):

2
testing/python/fixture.py
View File

@ -1487,7 +1487,7 @@ class TestAutouseManagement:
reprec = testdir.inline_run("-v","-s")
reprec.assertoutcome(passed=8)
config = reprec.getcalls("pytest_unconfigure")[0].config
l = config._conftest.getconftestmodules(p)[0].l
l = config.pluginmanager._getconftestmodules(p)[0].l
assert l == ["fin_a1", "fin_a2", "fin_b1", "fin_b2"] * 2
def test_scope_ordering(self, testdir):

75
testing/test_conftest.py
View File

@ -1,7 +1,6 @@
from textwrap import dedent
import py, pytest
from _pytest.config import Conftest
from _pytest.config import PytestPluginManager
@pytest.fixture(scope="module", params=["global", "inpackage"])
@ -16,7 +15,7 @@ def basedir(request):
return tmpdir
def ConftestWithSetinitial(path):
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, [path])
return conftest
@ -25,51 +24,41 @@ def conftest_setinitial(conftest, args, confcutdir=None):
def __init__(self):
self.file_or_dir = args
self.confcutdir = str(confcutdir)
conftest.setinitial(Namespace())
conftest._set_initial_conftests(Namespace())
class TestConftestValueAccessGlobal:
def test_basic_init(self, basedir):
conftest = Conftest()
conftest = PytestPluginManager()
p = basedir.join("adir")
assert conftest.rget_with_confmod("a", p)[1] == 1
def test_onimport(self, basedir):
l = []
conftest = Conftest(onimport=l.append)
adir = basedir.join("adir")
conftest_setinitial(conftest, [adir], confcutdir=basedir)
assert len(l) == 1
assert conftest.rget_with_confmod("a", adir)[1] == 1
assert conftest.rget_with_confmod("b", adir.join("b"))[1] == 2
assert len(l) == 2
assert conftest._rget_with_confmod("a", p)[1] == 1
def test_immediate_initialiation_and_incremental_are_the_same(self, basedir):
conftest = Conftest()
conftest = PytestPluginManager()
len(conftest._path2confmods)
conftest.getconftestmodules(basedir)
conftest._getconftestmodules(basedir)
snap1 = len(conftest._path2confmods)
#assert len(conftest._path2confmods) == snap1 + 1
conftest.getconftestmodules(basedir.join('adir'))
conftest._getconftestmodules(basedir.join('adir'))
assert len(conftest._path2confmods) == snap1 + 1
conftest.getconftestmodules(basedir.join('b'))
conftest._getconftestmodules(basedir.join('b'))
assert len(conftest._path2confmods) == snap1 + 2
def test_value_access_not_existing(self, basedir):
conftest = ConftestWithSetinitial(basedir)
with pytest.raises(KeyError):
conftest.rget_with_confmod('a', basedir)
conftest._rget_with_confmod('a', basedir)
def test_value_access_by_path(self, basedir):
conftest = ConftestWithSetinitial(basedir)
adir = basedir.join("adir")
assert conftest.rget_with_confmod("a", adir)[1] == 1
assert conftest.rget_with_confmod("a", adir.join("b"))[1] == 1.5
assert conftest._rget_with_confmod("a", adir)[1] == 1
assert conftest._rget_with_confmod("a", adir.join("b"))[1] == 1.5
def test_value_access_with_confmod(self, basedir):
startdir = basedir.join("adir", "b")
startdir.ensure("xx", dir=True)
conftest = ConftestWithSetinitial(startdir)
mod, value = conftest.rget_with_confmod("a", startdir)
mod, value = conftest._rget_with_confmod("a", startdir)
assert value == 1.5
path = py.path.local(mod.__file__)
assert path.dirpath() == basedir.join("adir", "b")
@ -85,9 +74,9 @@ def test_conftest_in_nonpkg_with_init(tmpdir):
def test_doubledash_considered(testdir):
conf = testdir.mkdir("--option")
conf.join("conftest.py").ensure()
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, [conf.basename, conf.basename])
l = conftest.getconftestmodules(conf)
l = conftest._getconftestmodules(conf)
assert len(l) == 1
def test_issue151_load_all_conftests(testdir):
@ -96,7 +85,7 @@ def test_issue151_load_all_conftests(testdir):
p = testdir.mkdir(name)
p.ensure("conftest.py")
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, names)
d = list(conftest._conftestpath2mod.values())
assert len(d) == len(names)
@ -105,15 +94,15 @@ def test_conftest_global_import(testdir):
testdir.makeconftest("x=3")
p = testdir.makepyfile("""
import py, pytest
from _pytest.config import Conftest
conf = Conftest()
mod = conf.importconftest(py.path.local("conftest.py"))
from _pytest.config import PytestPluginManager
conf = PytestPluginManager()
mod = conf._importconftest(py.path.local("conftest.py"))
assert mod.x == 3
import conftest
assert conftest is mod, (conftest, mod)
subconf = py.path.local().ensure("sub", "conftest.py")
subconf.write("y=4")
mod2 = conf.importconftest(subconf)
mod2 = conf._importconftest(subconf)
assert mod != mod2
assert mod2.y == 4
import conftest
@ -125,27 +114,27 @@ def test_conftest_global_import(testdir):
def test_conftestcutdir(testdir):
conf = testdir.makeconftest("")
p = testdir.mkdir("x")
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, [testdir.tmpdir], confcutdir=p)
l = conftest.getconftestmodules(p)
l = conftest._getconftestmodules(p)
assert len(l) == 0
l = conftest.getconftestmodules(conf.dirpath())
l = conftest._getconftestmodules(conf.dirpath())
assert len(l) == 0
assert conf not in conftest._conftestpath2mod
# but we can still import a conftest directly
conftest.importconftest(conf)
l = conftest.getconftestmodules(conf.dirpath())
conftest._importconftest(conf)
l = conftest._getconftestmodules(conf.dirpath())
assert l[0].__file__.startswith(str(conf))
# and all sub paths get updated properly
l = conftest.getconftestmodules(p)
l = conftest._getconftestmodules(p)
assert len(l) == 1
assert l[0].__file__.startswith(str(conf))
def test_conftestcutdir_inplace_considered(testdir):
conf = testdir.makeconftest("")
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, [conf.dirpath()], confcutdir=conf.dirpath())
l = conftest.getconftestmodules(conf.dirpath())
l = conftest._getconftestmodules(conf.dirpath())
assert len(l) == 1
assert l[0].__file__.startswith(str(conf))
@ -153,7 +142,7 @@ def test_conftestcutdir_inplace_considered(testdir):
def test_setinitial_conftest_subdirs(testdir, name):
sub = testdir.mkdir(name)
subconftest = sub.ensure("conftest.py")
conftest = Conftest()
conftest = PytestPluginManager()
conftest_setinitial(conftest, [sub.dirpath()], confcutdir=testdir.tmpdir)
if name not in ('whatever', '.dotdir'):
assert subconftest in conftest._conftestpath2mod
@ -199,9 +188,9 @@ def test_conftest_import_order(testdir, monkeypatch):
ct2.write("")
def impct(p):
return p
conftest = Conftest()
monkeypatch.setattr(conftest, 'importconftest', impct)
assert conftest.getconftestmodules(sub) == [ct1, ct2]
conftest = PytestPluginManager()
monkeypatch.setattr(conftest, '_importconftest', impct)
assert conftest._getconftestmodules(sub) == [ct1, ct2]
def test_fixture_dependency(testdir, monkeypatch):

491
testing/test_core.py
View File

@ -3,234 +3,48 @@ from _pytest.core import * # noqa
from _pytest.config import get_plugin_manager
class TestBootstrapping:
def test_consider_env_fails_to_import(self, monkeypatch):
pluginmanager = PluginManager()
monkeypatch.setenv('PYTEST_PLUGINS', 'nonexisting', prepend=",")
pytest.raises(ImportError, lambda: pluginmanager.consider_env())
@pytest.fixture
def pm():
return PluginManager("he")
def test_preparse_args(self):
pluginmanager = PluginManager()
pytest.raises(ImportError, lambda:
pluginmanager.consider_preparse(["xyz", "-p", "hello123"]))
@pytest.fixture
def pytestpm():
return PytestPluginManager()
def test_plugin_prevent_register(self):
pluginmanager = PluginManager()
pluginmanager.consider_preparse(["xyz", "-p", "no:abc"])
l1 = pluginmanager.getplugins()
pluginmanager.register(42, name="abc")
l2 = pluginmanager.getplugins()
assert len(l2) == len(l1)
def test_plugin_prevent_register_unregistered_alredy_registered(self):
pluginmanager = PluginManager()
pluginmanager.register(42, name="abc")
l1 = pluginmanager.getplugins()
assert 42 in l1
pluginmanager.consider_preparse(["xyz", "-p", "no:abc"])
l2 = pluginmanager.getplugins()
assert 42 not in l2
def test_plugin_double_register(self):
pm = PluginManager()
class TestPluginManager:
def test_plugin_double_register(self, pm):
pm.register(42, name="abc")
pytest.raises(ValueError, lambda: pm.register(42, name="abc"))
with pytest.raises(ValueError):
pm.register(42, name="abc")
def test_plugin_skip(self, testdir, monkeypatch):
p = testdir.makepyfile(skipping1="""
import pytest
pytest.skip("hello")
""")
p.copy(p.dirpath("skipping2.py"))
monkeypatch.setenv("PYTEST_PLUGINS", "skipping2")
result = testdir.runpytest("-rw", "-p", "skipping1", "--traceconfig")
assert result.ret == 0
result.stdout.fnmatch_lines([
"WI1*skipped plugin*skipping1*hello*",
"WI1*skipped plugin*skipping2*hello*",
])
def test_consider_env_plugin_instantiation(self, testdir, monkeypatch):
pluginmanager = PluginManager()
testdir.syspathinsert()
testdir.makepyfile(xy123="#")
monkeypatch.setitem(os.environ, 'PYTEST_PLUGINS', 'xy123')
l1 = len(pluginmanager.getplugins())
pluginmanager.consider_env()
l2 = len(pluginmanager.getplugins())
assert l2 == l1 + 1
assert pluginmanager.getplugin('xy123')
pluginmanager.consider_env()
l3 = len(pluginmanager.getplugins())
assert l2 == l3
def test_consider_setuptools_instantiation(self, monkeypatch):
pkg_resources = pytest.importorskip("pkg_resources")
def my_iter(name):
assert name == "pytest11"
class EntryPoint:
name = "pytest_mytestplugin"
dist = None
def load(self):
class PseudoPlugin:
x = 42
return PseudoPlugin()
return iter([EntryPoint()])
monkeypatch.setattr(pkg_resources, 'iter_entry_points', my_iter)
pluginmanager = PluginManager()
pluginmanager.consider_setuptools_entrypoints()
plugin = pluginmanager.getplugin("mytestplugin")
assert plugin.x == 42
def test_consider_setuptools_not_installed(self, monkeypatch):
monkeypatch.setitem(py.std.sys.modules, 'pkg_resources',
py.std.types.ModuleType("pkg_resources"))
pluginmanager = PluginManager()
pluginmanager.consider_setuptools_entrypoints()
# ok, we did not explode
def test_pluginmanager_ENV_startup(self, testdir, monkeypatch):
testdir.makepyfile(pytest_x500="#")
p = testdir.makepyfile("""
import pytest
def test_hello(pytestconfig):
plugin = pytestconfig.pluginmanager.getplugin('pytest_x500')
assert plugin is not None
""")
monkeypatch.setenv('PYTEST_PLUGINS', 'pytest_x500', prepend=",")
result = testdir.runpytest(p)
assert result.ret == 0
result.stdout.fnmatch_lines(["*1 passed in*"])
def test_import_plugin_importname(self, testdir):
pluginmanager = PluginManager()
pytest.raises(ImportError, 'pluginmanager.import_plugin("qweqwex.y")')
pytest.raises(ImportError, 'pluginmanager.import_plugin("pytest_qweqwx.y")')
testdir.syspathinsert()
pluginname = "pytest_hello"
testdir.makepyfile(**{pluginname: ""})
pluginmanager.import_plugin("pytest_hello")
len1 = len(pluginmanager.getplugins())
pluginmanager.import_plugin("pytest_hello")
len2 = len(pluginmanager.getplugins())
assert len1 == len2
plugin1 = pluginmanager.getplugin("pytest_hello")
assert plugin1.__name__.endswith('pytest_hello')
plugin2 = pluginmanager.getplugin("pytest_hello")
assert plugin2 is plugin1
def test_import_plugin_dotted_name(self, testdir):
pluginmanager = PluginManager()
pytest.raises(ImportError, 'pluginmanager.import_plugin("qweqwex.y")')
pytest.raises(ImportError, 'pluginmanager.import_plugin("pytest_qweqwex.y")')
testdir.syspathinsert()
testdir.mkpydir("pkg").join("plug.py").write("x=3")
pluginname = "pkg.plug"
pluginmanager.import_plugin(pluginname)
mod = pluginmanager.getplugin("pkg.plug")
assert mod.x == 3
def test_consider_module(self, testdir):
pluginmanager = PluginManager()
testdir.syspathinsert()
testdir.makepyfile(pytest_p1="#")
testdir.makepyfile(pytest_p2="#")
mod = py.std.types.ModuleType("temp")
mod.pytest_plugins = ["pytest_p1", "pytest_p2"]
pluginmanager.consider_module(mod)
assert pluginmanager.getplugin("pytest_p1").__name__ == "pytest_p1"
assert pluginmanager.getplugin("pytest_p2").__name__ == "pytest_p2"
def test_consider_module_import_module(self, testdir):
mod = py.std.types.ModuleType("x")
mod.pytest_plugins = "pytest_a"
aplugin = testdir.makepyfile(pytest_a="#")
pluginmanager = get_plugin_manager()
reprec = testdir.make_hook_recorder(pluginmanager)
#syspath.prepend(aplugin.dirpath())
py.std.sys.path.insert(0, str(aplugin.dirpath()))
pluginmanager.consider_module(mod)
call = reprec.getcall(pluginmanager.hook.pytest_plugin_registered.name)
assert call.plugin.__name__ == "pytest_a"
# check that it is not registered twice
pluginmanager.consider_module(mod)
l = reprec.getcalls("pytest_plugin_registered")
assert len(l) == 1
def test_config_sets_conftesthandle_onimport(self, testdir):
config = testdir.parseconfig([])
assert config._conftest._onimport == config._onimportconftest
def test_consider_conftest_deps(self, testdir):
mod = testdir.makepyfile("pytest_plugins='xyz'").pyimport()
pp = PluginManager()
pytest.raises(ImportError, lambda: pp.consider_conftest(mod))
def test_pm(self):
pp = PluginManager()
def test_pm(self, pm):
class A: pass
a1, a2 = A(), A()
pp.register(a1)
assert pp.isregistered(a1)
pp.register(a2, "hello")
assert pp.isregistered(a2)
l = pp.getplugins()
pm.register(a1)
assert pm.isregistered(a1)
pm.register(a2, "hello")
assert pm.isregistered(a2)
l = pm.getplugins()
assert a1 in l
assert a2 in l
assert pp.getplugin('hello') == a2
pp.unregister(a1)
assert not pp.isregistered(a1)
def test_pm_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")
pp.register(mod)
assert pp.isregistered(mod)
l = pp.getplugins()
assert mod in l
pytest.raises(ValueError, "pp.register(mod)")
pytest.raises(ValueError, lambda: pp.register(mod))
#assert not pp.isregistered(mod2)
assert pp.getplugins() == l
def test_canonical_import(self, monkeypatch):
mod = py.std.types.ModuleType("pytest_xyz")
monkeypatch.setitem(py.std.sys.modules, 'pytest_xyz', mod)
pp = PluginManager()
pp.import_plugin('pytest_xyz')
assert pp.getplugin('pytest_xyz') == mod
assert pp.isregistered(mod)
assert pm.getplugin('hello') == a2
pm.unregister(a1)
assert not pm.isregistered(a1)
def test_register_mismatch_method(self):
pp = get_plugin_manager()
pm = get_plugin_manager()
class hello:
def pytest_gurgel(self):
pass
pytest.raises(Exception, lambda: pp.register(hello()))
pytest.raises(Exception, lambda: pm.register(hello()))
def test_register_mismatch_arg(self):
pp = get_plugin_manager()
pm = get_plugin_manager()
class hello:
def pytest_configure(self, asd):
pass
pytest.raises(Exception, lambda: pp.register(hello()))
pytest.raises(Exception, lambda: pm.register(hello()))
def test_register(self):
pm = get_plugin_manager()
@ -250,7 +64,7 @@ class TestBootstrapping:
assert pm.getplugins()[-1:] == [my2]
def test_listattr(self):
plugins = PluginManager()
plugins = PluginManager("xyz")
class api1:
x = 41
class api2:
@ -263,27 +77,6 @@ class TestBootstrapping:
l = list(plugins.listattr('x'))
assert l == [41, 42, 43]
def test_hook_tracing(self):
pm = get_plugin_manager()
saveindent = []
class api1:
x = 41
def pytest_plugin_registered(self, plugin):
saveindent.append(pm.trace.root.indent)
raise ValueError(42)
l = []
pm.set_tracing(l.append)
indent = pm.trace.root.indent
p = api1()
pm.register(p)
assert pm.trace.root.indent == indent
assert len(l) == 2
assert 'pytest_plugin_registered' in l[0]
assert 'finish' in l[1]
pytest.raises(ValueError, lambda: pm.register(api1()))
assert pm.trace.root.indent == indent
assert saveindent[0] > indent
class TestPytestPluginInteractions:
@ -301,7 +94,7 @@ class TestPytestPluginInteractions:
return xyz + 1
""")
config = get_plugin_manager().config
config._conftest.importconftest(conf)
config.pluginmanager._importconftest(conf)
print(config.pluginmanager.getplugins())
res = config.hook.pytest_myhook(xyz=10)
assert res == [11]
@ -350,7 +143,7 @@ class TestPytestPluginInteractions:
parser.addoption('--test123', action="store_true",
default=True)
""")
config._conftest.importconftest(p)
config.pluginmanager._importconftest(p)
assert config.option.test123
def test_configure(self, testdir):
@ -362,20 +155,43 @@ class TestPytestPluginInteractions:
config.pluginmanager.register(A())
assert len(l) == 0
config.do_configure()
config._do_configure()
assert len(l) == 1
config.pluginmanager.register(A()) # leads to a configured() plugin
assert len(l) == 2
assert l[0] != l[1]
config.do_unconfigure()
config._ensure_unconfigure()
config.pluginmanager.register(A())
assert len(l) == 2
def test_hook_tracing(self):
pytestpm = get_plugin_manager() # fully initialized with plugins
saveindent = []
class api1:
x = 41
def pytest_plugin_registered(self, plugin):
saveindent.append(pytestpm.trace.root.indent)
raise ValueError(42)
l = []
pytestpm.set_tracing(l.append)
indent = pytestpm.trace.root.indent
p = api1()
pytestpm.register(p)
assert pytestpm.trace.root.indent == indent
assert len(l) == 2
assert 'pytest_plugin_registered' in l[0]
assert 'finish' in l[1]
with pytest.raises(ValueError):
pytestpm.register(api1())
assert pytestpm.trace.root.indent == indent
assert saveindent[0] > indent
# lower level API
def test_listattr(self):
pluginmanager = PluginManager()
pluginmanager = PluginManager("xyz")
class My2:
x = 42
pluginmanager.register(My2())
@ -395,7 +211,7 @@ class TestPytestPluginInteractions:
def m(self):
return 19
pluginmanager = PluginManager()
pluginmanager = PluginManager("xyz")
p1 = P1()
p2 = P2()
p3 = P3()
@ -572,7 +388,7 @@ class TestMultiCall:
def m(self):
return 19
pluginmanager = PluginManager()
pluginmanager = PluginManager("xyz")
p1 = P1()
p2 = P2()
p3 = P3()
@ -624,11 +440,12 @@ class TestMultiCall:
class TestHookRelay:
def test_happypath(self):
def test_hapmypath(self):
class Api:
def hello(self, arg):
"api hook 1"
pm = PluginManager([Api], prefix="he")
pm = PluginManager("he")
pm.addhooks(Api)
hook = pm.hook
assert hasattr(hook, 'hello')
assert repr(hook.hello).find("hello") != -1
@ -647,7 +464,8 @@ class TestHookRelay:
class Api:
def hello(self, arg):
"api hook 1"
pm = PluginManager(Api, prefix="he")
pm = PluginManager("he")
pm.addhooks(Api)
class Plugin:
def hello(self, argwrong):
return arg + 1
@ -656,19 +474,20 @@ class TestHookRelay:
assert "argwrong" in str(exc.value)
def test_only_kwargs(self):
pm = PluginManager()
pm = PluginManager("he")
class Api:
def hello(self, arg):
"api hook 1"
mcm = HookRelay(hookspecs=Api, pm=pm, prefix="he")
pytest.raises(TypeError, lambda: mcm.hello(3))
pm.addhooks(Api)
pytest.raises(TypeError, lambda: pm.hook.hello(3))
def test_firstresult_definition(self):
class Api:
def hello(self, arg):
"api hook 1"
hello.firstresult = True
pm = PluginManager([Api], "he")
pm = PluginManager("he")
pm.addhooks(Api)
class Plugin:
def hello(self, arg):
return arg + 1
@ -771,15 +590,16 @@ def test_default_markers(testdir):
"*trylast*last*",
])
def test_importplugin_issue375(testdir):
def test_importplugin_issue375(testdir, pytestpm):
testdir.syspathinsert(testdir.tmpdir)
testdir.makepyfile(qwe="import aaaa")
excinfo = pytest.raises(ImportError, lambda: importplugin("qwe"))
with pytest.raises(ImportError) as excinfo:
pytestpm.import_plugin("qwe")
assert "qwe" not in str(excinfo.value)
assert "aaaa" in str(excinfo.value)
class TestWrapMethod:
def test_basic_happypath(self):
def test_basic_hapmypath(self):
class A:
def f(self):
return "A.f"
@ -880,3 +700,178 @@ class TestWrapMethod:
with pytest.raises(ValueError):
A().error()
assert l == [1]
### to be shifted to own test file
from _pytest.config import PytestPluginManager
class TestPytestPluginManager:
def test_register_imported_modules(self):
pm = PytestPluginManager()
mod = py.std.types.ModuleType("x.y.pytest_hello")
pm.register(mod)
assert pm.isregistered(mod)
l = pm.getplugins()
assert mod in l
pytest.raises(ValueError, "pm.register(mod)")
pytest.raises(ValueError, lambda: pm.register(mod))
#assert not pm.isregistered(mod2)
assert pm.getplugins() == l
def test_canonical_import(self, monkeypatch):
mod = py.std.types.ModuleType("pytest_xyz")
monkeypatch.setitem(py.std.sys.modules, 'pytest_xyz', mod)
pm = PytestPluginManager()
pm.import_plugin('pytest_xyz')
assert pm.getplugin('pytest_xyz') == mod
assert pm.isregistered(mod)
def test_consider_module(self, testdir, pytestpm):
testdir.syspathinsert()
testdir.makepyfile(pytest_p1="#")
testdir.makepyfile(pytest_p2="#")
mod = py.std.types.ModuleType("temp")
mod.pytest_plugins = ["pytest_p1", "pytest_p2"]
pytestpm.consider_module(mod)
assert pytestpm.getplugin("pytest_p1").__name__ == "pytest_p1"
assert pytestpm.getplugin("pytest_p2").__name__ == "pytest_p2"
def test_consider_module_import_module(self, testdir):
pytestpm = get_plugin_manager()
mod = py.std.types.ModuleType("x")
mod.pytest_plugins = "pytest_a"
aplugin = testdir.makepyfile(pytest_a="#")
reprec = testdir.make_hook_recorder(pytestpm)
#syspath.prepend(aplugin.dirpath())
py.std.sys.path.insert(0, str(aplugin.dirpath()))
pytestpm.consider_module(mod)
call = reprec.getcall(pytestpm.hook.pytest_plugin_registered.name)
assert call.plugin.__name__ == "pytest_a"
# check that it is not registered twice
pytestpm.consider_module(mod)
l = reprec.getcalls("pytest_plugin_registered")
assert len(l) == 1
def test_consider_env_fails_to_import(self, monkeypatch, pytestpm):
monkeypatch.setenv('PYTEST_PLUGINS', 'nonexisting', prepend=",")
with pytest.raises(ImportError):
pytestpm.consider_env()
def test_plugin_skip(self, testdir, monkeypatch):
p = testdir.makepyfile(skipping1="""
import pytest
pytest.skip("hello")
""")
p.copy(p.dirpath("skipping2.py"))
monkeypatch.setenv("PYTEST_PLUGINS", "skipping2")
result = testdir.runpytest("-rw", "-p", "skipping1", "--traceconfig")
assert result.ret == 0
result.stdout.fnmatch_lines([
"WI1*skipped plugin*skipping1*hello*",
"WI1*skipped plugin*skipping2*hello*",
])
def test_consider_env_plugin_instantiation(self, testdir, monkeypatch, pytestpm):
testdir.syspathinsert()
testdir.makepyfile(xy123="#")
monkeypatch.setitem(os.environ, 'PYTEST_PLUGINS', 'xy123')
l1 = len(pytestpm.getplugins())
pytestpm.consider_env()
l2 = len(pytestpm.getplugins())
assert l2 == l1 + 1
assert pytestpm.getplugin('xy123')
pytestpm.consider_env()
l3 = len(pytestpm.getplugins())
assert l2 == l3
def test_consider_setuptools_instantiation(self, monkeypatch, pytestpm):
pkg_resources = pytest.importorskip("pkg_resources")
def my_iter(name):
assert name == "pytest11"
class EntryPoint:
name = "pytest_mytestplugin"
dist = None
def load(self):
class PseudoPlugin:
x = 42
return PseudoPlugin()
return iter([EntryPoint()])
monkeypatch.setattr(pkg_resources, 'iter_entry_points', my_iter)
pytestpm.consider_setuptools_entrypoints()
plugin = pytestpm.getplugin("mytestplugin")
assert plugin.x == 42
def test_consider_setuptools_not_installed(self, monkeypatch, pytestpm):
monkeypatch.setitem(py.std.sys.modules, 'pkg_resources',
py.std.types.ModuleType("pkg_resources"))
pytestpm.consider_setuptools_entrypoints()
# ok, we did not explode
def test_pluginmanager_ENV_startup(self, testdir, monkeypatch):
testdir.makepyfile(pytest_x500="#")
p = testdir.makepyfile("""
import pytest
def test_hello(pytestconfig):
plugin = pytestconfig.pluginmanager.getplugin('pytest_x500')
assert plugin is not None
""")
monkeypatch.setenv('PYTEST_PLUGINS', 'pytest_x500', prepend=",")
result = testdir.runpytest(p)
assert result.ret == 0
result.stdout.fnmatch_lines(["*1 passed in*"])
def test_import_plugin_importname(self, testdir, pytestpm):
pytest.raises(ImportError, 'pytestpm.import_plugin("qweqwex.y")')
pytest.raises(ImportError, 'pytestpm.import_plugin("pytest_qweqwx.y")')
testdir.syspathinsert()
pluginname = "pytest_hello"
testdir.makepyfile(**{pluginname: ""})
pytestpm.import_plugin("pytest_hello")
len1 = len(pytestpm.getplugins())
pytestpm.import_plugin("pytest_hello")
len2 = len(pytestpm.getplugins())
assert len1 == len2
plugin1 = pytestpm.getplugin("pytest_hello")
assert plugin1.__name__.endswith('pytest_hello')
plugin2 = pytestpm.getplugin("pytest_hello")
assert plugin2 is plugin1
def test_import_plugin_dotted_name(self, testdir, pytestpm):
pytest.raises(ImportError, 'pytestpm.import_plugin("qweqwex.y")')
pytest.raises(ImportError, 'pytestpm.import_plugin("pytest_qweqwex.y")')
testdir.syspathinsert()
testdir.mkpydir("pkg").join("plug.py").write("x=3")
pluginname = "pkg.plug"
pytestpm.import_plugin(pluginname)
mod = pytestpm.getplugin("pkg.plug")
assert mod.x == 3
def test_consider_conftest_deps(self, testdir, pytestpm):
mod = testdir.makepyfile("pytest_plugins='xyz'").pyimport()
with pytest.raises(ImportError):
pytestpm.consider_conftest(mod)
class TestPytestPluginManagerBootstrapming:
def test_preparse_args(self, pytestpm):
pytest.raises(ImportError, lambda:
pytestpm.consider_preparse(["xyz", "-p", "hello123"]))
def test_plugin_prevent_register(self, pytestpm):
pytestpm.consider_preparse(["xyz", "-p", "no:abc"])
l1 = pytestpm.getplugins()
pytestpm.register(42, name="abc")
l2 = pytestpm.getplugins()
assert len(l2) == len(l1)
def test_plugin_prevent_register_unregistered_alredy_registered(self, pytestpm):
pytestpm.register(42, name="abc")
l1 = pytestpm.getplugins()
assert 42 in l1
pytestpm.consider_preparse(["xyz", "-p", "no:abc"])
l2 = pytestpm.getplugins()
assert 42 not in l2

8
testing/test_pytester.py
View File

@ -1,7 +1,7 @@
import pytest
import os
from _pytest.pytester import HookRecorder
from _pytest.core import PluginManager
from _pytest.config import PytestPluginManager
from _pytest.main import EXIT_OK, EXIT_TESTSFAILED
@ -64,7 +64,7 @@ def test_parseconfig(testdir):
def test_testdir_runs_with_plugin(testdir):
testdir.makepyfile("""
pytest_plugins = "pytest_pytester"
pytest_plugins = "pytester"
def test_hello(testdir):
assert 1
""")
@ -93,8 +93,8 @@ def make_holder():
@pytest.mark.parametrize("holder", make_holder())
def test_hookrecorder_basic(holder):
pm = PluginManager()
pm.hook._addhooks(holder, "pytest_")
pm = PytestPluginManager()
pm.addhooks(holder)
rec = HookRecorder(pm)
pm.hook.pytest_xyz(arg=123)
call = rec.popcall("pytest_xyz")

1
testing/test_recwarn.py
View File

@ -22,7 +22,6 @@ def test_WarningRecorder(recwarn):
def test_recwarn_functional(testdir):
reprec = testdir.inline_runsource("""
pytest_plugins = 'pytest_recwarn',
import warnings
oldwarn = warnings.showwarning
def test_method(recwarn):

4
testing/test_session.py
View File

@ -214,8 +214,8 @@ def test_plugin_specify(testdir):
def test_plugin_already_exists(testdir):
config = testdir.parseconfig("-p", "terminal")
assert config.option.plugins == ['terminal']
config.do_configure()
config.do_unconfigure()
config._do_configure()
config._ensure_unconfigure()
def test_exclude(testdir):
hellodir = testdir.mkdir("hello")

4
testing/test_terminal.py
View File

@ -457,7 +457,9 @@ class TestTerminalFunctional:
])
assert result.ret == 1
pytestconfig.pluginmanager.skipifmissing("xdist")
if not pytestconfig.pluginmanager.hasplugin("xdist"):
pytest.skip("xdist plugin not installed")
result = testdir.runpytest(p1, '-v', '-n 1')
result.stdout.fnmatch_lines([
"*FAIL*test_verbose_reporting.py::test_fail*",

2
testing/test_unittest.py
View File

@ -3,7 +3,6 @@ import pytest
def test_simple_unittest(testdir):
testpath = testdir.makepyfile("""
import unittest
pytest_plugins = "pytest_unittest"
class MyTestCase(unittest.TestCase):
def testpassing(self):
self.assertEquals('foo', 'foo')
@ -17,7 +16,6 @@ def test_simple_unittest(testdir):
def test_runTest_method(testdir):
testdir.makepyfile("""
import unittest
pytest_plugins = "pytest_unittest"
class MyTestCaseWithRunTest(unittest.TestCase):
def runTest(self):
self.assertEquals('foo', 'foo')