diff --git a/setup.py b/setup.py index d15f2ae..172192a 100644 --- a/setup.py +++ b/setup.py @@ -57,7 +57,12 @@ setup( # only_in_packages=False, # ), - entry_points={}, + entry_points={ + 'stevedore.test.extension': [ + 't1 = stevedore.tests.test_extension:FauxExtension', + 't2 = stevedore.tests.test_extension:FauxExtension', + ], + }, zip_safe=False, ) diff --git a/stevedore/extension.py b/stevedore/extension.py new file mode 100644 index 0000000..8771a2c --- /dev/null +++ b/stevedore/extension.py @@ -0,0 +1,43 @@ +"""ExtensionManager +""" + +import pkg_resources + +import logging + + +LOG = logging.getLogger(__name__) + + +class Extension(object): + + def __init__(self, name, entry_point, plugin, obj): + self.name = name + self.entry_point = entry_point + self.plugin = plugin + self.obj = obj + + +class ExtensionManager(object): + + def __init__(self, namespace, invoke_on_load=False, invoke_args=(), invoke_kwds={}): + self.namespace = namespace + self.extensions = [] + for ep in pkg_resources.iter_entry_points(self.namespace): + LOG.debug('found extension %r', ep.name) + try: + plugin = ep.load() + if invoke_on_load: + obj = plugin(ep.name, *invoke_args, **invoke_kwds) + else: + obj = None + except KeyboardInterrupt: + raise + except Exception as err: + LOG.error('Could not load %s: %s', ep.name, err) + else: + self.extensions.append(Extension(ep.name, ep, plugin, obj)) + return + + def names(self): + return [e.name for e in self.extensions] diff --git a/stevedore/tests/__init__.py b/stevedore/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/stevedore/tests/test_extension.py b/stevedore/tests/test_extension.py new file mode 100644 index 0000000..c12ca7c --- /dev/null +++ b/stevedore/tests/test_extension.py @@ -0,0 +1,29 @@ +"""Tests for stevedore.extension +""" + +from stevedore import extension + + +class FauxExtension(object): + def __init__(self, name, *args, **kwds): + self.name = name + self.args = args + self.kwds = kwds + + +def test_detect_plugins(): + em = extension.ExtensionManager('stevedore.test.extension') + names = sorted(em.names()) + assert names == ['t1', 't2'] + + +def test_invoke_on_load(): + em = extension.ExtensionManager('stevedore.test.extension', + invoke_on_load=True, + invoke_args=('a',), + invoke_kwds={'b': 'B'}, + ) + assert len(em.extensions) == 2 + for e in em.extensions: + assert e.obj.args == ('a',) + assert e.obj.kwds == {'b': 'B'}