[svn r62211] merge 60797:HEAD of pytestplugin branch:
this merge contains: * a new plugin architecture * a pluginized pytest core * many pytest related refactorings * refactorings/streamlining of pytest's own tests --HG-- branch : trunk
This commit is contained in:
150
py/_com.py
Normal file
150
py/_com.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
py lib plugins and events.
|
||||
|
||||
you can write plugins that extend the py lib API.
|
||||
currently this is mostly used by py.test
|
||||
|
||||
registering a plugin
|
||||
++++++++++++++++++++++++++++++++++
|
||||
|
||||
::
|
||||
>>> class MyPlugin:
|
||||
... def pyevent_plugin_registered(self, plugin):
|
||||
... print "registering", plugin.__class__.__name__
|
||||
...
|
||||
>>> import py
|
||||
>>> py._com.pyplugins.register(MyPlugin())
|
||||
registering MyPlugin
|
||||
|
||||
"""
|
||||
|
||||
import py
|
||||
|
||||
class MultiCall:
|
||||
""" Manage a specific call into many python functions/methods.
|
||||
|
||||
Simple example:
|
||||
MultiCall([list1.append, list2.append], 42).execute()
|
||||
"""
|
||||
NONEASRESULT = object()
|
||||
|
||||
def __init__(self, methods, *args, **kwargs):
|
||||
self.methods = methods
|
||||
self.args = args
|
||||
self.kwargs = kwargs
|
||||
self.results = []
|
||||
|
||||
def execute(self, firstresult=False):
|
||||
while self.methods:
|
||||
self.currentmethod = self.methods.pop()
|
||||
# provide call introspection if "__call__" is the first positional argument
|
||||
if hasattr(self.currentmethod, 'im_self'):
|
||||
varnames = self.currentmethod.im_func.func_code.co_varnames
|
||||
needscall = varnames[1:2] == ('__call__',)
|
||||
else:
|
||||
try:
|
||||
varnames = self.currentmethod.func_code.co_varnames
|
||||
except AttributeError:
|
||||
# builtin function
|
||||
varnames = ()
|
||||
needscall = varnames[:1] == ('__call__',)
|
||||
if needscall:
|
||||
res = self.currentmethod(self, *self.args, **self.kwargs)
|
||||
else:
|
||||
res = self.currentmethod(*self.args, **self.kwargs)
|
||||
if res is not None:
|
||||
if res is self.NONEASRESULT:
|
||||
res = None
|
||||
self.results.append(res)
|
||||
if firstresult:
|
||||
break
|
||||
if not firstresult:
|
||||
return self.results
|
||||
if self.results:
|
||||
return self.results[-1]
|
||||
|
||||
class PyPlugins:
|
||||
"""
|
||||
Manage Plugins: Load plugins and manage calls to plugins.
|
||||
"""
|
||||
MultiCall = MultiCall
|
||||
|
||||
def __init__(self, plugins=None):
|
||||
if plugins is None:
|
||||
plugins = []
|
||||
self._plugins = plugins
|
||||
self._callbacks = []
|
||||
|
||||
def import_module(self, modspec):
|
||||
# XXX allow modspec to specify version / lookup
|
||||
modpath = modspec
|
||||
self.notify("importingmodule", modpath)
|
||||
__import__(modpath)
|
||||
|
||||
def consider_env(self):
|
||||
""" consider ENV variable for loading modules. """
|
||||
for spec in self._envlist("PYLIB"):
|
||||
self.import_module(spec)
|
||||
|
||||
def _envlist(self, varname):
|
||||
val = py.std.os.environ.get(varname, None)
|
||||
if val is not None:
|
||||
return val.split(',')
|
||||
return ()
|
||||
|
||||
def consider_module(self, mod, varname="pylib"):
|
||||
speclist = getattr(mod, varname, ())
|
||||
if not isinstance(speclist, (list, tuple)):
|
||||
speclist = (speclist,)
|
||||
for spec in speclist:
|
||||
self.import_module(spec)
|
||||
|
||||
def register(self, plugin):
|
||||
assert not isinstance(plugin, str)
|
||||
self._plugins.append(plugin)
|
||||
self.notify("plugin_registered", plugin)
|
||||
|
||||
def unregister(self, plugin):
|
||||
self._plugins.remove(plugin)
|
||||
self.notify("plugin_unregistered", plugin)
|
||||
|
||||
def getplugins(self):
|
||||
return list(self._plugins)
|
||||
|
||||
def isregistered(self, plugin):
|
||||
return plugin in self._plugins
|
||||
|
||||
def listattr(self, attrname, plugins=None, extra=()):
|
||||
l = []
|
||||
if plugins is None:
|
||||
plugins = self._plugins
|
||||
if extra:
|
||||
plugins += list(extra)
|
||||
for plugin in plugins:
|
||||
try:
|
||||
l.append(getattr(plugin, attrname))
|
||||
except AttributeError:
|
||||
continue
|
||||
return l
|
||||
|
||||
def call_each(self, methname, *args, **kwargs):
|
||||
""" return call object for executing a plugin call. """
|
||||
return MultiCall(self.listattr(methname), *args, **kwargs).execute()
|
||||
|
||||
def call_firstresult(self, methname, *args, **kwargs):
|
||||
""" return first non-None result of a plugin method. """
|
||||
return MultiCall(self.listattr(methname), *args, **kwargs).execute(firstresult=True)
|
||||
|
||||
def call_plugin(self, plugin, methname, *args, **kwargs):
|
||||
return MultiCall(self.listattr(methname, plugins=[plugin]),
|
||||
*args, **kwargs).execute(firstresult=True)
|
||||
|
||||
def notify(self, eventname, *args, **kwargs):
|
||||
#print "notifying", eventname, args, kwargs
|
||||
MultiCall(self.listattr("pyevent_" + eventname),
|
||||
*args, **kwargs).execute()
|
||||
#print "calling anonymous hooks", args, kwargs
|
||||
MultiCall(self.listattr("pyevent"),
|
||||
eventname, *args, **kwargs).execute()
|
||||
|
||||
pyplugins = PyPlugins()
|
||||
Reference in New Issue
Block a user