|
""" Plugin Manager --------------
A plugin manager class is used to load plugins, manage the list of loaded plugins, and proxy calls to those plugins.
The plugin managers provided with nose are:
:class:`PluginManager` This manager doesn't implement loadPlugins, so it can only work with a static list of plugins.
:class:`BuiltinPluginManager` This manager loads plugins referenced in ``nose.plugins.builtin``.
:class:`EntryPointPluginManager` This manager uses setuptools entrypoints to load plugins.
:class:`ExtraPluginsPluginManager` This manager loads extra plugins specified with the keyword `addplugins`.
:class:`DefaultPluginMananger` This is the manager class that will be used by default. If setuptools is installed, it is a subclass of :class:`EntryPointPluginManager` and :class:`BuiltinPluginManager`; otherwise, an alias to :class:`BuiltinPluginManager`.
:class:`RestrictedPluginManager` This manager is for use in test runs where some plugin calls are not available, such as runs started with ``python setup.py test``, where the test runner is the default unittest :class:`TextTestRunner`. It is a subclass of :class:`DefaultPluginManager`.
Writing a plugin manager ========================
If you want to load plugins via some other means, you can write a plugin manager and pass an instance of your plugin manager class when instantiating the :class:`nose.config.Config` instance that you pass to :class:`TestProgram` (or :func:`main` or :func:`run`).
To implement your plugin loading scheme, implement ``loadPlugins()``, and in that method, call ``addPlugin()`` with an instance of each plugin you wish to make available. Make sure to call ``super(self).loadPlugins()`` as well if have subclassed a manager other than ``PluginManager``.
"""
except: import pickle except: from io import StringIO
'BuiltinPluginManager', 'RestrictedPluginManager']
"""Proxy for plugin calls. Essentially a closure bound to the given call and plugin list.
The plugin proxy also must be bound to a particular plugin interface specification, so that it knows what calls are available and any special handling that is required for each call. """ except AttributeError: raise AttributeError("%s is not a valid %s method" % (call, self.interface.__name__))
"""Add plugin to my list of plugins to call, if it has the attribute I'm bound to. """ len(inspect.getargspec(meth)[0]) == 2: orig_meth = meth meth = lambda module, path, **kwargs: orig_meth(module)
# special case -- load tests from names behaves somewhat differently # from other chainable calls, because plugins return a tuple, only # part of which can be chained to the next plugin.
# call all plugins and yield a flattened iterator of their results return self.chain else: # return a value from the first plugin that returns non-None
"""Call plugins in a chain, where the result of each plugin call is sent to the next plugin as input. The final output result is returned. """ result = None # extract the static arguments (if any) from arg so they can # be passed to each plugin call in the chain static = [a for (static, a) in zip(getattr(self.method, 'static_args', []), arg) if static] for p, meth in self.plugins: result = meth(*arg, **kw) arg = static[:] arg.append(result) return result
"""Call all plugins, yielding each item in each non-None result. """ result = None try: result = meth(*arg, **kw) if result is not None: for r in result: yield r except (KeyboardInterrupt, SystemExit): raise except: exc = sys.exc_info() yield Failure(*exc) continue
"""Call all plugins, returning the first non-None result. """ return result
"""Chainable but not quite normal. Plugins return a tuple of (tests, names) after processing the names. The tests are added to a suite that is accumulated throughout the full call, while names are input for the next plugin in the chain. """ result = meth(names, module=module) if result is not None: suite_part, names = result if suite_part: suite.extend(suite_part)
"""Null Plugin manager that has no plugins."""
return ()
pass
return ()
method = getattr(self.interface, call) if getattr(method, "generative", False): return self._emptyIterator else: return self._doNothing
raise NotImplementedError()
raise NotImplementedError()
pass
pass
pass
"""Base class for plugin managers. PluginManager is intended to be used only with a static list of plugins. The loadPlugins() implementation only reloads plugins from _extraplugins to prevent those from being overridden by a subclass.
The basic functionality of a plugin manager is to proxy all unknown attributes through a ``PluginProxy`` to a list of plugins.
Note that the list of plugins *may not* be changed after the first plugin call. """
self.addPlugins(plugins) self.proxyClass = proxyClass
return iter(self.plugins)
# allow, for instance, plugins loaded via entry points to # supplant builtin plugins. if getattr(p, 'name', None) != new_name]
"""extraplugins are maintained in a separate list and re-added by loadPlugins() to prevent their being overwritten by plugins added by a subclass of PluginManager """
"""Configure the set of plugins with the given options and config instance. After configuration, disabled plugins are removed from the plugins list. """
self.addPlugin(plug)
return self._plugins
"""Access the list of plugins managed by this plugin manager""")
"""Proxy for 0.9 plugins, adapts 0.10 calls to 0.9 standard. """ self.plugin = plugin
self.plugin.add_options(parser, env)
if not hasattr(self.plugin, 'addError'): return # switch off to addSkip, addDeprecated if those types from nose.exc import SkipTest, DeprecatedTest ec, ev, tb = err if issubclass(ec, SkipTest): if not hasattr(self.plugin, 'addSkip'): return return self.plugin.addSkip(test.test) elif issubclass(ec, DeprecatedTest): if not hasattr(self.plugin, 'addDeprecated'): return return self.plugin.addDeprecated(test.test) # add capt capt = test.capturedOutput return self.plugin.addError(test.test, err, capt)
if hasattr(self.plugin, 'loadTestsFromPath'): return self.plugin.loadTestsFromPath(filename)
if not hasattr(self.plugin, 'addFailure'): return # add capt and tbinfo capt = test.capturedOutput tbinfo = test.tbinfo return self.plugin.addFailure(test.test, err, capt, tbinfo)
if not hasattr(self.plugin, 'addSuccess'): return capt = test.capturedOutput self.plugin.addSuccess(test.test, capt)
if not hasattr(self.plugin, 'startTest'): return return self.plugin.startTest(test.test)
if not hasattr(self.plugin, 'stopTest'): return return self.plugin.stopTest(test.test)
return getattr(self.plugin, val)
"""Plugin manager that loads plugins from the `nose.plugins` and `nose.plugins.0.10` entry points. """ ('nose.plugins', ZeroNinePlugin))
"""Load plugins by iterating the `nose.plugins` entry point. """ if ep.name in loaded: continue loaded[ep.name] = True log.debug('%s load plugin %s', self.__class__.__name__, ep) try: plugcls = ep.load() except KeyboardInterrupt: raise except Exception as e: # never want a plugin load to kill the test run # but we can't log here because the logger is not yet # configured warn("Unable to load plugin %s: %s" % (ep, e), RuntimeWarning) continue if adapt: plug = adapt(plugcls()) else: plug = plugcls() self.addPlugin(plug)
"""Plugin manager that loads plugins from the list in `nose.plugins.builtin`. """ """Load plugins in nose.plugins.builtin """
except ImportError: class DefaultPluginManager(BuiltinPluginManager): pass
"""Plugin manager that restricts the plugin list to those not excluded by a list of exclude methods. Any plugin that implements an excluded method will be removed from the manager's plugin list after plugins are loaded. """ DefaultPluginManager.__init__(self, plugins) self.load = load self.exclude = exclude self.excluded = [] self._excludedOpts = None
if self._excludedOpts is None: from optparse import OptionParser self._excludedOpts = OptionParser(add_help_option=False) for plugin in self.excluded: plugin.options(self._excludedOpts, env={}) return self._excludedOpts.get_option('--' + name)
if self.load: DefaultPluginManager.loadPlugins(self) allow = [] for plugin in self.plugins: ok = True for method in self.exclude: if hasattr(plugin, method): ok = False self.excluded.append(plugin) break if ok: allow.append(plugin) self.plugins = allow |