Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self):
self.ext_plugins = stevedore.ExtensionManager(
'tempest.test_plugins', invoke_on_load=True,
propagate_map_exceptions=True,
on_load_failure_callback=self.failure_hook)
self._register_service_clients()
def _open_source_from_opt_group(self, group_name):
if not self._ext_mgr:
self._ext_mgr = stevedore.ExtensionManager(
"oslo.config.driver",
invoke_on_load=True)
self.register_opt(
StrOpt('driver',
choices=self._ext_mgr.names(),
help=_SOURCE_DRIVER_OPTION_HELP),
group=group_name)
try:
driver_name = self[group_name].driver
except ConfigFileValueError as err:
LOG.error(
"could not load configuration from %r. %s",
group_name, err.msg)
return None
def get_plugin_list():
"""Gather plugin list and cache it"""
global PLUGIN_LIST
if PLUGIN_LIST is None:
PLUGIN_LIST = stevedore.ExtensionManager(
base.PLUGIN_NAMESPACE,
invoke_on_load=False,
propagate_map_exceptions=True,
)
return PLUGIN_LIST
def load_strategies(self):
extension_manager = ExtensionManager(
namespace='watcher_strategies',
invoke_on_load=True,
)
return {ext.name: ext.plugin for ext in extension_manager.extensions}
def rule_conditions_manager():
"""Create a Stevedore extension manager for conditions in rules."""
global _CONDITIONS_MGR
if _CONDITIONS_MGR is None:
_CONDITIONS_MGR = stevedore.ExtensionManager(
'ironic_inspector.rules.conditions',
invoke_on_load=True)
return _CONDITIONS_MGR
def load_all_extensions():
"""Load all available modules with DF models"""
manager = stevedore.ExtensionManager( # noqa F841
'dragonflow.db.models',
)
def introspection_data_manager():
global _INTROSPECTION_DATA_MGR
if _INTROSPECTION_DATA_MGR is None:
_INTROSPECTION_DATA_MGR = stevedore.ExtensionManager(
'ironic_inspector.introspection_data.store',
invoke_on_load=True)
return _INTROSPECTION_DATA_MGR
def load(self, name):
if name in self._unloaded:
self.impls[name] = self._unloaded[name]()
return self.impls[name]
if name in self.impls:
return self.impls[name]
else: # pragma NO COVERAGE
if self._mgr is None:
self._mgr = stevedore.ExtensionManager(self.group)
try:
self.impls[name] = self._mgr[name].plugin
return self.impls[name]
except KeyError:
raise self.NotFound(
"Can't load plugin %s %s" % (self.group, name)
)
def __init__(self, dispatcher: EventDispatcher):
self._configs = {}
self._features = {}
self._pipelines = {}
self._variables = OrderedDict()
self.dispatcher = dispatcher
self.resources = OrderedDict()
def register_feature(ext):
self._features[ext.name] = ext.plugin
def on_load_feature_failure(mgr, entrypoint, err):
logger.exception("Exception caught while loading {}.".format(entrypoint), err)
mgr = ExtensionManager(namespace="medikit.feature", on_load_failure_callback=on_load_feature_failure)
mgr.map(register_feature)
dispatcher.add_listener(medikit.on_end, self.write_resources)
'e3.anod.sandbox.sandbox_action': [
'foo = e3_contrib.sandbox_actions.SandBoxFoo']
}
:param get_argument_parser: return e3.main.Main argument_parser instead
of running the action.
"""
m = Main()
m.parse_args(known_args_only=True)
subparsers = m.argument_parser.add_subparsers(
title="action", description="valid actions"
)
# Load all sandbox actions plugins
ext = stevedore.ExtensionManager(
namespace="e3.anod.sandbox.sandbox_action",
invoke_on_load=True,
invoke_args=(subparsers,),
)
if len(ext.names()) != len(ext.entry_points_names()):
raise SandBoxError(
"an error occured when loading sandbox_action entry points %s"
% ",".join(ext.entry_points_names())
) # defensive code
if get_argument_parser:
return m.argument_parser
args = m.argument_parser.parse_args()