from quantum.openstack.common import importutils
from quantum.plugins.cisco.common import cisco_constants as const
+from quantum.plugins.cisco.common import cisco_credentials_v2 as cred
from quantum.plugins.cisco.db import network_db_v2 as cdb
from quantum.plugins.cisco import l2network_plugin_configuration as conf
from quantum import quantum_plugin_base_v2
configured, and load the inventories those device plugins for which the
inventory is configured
"""
+ cdb.initialize()
+ cred.Store.initialize()
self._vlan_mgr = importutils.import_object(conf.MANAGER_CLASS)
for key in conf.PLUGINS[const.PLUGINS].keys():
plugin_obj = conf.PLUGINS[const.PLUGINS][key]
One or more servers to a nexus switch.
"""
MANAGE_STATE = True
+ __native_bulk_support = True
supported_extension_aliases = []
_plugins = {}
_inventory = {}
_methods_to_delegate = ['update_network', 'get_network', 'get_networks',
- 'create_port', 'delete_port', 'update_port',
- 'get_port', 'get_ports',
- 'create_subnet', 'delete_subnet', 'update_subnet',
- 'get_subnet', 'get_subnets']
+ 'create_port', 'create_port_bulk', 'delete_port',
+ 'update_port', 'get_port', 'get_ports',
+ 'create_subnet', 'create_subnet_bulk',
+ 'delete_subnet', 'update_subnet', 'get_subnet',
+ 'get_subnets', ]
def __init__(self):
"""
LOG.debug("Loaded device inventory %s\n" %
conf.PLUGINS[const.INVENTORY][key])
+ if hasattr(self._plugins[const.VSWITCH_PLUGIN],
+ "supported_extension_aliases"):
+ self.supported_extension_aliases.extend(
+ self._plugins[const.VSWITCH_PLUGIN].
+ supported_extension_aliases)
+
LOG.debug("%s.%s init done" % (__name__, self.__class__.__name__))
def __getattribute__(self, name):
- methods = object.__getattribute__(self, "_methods_to_delegate")
+ """
+ This delegates the calls to the methods implemented only by the OVS
+ sub-plugin.
+ """
+ super_getattr = super(VirtualPhysicalSwitchModelV2,
+ self).__getattribute__
+ methods = super_getattr('_methods_to_delegate')
+
if name in methods:
- return getattr(object.__getattribute__(self, "_plugins")
- [const.VSWITCH_PLUGIN], name)
- else:
- return object.__getattribute__(self, name)
+ plugin = super_getattr('_plugins')[const.VSWITCH_PLUGIN]
+ return getattr(plugin, name)
+
+ try:
+ return super_getattr(name)
+ except AttributeError:
+ plugin = super_getattr('_plugins')[const.VSWITCH_PLUGIN]
+ return getattr(plugin, name)
def _func_name(self, offset=0):
"""Get the name of the calling function"""
# TODO (Sumit): Check if we need to perform any rollback here
raise
+ def create_network_bulk(self, context, networks):
+ """
+ Perform this operation in the context of the configured device
+ plugins.
+ """
+ LOG.debug("create_network_bulk() called\n")
+ try:
+ args = [context, networks]
+ ovs_output = self._plugins[
+ const.VSWITCH_PLUGIN].create_network_bulk(context, networks)
+ vlan_ids = odb.get_vlans()
+ vlanids = ''
+ for v_id in vlan_ids:
+ vlanids = str(v_id[0]) + ',' + vlanids
+ vlanids = vlanids.strip(',')
+ LOG.debug("ovs_output: %s\n " % ovs_output)
+ ovs_networks = ovs_output
+ for ovs_network in ovs_networks:
+ vlan_id = odb.get_vlan(ovs_network['id'])
+ vlan_name = conf.VLAN_NAME_PREFIX + str(vlan_id)
+ args = [ovs_network['tenant_id'], ovs_network['name'],
+ ovs_network['id'], vlan_name, vlan_id,
+ {'vlan_ids':vlanids}]
+ nexus_output = self._invoke_plugin_per_device(
+ const.NEXUS_PLUGIN, "create_network", args)
+ return ovs_output
+ except:
+ # TODO (Sumit): Check if we need to perform any rollback here
+ raise
+
def update_network(self, context, id, network):
"""For this model this method will be delegated to vswitch plugin"""
pass
supported_extension_aliases = ["Cisco Credential", "Cisco Port Profile",
"Cisco qos", "Cisco Nova Tenant",
"Cisco Multiport"]
- _methods_to_delegate = ['create_network', 'delete_network',
- 'update_network', 'get_network', 'get_networks',
- 'create_port', 'delete_port', 'update_port',
- 'get_port', 'get_ports',
- 'create_subnet', 'delete_subnet', 'update_subnet',
- 'get_subnet', 'get_subnets']
+ _methods_to_delegate = ['create_network', 'create_network_bulk',
+ 'delete_network', 'update_network', 'get_network',
+ 'get_networks',
+ 'create_port', 'create_port_bulk', 'delete_port',
+ 'update_port', 'get_port', 'get_ports',
+ 'create_subnet', 'create_subnet_bulk',
+ 'delete_subnet', 'update_subnet',
+ 'get_subnet', 'get_subnets', ]
_master = True
def __init__(self):
"""
- Loads the model class, initializes the DB, and credential store.
+ Loads the model class.
"""
self._model = importutils.import_object(conf.MODEL_CLASS)
if hasattr(self._model, "MANAGE_STATE") and self._model.MANAGE_STATE:
self._master = False
LOG.debug("Model %s manages state" % conf.MODEL_CLASS)
- else:
- cdb.initialize()
- cred.Store.initialize()
+ native_bulk_attr_name = ("_%s__native_bulk_support"
+ % self._model.__class__.__name__)
+ self.__native_bulk_support = getattr(self._model,
+ native_bulk_attr_name, False)
if hasattr(self._model, "supported_extension_aliases"):
self.supported_extension_aliases.extend(
pipeline = extensions extensions_test_app
[filter:extensions]
-paste.filter_factory = quantum.extensions.extensions:plugin_aware_extension_middleware_factory
+paste.filter_factory = quantum.common.extensions:plugin_aware_extension_middleware_factory
[app:extensions_test_app]
paste.app_factory = quantum.plugins.cisco.tests.unit.test_cisco_extension:app_factory
--- /dev/null
+[pipeline:extensions_app_with_filter]
+pipeline = extensions extensions_test_app
+
+[filter:extensions]
+paste.filter_factory = quantum.common.extensions:plugin_aware_extension_middleware_factory
+
+[app:extensions_test_app]
+paste.app_factory = quantum.plugins.cisco.tests.unit.test_cisco_extension:app_factory
pass
def create_vlan(self, vlan_name, vlan_id, nexus_host, nexus_user,
- nexus_password, nexus_ports, nexus_ssh_port):
+ nexus_password, nexus_ports, nexus_ssh_port, vlan_ids):
"""
Creates a VLAN and Enable on trunk mode an interface on Nexus Switch
given the VLAN ID and Name and Interface Number
bind_port = 9696
# Path to the extensions
-api_extensions_path = ../../../../extensions
+api_extensions_path = ../../../../../extensions
# Paste configuration file
api_paste_config = api-paste.ini.cisco.test
core_plugin = quantum.plugins.cisco.network_plugin.PluginV2
+
+# The messaging module to use, defaults to kombu.
+rpc_backend = quantum.openstack.common.rpc.impl_fake
+
+[QUOTAS]
+# resource name(s) that are supported in quota features
+quota_items = network,subnet,port
+
+# default number of resource allowed per tenant, minus for unlimited
+default_quota = -1
+
+# number of networks allowed per tenant, and minus means unlimited
+# quota_network = 10
+
+# number of subnets allowed per tenant, and minus means unlimited
+# quota_subnet = 10
+
+# number of ports allowed per tenant, and minus means unlimited
+# quota_port = 50
+
+# default driver to use for quota checks
+# quota_driver = quantum.quota.ConfDriver
+quota_driver = quantum.extensions._quotav2_driver.DbQuotaDriver
from quantum.api.v2 import router
from quantum.common import config
+from quantum.extensions.extensions import PluginAwareExtensionManager
+from quantum.manager import QuantumManager
from quantum.openstack.common import cfg
from quantum.tests.unit import test_api_v2
def setUp(self):
plugin = 'quantum.plugins.cisco.network_plugin.PluginV2'
+ # Ensure 'stale' patched copies of the plugin are never returned
+ QuantumManager._instance = None
+ # Ensure existing ExtensionManager is not used
+ PluginAwareExtensionManager._instance = None
# Create the default configurations
args = ['--config-file', curdir('quantumv2.conf.cisco.test')]
config.parse(args=args)
from quantum.api.v2.router import APIRouter
from quantum.common import config
+from quantum.common.test_lib import test_config
+from quantum import context
from quantum.db import api as db
+from quantum.extensions import _quotav2_model as quotav2_model
from quantum.manager import QuantumManager
+from quantum.plugins.cisco.common import cisco_constants as const
+from quantum.plugins.cisco.db import network_db_v2
from quantum.plugins.cisco.db import network_models_v2
from quantum.openstack.common import cfg
from quantum.tests.unit import test_db_plugin
plugin = 'quantum.plugins.cisco.network_plugin.PluginV2'
# Create the default configurations
args = ['--config-file', curdir('quantumv2.conf.cisco.test')]
+ # If test_config specifies some config-file, use it, as well
+ for config_file in test_config.get('config_files', []):
+ args.extend(['--config-file', config_file])
config.parse(args=args)
# Update the plugin
cfg.CONF.set_override('core_plugin', plugin)
cfg.CONF.set_override('base_mac', "12:34:56:78:90:ab")
- self.api = APIRouter()
+ cfg.CONF.max_dns_nameservers = 2
+ cfg.CONF.max_subnet_host_routes = 2
+
+ def new_init():
+ db.configure_db({'sql_connection': 'sqlite://',
+ 'base': network_models_v2.model_base.BASEV2})
+
+ with mock.patch.object(network_db_v2,
+ 'initialize', new=new_init):
+ self.api = APIRouter()
+
+ def _is_native_bulk_supported():
+ plugin_obj = QuantumManager.get_plugin()
+ native_bulk_attr_name = ("_%s__native_bulk_support"
+ % plugin_obj.__class__.__name__)
+ return getattr(plugin_obj, native_bulk_attr_name, False)
+
+ self._skip_native_bulk = not _is_native_bulk_supported()
+
+ ext_mgr = test_config.get('extension_manager', None)
+ if ext_mgr:
+ self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
LOG.debug("%s.%s.%s done" % (__name__, self.__class__.__name__,
inspect.stack()[0][3]))
cfg.CONF.reset()
+ def _get_plugin_ref(self):
+ plugin_obj = QuantumManager.get_plugin()
+ if getattr(plugin_obj, "_master"):
+ plugin_ref = plugin_obj
+ else:
+ plugin_ref = getattr(plugin_obj, "_model").\
+ _plugins[const.VSWITCH_PLUGIN]
+
+ return plugin_ref
+
class TestV2HTTPResponse(NetworkPluginV2TestCase,
test_db_plugin.TestV2HTTPResponse):
class TestPortsV2(NetworkPluginV2TestCase, test_db_plugin.TestPortsV2):
- pass
+ def test_create_ports_bulk_emulated_plugin_failure(self):
+ real_has_attr = hasattr
+
+ #ensures the API choose the emulation code path
+ def fakehasattr(item, attr):
+ if attr.endswith('__native_bulk_support'):
+ return False
+ return real_has_attr(item, attr)
+
+ with mock.patch('__builtin__.hasattr',
+ new=fakehasattr):
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_port
+ with mock.patch.object(plugin_ref,
+ 'create_port') as patched_plugin:
+
+ def side_effect(*args, **kwargs):
+ return self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+
+ patched_plugin.side_effect = side_effect
+ with self.network() as net:
+ res = self._create_port_bulk('json', 2,
+ net['network']['id'],
+ 'test',
+ True)
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'ports')
+
+ def test_create_ports_bulk_native_plugin_failure(self):
+ if self._skip_native_bulk:
+ self.skipTest("Plugin does not support native bulk port create")
+ ctx = context.get_admin_context()
+ with self.network() as net:
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_port
+ with mock.patch.object(plugin_ref,
+ 'create_port') as patched_plugin:
+
+ def side_effect(*args, **kwargs):
+ return self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+
+ patched_plugin.side_effect = side_effect
+ res = self._create_port_bulk('json', 2, net['network']['id'],
+ 'test', True, context=ctx)
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'ports')
class TestNetworksV2(NetworkPluginV2TestCase, test_db_plugin.TestNetworksV2):
- pass
+ def test_create_networks_bulk_emulated_plugin_failure(self):
+ real_has_attr = hasattr
+
+ def fakehasattr(item, attr):
+ if attr.endswith('__native_bulk_support'):
+ return False
+ return real_has_attr(item, attr)
+
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_network
+ #ensures the API choose the emulation code path
+ with mock.patch('__builtin__.hasattr',
+ new=fakehasattr):
+ with mock.patch.object(plugin_ref,
+ 'create_network') as patched_plugin:
+ def side_effect(*args, **kwargs):
+ return self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+ patched_plugin.side_effect = side_effect
+ res = self._create_network_bulk('json', 2, 'test', True)
+ LOG.debug("response is %s" % res)
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'networks')
+
+ def test_create_networks_bulk_native_plugin_failure(self):
+ if self._skip_native_bulk:
+ self.skipTest("Plugin does not support native bulk network create")
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_network
+ with mock.patch.object(plugin_ref,
+ 'create_network') as patched_plugin:
+
+ def side_effect(*args, **kwargs):
+ return self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+
+ patched_plugin.side_effect = side_effect
+ res = self._create_network_bulk('json', 2, 'test', True)
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'networks')
class TestSubnetsV2(NetworkPluginV2TestCase, test_db_plugin.TestSubnetsV2):
- pass
+ def test_create_subnets_bulk_emulated_plugin_failure(self):
+ real_has_attr = hasattr
+
+ #ensures the API choose the emulation code path
+ def fakehasattr(item, attr):
+ if attr.endswith('__native_bulk_support'):
+ return False
+ return real_has_attr(item, attr)
+
+ with mock.patch('__builtin__.hasattr',
+ new=fakehasattr):
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_subnet
+ with mock.patch.object(plugin_ref,
+ 'create_subnet') as patched_plugin:
+
+ def side_effect(*args, **kwargs):
+ self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+
+ patched_plugin.side_effect = side_effect
+ with self.network() as net:
+ res = self._create_subnet_bulk('json', 2,
+ net['network']['id'],
+ 'test')
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'subnets')
+
+ def test_create_subnets_bulk_native_plugin_failure(self):
+ if self._skip_native_bulk:
+ self.skipTest("Plugin does not support native bulk subnet create")
+ plugin_ref = self._get_plugin_ref()
+ orig = plugin_ref.create_subnet
+ with mock.patch.object(plugin_ref,
+ 'create_subnet') as patched_plugin:
+ def side_effect(*args, **kwargs):
+ return self._do_side_effect(patched_plugin, orig,
+ *args, **kwargs)
+
+ patched_plugin.side_effect = side_effect
+ with self.network() as net:
+ res = self._create_subnet_bulk('json', 2,
+ net['network']['id'],
+ 'test')
+
+ # We expect a 500 as we injected a fault in the plugin
+ self._validate_behavior_on_bulk_failure(res, 'subnets')
from quantum.openstack.common import importutils
from quantum.plugins.cisco.common import cisco_constants as const
-from quantum.plugins.cisco.common import cisco_credentials as cred
+from quantum.plugins.cisco.common import cisco_credentials_v2 as cred
from quantum.plugins.cisco.common import cisco_exceptions as cexc
from quantum.plugins.cisco.common import cisco_utils as cutil
from quantum.plugins.cisco.db import network_db_v2 as cdb
Code Review / openstack-build / neutron-build.git / commitdiff