+++ /dev/null
-Neutron ML2 Cisco Nexus Mechanism Driver README
-
-
-Notes:
-
-The initial version of this driver supports only a single physical
-network.
-
-For provider networks, extended configuration options are not
-currently supported.
-
-This driver's database may have duplicate entries also found in the
-core ML2 database. Since the Cisco Nexus DB code is a port from the
-plugins/cisco implementation this duplication will remain until the
-plugins/cisco code is deprecated.
-
-
-For more details on using Cisco Nexus switches under ML2 please refer to:
-http://wiki.openstack.org/wiki/Neutron/ML2/MechCiscoNexus
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation
-# All Rights Reserved.
-#
-# 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.
-
-from oslo_config import cfg
-
-
-ml2_cisco_opts = [
- cfg.StrOpt('vlan_name_prefix', default='q-',
- help=_("VLAN Name prefix")),
- cfg.BoolOpt('svi_round_robin', default=False,
- help=_("Distribute SVI interfaces over all switches")),
- cfg.StrOpt('managed_physical_network',
- help=_("The physical network managed by the switches.")),
-]
-
-
-cfg.CONF.register_opts(ml2_cisco_opts, "ml2_cisco")
-
-#
-# Format for ml2_conf_cisco.ini 'ml2_mech_cisco_nexus' is:
-# {('<device ipaddr>', '<keyword>'): '<value>', ...}
-#
-# Example:
-# {('1.1.1.1', 'username'): 'admin',
-# ('1.1.1.1', 'password'): 'mySecretPassword',
-# ('1.1.1.1', 'compute1'): '1/1', ...}
-#
-
-
-class ML2MechCiscoConfig(object):
- """ML2 Mechanism Driver Cisco Configuration class."""
- nexus_dict = {}
-
- def __init__(self):
- self._create_ml2_mech_device_cisco_dictionary()
-
- def _create_ml2_mech_device_cisco_dictionary(self):
- """Create the ML2 device cisco dictionary.
-
- Read data from the ml2_conf_cisco.ini device supported sections.
- """
- multi_parser = cfg.MultiConfigParser()
- read_ok = multi_parser.read(cfg.CONF.config_file)
-
- if len(read_ok) != len(cfg.CONF.config_file):
- raise cfg.Error(_("Some config files were not parsed properly"))
-
- for parsed_file in multi_parser.parsed:
- for parsed_item in parsed_file.keys():
- dev_id, sep, dev_ip = parsed_item.partition(':')
- if dev_id.lower() == 'ml2_mech_cisco_nexus':
- for dev_key, value in parsed_file[parsed_item].items():
- self.nexus_dict[dev_ip, dev_key] = value[0]
+++ /dev/null
-# Copyright 2011 OpenStack Foundation.
-# All rights reserved.
-#
-# 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.
-#
-
-
-CREDENTIAL_USERNAME = 'user_name'
-CREDENTIAL_PASSWORD = 'password'
-
-USERNAME = 'username'
-PASSWORD = 'password'
-
-NETWORK_ADMIN = 'network_admin'
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation
-# All Rights Reserved.
-#
-# 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.
-
-"""Exceptions used by Cisco Nexus ML2 mechanism driver."""
-
-from neutron.common import exceptions
-
-
-class CredentialNotFound(exceptions.NeutronException):
- """Credential with this ID cannot be found."""
- message = _("Credential %(credential_id)s could not be found.")
-
-
-class CredentialNameNotFound(exceptions.NeutronException):
- """Credential Name could not be found."""
- message = _("Credential %(credential_name)s could not be found.")
-
-
-class CredentialAlreadyExists(exceptions.NeutronException):
- """Credential name already exists."""
- message = _("Credential %(credential_name)s already exists "
- "for tenant %(tenant_id)s.")
-
-
-class NexusComputeHostNotConfigured(exceptions.NeutronException):
- """Connection to compute host is not configured."""
- message = _("Connection to %(host)s is not configured.")
-
-
-class NexusConnectFailed(exceptions.NeutronException):
- """Failed to connect to Nexus switch."""
- message = _("Unable to connect to Nexus %(nexus_host)s. Reason: %(exc)s.")
-
-
-class NexusConfigFailed(exceptions.NeutronException):
- """Failed to configure Nexus switch."""
- message = _("Failed to configure Nexus: %(config)s. Reason: %(exc)s.")
-
-
-class NexusPortBindingNotFound(exceptions.NeutronException):
- """NexusPort Binding is not present."""
- message = _("Nexus Port Binding (%(filters)s) is not present")
-
- def __init__(self, **kwargs):
- filters = ','.join('%s=%s' % i for i in kwargs.items())
- super(NexusPortBindingNotFound, self).__init__(filters=filters)
-
-
-class NexusMissingRequiredFields(exceptions.NeutronException):
- """Missing required fields to configure nexus switch."""
- message = _("Missing required field(s) to configure nexus switch: "
- "%(fields)s")
-
-
-class NoNexusSviSwitch(exceptions.NeutronException):
- """No usable nexus switch found."""
- message = _("No usable Nexus switch found to create SVI interface.")
-
-
-class SubnetNotSpecified(exceptions.NeutronException):
- """Subnet id not specified."""
- message = _("No subnet_id specified for router gateway.")
-
-
-class SubnetInterfacePresent(exceptions.NeutronException):
- """Subnet SVI interface already exists."""
- message = _("Subnet %(subnet_id)s has an interface on %(router_id)s.")
-
-
-class PortIdForNexusSvi(exceptions.NeutronException):
- """Port Id specified for Nexus SVI."""
- message = _('Nexus hardware router gateway only uses Subnet Ids.')
ML2 Mechanism Driver for Cisco Nexus platforms.
"""
-from oslo_config import cfg
+from networking_cisco.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
-from neutron.common import constants as n_const
-from neutron.extensions import portbindings
-from neutron.openstack.common import log as logging
-from neutron.plugins.common import constants as p_const
-from neutron.plugins.ml2 import driver_api as api
-from neutron.plugins.ml2.drivers.cisco.nexus import config as conf
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions as excep
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2 as nxos_db
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_network_driver
-LOG = logging.getLogger(__name__)
-
-
-class CiscoNexusMechanismDriver(api.MechanismDriver):
-
- """Cisco Nexus ML2 Mechanism Driver."""
-
- def initialize(self):
- # Create ML2 device dictionary from ml2_conf.ini entries.
- conf.ML2MechCiscoConfig()
-
- # Extract configuration parameters from the configuration file.
- self._nexus_switches = conf.ML2MechCiscoConfig.nexus_dict
- LOG.debug("nexus_switches found = %s", self._nexus_switches)
-
- self.driver = nexus_network_driver.CiscoNexusDriver()
-
- def _valid_network_segment(self, segment):
- return (cfg.CONF.ml2_cisco.managed_physical_network is None or
- cfg.CONF.ml2_cisco.managed_physical_network ==
- segment[api.PHYSICAL_NETWORK])
-
- def _get_vlanid(self, segment):
- if (segment and segment[api.NETWORK_TYPE] == p_const.TYPE_VLAN and
- self._valid_network_segment(segment)):
- return segment.get(api.SEGMENTATION_ID)
-
- def _is_deviceowner_compute(self, port):
- return port['device_owner'].startswith('compute')
-
- def _is_status_active(self, port):
- return port['status'] == n_const.PORT_STATUS_ACTIVE
-
- def _get_switch_info(self, host_id):
- host_connections = []
- for switch_ip, attr in self._nexus_switches:
- if str(attr) == str(host_id):
- for port_id in (
- self._nexus_switches[switch_ip, attr].split(',')):
- if ':' in port_id:
- intf_type, port = port_id.split(':')
- else:
- intf_type, port = 'ethernet', port_id
- host_connections.append((switch_ip, intf_type, port))
-
- if host_connections:
- return host_connections
- else:
- raise excep.NexusComputeHostNotConfigured(host=host_id)
-
- def _configure_nxos_db(self, vlan_id, device_id, host_id):
- """Create the nexus database entry.
-
- Called during update precommit port event.
- """
- host_connections = self._get_switch_info(host_id)
- for switch_ip, intf_type, nexus_port in host_connections:
- port_id = '%s:%s' % (intf_type, nexus_port)
- nxos_db.add_nexusport_binding(port_id, str(vlan_id), switch_ip,
- device_id)
-
- def _configure_switch_entry(self, vlan_id, device_id, host_id):
- """Create a nexus switch entry.
-
- if needed, create a VLAN in the appropriate switch/port and
- configure the appropriate interfaces for this VLAN.
-
- Called during update postcommit port event.
- """
- vlan_name = cfg.CONF.ml2_cisco.vlan_name_prefix + str(vlan_id)
- host_connections = self._get_switch_info(host_id)
-
- # (nexus_port,switch_ip) will be unique in each iteration.
- # But switch_ip will repeat if host has >1 connection to same switch.
- # So track which switch_ips already have vlan created in this loop.
- vlan_already_created = []
- for switch_ip, intf_type, nexus_port in host_connections:
-
- # The VLAN needs to be created on the switch if no other
- # instance has been placed in this VLAN on a different host
- # attached to this switch. Search the existing bindings in the
- # database. If all the instance_id in the database match the
- # current device_id, then create the VLAN, but only once per
- # switch_ip. Otherwise, just trunk.
- all_bindings = nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
- previous_bindings = [row for row in all_bindings
- if row.instance_id != device_id]
- if previous_bindings or (switch_ip in vlan_already_created):
- LOG.debug("Nexus: trunk vlan %s", vlan_name)
- self.driver.enable_vlan_on_trunk_int(switch_ip, vlan_id,
- intf_type, nexus_port)
- else:
- vlan_already_created.append(switch_ip)
- LOG.debug("Nexus: create & trunk vlan %s", vlan_name)
- self.driver.create_and_trunk_vlan(
- switch_ip, vlan_id, vlan_name, intf_type, nexus_port)
-
- def _delete_nxos_db(self, vlan_id, device_id, host_id):
- """Delete the nexus database entry.
-
- Called during delete precommit port event.
- """
- try:
- rows = nxos_db.get_nexusvm_bindings(vlan_id, device_id)
- for row in rows:
- nxos_db.remove_nexusport_binding(
- row.port_id, row.vlan_id, row.switch_ip, row.instance_id)
- except excep.NexusPortBindingNotFound:
- return
-
- def _delete_switch_entry(self, vlan_id, device_id, host_id):
- """Delete the nexus switch entry.
-
- By accessing the current db entries determine if switch
- configuration can be removed.
-
- Called during update postcommit port event.
- """
- host_connections = self._get_switch_info(host_id)
-
- # (nexus_port,switch_ip) will be unique in each iteration.
- # But switch_ip will repeat if host has >1 connection to same switch.
- # So track which switch_ips already have vlan removed in this loop.
- vlan_already_removed = []
- for switch_ip, intf_type, nexus_port in host_connections:
-
- # if there are no remaining db entries using this vlan on this
- # nexus switch port then remove vlan from the switchport trunk.
- port_id = '%s:%s' % (intf_type, nexus_port)
- try:
- nxos_db.get_port_vlan_switch_binding(port_id, vlan_id,
- switch_ip)
- except excep.NexusPortBindingNotFound:
- self.driver.disable_vlan_on_trunk_int(switch_ip, vlan_id,
- intf_type, nexus_port)
-
- # if there are no remaining db entries using this vlan on this
- # nexus switch then remove the vlan.
- try:
- nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
- except excep.NexusPortBindingNotFound:
-
- # Do not perform a second time on same switch
- if switch_ip not in vlan_already_removed:
- self.driver.delete_vlan(switch_ip, vlan_id)
- vlan_already_removed.append(switch_ip)
-
- def _is_vm_migration(self, context):
- if (not context.bottom_bound_segment and
- context.original_bottom_bound_segment):
- return context.host != context.original_host
-
- def _port_action(self, port, segment, func):
- """Verify configuration and then process event."""
- device_id = port.get('device_id')
- host_id = port.get(portbindings.HOST_ID)
- vlan_id = self._get_vlanid(segment)
-
- if vlan_id and device_id and host_id:
- func(vlan_id, device_id, host_id)
- else:
- fields = "vlan_id " if not vlan_id else ""
- fields += "device_id " if not device_id else ""
- fields += "host_id" if not host_id else ""
- raise excep.NexusMissingRequiredFields(fields=fields)
-
- def update_port_precommit(self, context):
- """Update port pre-database transaction commit event."""
-
- # if VM migration is occurring then remove previous database entry
- # else process update event.
- if self._is_vm_migration(context):
- self._port_action(context.original,
- context.original_bottom_bound_segment,
- self._delete_nxos_db)
- else:
- if (self._is_deviceowner_compute(context.current) and
- self._is_status_active(context.current)):
- self._port_action(context.current,
- context.bottom_bound_segment,
- self._configure_nxos_db)
-
- def update_port_postcommit(self, context):
- """Update port non-database commit event."""
-
- # if VM migration is occurring then remove previous nexus switch entry
- # else process update event.
- if self._is_vm_migration(context):
- self._port_action(context.original,
- context.original_bottom_bound_segment,
- self._delete_switch_entry)
- else:
- if (self._is_deviceowner_compute(context.current) and
- self._is_status_active(context.current)):
- self._port_action(context.current,
- context.bottom_bound_segment,
- self._configure_switch_entry)
-
- def delete_port_precommit(self, context):
- """Delete port pre-database commit event."""
- if self._is_deviceowner_compute(context.current):
- self._port_action(context.current,
- context.bottom_bound_segment,
- self._delete_nxos_db)
-
- def delete_port_postcommit(self, context):
- """Delete port non-database commit event."""
- if self._is_deviceowner_compute(context.current):
- self._port_action(context.current,
- context.bottom_bound_segment,
- self._delete_switch_entry)
+class CiscoNexusMechanismDriver(mech_cisco_nexus.CiscoNexusMechanismDriver):
+ pass
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation
-# All Rights Reserved.
-#
-# 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 sqlalchemy.orm.exc as sa_exc
-
-import neutron.db.api as db
-from neutron.i18n import _LW
-from neutron.openstack.common import log as logging
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions as c_exc
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_models_v2
-
-
-LOG = logging.getLogger(__name__)
-
-
-def get_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Lists a nexusport binding."""
- LOG.debug("get_nexusport_binding() called")
- return _lookup_all_nexus_bindings(port_id=port_id,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- instance_id=instance_id)
-
-
-def get_nexusvlan_binding(vlan_id, switch_ip):
- """Lists a vlan and switch binding."""
- LOG.debug("get_nexusvlan_binding() called")
- return _lookup_all_nexus_bindings(vlan_id=vlan_id, switch_ip=switch_ip)
-
-
-def add_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Adds a nexusport binding."""
- LOG.debug("add_nexusport_binding() called")
- session = db.get_session()
- binding = nexus_models_v2.NexusPortBinding(port_id=port_id,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- instance_id=instance_id)
- session.add(binding)
- session.flush()
- return binding
-
-
-def remove_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Removes a nexusport binding."""
- LOG.debug("remove_nexusport_binding() called")
- session = db.get_session()
- binding = _lookup_all_nexus_bindings(session=session,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- port_id=port_id,
- instance_id=instance_id)
- for bind in binding:
- session.delete(bind)
- session.flush()
- return binding
-
-
-def update_nexusport_binding(port_id, new_vlan_id):
- """Updates nexusport binding."""
- if not new_vlan_id:
- LOG.warning(_LW("update_nexusport_binding called with no vlan"))
- return
- LOG.debug("update_nexusport_binding called")
- session = db.get_session()
- binding = _lookup_one_nexus_binding(session=session, port_id=port_id)
- binding.vlan_id = new_vlan_id
- session.merge(binding)
- session.flush()
- return binding
-
-
-def get_nexusvm_bindings(vlan_id, instance_id):
- """Lists nexusvm bindings."""
- LOG.debug("get_nexusvm_bindings() called")
- return _lookup_all_nexus_bindings(instance_id=instance_id,
- vlan_id=vlan_id)
-
-
-def get_port_vlan_switch_binding(port_id, vlan_id, switch_ip):
- """Lists nexusvm bindings."""
- LOG.debug("get_port_vlan_switch_binding() called")
- return _lookup_all_nexus_bindings(port_id=port_id,
- switch_ip=switch_ip,
- vlan_id=vlan_id)
-
-
-def get_port_switch_bindings(port_id, switch_ip):
- """List all vm/vlan bindings on a Nexus switch port."""
- LOG.debug("get_port_switch_bindings() called, "
- "port:'%(port_id)s', switch:'%(switch_ip)s'",
- {'port_id': port_id, 'switch_ip': switch_ip})
- try:
- return _lookup_all_nexus_bindings(port_id=port_id,
- switch_ip=switch_ip)
- except c_exc.NexusPortBindingNotFound:
- pass
-
-
-def _lookup_nexus_bindings(query_type, session=None, **bfilter):
- """Look up 'query_type' Nexus bindings matching the filter.
-
- :param query_type: 'all', 'one' or 'first'
- :param session: db session
- :param bfilter: filter for bindings query
- :return: bindings if query gave a result, else
- raise NexusPortBindingNotFound.
- """
- if session is None:
- session = db.get_session()
- query_method = getattr(session.query(
- nexus_models_v2.NexusPortBinding).filter_by(**bfilter), query_type)
- try:
- bindings = query_method()
- if bindings:
- return bindings
- except sa_exc.NoResultFound:
- pass
- raise c_exc.NexusPortBindingNotFound(**bfilter)
-
-
-def _lookup_all_nexus_bindings(session=None, **bfilter):
- return _lookup_nexus_bindings('all', session, **bfilter)
-
-
-def _lookup_one_nexus_binding(session=None, **bfilter):
- return _lookup_nexus_bindings('one', session, **bfilter)
-
-
-def _lookup_first_nexus_binding(session=None, **bfilter):
- return _lookup_nexus_bindings('first', session, **bfilter)
+++ /dev/null
-# Copyright 2013 OpenStack Foundation
-# All rights reserved.
-#
-# 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.
-
-"""
-Implements a Nexus-OS NETCONF over SSHv2 API Client
-"""
-
-from oslo_utils import excutils
-from oslo_utils import importutils
-
-from neutron.openstack.common import log as logging
-from neutron.plugins.ml2.drivers.cisco.nexus import config as conf
-from neutron.plugins.ml2.drivers.cisco.nexus import constants as const
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions as cexc
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_snippets as snipp
-
-LOG = logging.getLogger(__name__)
-
-
-class CiscoNexusDriver(object):
- """Nexus Driver Main Class."""
- def __init__(self):
- self.ncclient = None
- self.nexus_switches = conf.ML2MechCiscoConfig.nexus_dict
- self.connections = {}
-
- def _import_ncclient(self):
- """Import the NETCONF client (ncclient) module.
-
- The ncclient module is not installed as part of the normal Neutron
- distributions. It is imported dynamically in this module so that
- the import can be mocked, allowing unit testing without requiring
- the installation of ncclient.
-
- """
- return importutils.import_module('ncclient.manager')
-
- def _edit_config(self, nexus_host, target='running', config='',
- allowed_exc_strs=None):
- """Modify switch config for a target config type.
-
- :param nexus_host: IP address of switch to configure
- :param target: Target config type
- :param config: Configuration string in XML format
- :param allowed_exc_strs: Exceptions which have any of these strings
- as a subset of their exception message
- (str(exception)) can be ignored
-
- :returns None: if config was edited successfully
- Exception object: if _edit_config() encountered an exception
- containing one of allowed_exc_strs
-
- :raises: NexusConfigFailed: if _edit_config() encountered an exception
- not containing one of allowed_exc_strs
-
- """
- if not allowed_exc_strs:
- allowed_exc_strs = []
- mgr = self.nxos_connect(nexus_host)
- try:
- LOG.debug("NexusDriver config: %s", config)
- mgr.edit_config(target=target, config=config)
- except Exception as e:
- for exc_str in allowed_exc_strs:
- if exc_str in str(e):
- return e
- # Raise a Neutron exception. Include a description of
- # the original ncclient exception.
- raise cexc.NexusConfigFailed(config=config, exc=e)
-
- def nxos_connect(self, nexus_host):
- """Make SSH connection to the Nexus Switch."""
- if getattr(self.connections.get(nexus_host), 'connected', None):
- return self.connections[nexus_host]
-
- if not self.ncclient:
- self.ncclient = self._import_ncclient()
- nexus_ssh_port = int(self.nexus_switches[nexus_host, 'ssh_port'])
- nexus_user = self.nexus_switches[nexus_host, const.USERNAME]
- nexus_password = self.nexus_switches[nexus_host, const.PASSWORD]
- try:
- try:
- # With new ncclient version, we can pass device_params...
- man = self.ncclient.connect(host=nexus_host,
- port=nexus_ssh_port,
- username=nexus_user,
- password=nexus_password,
- device_params={"name": "nexus"})
- except TypeError:
- # ... but if that causes an error, we appear to have the old
- # ncclient installed, which doesn't understand this parameter.
- man = self.ncclient.connect(host=nexus_host,
- port=nexus_ssh_port,
- username=nexus_user,
- password=nexus_password)
- except Exception as e:
- # Raise a Neutron exception. Include a description of
- # the original ncclient exception.
- raise cexc.NexusConnectFailed(nexus_host=nexus_host, exc=e)
-
- self.connections[nexus_host] = man
- return self.connections[nexus_host]
-
- def create_xml_snippet(self, customized_config):
- """Create XML snippet.
-
- Creates the Proper XML structure for the Nexus Switch Configuration.
- """
- conf_xml_snippet = snipp.EXEC_CONF_SNIPPET % (customized_config)
- return conf_xml_snippet
-
- def create_vlan(self, nexus_host, vlanid, vlanname):
- """Create a VLAN on Nexus Switch given the VLAN ID and Name."""
- confstr = self.create_xml_snippet(
- snipp.CMD_VLAN_CONF_SNIPPET % (vlanid, vlanname))
- self._edit_config(nexus_host, target='running', config=confstr)
-
- # Enable VLAN active and no-shutdown states. Some versions of
- # Nexus switch do not allow state changes for the extended VLAN
- # range (1006-4094), but these errors can be ignored (default
- # values are appropriate).
- for snippet in [snipp.CMD_VLAN_ACTIVE_SNIPPET,
- snipp.CMD_VLAN_NO_SHUTDOWN_SNIPPET]:
- try:
- confstr = self.create_xml_snippet(snippet % vlanid)
- self._edit_config(
- nexus_host,
- target='running',
- config=confstr,
- allowed_exc_strs=["Can't modify state for extended",
- "Command is only allowed on VLAN"])
- except cexc.NexusConfigFailed:
- with excutils.save_and_reraise_exception():
- self.delete_vlan(nexus_host, vlanid)
-
- def delete_vlan(self, nexus_host, vlanid):
- """Delete a VLAN on Nexus Switch given the VLAN ID."""
- confstr = snipp.CMD_NO_VLAN_CONF_SNIPPET % vlanid
- confstr = self.create_xml_snippet(confstr)
- self._edit_config(nexus_host, target='running', config=confstr)
-
- def build_intf_confstr(self, snippet, intf_type, interface, vlanid):
- """Build the VLAN config string xml snippet to be used."""
- confstr = snippet % (intf_type, interface, vlanid, intf_type)
- confstr = self.create_xml_snippet(confstr)
- LOG.debug("NexusDriver: %s", confstr)
- return confstr
-
- def enable_vlan_on_trunk_int(self, nexus_host, vlanid, intf_type,
- interface):
- """Enable a VLAN on a trunk interface."""
- # Configure a new VLAN into the interface using 'ADD' keyword
- confstr = self.build_intf_confstr(
- snippet=snipp.CMD_INT_VLAN_ADD_SNIPPET,
- intf_type=intf_type,
- interface=interface,
- vlanid=vlanid
- )
- exc_str = ["switchport trunk allowed vlan list is empty"]
- ret_exc = self._edit_config(nexus_host, target='running',
- config=confstr,
- allowed_exc_strs=exc_str)
- if ret_exc:
- # If no switchports have been configured on the switch
- # before the new 'ADD', configure the VLAN into the
- # interface without the keyword so as to create a vlan list
- confstr = self.build_intf_confstr(
- snippet=snipp.CMD_INT_VLAN_SNIPPET,
- intf_type=intf_type,
- interface=interface,
- vlanid=vlanid
- )
- self._edit_config(nexus_host, target='running',
- config=confstr)
-
- def disable_vlan_on_trunk_int(self, nexus_host, vlanid, intf_type,
- interface):
- """Disable a VLAN on a trunk interface."""
- confstr = (snipp.CMD_NO_VLAN_INT_SNIPPET %
- (intf_type, interface, vlanid, intf_type))
- confstr = self.create_xml_snippet(confstr)
- self._edit_config(nexus_host, target='running', config=confstr)
-
- def create_and_trunk_vlan(self, nexus_host, vlan_id, vlan_name,
- intf_type, nexus_port):
- """Create VLAN and trunk it on the specified ports."""
- self.create_vlan(nexus_host, vlan_id, vlan_name)
- LOG.debug("NexusDriver created VLAN: %s", vlan_id)
- if nexus_port:
- self.enable_vlan_on_trunk_int(nexus_host, vlan_id, intf_type,
- nexus_port)
+++ /dev/null
-# Copyright 2013 OpenStack Foundation.
-# All rights reserved.
-#
-# 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.
-
-
-"""
-Cisco Nexus-OS XML-based configuration snippets.
-"""
-
-
-# The following are standard strings, messages used to communicate with Nexus.
-EXEC_CONF_SNIPPET = """
- <config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
- <configure>
- <__XML__MODE__exec_configure>%s
- </__XML__MODE__exec_configure>
- </configure>
- </config>
-"""
-
-CMD_VLAN_CONF_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <name>
- <vlan-name>%s</vlan-name>
- </name>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_VLAN_ACTIVE_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <state>
- <vstate>active</vstate>
- </state>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_VLAN_NO_SHUTDOWN_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <no>
- <shutdown/>
- </no>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_NO_VLAN_CONF_SNIPPET = """
- <no>
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- </vlan-id-create-delete>
- </vlan>
- </no>
-"""
-
-CMD_INT_VLAN_HEADER = """
- <interface>
- <%s>
- <interface>%s</interface>
- <__XML__MODE_if-ethernet-switch>
- <switchport>
- <trunk>
- <allowed>
- <vlan>"""
-
-CMD_VLAN_ID = """
- <vlan_id>%s</vlan_id>"""
-
-CMD_VLAN_ADD_ID = """
- <add>%s
- </add>""" % CMD_VLAN_ID
-
-CMD_INT_VLAN_TRAILER = """
- </vlan>
- </allowed>
- </trunk>
- </switchport>
- </__XML__MODE_if-ethernet-switch>
- </%s>
- </interface>
-"""
-
-CMD_INT_VLAN_SNIPPET = (CMD_INT_VLAN_HEADER +
- CMD_VLAN_ID +
- CMD_INT_VLAN_TRAILER)
-
-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>
- <interface>%s</interface>
- <__XML__MODE_if-ethernet-switch>
- <switchport></switchport>
- <switchport>
- <trunk>
- <allowed>
- <vlan>
- <remove>
- <vlan>%s</vlan>
- </remove>
- </vlan>
- </allowed>
- </trunk>
- </switchport>
- </__XML__MODE_if-ethernet-switch>
- </%s>
- </interface>
-"""
-
-CMD_VLAN_SVI_SNIPPET = """
-<interface>
- <vlan>
- <vlan>%s</vlan>
- <__XML__MODE_vlan>
- <no>
- <shutdown/>
- </no>
- <ip>
- <address>
- <address>%s</address>
- </address>
- </ip>
- </__XML__MODE_vlan>
- </vlan>
-</interface>
-"""
-
-CMD_NO_VLAN_SVI_SNIPPET = """
-<no>
- <interface>
- <vlan>
- <vlan>%s</vlan>
- </vlan>
- </interface>
-</no>
-"""
--- /dev/null
+networking-cisco
+++ /dev/null
-# Copyright (c) 2014 Cisco Systems, Inc.
-# All rights reserved.
-#
-# 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 mock
-from oslo_config import cfg
-
-from neutron.plugins.ml2.drivers.cisco.nexus import config as cisco_config
-from neutron.tests import base
-
-
-class TestCiscoNexusPluginConfig(base.BaseTestCase):
-
- def setUp(self):
- self.config_parse()
- super(TestCiscoNexusPluginConfig, self).setUp()
-
- def test_config_parse_error(self):
- """Check that config error is raised upon config parser failure."""
- with mock.patch.object(cfg, 'MultiConfigParser') as parser:
- parser.return_value.read.return_value = []
- self.assertRaises(cfg.Error, cisco_config.ML2MechCiscoConfig)
-
- def test_create_device_dictionary(self):
- """Test creation of the device dictionary based on nexus config."""
- test_config = {
- 'ml2_mech_cisco_nexus:1.1.1.1': {
- 'username': ['admin'],
- 'password': ['mySecretPassword'],
- 'ssh_port': [22],
- 'compute1': ['1/1'],
- 'compute2': ['1/2'],
- 'compute5': ['1/3,1/4']
- },
- 'ml2_mech_cisco_nexus:2.2.2.2': {
- 'username': ['admin'],
- 'password': ['mySecretPassword'],
- 'ssh_port': [22],
- 'compute3': ['1/1'],
- 'compute4': ['1/2'],
- 'compute5': ['portchannel:20,portchannel:30']
- },
- }
- expected_dev_dict = {
- ('1.1.1.1', 'username'): 'admin',
- ('1.1.1.1', 'password'): 'mySecretPassword',
- ('1.1.1.1', 'ssh_port'): 22,
- ('1.1.1.1', 'compute1'): '1/1',
- ('1.1.1.1', 'compute2'): '1/2',
- ('1.1.1.1', 'compute5'): '1/3,1/4',
- ('2.2.2.2', 'username'): 'admin',
- ('2.2.2.2', 'password'): 'mySecretPassword',
- ('2.2.2.2', 'ssh_port'): 22,
- ('2.2.2.2', 'compute3'): '1/1',
- ('2.2.2.2', 'compute4'): '1/2',
- ('2.2.2.2', 'compute5'): 'portchannel:20,portchannel:30',
- }
- with mock.patch.object(cfg, 'MultiConfigParser') as parser:
- parser.return_value.read.return_value = cfg.CONF.config_file
- parser.return_value.parsed = [test_config]
- cisco_config.ML2MechCiscoConfig()
- self.assertEqual(expected_dev_dict,
- cisco_config.ML2MechCiscoConfig.nexus_dict)
+++ /dev/null
-# Copyright (c) 2012 OpenStack Foundation.
-#
-# 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 contextlib
-import mock
-
-import webob.exc as wexc
-
-from neutron.api.v2 import base
-from neutron import context
-from neutron.extensions import portbindings
-from neutron import manager
-from neutron.openstack.common import log as logging
-from neutron.plugins.common import constants as p_const
-from neutron.plugins.ml2 import driver_api as api
-from neutron.plugins.ml2 import driver_context
-from neutron.plugins.ml2.drivers.cisco.nexus import config as cisco_config
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions as c_exc
-from neutron.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_network_driver
-from neutron.tests.unit.ml2 import test_ml2_plugin
-
-
-LOG = logging.getLogger(__name__)
-PHYS_NET = 'physnet1'
-COMP_HOST_NAME = 'testhost'
-COMP_HOST_NAME_2 = 'testhost_2'
-VLAN_START = 1000
-VLAN_END = 1100
-NEXUS_IP_ADDR = '1.1.1.1'
-NETWORK_NAME = 'test_network'
-NETWORK_NAME_2 = 'test_network_2'
-NEXUS_INTERFACE = '1/1'
-NEXUS_INTERFACE_2 = '1/2'
-CIDR_1 = '10.0.0.0/24'
-CIDR_2 = '10.0.1.0/24'
-DEVICE_ID_1 = '11111111-1111-1111-1111-111111111111'
-DEVICE_ID_2 = '22222222-2222-2222-2222-222222222222'
-DEVICE_OWNER = 'compute:None'
-BOUND_SEGMENT1 = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
- api.PHYSICAL_NETWORK: PHYS_NET,
- api.SEGMENTATION_ID: VLAN_START}
-BOUND_SEGMENT2 = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
- api.PHYSICAL_NETWORK: PHYS_NET,
- api.SEGMENTATION_ID: VLAN_START + 1}
-
-
-class CiscoML2MechanismTestCase(test_ml2_plugin.Ml2PluginV2TestCase):
- _mechanism_drivers = ['cisco_nexus']
-
- def setUp(self):
- """Configure for end-to-end neutron testing using a mock ncclient.
-
- This setup includes:
- - Configure the ML2 plugin to use VLANs in the range of 1000-1100.
- - Configure the Cisco mechanism driver to use an imaginary switch
- at NEXUS_IP_ADDR.
- - Create a mock NETCONF client (ncclient) for the Cisco mechanism
- driver
-
- """
-
- # Configure the Cisco Nexus mechanism driver
- nexus_config = {
- (NEXUS_IP_ADDR, 'username'): 'admin',
- (NEXUS_IP_ADDR, 'password'): 'mySecretPassword',
- (NEXUS_IP_ADDR, 'ssh_port'): 22,
- (NEXUS_IP_ADDR, COMP_HOST_NAME): NEXUS_INTERFACE,
- (NEXUS_IP_ADDR, COMP_HOST_NAME_2): NEXUS_INTERFACE_2}
- nexus_patch = mock.patch.dict(
- cisco_config.ML2MechCiscoConfig.nexus_dict,
- nexus_config)
- nexus_patch.start()
- self.addCleanup(nexus_patch.stop)
-
- # The NETCONF client module is not included in the DevStack
- # distribution, so mock this module for unit testing.
- self.mock_ncclient = mock.Mock()
- mock.patch.object(nexus_network_driver.CiscoNexusDriver,
- '_import_ncclient',
- return_value=self.mock_ncclient).start()
-
- # Mock port context values for bound_segments and 'status'.
- self.mock_bound_segment = mock.patch.object(
- driver_context.PortContext,
- 'bottom_bound_segment',
- new_callable=mock.PropertyMock).start()
- self.mock_bound_segment.return_value = BOUND_SEGMENT1
-
- self.mock_original_bound_segment = mock.patch.object(
- driver_context.PortContext,
- 'original_bottom_bound_segment',
- new_callable=mock.PropertyMock).start()
- self.mock_original_bound_segment.return_value = None
-
- # Use _is_status_active method to determine bind state.
- def _mock_check_bind_state(port_context):
- if (port_context[portbindings.VIF_TYPE] !=
- portbindings.VIF_TYPE_UNBOUND):
- return True
- else:
- return False
-
- self.mock_status = mock.patch.object(
- mech_cisco_nexus.CiscoNexusMechanismDriver,
- '_is_status_active').start()
- self.mock_status.side_effect = _mock_check_bind_state
-
- super(CiscoML2MechanismTestCase, self).setUp()
-
- self.port_create_status = 'DOWN'
-
- def _create_deviceowner_mock(self):
- # Mock deviceowner method for UT's that expect update precommit
- # failures. This allows control of delete_port_pre/postcommit()
- # actions.
- mock_deviceowner = mock.patch.object(
- mech_cisco_nexus.CiscoNexusMechanismDriver,
- '_is_deviceowner_compute').start()
- mock_deviceowner.return_value = False
- self.addCleanup(mock_deviceowner.stop)
-
- @contextlib.contextmanager
- def _patch_ncclient(self, attr, value):
- """Configure an attribute on the mock ncclient module.
-
- This method can be used to inject errors by setting a side effect
- or a return value for an ncclient method.
-
- :param attr: ncclient attribute (typically method) to be configured.
- :param value: Value to be configured on the attribute.
-
- """
- # Configure attribute.
- config = {attr: value}
- self.mock_ncclient.configure_mock(**config)
- # Continue testing
- yield
- # Unconfigure attribute
- 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.connect.return_value.
- edit_config.mock_calls):
- configlet = call[2]['config']
- if all(word in configlet for word in words):
- return True
- return False
-
- def _is_in_last_nexus_cfg(self, words):
- """Confirm last config sent to Nexus contains specified keywords."""
- last_cfg = (self.mock_ncclient.connect.return_value.
- edit_config.mock_calls[-1][2]['config'])
- return all(word in last_cfg for word in words)
-
- def _is_vlan_configured(self, vlan_creation_expected=True,
- first_vlan_addition=False):
- """Confirm if VLAN was configured or not."""
- vlan_created = self._is_in_nexus_cfg(['vlan', 'vlan-name'])
- add_appears = self._is_in_last_nexus_cfg(['add'])
- # The first VLAN being configured should be done without the
- # ADD keyword. Thereafter additional VLANs to be configured
- # should be done with the ADD keyword.
- add_keyword_expected = not first_vlan_addition
- return (self._is_in_last_nexus_cfg(['allowed', 'vlan']) and
- 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(
- ['no', 'vlan', 'vlan-id-create-delete'])
- return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and
- vlan_deleted == vlan_deletion_expected)
-
-
-class TestCiscoBasicGet(CiscoML2MechanismTestCase,
- test_ml2_plugin.TestMl2BasicGet):
-
- pass
-
-
-class TestCiscoV2HTTPResponse(CiscoML2MechanismTestCase,
- test_ml2_plugin.TestMl2V2HTTPResponse):
-
- pass
-
-
-class TestCiscoPortsV2(CiscoML2MechanismTestCase,
- test_ml2_plugin.TestMl2PortsV2):
-
- @contextlib.contextmanager
- def _create_resources(self, name=NETWORK_NAME, cidr=CIDR_1,
- device_id=DEVICE_ID_1,
- host_id=COMP_HOST_NAME,
- expected_failure=False):
- """Create network, subnet, and port resources for test cases.
-
- Create a network, subnet, port and then update the port, yield the
- result, then delete the port, subnet and network.
-
- :param name: Name of network to be created.
- :param cidr: cidr address of subnetwork to be created.
- :param device_id: Device ID to use for port to be created/updated.
- :param host_id: Host ID to use for port create/update.
- :param expected_failure: Set to True when an update_port_precommit
- failure is expected. Results in no actions being taken in
- delete_port_pre/postcommit() methods.
- """
- with self.network(name=name) as network:
- with self.subnet(network=network, cidr=cidr) as subnet:
- with self.port(subnet=subnet, cidr=cidr) as port:
-
- data = {'port': {portbindings.HOST_ID: host_id,
- 'device_id': device_id,
- 'device_owner': DEVICE_OWNER,
- 'admin_state_up': True}}
- req = self.new_update_request('ports', data,
- port['port']['id'])
- yield req.get_response(self.api)
- if expected_failure:
- self._create_deviceowner_mock()
- self._delete('ports', port['port']['id'])
- self._delete('networks', network['network']['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 _mock_config_first_trunk(self):
- """Mock the behavior for the first VLAN addition.
-
- When the first VLAN is being added to the interface the usage of
- the ADD keyword should raise an exception specifying that the ADD
- keyword cannot be used till a VLAN list is created and to create
- a VLAN list the configuration should not contain the ADD keyword.
-
- """
- config = "switchport trunk allowed vlan add"
- exc = Exception("switchport trunk allowed vlan list is empty, "
- "please specify a list using "
- "'switchport trunk allowed vlan X' "
- "before using the add option")
- return (self._patch_ncclient(
- 'connect.return_value.edit_config.side_effect',
- self._config_dependent_side_effect(config, exc)))
-
- def test_create_ports_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- #ensures the API chooses the emulation code path
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_port
- with mock.patch.object(plugin_obj,
- '_create_port_db') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_port_bulk(self.fmt, 2,
- net['network']['id'],
- 'test',
- True)
- # Expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'ports',
- wexc.HTTPInternalServerError.code)
-
- def test_create_ports_bulk_native(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
-
- def test_create_ports_bulk_emulated(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
-
- def test_create_ports_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
- ctx = context.get_admin_context()
- with self.network() as net:
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_port
- with mock.patch.object(plugin_obj,
- '_create_port_db') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- res = self._create_port_bulk(self.fmt, 2, net['network']['id'],
- 'test', True, context=ctx)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'ports',
- wexc.HTTPInternalServerError.code)
-
- def test_nexus_enable_vlan_cmd_on_same_host(self):
- """Verify the syntax of the command to enable a vlan on an intf.
-
- Test of the following ml2_conf_cisco_ini config:
- [ml2_mech_cisco_nexus:1.1.1.1]
- Resource A on host=COMP_HOST_NAME with vlan_id = 1000
- Resource B on host=COMP_HOST_NAME with vlan_id = 1001
-
- Confirm that when configuring the first VLAN on a Nexus interface,
- the final command string sent to the switch does not contain the
- keyword 'add'. The initial attempt will contain 'add' but when
- the switch rejects it, the re-attempt shouldn't contain the 'add'.
-
- Confirm that for the second VLAN configured on a Nexus interface,
- the command string sent to the switch contains the keyword 'add'
- since it is on the same host.
-
- """
- # First vlan should be configured without 'add' keyword and an
- # exception should be raised when it is done with the 'add'
- # thereby triggering a re-attempt without the 'add'.
- with self._mock_config_first_trunk():
- with self._create_resources():
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- first_vlan_addition=True))
- self.mock_ncclient.reset_mock()
- self.mock_bound_segment.return_value = BOUND_SEGMENT2
-
- # Second vlan should be configured with the 'add' keyword
- # when on first host.
- with(self._patch_ncclient(
- 'connect.return_value.edit_config.side_effect',
- None)):
- with self._create_resources(name=NETWORK_NAME_2,
- device_id=DEVICE_ID_2,
- cidr=CIDR_2,
- host_id=COMP_HOST_NAME):
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- first_vlan_addition=False
- ))
-
- # Return to first segment for delete port calls.
- self.mock_bound_segment.return_value = BOUND_SEGMENT1
-
- def test_nexus_enable_vlan_cmd_on_different_hosts(self):
- """Verify the syntax of the command to enable a vlan on an intf.
-
- Test of the following ml2_conf_cisco_ini config:
- [ml2_mech_cisco_nexus:1.1.1.1]
- Resource A on host=COMP_HOST_NAME with vlan_id = 1000
- Resource B on host=COMP_HOST_NAME_2 with vlan_id = 1001
-
- Confirm that when configuring the first VLAN on a Nexus interface,
- the final command string sent to the switch does not contain the
- keyword 'add'. The initial attempt will contain 'add' but when
- the switch rejects it, the re-attempt shouldn't contain the 'add'.
-
- Confirm that for the second VLAN configured on a Nexus interface,
- the command string sent to the switch does not contain the
- keyword 'add' since it is on a different host.
-
- """
- # First vlan should be configured without 'add' keyword and an
- # exception should be raised when it is done with the 'add'
- # thereby triggering a re-attempt without the 'add'.
- with self._mock_config_first_trunk():
- with self._create_resources():
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- first_vlan_addition=True))
- self.mock_ncclient.reset_mock()
- self.mock_bound_segment.return_value = BOUND_SEGMENT2
-
- # Second vlan should be configured without the 'add' keyword
- # when on second host.
- with self._create_resources(name=NETWORK_NAME_2,
- device_id=DEVICE_ID_2,
- cidr=CIDR_2,
- host_id=COMP_HOST_NAME_2):
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- first_vlan_addition=True
- ))
-
- # Return to first segment for delete port calls.
- self.mock_bound_segment.return_value = BOUND_SEGMENT1
-
- def test_ncclient_version_detect(self):
- """Test ability to handle connection to old and new-style ncclient.
-
- We used to require a custom version of the ncclient library. However,
- recent contributions to the ncclient make this unnecessary. Our
- driver was modified to be able to establish a connection via both
- the old and new type of ncclient.
-
- The new style ncclient.connect() function takes one additional
- parameter.
-
- The ML2 driver uses this to detect whether we are dealing with an
- old or new ncclient installation.
-
- """
- # The code we are exercising calls connect() twice, if there is a
- # TypeError on the first call (if the old ncclient is installed).
- # The second call should succeed. That's what we are simulating here.
- orig_connect_return_val = self.mock_ncclient.connect.return_value
- with self._patch_ncclient('connect.side_effect',
- [TypeError, orig_connect_return_val]):
- with self._create_resources() as result:
- self.assertEqual(result.status_int,
- wexc.HTTPOk.code)
-
- def test_ncclient_fail_on_second_connect(self):
- """Test that other errors during connect() sequences are still handled.
-
- If the old ncclient is installed, we expect to get a TypeError first,
- but should still handle other errors in the usual way, whether they
- appear on the first or second call to connect().
-
- """
- with self._patch_ncclient('connect.side_effect',
- [TypeError, IOError]):
- with self._create_resources() as result:
- self._assertExpectedHTTP(result.status_int,
- c_exc.NexusConnectFailed)
-
- def test_nexus_connect_fail(self):
- """Test failure to connect to a Nexus switch.
-
- While creating a network, subnet, and port, simulate a connection
- failure to a nexus switch. Confirm that the expected HTTP code
- is returned for the create port operation.
-
- """
- with self._patch_ncclient('connect.side_effect',
- AttributeError):
- with self._create_resources() as result:
- self._assertExpectedHTTP(result.status_int,
- c_exc.NexusConnectFailed)
-
- def test_nexus_vlan_config_two_hosts(self):
- """Verify config/unconfig of vlan on two compute hosts."""
-
- @contextlib.contextmanager
- def _create_port_check_vlan(comp_host_name, device_id,
- vlan_creation_expected=True):
- with self.port(subnet=subnet, fmt=self.fmt) as port:
- data = {'port': {portbindings.HOST_ID: comp_host_name,
- 'device_id': device_id,
- 'device_owner': DEVICE_OWNER,
- 'admin_state_up': True}}
- req = self.new_update_request('ports', data,
- port['port']['id'])
- req.get_response(self.api)
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=vlan_creation_expected,
- first_vlan_addition=True))
- self.mock_ncclient.reset_mock()
- yield
- self._delete('ports', port['port']['id'])
-
- # Create network and subnet
- with self._mock_config_first_trunk():
- with self.network(name=NETWORK_NAME) as network:
- with self.subnet(network=network, cidr=CIDR_1) as subnet:
-
- # Create an instance on first compute host
- with _create_port_check_vlan(COMP_HOST_NAME, DEVICE_ID_1,
- vlan_creation_expected=True):
- # Create an instance on second compute host
- with _create_port_check_vlan(
- COMP_HOST_NAME_2,
- DEVICE_ID_2,
- vlan_creation_expected=False):
- pass
-
- # Instance on second host is now terminated.
- # Vlan should be untrunked from port, but vlan should
- # still exist on the switch.
- self.assertTrue(self._is_vlan_unconfigured(
- vlan_deletion_expected=False))
- self.mock_ncclient.reset_mock()
-
- # Instance on first host is now terminated.
- # Vlan should be untrunked from port and vlan should have
- # been deleted from the switch.
- self.assertTrue(self._is_vlan_unconfigured(
- vlan_deletion_expected=True))
-
- def test_nexus_vm_migration(self):
- """Verify VM (live) migration.
-
- Simulate the following:
- Nova informs neutron of live-migration with port-update(new host).
- This should trigger two update_port_pre/postcommit() calls.
-
- The first one should only change the current host_id and remove the
- binding resulting in the mechanism drivers receiving:
- PortContext.original['binding:host_id']: previous value
- PortContext.original_bottom_bound_segment: previous value
- PortContext.current['binding:host_id']: current (new) value
- PortContext.bottom_bound_segment: None
-
- The second one binds the new host resulting in the mechanism
- drivers receiving:
- PortContext.original['binding:host_id']: previous value
- PortContext.original_bottom_bound_segment: None
- PortContext.current['binding:host_id']: previous value
- PortContext.bottom_bound_segment: new value
- """
-
- # Create network, subnet and port.
- with self._create_resources() as result:
- # Verify initial database entry.
- # Use port_id to verify that 1st host name was used.
- binding = nexus_db_v2.get_nexusvm_bindings(VLAN_START,
- DEVICE_ID_1)[0]
- intf_type, nexus_port = binding.port_id.split(':')
- self.assertEqual(nexus_port, NEXUS_INTERFACE)
-
- port = self.deserialize(self.fmt, result)
- port_id = port['port']['id']
-
- # Trigger update event to unbind segment.
- # Results in port being deleted from nexus DB and switch.
- data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
- self.mock_bound_segment.return_value = None
- self.mock_original_bound_segment.return_value = BOUND_SEGMENT1
- self.new_update_request('ports', data,
- port_id).get_response(self.api)
-
- # Verify that port entry has been deleted.
- self.assertRaises(c_exc.NexusPortBindingNotFound,
- nexus_db_v2.get_nexusvm_bindings,
- VLAN_START, DEVICE_ID_1)
-
- # Trigger update event to bind segment with new host.
- self.mock_bound_segment.return_value = BOUND_SEGMENT1
- self.mock_original_bound_segment.return_value = None
- self.new_update_request('ports', data,
- port_id).get_response(self.api)
-
- # Verify that port entry has been added using new host name.
- # Use port_id to verify that 2nd host name was used.
- binding = nexus_db_v2.get_nexusvm_bindings(VLAN_START,
- DEVICE_ID_1)[0]
- intf_type, nexus_port = binding.port_id.split(':')
- self.assertEqual(nexus_port, NEXUS_INTERFACE_2)
-
- def test_nexus_config_fail(self):
- """Test a Nexus switch configuration failure.
-
- While creating a network, subnet, and port, simulate a nexus
- switch configuration error. Confirm that the expected HTTP code
- is returned for the create port operation.
-
- """
- with self._patch_ncclient(
- 'connect.return_value.edit_config.side_effect',
- AttributeError):
- with self._create_resources() as result:
- self._assertExpectedHTTP(result.status_int,
- c_exc.NexusConfigFailed)
-
- def test_nexus_extended_vlan_range_failure(self):
- """Test that extended VLAN range config errors are ignored.
-
- Some versions of Nexus switch do not allow state changes for
- the extended VLAN range (1006-4094), but these errors can be
- ignored (default values are appropriate). Test that such errors
- 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(
- 'connect.return_value.edit_config.side_effect',
- mock_edit_config_a):
- with self._create_resources() as result:
- self.assertEqual(result.status_int, wexc.HTTPOk.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(
- 'connect.return_value.edit_config.side_effect',
- mock_edit_config_b):
- with self._create_resources() as result:
- self.assertEqual(result.status_int, wexc.HTTPOk.code)
-
- def test_nexus_vlan_config_rollback(self):
- """Test rollback following Nexus VLAN state config failure.
-
- Test that the Cisco Nexus plugin correctly deletes the VLAN
- on the Nexus switch when the 'state active' command fails (for
- a reason other than state configuration change is rejected
- for the extended VLAN range).
-
- """
- vlan_state_configs = ['state active', 'no shutdown']
- for config in vlan_state_configs:
- with self._patch_ncclient(
- 'connect.return_value.edit_config.side_effect',
- self._config_dependent_side_effect(config, ValueError)):
- with self._create_resources() as result:
- # 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(result.status_int,
- c_exc.NexusConfigFailed)
-
- def test_nexus_host_not_configured(self):
- """Test handling of a NexusComputeHostNotConfigured exception.
-
- Test the Cisco NexusComputeHostNotConfigured exception by using
- a fictitious host name during port creation.
-
- """
- with self._create_resources(host_id='fake_host',
- expected_failure=True) as result:
- self._assertExpectedHTTP(result.status_int,
- c_exc.NexusComputeHostNotConfigured)
-
- def test_nexus_missing_fields(self):
- """Test handling of a NexusMissingRequiredFields exception.
-
- Test the Cisco NexusMissingRequiredFields exception by using
- empty device_id value during port creation.
-
- """
- with self._create_resources(device_id='',
- expected_failure=True) as result:
- self._assertExpectedHTTP(result.status_int,
- c_exc.NexusMissingRequiredFields)
-
- def test_update_port_mac(self):
- # REVISIT: test passes, but is back-end OK?
- host_arg = {
- portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- }
- arg_list = (portbindings.HOST_ID, 'device_id',)
- self.check_update_port_mac(host_arg=host_arg, arg_list=arg_list)
-
-
-class TestCiscoNetworksV2(CiscoML2MechanismTestCase,
- test_ml2_plugin.TestMl2NetworksV2):
-
- def test_create_networks_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_network
- #ensures the API choose the emulation code path
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- with mock.patch.object(plugin_obj,
- '_create_network_db') as patched_plugin:
- def side_effect(*args, **kwargs):
- return self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
- patched_plugin.side_effect = side_effect
- res = self._create_network_bulk(self.fmt, 2, 'test', True)
- LOG.debug("response is %s" % res)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'networks',
- wexc.HTTPInternalServerError.code)
-
- def test_create_networks_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk network create")
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_network
- with mock.patch.object(plugin_obj,
- '_create_network_db') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- res = self._create_network_bulk(self.fmt, 2, 'test', True)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'networks',
- wexc.HTTPInternalServerError.code)
-
-
-class TestCiscoSubnetsV2(CiscoML2MechanismTestCase,
- test_ml2_plugin.TestMl2SubnetsV2):
-
- def test_create_subnets_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- #ensures the API choose the emulation code path
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_subnet
- with mock.patch.object(plugin_obj,
- '_create_subnet_db') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_subnet_bulk(self.fmt, 2,
- net['network']['id'],
- 'test')
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'subnets',
- wexc.HTTPInternalServerError.code)
-
- def test_create_subnets_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk subnet create")
- plugin_obj = manager.NeutronManager.get_plugin()
- orig = plugin_obj.create_subnet
- with mock.patch.object(plugin_obj,
- '_create_subnet_db') as patched_plugin:
- def side_effect(*args, **kwargs):
- return self._fail_second_call(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_subnet_bulk(self.fmt, 2,
- net['network']['id'],
- 'test')
-
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'subnets',
- wexc.HTTPInternalServerError.code)
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation.
-#
-# 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 collections
-import mock
-from oslo_utils import importutils
-import testtools
-
-from neutron.common import constants as n_const
-from neutron.extensions import portbindings
-from neutron.plugins.ml2 import driver_api as api
-from neutron.plugins.ml2.drivers.cisco.nexus import constants
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions
-from neutron.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_network_driver
-from neutron.tests.unit import testlib_api
-
-
-NEXUS_IP_ADDRESS = '1.1.1.1'
-NEXUS_IP_ADDRESS_PC = '2.2.2.2'
-NEXUS_IP_ADDRESS_DUAL = '3.3.3.3'
-HOST_NAME_1 = 'testhost1'
-HOST_NAME_2 = 'testhost2'
-HOST_NAME_PC = 'testpchost'
-HOST_NAME_DUAL = 'testdualhost'
-INSTANCE_1 = 'testvm1'
-INSTANCE_2 = 'testvm2'
-INSTANCE_PC = 'testpcvm'
-INSTANCE_DUAL = 'testdualvm'
-NEXUS_PORT_1 = 'ethernet:1/10'
-NEXUS_PORT_2 = 'ethernet:1/20'
-NEXUS_PORTCHANNELS = 'portchannel:2'
-NEXUS_DUAL = 'ethernet:1/3,portchannel:2'
-VLAN_ID_1 = 267
-VLAN_ID_2 = 265
-VLAN_ID_PC = 268
-VLAN_ID_DUAL = 269
-DEVICE_OWNER = 'compute:test'
-NEXUS_SSH_PORT = '22'
-PORT_STATE = n_const.PORT_STATUS_ACTIVE
-NETWORK_TYPE = 'vlan'
-NEXUS_DRIVER = ('neutron.plugins.ml2.drivers.cisco.nexus.'
- 'nexus_network_driver.CiscoNexusDriver')
-
-
-class FakeNetworkContext(object):
-
- """Network context for testing purposes only."""
-
- def __init__(self, segment_id):
- self._network_segments = {api.SEGMENTATION_ID: segment_id,
- api.NETWORK_TYPE: NETWORK_TYPE}
-
- @property
- def network_segments(self):
- return self._network_segments
-
-
-class FakePortContext(object):
-
- """Port context for testing purposes only."""
-
- def __init__(self, device_id, host_name, network_context):
- self._port = {
- 'status': PORT_STATE,
- 'device_id': device_id,
- 'device_owner': DEVICE_OWNER,
- portbindings.HOST_ID: host_name,
- portbindings.VIF_TYPE: portbindings.VIF_TYPE_OVS
- }
- self._network = network_context
- self._segment = network_context.network_segments
-
- @property
- def current(self):
- return self._port
-
- @property
- def network(self):
- return self._network
-
- @property
- def bottom_bound_segment(self):
- return self._segment
-
-
-class TestCiscoNexusDevice(testlib_api.SqlTestCase):
-
- """Unit tests for Cisco ML2 Nexus device driver."""
-
- TestConfigObj = collections.namedtuple(
- 'TestConfigObj',
- 'nexus_ip_addr host_name nexus_port instance_id vlan_id')
-
- test_configs = {
- 'test_config1': TestConfigObj(
- NEXUS_IP_ADDRESS,
- HOST_NAME_1,
- NEXUS_PORT_1,
- INSTANCE_1,
- VLAN_ID_1),
- 'test_config2': TestConfigObj(
- NEXUS_IP_ADDRESS,
- HOST_NAME_2,
- NEXUS_PORT_2,
- INSTANCE_2,
- VLAN_ID_2),
- 'test_config_portchannel': TestConfigObj(
- NEXUS_IP_ADDRESS_PC,
- HOST_NAME_PC,
- NEXUS_PORTCHANNELS,
- INSTANCE_PC,
- VLAN_ID_PC),
- 'test_config_dual': TestConfigObj(
- NEXUS_IP_ADDRESS_DUAL,
- HOST_NAME_DUAL,
- NEXUS_DUAL,
- INSTANCE_DUAL,
- VLAN_ID_DUAL),
- }
-
- def setUp(self):
- """Sets up mock ncclient, and switch and credentials dictionaries."""
- super(TestCiscoNexusDevice, self).setUp()
-
- # Use a mock netconf client
- mock_ncclient = mock.Mock()
- mock.patch.object(nexus_network_driver.CiscoNexusDriver,
- '_import_ncclient',
- return_value=mock_ncclient).start()
-
- def new_nexus_init(mech_instance):
- mech_instance.driver = importutils.import_object(NEXUS_DRIVER)
-
- mech_instance._nexus_switches = {}
- for name, config in TestCiscoNexusDevice.test_configs.items():
- ip_addr = config.nexus_ip_addr
- host_name = config.host_name
- nexus_port = config.nexus_port
- mech_instance._nexus_switches[(ip_addr,
- host_name)] = nexus_port
- mech_instance._nexus_switches[(ip_addr,
- 'ssh_port')] = NEXUS_SSH_PORT
- mech_instance._nexus_switches[(ip_addr,
- constants.USERNAME)] = 'admin'
- mech_instance._nexus_switches[(ip_addr,
- constants.PASSWORD)] = 'password'
- mech_instance.driver.nexus_switches = (
- mech_instance._nexus_switches)
-
- mock.patch.object(mech_cisco_nexus.CiscoNexusMechanismDriver,
- '__init__', new=new_nexus_init).start()
- self._cisco_mech_driver = (mech_cisco_nexus.
- CiscoNexusMechanismDriver())
-
- def _create_delete_port(self, port_config):
- """Tests creation and deletion of a virtual port."""
- nexus_ip_addr = port_config.nexus_ip_addr
- host_name = port_config.host_name
- nexus_port = port_config.nexus_port
- instance_id = port_config.instance_id
- vlan_id = port_config.vlan_id
-
- network_context = FakeNetworkContext(vlan_id)
- port_context = FakePortContext(instance_id, host_name,
- network_context)
-
- self._cisco_mech_driver.update_port_precommit(port_context)
- self._cisco_mech_driver.update_port_postcommit(port_context)
- for port_id in nexus_port.split(','):
- bindings = nexus_db_v2.get_nexusport_binding(port_id,
- vlan_id,
- nexus_ip_addr,
- instance_id)
- self.assertEqual(len(bindings), 1)
-
- self._cisco_mech_driver.delete_port_precommit(port_context)
- self._cisco_mech_driver.delete_port_postcommit(port_context)
- for port_id in nexus_port.split(','):
- with testtools.ExpectedException(
- exceptions.NexusPortBindingNotFound):
- nexus_db_v2.get_nexusport_binding(port_id,
- vlan_id,
- nexus_ip_addr,
- instance_id)
-
- def test_create_delete_ports(self):
- """Tests creation and deletion of two new virtual Ports."""
- self._create_delete_port(
- TestCiscoNexusDevice.test_configs['test_config1'])
-
- self._create_delete_port(
- TestCiscoNexusDevice.test_configs['test_config2'])
-
- def test_create_delete_portchannel(self):
- """Tests creation of a port over a portchannel."""
- self._create_delete_port(
- TestCiscoNexusDevice.test_configs['test_config_portchannel'])
-
- def test_create_delete_dual(self):
- """Tests creation and deletion of dual ports for single server"""
- self._create_delete_port(
- TestCiscoNexusDevice.test_configs['test_config_dual'])
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation
-# All Rights Reserved.
-#
-# 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 collections
-import testtools
-
-from neutron.plugins.ml2.drivers.cisco.nexus import exceptions
-from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
-from neutron.tests.unit import testlib_api
-
-
-class CiscoNexusDbTest(testlib_api.SqlTestCase):
-
- """Unit tests for Cisco mechanism driver's Nexus port binding database."""
-
- NpbObj = collections.namedtuple('NpbObj', 'port vlan switch instance')
-
- def _npb_test_obj(self, pnum, vnum, switch='10.9.8.7', instance=None):
- """Creates a Nexus port binding test object from a pair of numbers."""
- if pnum is 'router':
- port = pnum
- else:
- port = '1/%s' % pnum
- if instance is None:
- instance = 'instance_%s_%s' % (pnum, vnum)
- return self.NpbObj(port, vnum, switch, instance)
-
- def _assert_bindings_match(self, npb, npb_obj):
- """Asserts that a port binding matches a port binding test obj."""
- self.assertEqual(npb.port_id, npb_obj.port)
- self.assertEqual(npb.vlan_id, npb_obj.vlan)
- self.assertEqual(npb.switch_ip, npb_obj.switch)
- self.assertEqual(npb.instance_id, npb_obj.instance)
-
- def _add_binding_to_db(self, npb):
- """Adds a port binding to the Nexus database."""
- return nexus_db_v2.add_nexusport_binding(
- npb.port, npb.vlan, npb.switch, npb.instance)
-
- def _add_bindings_to_db(self, npbs):
- """Adds a list of port bindings to the Nexus database."""
- for npb in npbs:
- nexus_db_v2.add_nexusport_binding(
- npb.port, npb.vlan, npb.switch, npb.instance)
-
- def _remove_binding_from_db(self, npb):
- """Removes a port binding from the Nexus database."""
- return nexus_db_v2.remove_nexusport_binding(
- npb.port, npb.vlan, npb.switch, npb.instance)
-
- def _get_nexusport_binding(self, npb):
- """Gets a port binding based on port, vlan, switch, and instance."""
- return nexus_db_v2.get_nexusport_binding(
- npb.port, npb.vlan, npb.switch, npb.instance)
-
- def _get_nexusvlan_binding(self, npb):
- """Gets port bindings based on vlan and switch."""
- return nexus_db_v2.get_nexusvlan_binding(npb.vlan, npb.switch)
-
- def _get_nexusvm_binding(self, npb):
- """Gets port binding based on vlan and instance."""
- return nexus_db_v2.get_nexusvm_bindings(npb.vlan, npb.instance)[0]
-
- def _get_port_vlan_switch_binding(self, npb):
- """Gets port bindings based on port, vlan, and switch."""
- return nexus_db_v2.get_port_vlan_switch_binding(
- npb.port, npb.vlan, npb.switch)
-
- def _get_port_switch_bindings(self, npb):
- """Get port bindings based on port and switch."""
- return nexus_db_v2.get_port_switch_bindings(npb.port, npb.switch)
-
- def test_nexusportbinding_add_remove(self):
- """Tests add and removal of port bindings from the Nexus database."""
- npb11 = self._npb_test_obj(10, 100)
- npb = self._add_binding_to_db(npb11)
- self._assert_bindings_match(npb, npb11)
- npb = self._remove_binding_from_db(npb11)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb11)
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- self._remove_binding_from_db(npb11)
-
- def test_nexusportbinding_get(self):
- """Tests get of specific port bindings from the database."""
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_bindings_to_db([npb11, npb21, npb22])
-
- npb = self._get_nexusport_binding(npb11)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb11)
- npb = self._get_nexusport_binding(npb21)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb21)
- npb = self._get_nexusport_binding(npb22)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb22)
-
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- nexus_db_v2.get_nexusport_binding(
- npb21.port, npb21.vlan, npb21.switch, "dummyInstance")
-
- def test_nexusvlanbinding_get(self):
- """Test get of port bindings based on vlan and switch."""
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_bindings_to_db([npb11, npb21, npb22])
-
- npb_all_v100 = self._get_nexusvlan_binding(npb11)
- self.assertEqual(len(npb_all_v100), 2)
- npb_v200 = self._get_nexusvlan_binding(npb22)
- self.assertEqual(len(npb_v200), 1)
- self._assert_bindings_match(npb_v200[0], npb22)
-
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- nexus_db_v2.get_nexusvlan_binding(npb21.vlan, "dummySwitch")
-
- def test_nexusvmbinding_get(self):
- """Test get of port bindings based on vlan and instance."""
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_bindings_to_db([npb11, npb21, npb22])
-
- npb = self._get_nexusvm_binding(npb21)
- self._assert_bindings_match(npb, npb21)
- npb = self._get_nexusvm_binding(npb22)
- self._assert_bindings_match(npb, npb22)
-
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- nexus_db_v2.get_nexusvm_bindings(npb21.vlan, "dummyInstance")[0]
-
- def test_nexusportvlanswitchbinding_get(self):
- """Tests get of port bindings based on port, vlan, and switch."""
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- self._add_bindings_to_db([npb11, npb21])
-
- npb = self._get_port_vlan_switch_binding(npb11)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb11)
-
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- nexus_db_v2.get_port_vlan_switch_binding(
- npb21.port, npb21.vlan, "dummySwitch")
-
- def test_nexusportswitchbinding_get(self):
- """Tests get of port bindings based on port and switch."""
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100, switch='2.2.2.2')
- npb22 = self._npb_test_obj(20, 200, switch='2.2.2.2')
- self._add_bindings_to_db([npb11, npb21, npb22])
-
- npb = self._get_port_switch_bindings(npb11)
- self.assertEqual(len(npb), 1)
- self._assert_bindings_match(npb[0], npb11)
- npb_all_p20 = self._get_port_switch_bindings(npb21)
- self.assertEqual(len(npb_all_p20), 2)
-
- npb = nexus_db_v2.get_port_switch_bindings(npb21.port, "dummySwitch")
- self.assertIsNone(npb)
-
- def test_nexusbinding_update(self):
- """Tests update of vlan IDs for port bindings."""
- npb11 = self._npb_test_obj(10, 100, switch='1.1.1.1', instance='test')
- npb21 = self._npb_test_obj(20, 100, switch='1.1.1.1', instance='test')
- self._add_bindings_to_db([npb11, npb21])
-
- npb_all_v100 = nexus_db_v2.get_nexusvlan_binding(100, '1.1.1.1')
- self.assertEqual(len(npb_all_v100), 2)
-
- npb22 = self._npb_test_obj(20, 200, switch='1.1.1.1', instance='test')
- npb = nexus_db_v2.update_nexusport_binding(npb21.port, 200)
- self._assert_bindings_match(npb, npb22)
-
- npb_all_v100 = nexus_db_v2.get_nexusvlan_binding(100, '1.1.1.1')
- self.assertEqual(len(npb_all_v100), 1)
- self._assert_bindings_match(npb_all_v100[0], npb11)
-
- npb = nexus_db_v2.update_nexusport_binding(npb21.port, 0)
- self.assertIsNone(npb)
-
- npb33 = self._npb_test_obj(30, 300, switch='1.1.1.1', instance='test')
- with testtools.ExpectedException(exceptions.NexusPortBindingNotFound):
- nexus_db_v2.update_nexusport_binding(npb33.port, 200)
Code Review / openstack-build / neutron-build.git / commitdiff