some more separation of core pluginmanager from pytest specific functionality.

Idea is to have the PluginManager be re-useable from other projects at some point.
This commit is contained in:
holger krekel 2013-09-30 13:14:14 +02:00
parent d946299b0a
commit 4b709037ab
9 changed files with 157 additions and 134 deletions

View File

@ -2,6 +2,87 @@
import py
import sys, os
from _pytest import hookspec # the extension point definitions
from _pytest.core import PluginManager
# pytest startup
def main(args=None, plugins=None):
""" return exit code, after performing an in-process test run.
:arg args: list of command line arguments.
:arg plugins: list of plugin objects to be auto-registered during
initialization.
"""
config = _prepareconfig(args, plugins)
exitstatus = config.hook.pytest_cmdline_main(config=config)
return exitstatus
class cmdline: # compatibility namespace
main = staticmethod(main)
class UsageError(Exception):
""" error in py.test usage or invocation"""
_preinit = []
default_plugins = (
"mark main terminal runner python pdb unittest capture skipping "
"tmpdir monkeypatch recwarn pastebin helpconfig nose assertion genscript "
"junitxml resultlog doctest").split()
def _preloadplugins():
assert not _preinit
_preinit.append(get_plugin_manager())
def get_plugin_manager():
if _preinit:
return _preinit.pop(0)
# subsequent calls to main will create a fresh instance
pluginmanager = PytestPluginManager()
pluginmanager.config = config = Config(pluginmanager) # XXX attr needed?
for spec in default_plugins:
pluginmanager.import_plugin(spec)
return pluginmanager
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)
pluginmanager = get_plugin_manager()
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
class PytestPluginManager(PluginManager):
def __init__(self, hookspecs=[hookspec]):
super(PytestPluginManager, self).__init__(hookspecs=hookspecs)
self.register(self)
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)
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.")
class Parser:
""" Parser for command line arguments and ini-file values. """
@ -494,10 +575,15 @@ class Config(object):
self._opt2dest = {}
self._cleanup = []
self.pluginmanager.register(self, "pytestconfig")
self.pluginmanager.set_register_callback(self._register_plugin)
self._configured = False
def pytest_plugin_registered(self, plugin):
def _register_plugin(self, plugin, name):
call_plugin = self.pluginmanager.call_plugin
call_plugin(plugin, "pytest_addhooks",
{'pluginmanager': self.pluginmanager})
self.hook.pytest_plugin_registered(plugin=plugin,
manager=self.pluginmanager)
dic = call_plugin(plugin, "pytest_namespace", {}) or {}
if dic:
import pytest
@ -527,10 +613,26 @@ class Config(object):
fin = config._cleanup.pop()
fin()
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,
excinfo=excinfo)
if not py.builtin.any(res):
for line in str(excrepr).split("\n"):
sys.stderr.write("INTERNALERROR> %s\n" %line)
sys.stderr.flush()
@classmethod
def fromdictargs(cls, option_dict, args):
""" constructor useable for subprocesses. """
from _pytest.core import get_plugin_manager
pluginmanager = get_plugin_manager()
config = pluginmanager.config
# XXX slightly crude way to initialize capturing

View File

@ -4,17 +4,10 @@ pytest PluginManager, basic initialization and tracing.
import sys, os
import inspect
import py
from _pytest import hookspec # the extension point definitions
from _pytest.config import Config
assert py.__version__.split(".")[:2] >= ['1', '4'], ("installation problem: "
"%s is too old, remove or upgrade 'py'" % (py.__version__))
default_plugins = (
"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 = {}
@ -73,7 +66,7 @@ class TagTracerSub:
return self.__class__(self.root, self.tags + (name,))
class PluginManager(object):
def __init__(self, load=False):
def __init__(self, hookspecs=None):
self._name2plugin = {}
self._listattrcache = {}
self._plugins = []
@ -81,20 +74,11 @@ class PluginManager(object):
self.trace = TagTracer().get("pluginmanage")
self._plugin_distinfo = []
self._shutdown = []
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)
self.config = Config(self) # XXX unclear if the attr is needed
if load:
for spec in default_plugins:
self.import_plugin(spec)
self.hook = HookRelay(hookspecs or [], pm=self)
def set_register_callback(self, callback):
assert not hasattr(self, "_registercallback")
self._registercallback = callback
def register(self, plugin, name=None, prepend=False):
if self._name2plugin.get(name, None) == -1:
@ -105,8 +89,9 @@ class PluginManager(object):
name, plugin, self._name2plugin))
#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)
reg = getattr(self, "_registercallback", None)
if reg is not None:
reg(plugin, name)
if not prepend:
self._plugins.append(plugin)
else:
@ -139,8 +124,8 @@ class PluginManager(object):
if plugin == val:
return True
def addhooks(self, spec):
self.hook._addhooks(spec, prefix="pytest_")
def addhooks(self, spec, prefix="pytest_"):
self.hook._addhooks(spec, prefix=prefix)
def getplugins(self):
return list(self._plugins)
@ -240,36 +225,6 @@ class PluginManager(object):
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_terminal_summary(self, terminalreporter):
tw = terminalreporter._tw
if terminalreporter.config.option.traceconfig:
for hint in self._hints:
tw.line("hint: %s" % hint)
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,
excinfo=excinfo)
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
@ -424,46 +379,3 @@ class HookCaller:
self.trace.root.indent -= 1
return res
_preinit = []
def _preloadplugins():
assert not _preinit
_preinit.append(PluginManager(load=True))
def get_plugin_manager():
if _preinit:
return _preinit.pop(0)
else: # subsequent calls to main will create a fresh instance
return 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)
pluginmanager = get_plugin_manager()
if plugins:
for plugin in plugins:
pluginmanager.register(plugin)
return pluginmanager.hook.pytest_cmdline_parse(
pluginmanager=pluginmanager, args=args)
def main(args=None, plugins=None):
""" return exit code, after performing an in-process test run.
:arg args: list of command line arguments.
:arg plugins: list of plugin objects to be auto-registered during
initialization.
"""
config = _prepareconfig(args, plugins)
exitstatus = config.hook.pytest_cmdline_main(config=config)
return exitstatus
class UsageError(Exception):
""" error in py.test usage or invocation"""

View File

@ -91,7 +91,7 @@ def wrap_session(config, doit):
session.exitstatus = EXIT_INTERRUPTED
except:
excinfo = py.code.ExceptionInfo()
config.pluginmanager.notify_exception(excinfo, config.option)
config.notify_exception(excinfo, config.option)
session.exitstatus = EXIT_INTERNALERROR
if excinfo.errisinstance(SystemExit):
sys.stderr.write("mainloop: caught Spurious SystemExit!\n")

View File

@ -375,8 +375,8 @@ class TmpTestdir:
break
else:
args.append("--basetemp=%s" % self.tmpdir.dirpath('basetemp'))
import _pytest.core
config = _pytest.core._prepareconfig(args, self.plugins)
import _pytest.config
config = _pytest.config._prepareconfig(args, self.plugins)
# 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)

View File

@ -342,6 +342,7 @@ class TerminalReporter:
if exitstatus in (0, 1, 2, 4):
self.summary_errors()
self.summary_failures()
self.summary_hints()
self.config.hook.pytest_terminal_summary(terminalreporter=self)
if exitstatus == 2:
self._report_keyboardinterrupt()
@ -407,6 +408,11 @@ class TerminalReporter:
l.append(x)
return l
def summary_hints(self):
if self.config.option.traceconfig:
for hint in self.config.pluginmanager._hints:
self._tw.line("hint: %s" % hint)
def summary_failures(self):
if self.config.option.tbstyle != "no":
reports = self.getreports('failed')

View File

@ -8,9 +8,11 @@ if __name__ == '__main__': # if run as a script or by 'python -m pytest'
# we trigger the below "else" condition by the following import
import pytest
raise SystemExit(pytest.main())
else:
# we are simply imported
from _pytest.core import main, UsageError, _preloadplugins
from _pytest import core as cmdline
from _pytest import __version__
_preloadplugins() # to populate pytest.* namespace so help(pytest) works
# else we are imported
from _pytest.config import main, UsageError, _preloadplugins, cmdline
from _pytest import __version__
_preloadplugins() # to populate pytest.* namespace so help(pytest) works

View File

@ -320,3 +320,18 @@ def test_cmdline_processargs_simple(testdir):
def test_toolongargs_issue224(testdir):
result = testdir.runpytest("-m", "hello" * 500)
assert result.ret == 0
def test_notify_exception(testdir, capfd):
config = testdir.parseconfig()
excinfo = pytest.raises(ValueError, "raise ValueError(1)")
config.notify_exception(excinfo)
out, err = capfd.readouterr()
assert "ValueError" in err
class A:
def pytest_internalerror(self, excrepr):
return True
config.pluginmanager.register(A())
config.notify_exception(excinfo)
out, err = capfd.readouterr()
assert not err

View File

@ -1,6 +1,7 @@
import pytest, py, os
from _pytest.core import PluginManager
from _pytest.core import MultiCall, HookRelay, varnames
from _pytest.config import get_plugin_manager
class TestBootstrapping:
@ -149,7 +150,7 @@ class TestBootstrapping:
mod = py.std.types.ModuleType("x")
mod.pytest_plugins = "pytest_a"
aplugin = testdir.makepyfile(pytest_a="#")
pluginmanager = PluginManager()
pluginmanager = get_plugin_manager()
reprec = testdir.getreportrecorder(pluginmanager)
#syspath.prepend(aplugin.dirpath())
py.std.sys.path.insert(0, str(aplugin.dirpath()))
@ -224,36 +225,21 @@ class TestBootstrapping:
assert pp.isregistered(mod)
def test_register_mismatch_method(self):
pp = PluginManager(load=True)
pp = get_plugin_manager()
class hello:
def pytest_gurgel(self):
pass
pytest.raises(Exception, "pp.register(hello())")
def test_register_mismatch_arg(self):
pp = PluginManager(load=True)
pp = get_plugin_manager()
class hello:
def pytest_configure(self, asd):
pass
excinfo = pytest.raises(Exception, "pp.register(hello())")
def test_notify_exception(self, capfd):
pp = PluginManager()
excinfo = pytest.raises(ValueError, "raise ValueError(1)")
pp.notify_exception(excinfo)
out, err = capfd.readouterr()
assert "ValueError" in err
class A:
def pytest_internalerror(self, excrepr):
return True
pp.register(A())
pp.notify_exception(excinfo)
out, err = capfd.readouterr()
assert not err
def test_register(self):
pm = PluginManager(load=False)
pm = get_plugin_manager()
class MyPlugin:
pass
my = MyPlugin()
@ -261,13 +247,13 @@ class TestBootstrapping:
assert pm.getplugins()
my2 = MyPlugin()
pm.register(my2)
assert pm.getplugins()[2:] == [my, my2]
assert pm.getplugins()[-2:] == [my, my2]
assert pm.isregistered(my)
assert pm.isregistered(my2)
pm.unregister(my)
assert not pm.isregistered(my)
assert pm.getplugins()[2:] == [my2]
assert pm.getplugins()[-1:] == [my2]
def test_listattr(self):
plugins = PluginManager()
@ -284,7 +270,7 @@ class TestBootstrapping:
assert l == [41, 42, 43]
def test_hook_tracing(self):
pm = PluginManager()
pm = get_plugin_manager()
saveindent = []
class api1:
x = 41
@ -319,7 +305,7 @@ class TestPytestPluginInteractions:
def pytest_myhook(xyz):
return xyz + 1
""")
config = PluginManager(load=True).config
config = get_plugin_manager().config
config._conftest.importconftest(conf)
print(config.pluginmanager.getplugins())
res = config.hook.pytest_myhook(xyz=10)

View File

@ -104,7 +104,7 @@ def test_functional(testdir, linecomp):
def test_func(_pytest):
class ApiClass:
def pytest_xyz(self, arg): "x"
hook = HookRelay([ApiClass], PluginManager(load=False))
hook = HookRelay([ApiClass], PluginManager())
rec = _pytest.gethookrecorder(hook)
class Plugin:
def pytest_xyz(self, arg):