"""
__native_bulk_support = True
supported_extension_aliases = ["provider", "binding"]
- _plugins = {}
_methods_to_delegate = ['create_network_bulk',
'get_network', 'get_networks',
'create_port_bulk',
"""
conf.CiscoConfigOptions()
+ self._plugins = {}
for key in conf.CISCO_PLUGINS.keys():
plugin_obj = conf.CISCO_PLUGINS[key]
if plugin_obj is not None:
LOG.info(_("No %s Plugin loaded"), plugin_key)
LOG.info(_("%(plugin_key)s: %(function_name)s with args %(args)s "
"ignored"),
- {'plugin_key': plugin_key, 'function_name': function_name,
+ {'plugin_key': plugin_key,
+ 'function_name': function_name,
'args': args})
- return
- return [self._invoke_plugin(plugin_key, function_name, args, kwargs)]
-
- def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
- """Invoke plugin.
-
- Invokes the relevant function on a device plugin's
- implementation for completing this operation.
- """
- func = getattr(self._plugins[plugin_key], function_name)
- func_args_len = int(inspect.getargspec(func).args.__len__()) - 1
- if args.__len__() > func_args_len:
- func_args = args[:func_args_len]
- extra_args = args[func_args_len:]
- for dict_arg in extra_args:
- for k, v in dict_arg.iteritems():
- kwargs[k] = v
- return func(*func_args, **kwargs)
else:
- return func(*args, **kwargs)
+ func = getattr(self._plugins[plugin_key], function_name)
+ return [func(*args, **kwargs)]
def _get_segmentation_id(self, network_id):
binding_seg_id = odb.get_network_binding(None, network_id)
return ovs_output[0]
def get_network(self, context, id, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Get network. This method is delegated to the vswitch plugin.
- def get_networks(self, context, filters=None, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
+
+ def get_networks(self, context, filters=None, fields=None,
+ sorts=None, limit=None, marker=None, page_reverse=False):
+ """Get networks. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def _invoke_nexus_for_net_create(self, context, tenant_id, net_id,
instance_id, host_id):
return ovs_output[0]
def get_port(self, context, id, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Get port. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def get_ports(self, context, filters=None, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Get ports. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def _check_nexus_net_create_needed(self, new_port, old_port):
"""Check if nexus plugin should be invoked for net_create.
n_args)
def create_subnet(self, context, subnet):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Create subnet. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def update_subnet(self, context, id, subnet):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Update subnet. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def get_subnet(self, context, id, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Get subnet. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
def delete_subnet(self, context, id, kwargs):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ """Delete subnet. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
- def get_subnets(self, context, filters=None, fields=None):
- """For this model this method will be delegated to vswitch plugin."""
- pass
+ def get_subnets(self, context, filters=None, fields=None,
+ sorts=None, limit=None, marker=None, page_reverse=False):
+ """Get subnets. This method is delegated to the vswitch plugin.
+
+ This method is included here to satisfy abstract method requirements.
+ """
+ pass # pragma no cover
return (vlan_deleted == vlan_deletion_expected and
vlan_untrunked == vlan_untrunk_expected)
+ def _assertExpectedHTTP(self, status, exc):
+ """Confirm that an HTTP status corresponds to an expected exception.
+
+ Confirm that an HTTP status which has been returned for an
+ neutron API request matches the HTTP status corresponding
+ to an expected exception.
+
+ :param status: HTTP status
+ :param exc: Expected exception
+
+ """
+ if exc in base.FAULT_MAP:
+ expected_http = base.FAULT_MAP[exc].code
+ else:
+ expected_http = wexc.HTTPInternalServerError.code
+ self.assertEqual(status, expected_http)
+
class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase,
test_db_plugin.TestBasicGet):
if do_delete:
self._delete('ports', port['port']['id'])
- def _assertExpectedHTTP(self, status, exc):
- """Confirm that an HTTP status corresponds to an expected exception.
-
- Confirm that an HTTP status which has been returned for an
- neutron API request matches the HTTP status corresponding
- to an expected exception.
-
- :param status: HTTP status
- :param exc: Expected exception
-
- """
- if exc in base.FAULT_MAP:
- expected_http = base.FAULT_MAP[exc].code
- else:
- expected_http = wexc.HTTPInternalServerError.code
- self.assertEqual(status, expected_http)
-
def test_create_ports_bulk_emulated_plugin_failure(self):
real_has_attr = hasattr
'networks',
wexc.HTTPInternalServerError.code)
- def test_create_provider_vlan_network(self):
+ @contextlib.contextmanager
+ def _provider_vlan_network(self, phys_net, segment_id, net_name):
provider_attrs = {provider.NETWORK_TYPE: 'vlan',
- provider.PHYSICAL_NETWORK: PHYS_NET,
- provider.SEGMENTATION_ID: '1234'}
+ provider.PHYSICAL_NETWORK: phys_net,
+ provider.SEGMENTATION_ID: segment_id}
arg_list = tuple(provider_attrs.keys())
- res = self._create_network(self.fmt, 'pvnet1', True,
+ res = self._create_network(self.fmt, net_name, True,
arg_list=arg_list, **provider_attrs)
- net = self.deserialize(self.fmt, res)
- expected = [('name', 'pvnet1'),
- ('admin_state_up', True),
- ('status', 'ACTIVE'),
- ('shared', False),
- (provider.NETWORK_TYPE, 'vlan'),
- (provider.PHYSICAL_NETWORK, PHYS_NET),
- (provider.SEGMENTATION_ID, 1234)]
- for k, v in expected:
- self.assertEqual(net['network'][k], v)
+ network = self.deserialize(self.fmt, res)['network']
+ try:
+ yield network
+ finally:
+ req = self.new_delete_request('networks', network['id'])
+ req.get_response(self.api)
+
+ def test_create_provider_vlan_network(self):
+ with self._provider_vlan_network(PHYS_NET, '1234',
+ 'pvnet1') as network:
+ expected = [('name', 'pvnet1'),
+ ('admin_state_up', True),
+ ('status', 'ACTIVE'),
+ ('shared', False),
+ (provider.NETWORK_TYPE, 'vlan'),
+ (provider.PHYSICAL_NETWORK, PHYS_NET),
+ (provider.SEGMENTATION_ID, 1234)]
+ for k, v in expected:
+ self.assertEqual(network[k], v)
+ self.assertTrue(network_db_v2.is_provider_network(network['id']))
+
+ def test_delete_provider_vlan_network(self):
+ with self._provider_vlan_network(PHYS_NET, '1234',
+ 'pvnet1') as network:
+ network_id = network['id']
+ # Provider network should now be deleted
+ self.assertFalse(network_db_v2.is_provider_network(network_id))
class TestCiscoSubnetsV2(CiscoNetworkPluginV2TestCase,
super(TestCiscoRouterInterfacesV2, self).setUp()
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
+ self.addCleanup(cisco_config.CONF.reset)
@contextlib.contextmanager
- def _router(self, subnet):
- """Create a virtual router, yield it for testing, then delete it."""
- data = {'router': {'tenant_id': 'test_tenant_id'}}
- router_req = self.new_create_request('routers', data, self.fmt)
- res = router_req.get_response(self.ext_api)
- router = self.deserialize(self.fmt, res)
- try:
- yield router
- finally:
- self._delete('routers', router['router']['id'])
+ def _network_subnet_router(self):
+ """Context mgr for creating/deleting a net, subnet, and router."""
+ with self.network() as network:
+ with self.subnet(network=network) as subnet:
+ data = {'router': {'tenant_id': 'test_tenant_id'}}
+ request = self.new_create_request('routers', data, self.fmt)
+ response = request.get_response(self.ext_api)
+ router = self.deserialize(self.fmt, response)
+ try:
+ yield network, subnet, router
+ finally:
+ self._delete('routers', router['router']['id'])
@contextlib.contextmanager
- def _router_interface(self, router, subnet):
- """Create a router interface, yield for testing, then delete it."""
- interface_data = {'subnet_id': subnet['subnet']['id']}
- req = self.new_action_request('routers', interface_data,
- router['router']['id'],
- 'add_router_interface')
- req.get_response(self.ext_api)
+ def _router_interface(self, router, subnet, **kwargs):
+ """Create a router interface, yield the response, then delete it."""
+ interface_data = {}
+ if subnet:
+ interface_data['subnet_id'] = subnet['subnet']['id']
+ interface_data.update(kwargs)
+ request = self.new_action_request('routers', interface_data,
+ router['router']['id'],
+ 'add_router_interface')
+ response = request.get_response(self.ext_api)
try:
- yield
+ yield response
finally:
- req = self.new_action_request('routers', interface_data,
- router['router']['id'],
- 'remove_router_interface')
- req.get_response(self.ext_api)
+ # If router interface was created successfully, delete it now.
+ if response.status_int == wexc.HTTPOk.code:
+ request = self.new_action_request('routers', interface_data,
+ router['router']['id'],
+ 'remove_router_interface')
+ request.get_response(self.ext_api)
- def test_nexus_l3_enable_config(self):
- """Verify proper operation of the Nexus L3 enable configuration."""
- self.addCleanup(cisco_config.CONF.reset)
- with self.network() as network:
- with self.subnet(network=network) as subnet:
- with self._router(subnet) as router:
- # With 'nexus_l3_enable' configured to True, confirm that
- # a switched virtual interface (SVI) is created/deleted
- # on the Nexus switch when a virtual router interface is
- # created/deleted.
- cisco_config.CONF.set_override('nexus_l3_enable',
- True, 'CISCO')
- with self._router_interface(router, subnet):
- self.assertTrue(self._is_in_last_nexus_cfg(
- ['interface', 'vlan', 'ip', 'address']))
- self.assertTrue(self._is_in_nexus_cfg(
- ['no', 'interface', 'vlan']))
- self.assertTrue(self._is_in_last_nexus_cfg(
- ['no', 'vlan']))
-
- # With 'nexus_l3_enable' configured to False, confirm
- # that no changes are made to the Nexus switch running
- # configuration when a virtual router interface is
- # created and then deleted.
- cisco_config.CONF.set_override('nexus_l3_enable',
- False, 'CISCO')
- self.mock_ncclient.reset_mock()
- self._router_interface(router, subnet)
- self.assertFalse(self.mock_ncclient.manager.connect.
- return_value.edit_config.called)
+ @contextlib.contextmanager
+ def _network_subnet_router_interface(self, **kwargs):
+ """Context mgr for create/deleting a net, subnet, router and intf."""
+ with self._network_subnet_router() as (network, subnet, router):
+ with self._router_interface(router, subnet,
+ **kwargs) as response:
+ yield response
+
+ def test_add_remove_router_intf_with_nexus_l3_enabled(self):
+ """Verifies proper add/remove intf operation with Nexus L3 enabled.
+
+ With 'nexus_l3_enable' configured to True, confirm that a switched
+ virtual interface (SVI) is created/deleted on the Nexus switch when
+ a virtual router interface is created/deleted.
+ """
+ cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
+ with self._network_subnet_router_interface():
+ self.assertTrue(self._is_in_last_nexus_cfg(
+ ['interface', 'vlan', 'ip', 'address']))
+ # Clear list of calls made to mock ncclient
+ self.mock_ncclient.reset()
+ # Router interface is now deleted. Confirm that SVI
+ # has been deleted from the Nexus switch.
+ self.assertTrue(self._is_in_nexus_cfg(['no', 'interface', 'vlan']))
+ self.assertTrue(self._is_in_last_nexus_cfg(['no', 'vlan']))
+
+ def test_add_remove_router_intf_with_nexus_l3_disabled(self):
+ """Verifies proper add/remove intf operation with Nexus L3 disabled.
+
+ With 'nexus_l3_enable' configured to False, confirm that no changes
+ are made to the Nexus switch running configuration when a virtual
+ router interface is created and then deleted.
+ """
+ cisco_config.CONF.set_override('nexus_l3_enable', False, 'CISCO')
+ with self._network_subnet_router_interface():
+ self.assertFalse(self.mock_ncclient.manager.connect.
+ return_value.edit_config.called)
+
+ def test_create_svi_but_subnet_not_specified_exception(self):
+ """Tests raising of SubnetNotSpecified exception.
+
+ Tests that a SubnetNotSpecified exception is raised when an
+ add_router_interface request is made for creating a switch virtual
+ interface (SVI), but the request does not specify a subnet.
+ """
+ cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
+ with self._network_subnet_router() as (network, subnet, router):
+ with self._router_interface(router, subnet=None) as response:
+ self._assertExpectedHTTP(response.status_int,
+ c_exc.SubnetNotSpecified)
+
+ def test_create_svi_but_port_id_included_exception(self):
+ """Tests raising of PortIdForNexusSvi exception.
+
+ Tests that a PortIdForNexusSvi exception is raised when an
+ add_router_interface request is made for creating a switch virtual
+ interface (SVI), but the request includes a virtual port ID.
+ """
+ cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
+ with self._network_subnet_router_interface(
+ port_id='my_port_id') as response:
+ self._assertExpectedHTTP(response.status_int,
+ c_exc.PortIdForNexusSvi)
class TestCiscoPortsV2XML(TestCiscoPortsV2):
--- /dev/null
+# Copyright 2014 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+import mock
+
+from neutron.common import config
+from neutron import context
+from neutron.plugins.cisco.common import cisco_constants as const
+from neutron.plugins.cisco.common import config as cisco_config
+from neutron.plugins.cisco.models import virt_phy_sw_v2
+from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
+from neutron.tests import base
+from neutron.tests.unit import test_api_v2
+
+
+class TestCiscoPluginModel(base.BaseTestCase):
+
+ def setUp(self):
+ # Point config file to: neutron/tests/etc/neutron.conf.test
+ args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
+ config.parse(args=args)
+
+ super(TestCiscoPluginModel, self).setUp()
+
+ self.addCleanup(cisco_config.CONF.reset)
+
+ def test_non_nexus_device_driver(self):
+ """Tests handling of an non-Nexus device driver being configured."""
+ with mock.patch.dict(sys.modules, {'mock_driver': mock.Mock()}):
+ cisco_config.CONF.set_override('nexus_driver',
+ 'mock_driver.Non_Nexus_Driver',
+ 'CISCO')
+ # Plugin model instance should have is_nexus_plugin set to False
+ model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
+ self.assertFalse(model.is_nexus_plugin)
+
+ # Model's _invoke_nexus_for_net_create should just return False
+ user_id = 'user_id'
+ tenant_id = 'tenant_id'
+ ctx = context.Context(user_id, tenant_id)
+ self.assertFalse(model._invoke_nexus_for_net_create(
+ ctx, tenant_id, net_id='net_id',
+ instance_id='instance_id', host_id='host_id'))
+
+ def test_nexus_plugin_calls_ignored_if_plugin_not_loaded(self):
+ """Verifies Nexus plugin calls are ignored if plugin is not loaded."""
+ cisco_config.CONF.set_override(const.NEXUS_PLUGIN,
+ None, 'CISCO_PLUGINS')
+ with mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
+ 'create_network') as mock_create_network:
+ model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
+ model._invoke_plugin_per_device(model, const.NEXUS_PLUGIN,
+ 'create_network')
+ self.assertFalse(mock_create_network.called)
Code Review / openstack-build / neutron-build.git / commitdiff