implement a new hook type: hook wrappers using a "yield" to distinguish

between working at the front and at the end of a hook call chain.
The idea is to make it easier for a plugin to "wrap" a certain hook
call and use context managers, in particular allow a major cleanup of
capturing.
This commit is contained in:
holger krekel 2014-03-14 12:49:35 +01:00
parent b47fdbe0a7
commit f43cda9681
2 changed files with 140 additions and 16 deletions

View File

@ -240,18 +240,22 @@ class PluginManager(object):
pass
l = []
last = []
wrappers = []
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
if hasattr(meth, 'hookwrapper'):
wrappers.append(meth)
elif hasattr(meth, 'tryfirst'):
last.append(meth)
elif hasattr(meth, 'trylast'):
l.insert(0, meth)
else:
l.append(meth)
l.extend(last)
l.extend(wrappers)
self._listattrcache[key] = list(l)
return l
@ -272,6 +276,14 @@ def importplugin(importspec):
class MultiCall:
""" execute a call into multiple python functions/methods. """
class WrongHookWrapper(Exception):
""" a hook wrapper does not behave correctly. """
def __init__(self, func, message):
Exception.__init__(self, func, message)
self.func = func
self.message = message
def __init__(self, methods, kwargs, firstresult=False):
self.methods = list(methods)
self.kwargs = kwargs
@ -283,16 +295,39 @@ class MultiCall:
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
next_finalizers = []
try:
while self.methods:
method = self.methods.pop()
kwargs = self.getkwargs(method)
if hasattr(method, "hookwrapper"):
it = method(**kwargs)
next = getattr(it, "next", None)
if next is None:
next = getattr(it, "__next__", None)
if next is None:
raise self.WrongHookWrapper(method,
"wrapper does not contain a yield")
res = next()
next_finalizers.append((method, next))
else:
res = method(**kwargs)
if res is not None:
self.results.append(res)
if self.firstresult:
return res
if not self.firstresult:
return self.results
finally:
for method, fin in reversed(next_finalizers):
try:
fin()
except StopIteration:
pass
else:
raise self.WrongHookWrapper(method,
"wrapper contain more than one yield")
def getkwargs(self, method):
kwargs = {}

View File

@ -523,6 +523,95 @@ class TestMultiCall:
res = MultiCall([m1, m2], {}).execute()
assert res == [1]
def test_hookwrapper(self):
l = []
def m1():
l.append("m1 init")
yield None
l.append("m1 finish")
m1.hookwrapper = True
def m2():
l.append("m2")
return 2
res = MultiCall([m2, m1], {}).execute()
assert res == [2]
assert l == ["m1 init", "m2", "m1 finish"]
l[:] = []
res = MultiCall([m2, m1], {}, firstresult=True).execute()
assert res == 2
assert l == ["m1 init", "m2", "m1 finish"]
def test_hookwrapper_order(self):
l = []
def m1():
l.append("m1 init")
yield 1
l.append("m1 finish")
m1.hookwrapper = True
def m2():
l.append("m2 init")
yield 2
l.append("m2 finish")
m2.hookwrapper = True
res = MultiCall([m2, m1], {}).execute()
assert res == [1, 2]
assert l == ["m1 init", "m2 init", "m2 finish", "m1 finish"]
def test_listattr_hookwrapper_ordering(self):
class P1:
@pytest.mark.hookwrapper
def m(self):
return 17
class P2:
def m(self):
return 23
class P3:
@pytest.mark.tryfirst
def m(self):
return 19
pluginmanager = PluginManager()
p1 = P1()
p2 = P2()
p3 = P3()
pluginmanager.register(p1)
pluginmanager.register(p2)
pluginmanager.register(p3)
methods = pluginmanager.listattr('m')
assert methods == [p2.m, p3.m, p1.m]
## listattr keeps a cache and deleting
## a function attribute requires clearing it
#pluginmanager._listattrcache.clear()
#del P1.m.__dict__['tryfirst']
def test_hookwrapper_not_yield(self):
def m1():
pass
m1.hookwrapper = True
mc = MultiCall([m1], {})
with pytest.raises(mc.WrongHookWrapper) as ex:
mc.execute()
assert ex.value.func == m1
assert ex.value.message
def test_hookwrapper_too_many_yield(self):
def m1():
yield 1
yield 2
m1.hookwrapper = True
mc = MultiCall([m1], {})
with pytest.raises(mc.WrongHookWrapper) as ex:
mc.execute()
assert ex.value.func == m1
assert ex.value.message
class TestHookRelay:
def test_happypath(self):
pm = PluginManager()