x/vmware-nsx
Code Issues Proposed changes

Improve unit test coverage for Cisco plugin nexus code

Browse Source 
Closes-Bug: #1190622

This fix improves the unit test coverage for the Cisco Nexus plugin's
cisco_nexus_network_driver_v2.py and cisco_nexus_plugin_v2.py
source files from:
cisco_nexus_network_driver_v2: 72%
cisco_nexus_plugin_v2:         79%
To:
cisco_nexus_network_driver_v2: 99%
cisco_nexus_plugin_v2:         94%

Much of what the remaining "uncovered" code (coverage tool
reports as partially covered) can be attributed to the
coverage tool not being aware that execution does not
continue at the end of a save_and_reraise_exception()
context block (i.e. the exception will be reraised, but
the coverage tool isn't aware of this).

This fix and coverage results are dependent on the fix for #1246080.

Change-Id: Ie0e11fc0a12502739fb39bb4b30deb04dd31b7b0
This commit is contained in:
Dane LeBlanc 2013-11-21 13:13:02 -05:00
parent a0a462f030
commit 26e9fad61f
7 changed files with 179 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
neutron/plugins/cisco/l2device_plugin_base.py
View File

@ -30,15 +30,6 @@ class L2DevicePluginBase(object):
__metaclass__ = ABCMeta __metaclass__ = ABCMeta
@abstractmethod
def get_all_networks(self, tenant_id, **kwargs):
"""Get newtorks.
:returns:
:raises:
"""
pass
@abstractmethod @abstractmethod
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id, def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
**kwargs): **kwargs):
@ -67,15 +58,6 @@ class L2DevicePluginBase(object):
""" """
pass pass
@abstractmethod
def get_all_ports(self, tenant_id, net_id, **kwargs):
"""Get ports.
:returns:
:raises:
"""
pass
@abstractmethod @abstractmethod
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs): def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
"""Create port. """Create port.
@ -103,15 +85,6 @@ class L2DevicePluginBase(object):
""" """
pass pass
@abstractmethod
def get_port_details(self, tenant_id, net_id, port_id, **kwargs):
"""Get port details.
:returns:
:raises:
"""
pass
@abstractmethod @abstractmethod
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id, def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs): **kwargs):

20
neutron/plugins/cisco/nexus/cisco_nexus_network_driver_v2.py
View File

@ -84,12 +84,6 @@ class CiscoNEXUSDriver():
} }
return self.credentials[nexus_ip] 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): def nxos_connect(self, nexus_host):
"""Make SSH connection to the Nexus Switch.""" """Make SSH connection to the Nexus Switch."""
if getattr(self.connections.get(nexus_host), 'connected', None): if getattr(self.connections.get(nexus_host), 'connected', None):
@ -151,20 +145,6 @@ class CiscoNEXUSDriver():
confstr = self.create_xml_snippet(confstr) confstr = self.create_xml_snippet(confstr)
self._edit_config(nexus_host, target='running', config=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): def enable_vlan_on_trunk_int(self, nexus_host, vlanid, etype, interface):
"""Enable a VLAN on a trunk interface.""" """Enable a VLAN on a trunk interface."""
# If one or more VLANs are already configured on this interface, # If one or more VLANs are already configured on this interface,

69
neutron/plugins/cisco/nexus/cisco_nexus_plugin_v2.py
View File

@ -26,7 +26,6 @@ PlugIn for Nexus OS driver
import logging import logging
from neutron.common import exceptions as exc
from neutron.openstack.common import excutils from neutron.openstack.common import excutils
from neutron.openstack.common import importutils from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const 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) LOG.debug(_("Loaded driver %s"), conf.CISCO.nexus_driver)
self._nexus_switches = conf.get_device_dictionary() 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): def create_network(self, network, attachment):
"""Create or update a network when an attachment is changed. """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): def delete_network(self, tenant_id, net_id, **kwargs):
"""Delete network. """Delete network.
Deletes the VLAN in all switches, and removes the VLAN configuration Not applicable to Nexus plugin. Defined here to satisfy abstract
from the relevant interfaces. 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): def update_network(self, tenant_id, net_id, **kwargs):
"""Update the properties of a particular Virtual Network.""" """Update the properties of a particular Virtual Network.
LOG.debug(_("NexusPlugin:update_network() called"))
def get_all_ports(self, tenant_id, net_id, **kwargs): Not applicable to Nexus plugin. Defined here to satisfy abstract
"""Get all ports. method requirements.
This is probably not applicable to the Nexus plugin.
Delete if not required.
""" """
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): def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
"""Create port. """Create port.
This is probably not applicable to the Nexus plugin. Not applicable to Nexus plugin. Defined here to satisfy abstract
Delete if not required. 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): def delete_port(self, device_id, vlan_id):
"""Delete port. """Delete port.
@ -337,40 +323,25 @@ class NexusPlugin(L2DevicePluginBase):
def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs): def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
"""Update port. """Update port.
This is probably not applicable to the Nexus plugin. Not applicable to Nexus plugin. Defined here to satisfy abstract
Delete if not required. method requirements.
""" """
LOG.debug(_("NexusPlugin:update_port() called")) LOG.debug(_("NexusPlugin:update_port() called")) # pragma no cover
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"))
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id, def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs): **kwargs):
"""Plug interfaces. """Plug interfaces.
This is probably not applicable to the Nexus plugin. Not applicable to Nexus plugin. Defined here to satisfy abstract
Delete if not required. 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): def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
"""Unplug interface. """Unplug interface.
This is probably not applicable to the Nexus plugin. Not applicable to Nexus plugin. Defined here to satisfy abstract
Delete if not required. method requirements.
""" """
LOG.debug(_("NexusPlugin:unplug_interface() called")) LOG.debug(_("NexusPlugin:unplug_interface() called")
) # pragma no cover
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}

35
neutron/plugins/cisco/nexus/cisco_nexus_snippets.py
View File

@ -37,7 +37,6 @@ EXEC_CONF_SNIPPET = """
</config> </config>
""" """
CMD_VLAN_CONF_SNIPPET = """ CMD_VLAN_CONF_SNIPPET = """
<vlan> <vlan>
<vlan-id-create-delete> <vlan-id-create-delete>
@ -122,38 +121,6 @@ CMD_INT_VLAN_ADD_SNIPPET = (CMD_INT_VLAN_HEADER +
CMD_VLAN_ADD_ID + CMD_VLAN_ADD_ID +
CMD_INT_VLAN_TRAILER) 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 = """ CMD_NO_VLAN_INT_SNIPPET = """
<interface> <interface>
<%s> <%s>
@ -176,7 +143,6 @@ CMD_NO_VLAN_INT_SNIPPET = """
</interface> </interface>
""" """
FILTER_SHOW_VLAN_BRIEF_SNIPPET = """ FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
<show xmlns="http://www.cisco.com/nxos:1.0:vlan_mgr_cli"> <show xmlns="http://www.cisco.com/nxos:1.0:vlan_mgr_cli">
<vlan> <vlan>
@ -185,7 +151,6 @@ FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
</show> </show>
""" """
CMD_VLAN_SVI_SNIPPET = """ CMD_VLAN_SVI_SNIPPET = """
<interface> <interface>
<vlan> <vlan>

4
neutron/plugins/cisco/test/nexus/fake_nexus_driver.py
View File

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

141
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_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc 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.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.db import nexus_db_v2
from neutron.plugins.cisco.models import virt_phy_sw_v2 from neutron.plugins.cisco.models import virt_phy_sw_v2
from neutron.plugins.openvswitch.common import config as ovs_config from neutron.plugins.openvswitch.common import config as ovs_config
@ -163,6 +164,29 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
config = {attr: None} config = {attr: None}
self.mock_ncclient.configure_mock(**config) 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): def _is_in_nexus_cfg(self, words):
"""Check if any config sent to Nexus contains all words in a list.""" """Check if any config sent to Nexus contains all words in a list."""
for call in (self.mock_ncclient.manager.connect.return_value. for call in (self.mock_ncclient.manager.connect.return_value.
@ -186,11 +210,13 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
vlan_created == vlan_creation_expected and vlan_created == vlan_creation_expected and
add_appears == add_keyword_expected) add_appears == add_keyword_expected)
def _is_vlan_unconfigured(self, vlan_deletion_expected=True): def _is_vlan_unconfigured(self, vlan_deletion_expected=True,
vlan_deleted = self._is_in_last_nexus_cfg( vlan_untrunk_expected=True):
vlan_deleted = self._is_in_nexus_cfg(
['no', 'vlan', 'vlan-id-create-delete']) ['no', 'vlan', 'vlan-id-create-delete'])
return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and vlan_untrunked = self._is_in_nexus_cfg(['allowed', 'vlan', 'remove'])
vlan_deleted == vlan_deletion_expected) return (vlan_deleted == vlan_deletion_expected and
vlan_untrunked == vlan_untrunk_expected)
class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase, class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase,
@ -419,25 +445,17 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
are ignored by the Nexus plugin. are ignored by the Nexus plugin.
""" """
def mock_edit_config_a(target, config): config_err_strings = {
if all(word in config for word in ['state', 'active']): "state active": "Can't modify state for extended",
raise Exception("Can't modify state for extended") "no shutdown": "Command is only allowed on VLAN",
}
with self._patch_ncclient( for config, err_string in config_err_strings.items():
'manager.connect.return_value.edit_config.side_effect', with self._patch_ncclient(
mock_edit_config_a): 'manager.connect.return_value.edit_config.side_effect',
with self._create_port_res() as res: self._config_dependent_side_effect(config,
self.assertEqual(res.status_int, wexc.HTTPCreated.code) Exception(err_string))):
with self._create_port_res() as res:
def mock_edit_config_b(target, config): self.assertEqual(res.status_int, wexc.HTTPCreated.code)
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)
def test_nexus_vlan_config_rollback(self): def test_nexus_vlan_config_rollback(self):
"""Test rollback following Nexus VLAN state config failure. """Test rollback following Nexus VLAN state config failure.
@ -448,20 +466,19 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
for the extended VLAN range). for the extended VLAN range).
""" """
def mock_edit_config(target, config): vlan_state_configs = ['state active', 'no shutdown']
if all(word in config for word in ['state', 'active']): for config in vlan_state_configs:
raise ValueError with self._patch_ncclient(
with self._patch_ncclient( 'manager.connect.return_value.edit_config.side_effect',
'manager.connect.return_value.edit_config.side_effect', self._config_dependent_side_effect(config, ValueError)):
mock_edit_config): with self._create_port_res(do_delete=False) as res:
with self._create_port_res(do_delete=False) as res: # Confirm that the last configuration sent to the Nexus
# Confirm that the last configuration sent to the Nexus # switch was deletion of the VLAN.
# switch was deletion of the VLAN. self.assertTrue(
self.assertTrue( self._is_in_last_nexus_cfg(['<no>', '<vlan>'])
self._is_in_last_nexus_cfg(['<no>', '<vlan>']) )
) self._assertExpectedHTTP(res.status_int,
self._assertExpectedHTTP(res.status_int, c_exc.NexusConfigFailed)
c_exc.NexusConfigFailed)
def test_get_seg_id_fail(self): def test_get_seg_id_fail(self):
"""Test handling of a NetworkSegmentIDNotFound exception. """Test handling of a NetworkSegmentIDNotFound exception.
@ -501,7 +518,9 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
self._assertExpectedHTTP(res.status_int, self._assertExpectedHTTP(res.status_int,
c_exc.NexusComputeHostNotConfigured) 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 for proper rollback following add Nexus DB binding failure.
Test that the Cisco Nexus plugin correctly rolls back the vlan Test that the Cisco Nexus plugin correctly rolls back the vlan
@ -509,15 +528,47 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
within the plugin's create_port() method. within the plugin's create_port() method.
""" """
inserted_exc = KeyError
with mock.patch.object(nexus_db_v2, 'add_nexusport_binding', 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: with self._create_port_res(do_delete=False) as res:
# Confirm that the last configuration sent to the Nexus # Confirm that the configuration sent to the Nexus
# switch was a removal of vlan from the test interface. # switch includes deletion of the vlan (if expected)
self.assertTrue( # and untrunking of the vlan from the ethernet interface
self._is_in_last_nexus_cfg(['<vlan>', '<remove>']) # (if expected).
) self.assertTrue(self._is_vlan_unconfigured(
self._assertExpectedHTTP(res.status_int, KeyError) 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): def test_model_update_port_rollback(self):
"""Test for proper rollback for Cisco model layer update port failure. """Test for proper rollback for Cisco model layer update port failure.

135
neutron/tests/unit/cisco/test_nexus_plugin.py
View File

@ -99,6 +99,10 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
const.NET_VLAN_NAME: 'q-268', const.NET_VLAN_NAME: 'q-268',
const.NET_VLAN_ID: '268', const.NET_VLAN_ID: '268',
} }
self.delete_port_args_1 = [
self.attachment1[const.INSTANCE_ID],
self.network1[const.NET_VLAN_ID],
]
self.providernet = { self.providernet = {
const.NET_ID: 9, const.NET_ID: 9,
const.NET_NAME: 'pnet1', const.NET_NAME: 'pnet1',
@ -181,44 +185,38 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
self.assertEqual(expected_instance_id, INSTANCE1) 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', with mock.patch.object(cdb, 'is_provider_vlan',
return_value=True) as mock_db: return_value=True) as mock_db:
# Create a provider network
new_net_dict = self._cisco_nexus_plugin.create_network( new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1) self.providernet, self.attachment1)
mock_db.assert_called_once() mock_db.assert_called_once()
for attr in NET_ATTRS: for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr]) 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): def test_create_delete_providernet(self):
cfg.CONF.set_override('provider_vlan_auto_create', True, 'CISCO') self._create_delete_providernet(auto_create=True, auto_trunk=True)
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_provider_vlan_network_cfg_man_auto(self): def test_create_delete_provider_vlan_network_cfg_auto_man(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO') self._create_delete_providernet(auto_create=True, auto_trunk=False)
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_provider_vlan_network_cfg_man_man(self): def test_create_delete_provider_vlan_network_cfg_man_auto(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO') self._create_delete_providernet(auto_create=False, auto_trunk=True)
cfg.CONF.set_override('provider_vlan_auto_trunk', False, 'CISCO')
self.addCleanup(cfg.CONF.reset) def test_create_delete_provider_vlan_network_cfg_man_man(self):
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True): self._create_delete_providernet(auto_create=False, auto_trunk=False)
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_network_portchannel(self): def test_create_delete_network_portchannel(self):
"""Tests creation of a network over a portchannel.""" """Tests creation of a network over a portchannel."""
@ -237,45 +235,48 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
INSTANCE3, self.network3[const.NET_VLAN_ID] 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): def test_nexus_add_remove_router_interface(self):
"""Tests addition of a router interface.""" """Tests addition of a router interface."""
vlan_name = self.vlan_name self.assertTrue(self._add_router_interface())
vlan_id = self.vlan_id self.assertEqual(self._remove_router_interface(), '00000R1')
gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
subnet_id = '00001'
result = self._cisco_nexus_plugin.add_router_interface(vlan_name, def test_nexus_dup_add_router_interface(self):
vlan_id, """Tests a duplicate add of a router interface."""
subnet_id, self._add_router_interface()
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)
try: try:
self.assertRaises( self.assertRaises(
cisco_exc.SubnetInterfacePresent, cisco_exc.SubnetInterfacePresent,
self._cisco_nexus_plugin.add_router_interface, self._add_router_interface)
vlan_name, vlan_id, subnet_id, gateway_ip, router_id)
finally: finally:
self._cisco_nexus_plugin.remove_router_interface(vlan_id, self._remove_router_interface()
router_id)
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): def test_nexus_add_port_after_router_interface(self):
"""Tests creating a port after a router interface. """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 been created. Only a trunk call should be invoked and the
plugin should not attempt to recreate the vlan. plugin should not attempt to recreate the vlan.
""" """
vlan_name = self.vlan_name self._add_router_interface()
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)
# Create a network on the switch # Create a network on the switch
self._cisco_nexus_plugin.create_network( self._cisco_nexus_plugin.create_network(
self.network1, self.attachment1) self.network1, self.attachment1)
# Grab a list of all mock calls from ncclient # 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) edit_config.mock_calls)
# The last ncclient call should be for trunking and the second # The last ncclient call should be for trunking and the second