Merge "Improve unit test coverage for Cisco plugin nexus code"

This commit is contained in:
Jenkins 2013-12-05 14:39:30 +00:00 committed by Gerrit Code Review
commit 2f5f2b3b88
7 changed files with 179 additions and 252 deletions

View File

@ -31,15 +31,6 @@ class L2DevicePluginBase(object):
the configuration on each device.
"""
@abstractmethod
def get_all_networks(self, tenant_id, **kwargs):
"""Get newtorks.
:returns:
:raises:
"""
pass
@abstractmethod
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
**kwargs):
@ -68,15 +59,6 @@ class L2DevicePluginBase(object):
"""
pass
@abstractmethod
def get_all_ports(self, tenant_id, net_id, **kwargs):
"""Get ports.
:returns:
:raises:
"""
pass
@abstractmethod
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
"""Create port.
@ -104,15 +86,6 @@ class L2DevicePluginBase(object):
"""
pass
@abstractmethod
def get_port_details(self, tenant_id, net_id, port_id, **kwargs):
"""Get port details.
:returns:
:raises:
"""
pass
@abstractmethod
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs):

View File

@ -84,12 +84,6 @@ class CiscoNEXUSDriver():
}
return self.credentials[nexus_ip]
def get_switch_and_port_id(self, host_name):
for switch_ip, attr in self.nexus_switches:
if str(attr) == host_name:
return switch_ip, self.nexus_switches[switch_ip, attr]
return None, None
def nxos_connect(self, nexus_host):
"""Make SSH connection to the Nexus Switch."""
if getattr(self.connections.get(nexus_host), 'connected', None):
@ -151,20 +145,6 @@ class CiscoNEXUSDriver():
confstr = self.create_xml_snippet(confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def enable_port_trunk(self, nexus_host, etype, interface):
"""Enable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_PORT_TRUNK % (etype, interface, etype)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def disable_switch_port(self, nexus_host, etype, interface):
"""Disable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_NO_SWITCHPORT % (etype, interface, etype)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def enable_vlan_on_trunk_int(self, nexus_host, vlanid, etype, interface):
"""Enable a VLAN on a trunk interface."""
# If one or more VLANs are already configured on this interface,

View File

@ -26,7 +26,6 @@ PlugIn for Nexus OS driver
import logging
from neutron.common import exceptions as exc
from neutron.openstack.common import excutils
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
@ -50,15 +49,6 @@ class NexusPlugin(L2DevicePluginBase):
LOG.debug(_("Loaded driver %s"), conf.CISCO.nexus_driver)
self._nexus_switches = conf.get_device_dictionary()
def get_all_networks(self, tenant_id):
"""Get all networks.
Returns a dictionary containing all <network_uuid, network_name> for
the specified tenant.
"""
LOG.debug(_("NexusPlugin:get_all_networks() called"))
return self._networks.values()
def create_network(self, network, attachment):
"""Create or update a network when an attachment is changed.
@ -240,30 +230,26 @@ class NexusPlugin(L2DevicePluginBase):
def delete_network(self, tenant_id, net_id, **kwargs):
"""Delete network.
Deletes the VLAN in all switches, and removes the VLAN configuration
from the relevant interfaces.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:delete_network() called"))
LOG.debug(_("NexusPlugin:delete_network() called")) # pragma no cover
def update_network(self, tenant_id, net_id, **kwargs):
"""Update the properties of a particular Virtual Network."""
LOG.debug(_("NexusPlugin:update_network() called"))
"""Update the properties of a particular Virtual Network.
def get_all_ports(self, tenant_id, net_id, **kwargs):
"""Get all ports.
This is probably not applicable to the Nexus plugin.
Delete if not required.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:get_all_ports() called"))
LOG.debug(_("NexusPlugin:update_network() called")) # pragma no cover
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
"""Create port.
This is probably not applicable to the Nexus plugin.
Delete if not required.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:create_port() called"))
LOG.debug(_("NexusPlugin:create_port() called")) # pragma no cover
def delete_port(self, device_id, vlan_id):
"""Delete port.
@ -337,40 +323,25 @@ class NexusPlugin(L2DevicePluginBase):
def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
"""Update port.
This is probably not applicable to the Nexus plugin.
Delete if not required.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:update_port() called"))
def get_port_details(self, tenant_id, net_id, port_id, **kwargs):
"""Get port details.
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
LOG.debug(_("NexusPlugin:get_port_details() called"))
LOG.debug(_("NexusPlugin:update_port() called")) # pragma no cover
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs):
"""Plug interfaces.
This is probably not applicable to the Nexus plugin.
Delete if not required.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:plug_interface() called"))
LOG.debug(_("NexusPlugin:plug_interface() called")) # pragma no cover
def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
"""Unplug interface.
This is probably not applicable to the Nexus plugin.
Delete if not required.
Not applicable to Nexus plugin. Defined here to satisfy abstract
method requirements.
"""
LOG.debug(_("NexusPlugin:unplug_interface() called"))
def _get_network(self, tenant_id, network_id, context, base_plugin_ref):
"""Get the Network ID."""
network = base_plugin_ref._get_network(context, network_id)
if not network:
raise exc.NetworkNotFound(net_id=network_id)
return {const.NET_ID: network_id, const.NET_NAME: network.name,
const.NET_PORTS: network.ports}
LOG.debug(_("NexusPlugin:unplug_interface() called")
) # pragma no cover

View File

@ -37,7 +37,6 @@ EXEC_CONF_SNIPPET = """
</config>
"""
CMD_VLAN_CONF_SNIPPET = """
<vlan>
<vlan-id-create-delete>
@ -122,38 +121,6 @@ CMD_INT_VLAN_ADD_SNIPPET = (CMD_INT_VLAN_HEADER +
CMD_VLAN_ADD_ID +
CMD_INT_VLAN_TRAILER)
CMD_PORT_TRUNK = """
<interface>
<%s>
<interface>%s</interface>
<__XML__MODE_if-ethernet-switch>
<switchport></switchport>
<switchport>
<mode>
<trunk>
</trunk>
</mode>
</switchport>
</__XML__MODE_if-ethernet-switch>
</%s>
</interface>
"""
CMD_NO_SWITCHPORT = """
<interface>
<%s>
<interface>%s</interface>
<__XML__MODE_if-ethernet-switch>
<no>
<switchport>
</switchport>
</no>
</__XML__MODE_if-ethernet-switch>
</%s>
</interface>
"""
CMD_NO_VLAN_INT_SNIPPET = """
<interface>
<%s>
@ -176,7 +143,6 @@ CMD_NO_VLAN_INT_SNIPPET = """
</interface>
"""
FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
<show xmlns="http://www.cisco.com/nxos:1.0:vlan_mgr_cli">
<vlan>
@ -185,7 +151,6 @@ FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
</show>
"""
CMD_VLAN_SVI_SNIPPET = """
<interface>
<vlan>

View File

@ -45,10 +45,6 @@ class CiscoNEXUSFakeDriver():
"""Delete a VLAN on Nexus Switch given the VLAN ID."""
pass
def enable_port_trunk(self, mgr, interface):
"""Enable trunk mode an interface on Nexus Switch."""
pass
def disable_switch_port(self, mgr, interface):
"""Disable trunk mode an interface on Nexus Switch."""
pass

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
@ -157,6 +158,29 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
config = {attr: None}
self.mock_ncclient.configure_mock(**config)
@staticmethod
def _config_dependent_side_effect(match_config, exc):
"""Generates a config-dependent side effect for ncclient edit_config.
This method generates a mock side-effect function which can be
configured on the mock ncclient module for the edit_config method.
This side effect will cause a given exception to be raised whenever
the XML config string that is passed to edit_config contains all
words in a given match config string.
:param match_config: String containing keywords to be matched
:param exc: Exception to be raised when match is found
:return: Side effect function for the mock ncclient module's
edit_config method.
"""
keywords = match_config.split()
def _side_effect_function(target, config):
if all(word in config for word in keywords):
raise exc
return _side_effect_function
def _is_in_nexus_cfg(self, words):
"""Check if any config sent to Nexus contains all words in a list."""
for call in (self.mock_ncclient.manager.connect.return_value.
@ -180,11 +204,13 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
vlan_created == vlan_creation_expected and
add_appears == add_keyword_expected)
def _is_vlan_unconfigured(self, vlan_deletion_expected=True):
vlan_deleted = self._is_in_last_nexus_cfg(
def _is_vlan_unconfigured(self, vlan_deletion_expected=True,
vlan_untrunk_expected=True):
vlan_deleted = self._is_in_nexus_cfg(
['no', 'vlan', 'vlan-id-create-delete'])
return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and
vlan_deleted == vlan_deletion_expected)
vlan_untrunked = self._is_in_nexus_cfg(['allowed', 'vlan', 'remove'])
return (vlan_deleted == vlan_deletion_expected and
vlan_untrunked == vlan_untrunk_expected)
class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase,
@ -413,25 +439,17 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
are ignored by the Nexus plugin.
"""
def mock_edit_config_a(target, config):
if all(word in config for word in ['state', 'active']):
raise Exception("Can't modify state for extended")
with self._patch_ncclient(
'manager.connect.return_value.edit_config.side_effect',
mock_edit_config_a):
with self._create_port_res() as res:
self.assertEqual(res.status_int, wexc.HTTPCreated.code)
def mock_edit_config_b(target, config):
if all(word in config for word in ['no', 'shutdown']):
raise Exception("Command is only allowed on VLAN")
with self._patch_ncclient(
'manager.connect.return_value.edit_config.side_effect',
mock_edit_config_b):
with self._create_port_res() as res:
self.assertEqual(res.status_int, wexc.HTTPCreated.code)
config_err_strings = {
"state active": "Can't modify state for extended",
"no shutdown": "Command is only allowed on VLAN",
}
for config, err_string in config_err_strings.items():
with self._patch_ncclient(
'manager.connect.return_value.edit_config.side_effect',
self._config_dependent_side_effect(config,
Exception(err_string))):
with self._create_port_res() as res:
self.assertEqual(res.status_int, wexc.HTTPCreated.code)
def test_nexus_vlan_config_rollback(self):
"""Test rollback following Nexus VLAN state config failure.
@ -442,20 +460,19 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
for the extended VLAN range).
"""
def mock_edit_config(target, config):
if all(word in config for word in ['state', 'active']):
raise ValueError
with self._patch_ncclient(
'manager.connect.return_value.edit_config.side_effect',
mock_edit_config):
with self._create_port_res(do_delete=False) as res:
# Confirm that the last configuration sent to the Nexus
# switch was deletion of the VLAN.
self.assertTrue(
self._is_in_last_nexus_cfg(['<no>', '<vlan>'])
)
self._assertExpectedHTTP(res.status_int,
c_exc.NexusConfigFailed)
vlan_state_configs = ['state active', 'no shutdown']
for config in vlan_state_configs:
with self._patch_ncclient(
'manager.connect.return_value.edit_config.side_effect',
self._config_dependent_side_effect(config, ValueError)):
with self._create_port_res(do_delete=False) as res:
# Confirm that the last configuration sent to the Nexus
# switch was deletion of the VLAN.
self.assertTrue(
self._is_in_last_nexus_cfg(['<no>', '<vlan>'])
)
self._assertExpectedHTTP(res.status_int,
c_exc.NexusConfigFailed)
def test_get_seg_id_fail(self):
"""Test handling of a NetworkSegmentIDNotFound exception.
@ -495,7 +512,9 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
self._assertExpectedHTTP(res.status_int,
c_exc.NexusComputeHostNotConfigured)
def test_nexus_bind_fail_rollback(self):
def _check_rollback_on_bind_failure(self,
vlan_deletion_expected,
vlan_untrunk_expected):
"""Test for proper rollback following add Nexus DB binding failure.
Test that the Cisco Nexus plugin correctly rolls back the vlan
@ -503,15 +522,47 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
within the plugin's create_port() method.
"""
inserted_exc = KeyError
with mock.patch.object(nexus_db_v2, 'add_nexusport_binding',
side_effect=KeyError):
side_effect=inserted_exc):
with self._create_port_res(do_delete=False) as res:
# Confirm that the last configuration sent to the Nexus
# switch was a removal of vlan from the test interface.
self.assertTrue(
self._is_in_last_nexus_cfg(['<vlan>', '<remove>'])
)
self._assertExpectedHTTP(res.status_int, KeyError)
# Confirm that the configuration sent to the Nexus
# switch includes deletion of the vlan (if expected)
# and untrunking of the vlan from the ethernet interface
# (if expected).
self.assertTrue(self._is_vlan_unconfigured(
vlan_deletion_expected=vlan_deletion_expected,
vlan_untrunk_expected=vlan_untrunk_expected))
self._assertExpectedHTTP(res.status_int, inserted_exc)
def test_nexus_rollback_on_bind_failure_non_provider_vlan(self):
"""Test rollback upon DB binding failure for non-provider vlan."""
self._check_rollback_on_bind_failure(vlan_deletion_expected=True,
vlan_untrunk_expected=True)
def test_nexus_rollback_on_bind_failure_prov_vlan_no_auto_create(self):
"""Test rollback on bind fail for prov vlan w auto-create disabled."""
with mock.patch.object(network_db_v2, 'is_provider_vlan',
return_value=True):
# Disable auto-create. This config change will be cleared based
# on cleanup scheduled in the CiscoNetworkPluginV2TestCase
# class' setUp() method.
cisco_config.CONF.set_override('provider_vlan_auto_create',
False, 'CISCO')
self._check_rollback_on_bind_failure(vlan_deletion_expected=False,
vlan_untrunk_expected=True)
def test_nexus_rollback_on_bind_failure_prov_vlan_no_auto_trunk(self):
"""Test rollback on bind fail for prov vlan w auto-trunk disabled."""
with mock.patch.object(network_db_v2, 'is_provider_vlan',
return_value=True):
# Disable auto-trunk. This config change will be cleared
# based on post-test cleanup scheduled in the
# CiscoNetworkPluginV2TestCase class' setUp() method.
cisco_config.CONF.set_override('provider_vlan_auto_trunk',
False, 'CISCO')
self._check_rollback_on_bind_failure(vlan_deletion_expected=True,
vlan_untrunk_expected=False)
def test_model_update_port_rollback(self):
"""Test for proper rollback for Cisco model layer update port failure.

View File

@ -99,6 +99,10 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
const.NET_VLAN_NAME: 'q-268',
const.NET_VLAN_ID: '268',
}
self.delete_port_args_1 = [
self.attachment1[const.INSTANCE_ID],
self.network1[const.NET_VLAN_ID],
]
self.providernet = {
const.NET_ID: 9,
const.NET_NAME: 'pnet1',
@ -181,44 +185,38 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
self.assertEqual(expected_instance_id, INSTANCE1)
def test_create_providernet(self):
def _create_delete_providernet(self, auto_create, auto_trunk):
cfg.CONF.set_override(
'provider_vlan_auto_create', auto_create, 'CISCO')
cfg.CONF.set_override(
'provider_vlan_auto_trunk', auto_trunk, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan',
return_value=True) as mock_db:
# Create a provider network
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
mock_db.assert_called_once()
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
# Delete the provider network
instance_id = self._cisco_nexus_plugin.delete_port(
self.attachment1[const.INSTANCE_ID],
self.providernet[const.NET_VLAN_ID])
self.assertEqual(instance_id,
self.attachment1[const.INSTANCE_ID])
def test_create_provider_vlan_network_cfg_auto_man(self):
cfg.CONF.set_override('provider_vlan_auto_create', True, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', False, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_create_delete_providernet(self):
self._create_delete_providernet(auto_create=True, auto_trunk=True)
def test_create_provider_vlan_network_cfg_man_auto(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', True, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_create_delete_provider_vlan_network_cfg_auto_man(self):
self._create_delete_providernet(auto_create=True, auto_trunk=False)
def test_create_provider_vlan_network_cfg_man_man(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', False, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_create_delete_provider_vlan_network_cfg_man_auto(self):
self._create_delete_providernet(auto_create=False, auto_trunk=True)
def test_create_delete_provider_vlan_network_cfg_man_man(self):
self._create_delete_providernet(auto_create=False, auto_trunk=False)
def test_create_delete_network_portchannel(self):
"""Tests creation of a network over a portchannel."""
@ -237,45 +235,48 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
INSTANCE3, self.network3[const.NET_VLAN_ID]
)
def _add_router_interface(self):
"""Add a router interface using fixed (canned) parameters."""
vlan_name = self.vlan_name
vlan_id = self.vlan_id
gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
subnet_id = '00001'
return self._cisco_nexus_plugin.add_router_interface(
vlan_name, vlan_id, subnet_id, gateway_ip, router_id)
def _remove_router_interface(self):
"""Remove a router interface created with _add_router_interface."""
vlan_id = self.vlan_id
router_id = '00000R1'
return self._cisco_nexus_plugin.remove_router_interface(vlan_id,
router_id)
def test_nexus_add_remove_router_interface(self):
"""Tests addition of a router interface."""
vlan_name = self.vlan_name
vlan_id = self.vlan_id
gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
subnet_id = '00001'
self.assertTrue(self._add_router_interface())
self.assertEqual(self._remove_router_interface(), '00000R1')
result = self._cisco_nexus_plugin.add_router_interface(vlan_name,
vlan_id,
subnet_id,
gateway_ip,
router_id)
self.assertTrue(result)
result = self._cisco_nexus_plugin.remove_router_interface(vlan_id,
router_id)
self.assertEqual(result, router_id)
def test_nexus_add_router_interface_fail(self):
"""Tests deletion of a router interface."""
vlan_name = self.vlan_name
vlan_id = self.vlan_id
gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
subnet_id = '00001'
self._cisco_nexus_plugin.add_router_interface(vlan_name,
vlan_id,
subnet_id,
gateway_ip,
router_id)
def test_nexus_dup_add_router_interface(self):
"""Tests a duplicate add of a router interface."""
self._add_router_interface()
try:
self.assertRaises(
cisco_exc.SubnetInterfacePresent,
self._cisco_nexus_plugin.add_router_interface,
vlan_name, vlan_id, subnet_id, gateway_ip, router_id)
self._add_router_interface)
finally:
self._cisco_nexus_plugin.remove_router_interface(vlan_id,
router_id)
self._remove_router_interface()
def test_nexus_no_svi_switch_exception(self):
"""Tests failure to find a Nexus switch for SVI placement."""
# Clear the Nexus switches dictionary.
with mock.patch.dict(self._cisco_nexus_plugin._client.nexus_switches,
{}, clear=True):
# Clear the first Nexus IP address discovered in config
with mock.patch.object(cisco_config, 'first_device_ip',
new=None):
self.assertRaises(cisco_exc.NoNexusSviSwitch,
self._add_router_interface)
def test_nexus_add_port_after_router_interface(self):
"""Tests creating a port after a router interface.
@ -284,23 +285,13 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
been created. Only a trunk call should be invoked and the
plugin should not attempt to recreate the vlan.
"""
vlan_name = self.vlan_name
vlan_id = self.vlan_id
gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
subnet_id = '00001'
self._cisco_nexus_plugin.add_router_interface(vlan_name,
vlan_id,
subnet_id,
gateway_ip,
router_id)
self._add_router_interface()
# Create a network on the switch
self._cisco_nexus_plugin.create_network(
self.network1, self.attachment1)
# Grab a list of all mock calls from ncclient
last_cfgs = (self.mock_ncclient.manager.connect().
last_cfgs = (self.mock_ncclient.manager.connect.return_value.
edit_config.mock_calls)
# The last ncclient call should be for trunking and the second