root/galaxy-central/eggs/nose-0.11.1-py2.6.egg/nose/plugins/manager.py

リビジョン 3, 14.3 KB (コミッタ: kohda, 14 年 前)

Install Unix tools  http://hannonlab.cshl.edu/galaxy_unix_tools/galaxy.html

行番号 
1"""
2Plugin Manager
3--------------
4
5A plugin manager class is used to load plugins, manage the list of
6loaded plugins, and proxy calls to those plugins.
7
8The plugin managers provided with nose are:
9
10:class:`PluginManager`
11    This manager doesn't implement loadPlugins, so it can only work
12    with a static list of plugins.
13
14:class:`BuiltinPluginManager`
15    This manager loads plugins referenced in ``nose.plugins.builtin``.
16
17:class:`EntryPointPluginManager`
18    This manager uses setuptools entrypoints to load plugins.
19
20:class:`DefaultPluginMananger`
21    This is the manager class that will be used by default. If
22    setuptools is installed, it is a subclass of
23    :class:`EntryPointPluginManager` and :class:`BuiltinPluginManager`;
24    otherwise, an alias to :class:`BuiltinPluginManager`.
25
26:class:`RestrictedPluginManager`
27    This manager is for use in test runs where some plugin calls are
28    not available, such as runs started with ``python setup.py test``,
29    where the test runner is the default unittest :class:`TextTestRunner`. It
30    is a subclass of :class:`DefaultPluginManager`.
31
32Writing a plugin manager
33========================
34
35If you want to load plugins via some other means, you can write a
36plugin manager and pass an instance of your plugin manager class when
37instantiating the :class:`nose.config.Config` instance that you pass to
38:class:`TestProgram` (or :func:`main` or :func:`run`).
39
40To implement your plugin loading scheme, implement ``loadPlugins()``,
41and in that method, call ``addPlugin()`` with an instance of each plugin
42you wish to make available. Make sure to call
43``super(self).loadPlugins()`` as well if have subclassed a manager
44other than ``PluginManager``.
45
46"""
47import inspect
48import logging
49import os
50import sys
51from warnings import warn
52from nose.failure import Failure
53from nose.plugins.base import IPluginInterface
54
55__all__ = ['DefaultPluginManager', 'PluginManager', 'EntryPointPluginManager',
56           'BuiltinPluginManager', 'RestrictedPluginManager']
57
58log = logging.getLogger(__name__)
59
60
61class PluginProxy(object):
62    """Proxy for plugin calls. Essentially a closure bound to the
63    given call and plugin list.
64
65    The plugin proxy also must be bound to a particular plugin
66    interface specification, so that it knows what calls are available
67    and any special handling that is required for each call.
68    """
69    interface = IPluginInterface
70    def __init__(self, call, plugins):
71        try:
72            self.method = getattr(self.interface, call)
73        except AttributeError:
74            raise AttributeError("%s is not a valid %s method"
75                                 % (call, self.interface.__name__))
76        self.call = self.makeCall(call)
77        self.plugins = []
78        for p in plugins:
79            self.addPlugin(p, call)
80   
81    def __call__(self, *arg, **kw):
82        return self.call(*arg, **kw)
83   
84    def addPlugin(self, plugin, call):
85        """Add plugin to my list of plugins to call, if it has the attribute
86        I'm bound to.
87        """
88        meth = getattr(plugin, call, None)
89        if meth is not None:
90            if call == 'loadTestsFromModule' and \
91                    len(inspect.getargspec(meth)[0]) == 2:
92                orig_meth = meth
93                meth = lambda module, path, **kwargs: orig_meth(module)
94            self.plugins.append((plugin, meth))
95
96    def makeCall(self, call):
97        if call == 'loadTestsFromNames':
98            # special case -- load tests from names behaves somewhat differently
99            # from other chainable calls, because plugins return a tuple, only
100            # part of which can be chained to the next plugin.
101            return self._loadTestsFromNames
102
103        meth = self.method
104        if getattr(meth, 'generative', False):
105            # call all plugins and yield a flattened iterator of their results
106            return lambda *arg, **kw: list(self.generate(*arg, **kw))
107        elif getattr(meth, 'chainable', False):
108            return self.chain
109        else:
110            # return a value from the first plugin that returns non-None
111            return self.simple       
112           
113    def chain(self, *arg, **kw):
114        """Call plugins in a chain, where the result of each plugin call is
115        sent to the next plugin as input. The final output result is returned.
116        """
117        result = None
118        # extract the static arguments (if any) from arg so they can
119        # be passed to each plugin call in the chain
120        static = [a for (static, a)
121                  in zip(getattr(self.method, 'static_args', []), arg)
122                  if static]
123        for p, meth in self.plugins:
124            result = meth(*arg, **kw)
125            arg = static[:]
126            arg.append(result)
127        return result
128
129    def generate(self, *arg, **kw):
130        """Call all plugins, yielding each item in each non-None result.
131        """
132        for p, meth in self.plugins:
133            result = None
134            try:
135                result = meth(*arg, **kw)
136                if result is not None:
137                    for r in result:
138                        yield r
139            except (KeyboardInterrupt, SystemExit):
140                raise
141            except:
142                exc = sys.exc_info()
143                yield Failure(*exc)
144                continue
145
146    def simple(self, *arg, **kw):
147        """Call all plugins, returning the first non-None result.
148        """
149        for p, meth in self.plugins:
150            result = meth(*arg, **kw)
151            if result is not None:
152                return result
153
154    def _loadTestsFromNames(self, names, module=None):
155        """Chainable but not quite normal. Plugins return a tuple of
156        (tests, names) after processing the names. The tests are added
157        to a suite that is accumulated throughout the full call, while
158        names are input for the next plugin in the chain.
159        """
160        suite = []
161        for p, meth in self.plugins:
162            result = meth(names, module=module)
163            if result is not None:
164                suite_part, names = result
165                if suite_part:
166                    suite.extend(suite_part)
167        return suite, names
168
169
170class NoPlugins(object):
171    """Null Plugin manager that has no plugins."""
172    interface = IPluginInterface
173    def __init__(self):
174        self.plugins = ()
175
176    def __iter__(self):
177        return ()
178
179    def _doNothing(self, *args, **kwds):
180        pass
181
182    def _emptyIterator(self, *args, **kwds):
183        return ()
184
185    def __getattr__(self, call):
186        method = getattr(self.interface, call)
187        if getattr(method, "generative", False):
188            return self._emptyIterator
189        else:
190            return self._doNothing
191
192    def addPlugin(self, plug):
193        raise NotImplementedError()
194
195    def addPlugins(self, plugins):
196        raise NotImplementedError()
197
198    def configure(self, options, config):
199        pass
200
201    def loadPlugins(self):
202        pass
203
204    def sort(self, cmpf=None):
205        pass
206
207
208class PluginManager(object):
209    """Base class for plugin managers. Does not implement loadPlugins, so it
210    may only be used with a static list of plugins.
211
212    The basic functionality of a plugin manager is to proxy all unknown
213    attributes through a ``PluginProxy`` to a list of plugins.
214
215    Note that the list of plugins *may not* be changed after the first plugin
216    call.
217    """
218    proxyClass = PluginProxy
219   
220    def __init__(self, plugins=(), proxyClass=None):
221        self._plugins = []
222        self._proxies = {}
223        if plugins:
224            self.addPlugins(plugins)
225        if proxyClass is not None:
226            self.proxyClass = proxyClass
227       
228    def __getattr__(self, call):
229        try:
230            return self._proxies[call]
231        except KeyError:
232            proxy = self.proxyClass(call, self._plugins)
233            self._proxies[call] = proxy
234        return proxy
235
236    def __iter__(self):
237        return iter(self.plugins)
238
239    def addPlugin(self, plug):
240        self._plugins.append(plug)
241
242    def addPlugins(self, plugins):
243        for plug in plugins:
244            self.addPlugin(plug)
245
246    def configure(self, options, config):
247        """Configure the set of plugins with the given options
248        and config instance. After configuration, disabled plugins
249        are removed from the plugins list.
250        """
251        log.debug("Configuring plugins")
252        self.config = config
253        cfg = PluginProxy('configure', self._plugins)
254        cfg(options, config)
255        enabled = [plug for plug in self._plugins if plug.enabled]
256        self.plugins = enabled
257        self.sort()
258        log.debug("Plugins enabled: %s", enabled)
259
260    def loadPlugins(self):
261        pass
262
263    def sort(self, cmpf=None):
264        if cmpf is None:
265            cmpf = lambda a, b: cmp(getattr(b, 'score', 1),
266                                    getattr(a, 'score', 1))
267        self._plugins.sort(cmpf)
268
269    def _get_plugins(self):
270        return self._plugins
271
272    def _set_plugins(self, plugins):
273        self._plugins = []
274        self.addPlugins(plugins)
275
276    plugins = property(_get_plugins, _set_plugins, None,
277                       """Access the list of plugins managed by
278                       this plugin manager""")
279
280
281class ZeroNinePlugin:
282    """Proxy for 0.9 plugins, adapts 0.10 calls to 0.9 standard.
283    """
284    def __init__(self, plugin):
285        self.plugin = plugin
286
287    def options(self, parser, env=os.environ):
288        self.plugin.add_options(parser, env)
289   
290    def addError(self, test, err):
291        if not hasattr(self.plugin, 'addError'):
292            return
293        # switch off to addSkip, addDeprecated if those types
294        from nose.exc import SkipTest, DeprecatedTest
295        ec, ev, tb = err
296        if issubclass(ec, SkipTest):
297            if not hasattr(self.plugin, 'addSkip'):
298                return
299            return self.plugin.addSkip(test.test)
300        elif issubclass(ec, DeprecatedTest):
301            if not hasattr(self.plugin, 'addDeprecated'):
302                return
303            return self.plugin.addDeprecated(test.test)           
304        # add capt
305        capt = test.capturedOutput
306        return self.plugin.addError(test.test, err, capt)
307
308    def loadTestsFromFile(self, filename):
309        if hasattr(self.plugin, 'loadTestsFromPath'):
310            return self.plugin.loadTestsFromPath(filename)
311
312    def addFailure(self, test, err):
313        if not hasattr(self.plugin, 'addFailure'):
314            return
315        # add capt and tbinfo
316        capt = test.capturedOutput
317        tbinfo = test.tbinfo
318        return self.plugin.addFailure(test.test, err, capt, tbinfo)
319
320    def addSuccess(self, test):
321        if not hasattr(self.plugin, 'addSuccess'):
322            return
323        capt = test.capturedOutput
324        self.plugin.addSuccess(test.test, capt)
325
326    def startTest(self, test):
327        if not hasattr(self.plugin, 'startTest'):
328            return
329        return self.plugin.startTest(test.test)
330
331    def stopTest(self, test):
332        if not hasattr(self.plugin, 'stopTest'):
333            return
334        return self.plugin.stopTest(test.test)
335
336    def __getattr__(self, val):
337        return getattr(self.plugin, val)
338
339           
340class EntryPointPluginManager(PluginManager):
341    """Plugin manager that loads plugins from the `nose.plugins` and
342    `nose.plugins.0.10` entry points.
343    """
344    entry_points = (('nose.plugins.0.10', None),
345                    ('nose.plugins', ZeroNinePlugin))
346   
347    def loadPlugins(self):
348        """Load plugins by iterating the `nose.plugins` entry point.
349        """
350        super(EntryPointPluginManager, self).loadPlugins()
351        from pkg_resources import iter_entry_points
352
353        loaded = {}
354        for entry_point, adapt in self.entry_points:
355            for ep in iter_entry_points(entry_point):
356                if ep.name in loaded:
357                    continue
358                loaded[ep.name] = True
359                log.debug('%s load plugin %s', self.__class__.__name__, ep)
360                try:
361                    plugcls = ep.load()
362                except KeyboardInterrupt:
363                    raise
364                except Exception, e:
365                    # never want a plugin load to kill the test run
366                    # but we can't log here because the logger is not yet
367                    # configured
368                    warn("Unable to load plugin %s: %s" % (ep, e),
369                         RuntimeWarning)
370                    continue
371                if adapt:
372                    plug = adapt(plugcls())
373                else:
374                    plug = plugcls()
375                self.addPlugin(plug)
376
377
378class BuiltinPluginManager(PluginManager):
379    """Plugin manager that loads plugins from the list in
380    `nose.plugins.builtin`.
381    """
382    def loadPlugins(self):
383        """Load plugins in nose.plugins.builtin
384        """
385        super(BuiltinPluginManager, self).loadPlugins()
386        from nose.plugins import builtin
387        for plug in builtin.plugins:
388            self.addPlugin(plug())
389       
390try:
391    import pkg_resources
392    class DefaultPluginManager(BuiltinPluginManager, EntryPointPluginManager):
393        pass
394except ImportError:
395    DefaultPluginManager = BuiltinPluginManager
396
397
398class RestrictedPluginManager(DefaultPluginManager):
399    """Plugin manager that restricts the plugin list to those not
400    excluded by a list of exclude methods. Any plugin that implements
401    an excluded method will be removed from the manager's plugin list
402    after plugins are loaded.
403    """
404    def __init__(self, plugins=(), exclude=(), load=True):
405        DefaultPluginManager.__init__(self, plugins)
406        self.load = load
407        self.exclude = exclude
408        self.excluded = []
409        self._excludedOpts = None
410       
411    def excludedOption(self, name):
412        if self._excludedOpts is None:
413            from optparse import OptionParser
414            self._excludedOpts = OptionParser(add_help_option=False)
415            for plugin in self.excluded:
416                plugin.options(self._excludedOpts, env={})
417        return self._excludedOpts.get_option('--' + name)
418       
419    def loadPlugins(self):
420        if self.load:
421            DefaultPluginManager.loadPlugins(self)
422        allow = []
423        for plugin in self.plugins:
424            ok = True
425            for method in self.exclude:
426                if hasattr(plugin, method):
427                    ok = False
428                    self.excluded.append(plugin)
429                    break
430            if ok:
431                allow.append(plugin)
432        self.plugins = allow
433
434   
Note: リポジトリブラウザについてのヘルプは TracBrowser を参照してください。