__metaclass__ = ABCMeta
- @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):
"""
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.
"""
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):
}
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):
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,
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
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.
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"))
-
- def get_all_ports(self, tenant_id, net_id, **kwargs):
- """Get all ports.
+ """Update the properties of a particular Virtual Network.
- 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.
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.
- """
- 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.
+ Not applicable to Nexus plugin. Defined here to satisfy abstract
+ method requirements.
"""
- 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
</config>
"""
-
CMD_VLAN_CONF_SNIPPET = """
<vlan>
<vlan-id-create-delete>
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>
</interface>
"""
-
FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
<show xmlns="http://www.cisco.com/nxos:1.0:vlan_mgr_cli">
<vlan>
</show>
"""
-
CMD_VLAN_SVI_SNIPPET = """
<interface>
<vlan>
"""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
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
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.
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,
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.
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.
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
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.
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',
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."""
INSTANCE3, self.network3[const.NET_VLAN_ID]
)
- def test_nexus_add_remove_router_interface(self):
- """Tests addition of a router interface."""
+ 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)
- 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
+ def _remove_router_interface(self):
+ """Remove a router interface created with _add_router_interface."""
vlan_id = self.vlan_id
- gateway_ip = '10.0.0.1/24'
router_id = '00000R1'
- subnet_id = '00001'
+ return self._cisco_nexus_plugin.remove_router_interface(vlan_id,
+ router_id)
- self._cisco_nexus_plugin.add_router_interface(vlan_name,
- vlan_id,
- subnet_id,
- gateway_ip,
- router_id)
+ def test_nexus_add_remove_router_interface(self):
+ """Tests addition of a router interface."""
+ self.assertTrue(self._add_router_interface())
+ self.assertEqual(self._remove_router_interface(), '00000R1')
+
+ 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.
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
Code Review / openstack-build / neutron-build.git / commitdiff