from abc import ABCMeta
import imp
+import itertools
import os
from oslo.config import cfg
def __init__(self, path, plugins):
self.plugins = plugins
super(PluginAwareExtensionManager, self).__init__(path)
+ self.check_if_plugin_extensions_loaded()
def _check_extension(self, extension):
"""Check if an extension is supported by any plugin."""
NeutronManager.get_service_plugins())
return cls._instance
+ def check_if_plugin_extensions_loaded(self):
+ """Check if an extension supported by a plugin has been loaded."""
+ plugin_extensions = set(itertools.chain.from_iterable([
+ getattr(plugin, "supported_extension_aliases", [])
+ for plugin in self.plugins.values()]))
+ missing_aliases = plugin_extensions - set(self.extensions)
+ if missing_aliases:
+ raise exceptions.ExtensionsNotFound(
+ extensions=list(missing_aliases))
+
class RequestExtension(object):
"""Extend requests and responses of core Neutron OpenStack API controllers.
paths = ':'.join([cfg.CONF.api_extensions_path, paths])
return paths
+
+
+def append_api_extensions_path(paths):
+ paths = [cfg.CONF.api_extensions_path] + paths
+ cfg.CONF.set_override('api_extensions_path',
+ ':'.join([p for p in paths if p]))
message = _("Invalid extension environment: %(reason)s")
+class ExtensionsNotFound(NotFound):
+ message = _("Extensions not found: %(extensions)s")
+
+
class InvalidContentType(NeutronException):
message = _("Invalid content type %(content_type)s")
import copy
import httplib
import json
-import os
import socket
from oslo.config import cfg
+from neutron.api import extensions as neutron_extensions
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.common import constants as const
from neutron.common import exceptions
from neutron.openstack.common import log as logging
from neutron.openstack.common import rpc
from neutron.plugins.bigswitch.db import porttracker_db
+from neutron.plugins.bigswitch import extensions
from neutron.plugins.bigswitch import routerrule_db
from neutron.plugins.bigswitch.version import version_string_with_vcs
LOG = logging.getLogger(__name__)
-# Include the BigSwitch Extensions path in the api_extensions
-EXTENSIONS_PATH = os.path.join(os.path.dirname(__file__), 'extensions')
-if not cfg.CONF.api_extensions_path:
- cfg.CONF.set_override('api_extensions_path',
- EXTENSIONS_PATH)
-
restproxy_opts = [
cfg.StrOpt('servers', default='localhost:8800',
help=_("A comma separated list of BigSwitch or Floodlight "
# init DB, proxy's persistent store defaults to in-memory sql-lite DB
db.configure_db()
+ # Include the BigSwitch Extensions path in the api_extensions
+ neutron_extensions.append_api_extensions_path(extensions.__path__)
+
# 'servers' is the list of network controller REST end-points
# (used in order specified till one suceeds, and it is sticky
# till next failure). Use 'server_auth' to encode api-key
import eventlet
-from oslo.config import cfg as q_conf
-
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
# bulk operations.
__native_bulk_support = False
supported_extension_aliases = ["provider", "agent",
- "policy_profile_binding",
- "network_profile_binding",
"n1kv_profile", "network_profile",
"policy_profile", "external-net", "router",
"credential"]
n1kv_db_v2.initialize()
c_cred.Store.initialize()
self._initialize_network_ranges()
- # If no api_extensions_path is provided set the following
- if not q_conf.CONF.api_extensions_path:
- q_conf.CONF.set_override(
- 'api_extensions_path',
- 'extensions:neutron/plugins/cisco/extensions')
self._setup_vsm()
self._setup_rpc()
from sqlalchemy import orm
import webob.exc as wexc
+from neutron.api import extensions as neutron_extensions
from neutron.api.v2 import base
from neutron.common import exceptions as exc
from neutron.db import db_base_plugin_v2
from neutron.plugins.cisco.common import cisco_exceptions as cexc
from neutron.plugins.cisco.common import config
from neutron.plugins.cisco.db import network_db_v2 as cdb
+from neutron.plugins.cisco import extensions
LOG = logging.getLogger(__name__)
class PluginV2(db_base_plugin_v2.NeutronDbPluginV2):
"""Meta-Plugin with v2 API support for multiple sub-plugins."""
- supported_extension_aliases = ["Cisco Credential", "Cisco qos"]
+ supported_extension_aliases = ["credential", "Cisco qos"]
_methods_to_delegate = ['create_network',
'delete_network', 'update_network', 'get_network',
'get_networks',
self.supported_extension_aliases.extend(
self._model.supported_extension_aliases)
+ neutron_extensions.append_api_extensions_path(extensions.__path__)
+
# Extend the fault map
self._extend_fault_map()
def __init__(self, configfile=None):
LOG.debug(_("Start initializing metaplugin"))
- self.supported_extension_aliases = \
- cfg.CONF.META.supported_extension_aliases.split(',')
- self.supported_extension_aliases += ['flavor', 'external-net',
- 'router', 'ext-gw-mode',
- 'extraroute']
+ self.supported_extension_aliases = ['flavor', 'external-net',
+ 'router', 'ext-gw-mode',
+ 'extraroute']
+ if cfg.CONF.META.supported_extension_aliases:
+ cfg_aliases = cfg.CONF.META.supported_extension_aliases.split(',')
+ self.supported_extension_aliases += cfg_aliases
# Ignore config option overapping
def _is_opt_registered(opts, opt):
securitygroups_db.SecurityGroupDbMixin):
supported_extension_aliases = ['external-net', 'router', 'security-group',
- 'agent' 'dhcp_agent_scheduler', 'binding']
+ 'agent', 'dhcp_agent_scheduler', 'binding']
__native_bulk_support = False
def __init__(self):
class Packetfilter(extensions.ExtensionDescriptor):
-
@classmethod
def get_name(cls):
return ALIAS
COLLECTION, resource, attr_map=PACKET_FILTER_ATTR_PARAMS)
return [pf_ext]
- def update_attributes_map(self, attributes):
- super(Packetfilter, self).update_attributes_map(
- attributes, extension_attrs_map=PACKET_FILTER_ATTR_MAP)
-
def get_extended_resources(self, version):
if version == "2.0":
return PACKET_FILTER_ATTR_MAP
# License for the specific language governing permissions and limitations
# under the License.
-from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.openstack.common import log as logging
}
-class Router_provider(extensions.ExtensionDescriptor):
+class Router_provider(object):
@classmethod
def get_name(cls):
return "Router Provider"
# @author: Akihiro MOTOKI
from neutron.agent import securitygroups_rpc as sg_rpc
+from neutron.api import extensions as neutron_extensions
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.api.v2 import attributes as attrs
from neutron.common import constants as const
from neutron.plugins.nec.common import exceptions as nexc
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.db import router as rdb
+from neutron.plugins.nec import extensions
from neutron.plugins.nec import nec_router
from neutron.plugins.nec import ofc_manager
from neutron.plugins.nec import packet_filter
self.ofc = ofc_manager.OFCManager()
self.base_binding_dict = self._get_base_binding_dict()
portbindings_base.register_port_dict_function()
- # Set the plugin default extension path
- # if no api_extensions_path is specified.
- if not config.CONF.api_extensions_path:
- config.CONF.set_override('api_extensions_path',
- 'neutron/plugins/nec/extensions')
+
+ neutron_extensions.append_api_extensions_path(extensions.__path__)
self.setup_rpc()
self.l3_rpc_notifier = nec_router.L3AgentNotifyAPI()
from sqlalchemy.orm import exc as sa_exc
import webob.exc
+from neutron.api import extensions as neutron_extensions
from neutron.api.v2 import attributes as attr
from neutron.api.v2 import base
from neutron.common import constants
'default': self._nvp_delete_port}
}
- # If no api_extensions_path is provided set the following
- if not cfg.CONF.api_extensions_path:
- cfg.CONF.set_override('api_extensions_path', NVP_EXT_PATH)
+ neutron_extensions.append_api_extensions_path([NVP_EXT_PATH])
self.nvp_opts = cfg.CONF.NVP
self.nvp_sync_opts = cfg.CONF.NVP_SYNC
self.cluster = create_nvp_cluster(cfg.CONF,
# @author: Abhishek Raut, Cisco Systems Inc.
from mock import patch
-import os
from oslo.config import cfg
+from neutron.api import extensions as neutron_extensions
from neutron.api.v2 import attributes
from neutron.common.test_lib import test_config
from neutron import context
n1kv_neutron_plugin.N1kvNeutronPluginV2._setup_vsm = _fake_setup_vsm
test_config['plugin_name_v2'] = self._plugin_name
- cfg.CONF.set_override('api_extensions_path',
- os.path.dirname(extensions.__file__))
+ neutron_extensions.append_api_extensions_path(extensions.__path__)
self.addCleanup(cfg.CONF.reset)
ext_mgr = NetworkProfileTestExtensionManager()
test_config['extension_manager'] = ext_mgr
from neutron import manager
from neutron.plugins.nicira.dbexts import nicira_networkgw_db
from neutron.plugins.nicira.extensions import nvp_networkgw as networkgw
+from neutron.plugins.nicira.NeutronPlugin import NVP_EXT_PATH
from neutron import quota
from neutron.tests import base
from neutron.tests.unit import test_api_v2
supported_extension_aliases = ["network-gateway"]
+ def __init__(self, **args):
+ super(TestNetworkGatewayPlugin, self).__init__(**args)
+ extensions.append_api_extensions_path([NVP_EXT_PATH])
+
def delete_port(self, context, id, nw_gw_port_check=True):
if nw_gw_port_check:
port = self._get_port(context, id)
import os
+import mock
import routes
import webob
import webtest
from neutron.api import extensions
from neutron.common import config
+from neutron.common import exceptions
from neutron.db import db_base_plugin_v2
from neutron.openstack.common import jsonutils
from neutron.openstack.common import log as logging
def test_unsupported_extensions_are_not_loaded(self):
stub_plugin = ext_stubs.StubPlugin(supported_extensions=["e1", "e3"])
plugin_info = {constants.CORE: stub_plugin}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(ext_stubs.StubExtension("e1"))
- ext_mgr.add_extension(ext_stubs.StubExtension("e2"))
- ext_mgr.add_extension(ext_stubs.StubExtension("e3"))
+ ext_mgr.add_extension(ext_stubs.StubExtension("e1"))
+ ext_mgr.add_extension(ext_stubs.StubExtension("e2"))
+ ext_mgr.add_extension(ext_stubs.StubExtension("e3"))
- self.assertIn("e1", ext_mgr.extensions)
- self.assertNotIn("e2", ext_mgr.extensions)
- self.assertIn("e3", ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
+ self.assertNotIn("e2", ext_mgr.extensions)
+ self.assertIn("e3", ext_mgr.extensions)
def test_extensions_are_not_loaded_for_plugins_unaware_of_extensions(self):
class ExtensionUnawarePlugin(object):
supported_extension_aliases = ["supported_extension"]
plugin_info = {constants.CORE: PluginWithoutExpectedIface()}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(
- ext_stubs.ExtensionExpectingPluginInterface("supported_extension"))
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ ext_mgr.add_extension(ext_stubs.ExtensionExpectingPluginInterface(
+ "supported_extension"))
- self.assertNotIn("e1", ext_mgr.extensions)
+ self.assertNotIn("e1", ext_mgr.extensions)
def test_extensions_are_loaded_for_plugin_with_expected_interface(self):
pass
plugin_info = {constants.CORE: PluginWithExpectedInterface()}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(
- ext_stubs.ExtensionExpectingPluginInterface("supported_extension"))
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ ext_mgr.add_extension(ext_stubs.ExtensionExpectingPluginInterface(
+ "supported_extension"))
- self.assertIn("supported_extension", ext_mgr.extensions)
+ self.assertIn("supported_extension", ext_mgr.extensions)
def test_extensions_expecting_neutron_plugin_interface_are_loaded(self):
class ExtensionForQuamtumPluginInterface(ext_stubs.StubExtension):
pass
stub_plugin = ext_stubs.StubPlugin(supported_extensions=["e1"])
plugin_info = {constants.CORE: stub_plugin}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
- self.assertIn("e1", ext_mgr.extensions)
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
+
+ self.assertIn("e1", ext_mgr.extensions)
def test_extensions_without_need_for__plugin_interface_are_loaded(self):
class ExtensionWithNoNeedForPluginInterface(ext_stubs.StubExtension):
stub_plugin = ext_stubs.StubPlugin(supported_extensions=["e1"])
plugin_info = {constants.CORE: stub_plugin}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
- self.assertIn("e1", ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
def test_extension_loaded_for_non_core_plugin(self):
class NonCorePluginExtenstion(ext_stubs.StubExtension):
stub_plugin = ext_stubs.StubPlugin(supported_extensions=["e1"])
plugin_info = {constants.DUMMY: stub_plugin}
- ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
- ext_mgr.add_extension(NonCorePluginExtenstion("e1"))
+ with mock.patch("neutron.api.extensions.PluginAwareExtensionManager."
+ "check_if_plugin_extensions_loaded"):
+ ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
+ ext_mgr.add_extension(NonCorePluginExtenstion("e1"))
- self.assertIn("e1", ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
+
+ def test_unloaded_supported_extensions_raises_exception(self):
+ stub_plugin = ext_stubs.StubPlugin(
+ supported_extensions=["unloaded_extension"])
+ plugin_info = {constants.CORE: stub_plugin}
+ self.assertRaises(exceptions.ExtensionsNotFound,
+ extensions.PluginAwareExtensionManager,
+ '', plugin_info)
class ExtensionControllerTest(testlib_api.WebTestCase):