test_ok1/py/_com.py

"""
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 = currentmethod = self.methods.pop()
            # provide call introspection if "__call__" is the first positional argument 
            if hasattr(currentmethod, 'im_self'):
                varnames = currentmethod.im_func.func_code.co_varnames
                needscall = varnames[1:2] == ('__call__',)
            else:
                try:
                    varnames = currentmethod.func_code.co_varnames
                except AttributeError:
                    # builtin function
                    varnames = ()
                needscall = varnames[:1] == ('__call__',)
            if needscall:
                res = currentmethod(self, *self.args, **self.kwargs)
            else:
                res = currentmethod(*self.args, **self.kwargs)
            if hasattr(self, '_ex1'):
                self.results = [res]
                break
            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] 

    def exclude_other_results(self):
        self._ex1 = True

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.notify("plugin_unregistered", plugin)
        self._plugins.remove(plugin)

    def getplugins(self):
        return list(self._plugins)

    def isregistered(self, plugin):
        return plugin in self._plugins 

    def listattr(self, attrname, plugins=None, extra=(), reverse=False):
        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 
        if reverse:
            l.reverse()
        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()