openstack
/
neutron
Code Issues Proposed changes

Improve unit test coverage for Cisco plugin model code 

Closes-Bug: #1190620

This fix improves unit test coverage for:
quantum/plugins/cisco/models/virt_phy_sw_v2.py
Test coverage is improved from about 78% to 99%.

One change included in this fix is removal of some code in
the _invoke_plugin() method in virt_phy_sw_v2.py which looks
like it's attempting to handle the case where the number of
arguments being passed to _invoke_plugin() exceeds the number
of arguments expected for the target plugin method. This
section of code does not get executed for any existing
calls to _invoke_plugin(), and it doesn't appear that
this logic would work (except when the target plugin method
includes a **kwargs expansion).

Change-Id: Ibceb7a462a213f3ba693bcbe94e77d97b2e1440a
This commit is contained in:
Dane LeBlanc 2013-11-21 12:34:57 -05:00
parent 0b2d56cda7
commit b6a8aea4d1
4 changed files with 282 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
neutron/plugins/cisco/models/virt_phy_sw_v2.py
View File

@ -50,7 +50,6 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
MANAGE_STATE = True
__native_bulk_support = True
supported_extension_aliases = ["provider", "binding"]
_plugins = {}
_methods_to_delegate = ['create_network_bulk',
'get_network', 'get_networks',
'create_port_bulk',
@ -68,6 +67,7 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
"""
conf.CiscoConfigOptions()
self._plugins = {}
for key in conf.CISCO_PLUGINS.keys():
plugin_obj = conf.CISCO_PLUGINS[key]
if plugin_obj is not None:
@ -139,28 +139,12 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
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)
@ -244,12 +228,19 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
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):
@ -329,12 +320,18 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
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.
@ -522,21 +519,37 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
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.
def get_subnets(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_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

25
neutron/plugins/cisco/network_plugin.py
View File

@ -51,22 +51,25 @@ class PluginV2(db_base_plugin_v2.NeutronDbPluginV2):
_master = True
CISCO_FAULT_MAP = {
cexc.NetworkSegmentIDNotFound: wexc.HTTPNotFound,
cexc.NoMoreNics: wexc.HTTPBadRequest,
cexc.NetworkVlanBindingAlreadyExists: wexc.HTTPBadRequest,
cexc.VlanIDNotFound: wexc.HTTPNotFound,
cexc.VlanIDNotAvailable: wexc.HTTPNotFound,
cexc.QosNotFound: wexc.HTTPNotFound,
cexc.QosNameAlreadyExists: wexc.HTTPBadRequest,
cexc.CredentialNotFound: wexc.HTTPNotFound,
cexc.CredentialNameNotFound: wexc.HTTPNotFound,
cexc.CredentialAlreadyExists: wexc.HTTPBadRequest,
cexc.CredentialNameNotFound: wexc.HTTPNotFound,
cexc.CredentialNotFound: wexc.HTTPNotFound,
cexc.NetworkSegmentIDNotFound: wexc.HTTPNotFound,
cexc.NetworkVlanBindingAlreadyExists: wexc.HTTPBadRequest,
cexc.NexusComputeHostNotConfigured: wexc.HTTPNotFound,
cexc.NexusConnectFailed: wexc.HTTPServiceUnavailable,
cexc.NexusConfigFailed: wexc.HTTPBadRequest,
cexc.NexusConnectFailed: wexc.HTTPServiceUnavailable,
cexc.NexusPortBindingNotFound: wexc.HTTPNotFound,
cexc.NoMoreNics: wexc.HTTPBadRequest,
cexc.PortIdForNexusSvi: wexc.HTTPBadRequest,
cexc.PortVnicBindingAlreadyExists: wexc.HTTPBadRequest,
cexc.PortVnicNotFound: wexc.HTTPNotFound}
cexc.PortVnicNotFound: wexc.HTTPNotFound,
cexc.QosNameAlreadyExists: wexc.HTTPBadRequest,
cexc.QosNotFound: wexc.HTTPNotFound,
cexc.SubnetNotSpecified: wexc.HTTPBadRequest,
cexc.VlanIDNotAvailable: wexc.HTTPNotFound,
cexc.VlanIDNotFound: wexc.HTTPNotFound,
}
def __init__(self):
"""Load the model class."""

232
neutron/tests/unit/cisco/test_network_plugin.py
View File

@ -33,6 +33,7 @@ from neutron.manager import NeutronManager
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.common import config as cisco_config
from neutron.plugins.cisco.db import network_db_v2
from neutron.plugins.cisco.db import nexus_db_v2
from neutron.plugins.cisco.models import virt_phy_sw_v2
from neutron.plugins.openvswitch.common import config as ovs_config
@ -192,6 +193,23 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and
vlan_deleted == vlan_deletion_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):
@ -239,23 +257,6 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
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
@ -866,23 +867,41 @@ class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
'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,
@ -950,65 +969,106 @@ class TestCiscoRouterInterfacesV2(CiscoNetworkPluginV2TestCase):
super(TestCiscoRouterInterfacesV2, self).setUp()
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
@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'])
@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)
try:
yield
finally:
req = self.new_action_request('routers', interface_data,
router['router']['id'],
'remove_router_interface')
req.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)
@contextlib.contextmanager
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:
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']))
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'])
# 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 _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 response
finally:
# 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)
@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):

68
neutron/tests/unit/cisco/test_plugin_model.py Executable file
View File

@ -0,0 +1,68 @@
# 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)