openstack/stevedore
Code Issues Proposed changes

break up monolithic module

Browse Source
This commit is contained in:
Doug Hellmann 2012-06-19 15:59:07 -04:00
parent abeae9ef53
commit 6af087440b
7 changed files with 74 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
stevedore/__init__.py
View File

@ -0,0 +1,3 @@
from .extension import ExtensionManager
from .enabled import EnabledExtensionManager
from .named import NamedExtensionManager

23
stevedore/enabled.py Normal file
View File

@ -0,0 +1,23 @@
from .extension import ExtensionManager
class EnabledExtensionManager(ExtensionManager):
"""An ExtensionManager that only loads plugins that pass a check function.
"""
def __init__(self, namespace, check_func, invoke_on_load=False,
invoke_args=(), invoke_kwds={}):
self.check_func = check_func
super(EnabledExtensionManager, self).__init__(namespace,
invoke_on_load=invoke_on_load,
invoke_args=invoke_args,
invoke_kwds=invoke_kwds,
)
def _load_one_plugin(self, ep, invoke_on_load, invoke_args, invoke_kwds):
if not self.check_func(ep):
LOG.debug('ignoring extension %r', ep.name)
return None
return super(EnabledExtensionManager, self)._load_one_plugin(
ep, invoke_on_load, invoke_args, invoke_kwds,
)

41
stevedore/extension.py
View File

@ -81,44 +81,3 @@ class ExtensionManager(object):
def __iter__(self):
return iter(self.extensions)
class EnabledExtensionManager(ExtensionManager):
"""An ExtensionManager that only loads plugins that pass a check function.
"""
def __init__(self, namespace, check_func, invoke_on_load=False,
invoke_args=(), invoke_kwds={}):
self.check_func = check_func
super(EnabledExtensionManager, self).__init__(namespace,
invoke_on_load=invoke_on_load,
invoke_args=invoke_args,
invoke_kwds=invoke_kwds,
)
def _load_one_plugin(self, ep, invoke_on_load, invoke_args, invoke_kwds):
if not self.check_func(ep):
LOG.debug('ignoring extension %r', ep.name)
return None
return super(EnabledExtensionManager, self)._load_one_plugin(
ep, invoke_on_load, invoke_args, invoke_kwds,
)
class NamedExtensionManager(EnabledExtensionManager):
"""ExtensionManager that only loads the named extensions.
This is useful for explictly enabling extensions in a
configuration file, for example.
"""
def __init__(self, namespace, names=[],
invoke_on_load=False, invoke_args=(), invoke_kwds={}):
def check(ep):
return ep.name in names
super(NamedExtensionManager, self).__init__(namespace,
check,
invoke_on_load=invoke_on_load,
invoke_args=invoke_args,
invoke_kwds=invoke_kwds,
)

20
stevedore/named.py Normal file
View File

@ -0,0 +1,20 @@
from .enabled import EnabledExtensionManager
class NamedExtensionManager(EnabledExtensionManager):
"""ExtensionManager that only loads the named extensions.
This is useful for explictly enabling extensions in a
configuration file, for example.
"""
def __init__(self, namespace, names=[],
invoke_on_load=False, invoke_args=(), invoke_kwds={}):
def check(ep):
return ep.name in names
super(NamedExtensionManager, self).__init__(namespace,
check,
invoke_on_load=invoke_on_load,
invoke_args=invoke_args,
invoke_kwds=invoke_kwds,
)

15
stevedore/tests/test_enabled.py Normal file
View File

@ -0,0 +1,15 @@
from stevedore import enabled
def test_enabled():
def check_enabled(ep):
return ep.name == 't2'
em = enabled.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t2']

26
stevedore/tests/test_extension.py
View File

@ -88,29 +88,3 @@ def test_map_errors_when_no_plugins():
em.map(mapped, 1, 2, a='A', b='B')
except RuntimeError as err:
assert 'No stevedore.test.extension.none extensions found' == str(err)
def test_enabled():
def check_enabled(ep):
return ep.name == 't2'
em = extension.EnabledExtensionManager(
'stevedore.test.extension',
check_enabled,
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t2']
def test_named():
em = extension.NamedExtensionManager(
'stevedore.test.extension',
['t1'],
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t1']

13
stevedore/tests/test_named.py Normal file
View File

@ -0,0 +1,13 @@
from stevedore import named
def test_named():
em = named.NamedExtensionManager(
'stevedore.test.extension',
['t1'],
invoke_on_load=True,
invoke_args=('a',),
invoke_kwds={'b': 'B'},
)
assert len(em.extensions) == 1
assert em.names() == ['t1']