Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.server_action_regex,
self.attach_volume_to_server_regex,
]
self.resize_resource_actions = [
self.resize_server_action,
]
self.stop_resource_actions = [
self.stop_server_action,
]
self.start_resource_actions = [
self.start_server_action,
]
self.restore_resource_actions = [
self.restore_server_action,
]
self.product_items = extension.ExtensionManager(
namespace='gringotts.server.product_items',
invoke_on_load=True,
invoke_args=(self.gclient,))
def _register_dynamic_action_classes(namespace=''):
extensions = extension.ExtensionManager(
namespace='mistral.generators',
invoke_on_load=True
)
for ext in extensions:
for generator in ext.obj:
_register_actions(generator, namespace)
def load_namespace(self, ns):
"""Load commands from namespace.
:param ns: namespace name
:type ns: str
"""
mgr = extension.ExtensionManager(namespace=ns,
verify_requirements=True,
on_load_failure_callback=self._on_failure)
# Load commands from ns
for ext in mgr.extensions:
try:
obj = ext.plugin(ext.name)
except Exception as err:
self._on_failure(self, ext, err)
obj = None
ext.obj = obj
self.mgrs.append(mgr)
def get_evaluators():
global _EXP_EVALUATORS
if _EXP_EVALUATORS is None:
_EXP_EVALUATORS = {}
mgr = extension.ExtensionManager(
namespace=_EXP_EVALUATOR_NAMESPACE,
invoke_on_load=False
)
for name in mgr.names():
_EXP_EVALUATORS[name] = get_evaluator(name)
return _EXP_EVALUATORS
def get_transformers():
transformers = {}
transformer_exts = extension.ExtensionManager(
TRANSFORMERS_NAMESPACE,
invoke_on_load=True)
for transformer in transformer_exts:
t_name = transformer.name
t_obj = transformer.obj
transformers[t_name] = t_obj
return transformers
def get_available_backends():
"""
Return names of the available / installed authentication backends.
:rtype: ``list`` of ``str``
"""
from stevedore.extension import ExtensionManager
manager = ExtensionManager(namespace=BACKENDS_NAMESPACE, invoke_on_load=False)
return manager.names()
def load_plugins(self, plugins_namespace):
self.plugins_mgr = extension.ExtensionManager(
namespace=plugins_namespace,
invoke_on_load=False,
verify_requirements=False,
)
def test_has_id(plugin):
if not hasattr(plugin.plugin, "_test_id"):
# logger not setup yet, so using print
print("WARNING: Test '%s' has no ID, skipping." % plugin.name,
file=sys.stderr)
return False
return True
self.plugins = list(filter(test_has_id, list(self.plugins_mgr)))
self.plugin_names = [plugin.name for plugin in self.plugins]
self.plugins_by_id = {p.plugin._test_id: p for p in self.plugins}
def discover_auth_systems():
"""Discover the available auth-systems.
This won't take into account the old style auth-systems.
"""
global _discovered_plugins
_discovered_plugins = {}
def add_plugin(ext):
_discovered_plugins[ext.name] = ext.plugin
ep_namespace = "ironicclient.openstack.common.apiclient.auth"
mgr = extension.ExtensionManager(ep_namespace)
mgr.map(add_plugin)
def discover_auth_systems():
"""Discover the available auth-systems.
This won't take into account the old style auth-systems.
"""
global _discovered_plugins
_discovered_plugins = {}
def add_plugin(ext):
_discovered_plugins[ext.name] = ext.plugin
ep_namespace = "magnumclient.openstack.common.apiclient.auth"
mgr = extension.ExtensionManager(ep_namespace)
mgr.map(add_plugin)
def _get_setter_component_manager():
"""stevedore extension manager for setter components."""
return extension.ExtensionManager(
namespace=GenericTransformBuilder.
_MONASCA_TRANSFORM_SETTER_NAMESPACE,
on_load_failure_callback=GenericTransformBuilder.
log_load_extension_error,
invoke_on_load=False)