introduce an experimental approach for allowing dynamic addition of hooks from plugin. Plugins may register new hooks by implementing the new

pytest_registerhooks(pluginmanager)

and call

    pluginmanager.registerhooks(module)

with the referenced 'module' object containing the hooks.

The new pytest_registerhooks is called after pytest_addoption
and before pytest_configure.

--HG--
branch : trunk
This commit is contained in:
holger krekel 2010-04-22 11:57:57 +02:00
parent cbb4c0dadc
commit 85d35f7418
10 changed files with 97 additions and 69 deletions

View File

@ -1,6 +1,7 @@
Changes between 1.2.1 and XXX
=====================================
Changes between 1.2.1 and 1.2.2 (release pending)
==================================================
- new mechanism to allow plugins to register new hooks
- added links to the new capturelog and coverage plugins
- (issue87) fix unboundlocal error in assertionold code
- (issue86) improve documentation for looponfailing

View File

@ -8,9 +8,8 @@ dictionary or an import path.
(c) Holger Krekel and others, 2004-2010
"""
version = "1.2.1"
_version__ = version = "1.2.2"
__version__ = version = version or "1.2.x"
import py.apipkg
py.apipkg.initpkg(__name__, dict(

View File

@ -9,6 +9,9 @@ hook specifications for py.test plugins
def pytest_addoption(parser):
""" called before commandline parsing. """
def pytest_registerhooks(pluginmanager):
""" called after commandline parsing before pytest_configure. """
def pytest_namespace():
""" return dict of name->object which will get stored at py.test. namespace"""
@ -133,31 +136,6 @@ def pytest_doctest_prepare_content(content):
""" return processed content for a given doctest"""
pytest_doctest_prepare_content.firstresult = True
# -------------------------------------------------------------------------
# distributed testing
# -------------------------------------------------------------------------
def pytest_gwmanage_newgateway(gateway, platinfo):
""" called on new raw gateway creation. """
def pytest_gwmanage_rsyncstart(source, gateways):
""" called before rsyncing a directory to remote gateways takes place. """
def pytest_gwmanage_rsyncfinish(source, gateways):
""" called after rsyncing a directory to remote gateways takes place. """
def pytest_testnodeready(node):
""" Test Node is ready to operate. """
def pytest_testnodedown(node, error):
""" Test Node is down. """
def pytest_rescheduleitems(items):
""" reschedule Items from a node that went down. """
def pytest_looponfailinfo(failreports, rootdirs):
""" info for repeating failing tests. """
# -------------------------------------------------------------------------
# error handling and internal debugging hooks

View File

@ -34,15 +34,18 @@ class HookRecorder:
self._recorders = {}
def start_recording(self, hookspecs):
assert hookspecs not in self._recorders
class RecordCalls:
_recorder = self
for name, method in vars(hookspecs).items():
if name[0] != "_":
setattr(RecordCalls, name, self._makecallparser(method))
recorder = RecordCalls()
self._recorders[hookspecs] = recorder
self._registry.register(recorder)
if not isinstance(hookspecs, (list, tuple)):
hookspecs = [hookspecs]
for hookspec in hookspecs:
assert hookspec not in self._recorders
class RecordCalls:
_recorder = self
for name, method in vars(hookspec).items():
if name[0] != "_":
setattr(RecordCalls, name, self._makecallparser(method))
recorder = RecordCalls()
self._recorders[hookspec] = recorder
self._registry.register(recorder)
self.hook = HookRelay(hookspecs, registry=self._registry)
def finish_recording(self):

View File

@ -93,9 +93,11 @@ def pytest_report_header(config):
# =====================================================
def pytest_plugin_registered(manager, plugin):
hookspec = manager.hook._hookspecs
methods = collectattr(plugin)
hooks = collectattr(hookspec)
hooks = {}
for hookspec in manager.hook._hookspecs:
hooks.update(collectattr(hookspec))
stringio = py.io.TextIO()
def Print(*args):
if args:
@ -109,10 +111,13 @@ def pytest_plugin_registered(manager, plugin):
if isgenerichook(name):
continue
if name not in hooks:
Print("found unknown hook:", name)
fail = True
if not getattr(method, 'optionalhook', False):
Print("found unknown hook:", name)
fail = True
else:
#print "checking", method
method_args = getargs(method)
#print "method_args", method_args
if '__multicall__' in method_args:
method_args.remove('__multicall__')
hook = hooks[name]

View File

@ -6,6 +6,8 @@ This is a good source for looking at the various reporting hooks.
import py
import sys
optionalhook = py.test.mark.optionalhook
def pytest_addoption(parser):
group = parser.getgroup("terminal reporting", "reporting", after="general")
group._addoption('-v', '--verbose', action="count",
@ -130,6 +132,15 @@ class TerminalReporter:
for line in str(excrepr).split("\n"):
self.write_line("INTERNALERROR> " + line)
def pytest_plugin_registered(self, plugin):
if self.config.option.traceconfig:
msg = "PLUGIN registered: %s" %(plugin,)
# XXX this event may happen during setup/teardown time
# which unfortunately captures our output here
# which garbles our output if we use self.write_line
self.write_line(msg)
@optionalhook
def pytest_gwmanage_newgateway(self, gateway, platinfo):
#self.write_line("%s instantiated gateway from spec %r" %(gateway.id, gateway.spec._spec))
d = {}
@ -149,30 +160,37 @@ class TerminalReporter:
self.write_line(infoline)
self.gateway2info[gateway] = infoline
def pytest_plugin_registered(self, plugin):
if self.config.option.traceconfig:
msg = "PLUGIN registered: %s" %(plugin,)
# XXX this event may happen during setup/teardown time
# which unfortunately captures our output here
# which garbles our output if we use self.write_line
self.write_line(msg)
@optionalhook
def pytest_testnodeready(self, node):
self.write_line("[%s] txnode ready to receive tests" %(node.gateway.id,))
@optionalhook
def pytest_testnodedown(self, node, error):
if error:
self.write_line("[%s] node down, error: %s" %(node.gateway.id, error))
@optionalhook
def pytest_rescheduleitems(self, items):
if self.config.option.debug:
self.write_sep("!", "RESCHEDULING %s " %(items,))
@optionalhook
def pytest_looponfailinfo(self, failreports, rootdirs):
if failreports:
self.write_sep("#", "LOOPONFAILING", red=True)
for report in failreports:
loc = self._getcrashline(report)
self.write_line(loc, red=True)
self.write_sep("#", "waiting for changes")
for rootdir in rootdirs:
self.write_line("### Watching: %s" %(rootdir,), bold=True)
def pytest_trace(self, category, msg):
if self.config.option.debug or \
self.config.option.traceconfig and category.find("config") != -1:
self.write_line("[%s] %s" %(category, msg))
def pytest_rescheduleitems(self, items):
if self.config.option.debug:
self.write_sep("!", "RESCHEDULING %s " %(items,))
def pytest_deselected(self, items):
self.stats.setdefault('deselected', []).append(items)
@ -274,16 +292,6 @@ class TerminalReporter:
else:
excrepr.reprcrash.toterminal(self._tw)
def pytest_looponfailinfo(self, failreports, rootdirs):
if failreports:
self.write_sep("#", "LOOPONFAILING", red=True)
for report in failreports:
loc = self._getcrashline(report)
self.write_line(loc, red=True)
self.write_sep("#", "waiting for changes")
for rootdir in rootdirs:
self.write_line("### Watching: %s" %(rootdir,), bold=True)
def _getcrashline(self, report):
try:
return report.longrepr.reprcrash

View File

@ -7,7 +7,7 @@ from py._plugin import hookspec
from py._test.outcome import Skipped
default_plugins = (
"default runner capture terminal mark skipping tmpdir monkeypatch "
"default runner capture mark terminal skipping tmpdir monkeypatch "
"recwarn pdb pastebin unittest helpconfig nose assertion genscript "
"junitxml doctest").split()
@ -20,7 +20,7 @@ class PluginManager(object):
self.registry = Registry()
self._name2plugin = {}
self._hints = []
self.hook = HookRelay(hookspecs=hookspec, registry=self.registry)
self.hook = HookRelay([hookspec], registry=self.registry)
self.register(self)
for spec in default_plugins:
self.import_plugin(spec)
@ -40,6 +40,8 @@ class PluginManager(object):
if name in self._name2plugin:
return False
self._name2plugin[name] = plugin
self.call_plugin(plugin, "pytest_registerhooks",
{'pluginmanager': self})
self.hook.pytest_plugin_registered(manager=self, plugin=plugin)
self.registry.register(plugin)
return True
@ -58,6 +60,9 @@ class PluginManager(object):
if plugin == val:
return True
def registerhooks(self, spec):
self.hook._registerhooks(spec)
def getplugins(self):
return list(self.registry)
@ -301,13 +306,21 @@ class Registry:
class HookRelay:
def __init__(self, hookspecs, registry):
self._hookspecs = hookspecs
if not isinstance(hookspecs, list):
hookspecs = [hookspecs]
self._hookspecs = []
self._registry = registry
for hookspec in hookspecs:
self._registerhooks(hookspec)
def _registerhooks(self, hookspecs):
self._hookspecs.append(hookspecs)
for name, method in vars(hookspecs).items():
if name[:1] != "_":
firstresult = getattr(method, 'firstresult', False)
hc = HookCaller(self, name, firstresult=firstresult)
setattr(self, name, hc)
#print ("setting new hook", name)
def _performcall(self, name, multicall):
return multicall.execute()

View File

@ -27,7 +27,7 @@ def main():
name='py',
description='py.test and pylib: rapid testing and development utils.',
long_description = long_description,
version= trunk or '1.2.1',
version= trunk or '1.2.2',
url='http://pylib.org',
license='MIT license',
platforms=['unix', 'linux', 'osx', 'cygwin', 'win32'],

View File

@ -34,7 +34,7 @@ def test_functional(testdir, linecomp):
def test_func(_pytest):
class ApiClass:
def xyz(self, arg): pass
hook = HookRelay(ApiClass, Registry())
hook = HookRelay([ApiClass], Registry())
rec = _pytest.gethookrecorder(hook)
class Plugin:
def xyz(self, arg):

View File

@ -29,3 +29,24 @@ def test_collectattr():
methods = py.builtin.sorted(collectattr(B()))
assert list(methods) == ['pytest_hello', 'pytest_world']
def test_hookvalidation_unknown(testdir):
testdir.makeconftest("""
def pytest_hello(xyz):
pass
""")
result = testdir.runpytest()
assert result.ret != 0
assert result.stderr.fnmatch_lines([
'*unknown hook*pytest_hello*'
])
def test_hookvalidation_optional(testdir):
testdir.makeconftest("""
import py
@py.test.mark.optionalhook
def pytest_hello(xyz):
pass
""")
result = testdir.runpytest()
assert result.ret == 0