Merge "Cisco Nexus ML2 Vendor decomposition"

This commit is contained in:
Jenkins 2015-02-20 21:02:15 +00:00 committed by Gerrit Code Review
commit ac65bca3ba
14 changed files with 4 additions and 2265 deletions

View File

@ -1,19 +0,0 @@
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

View File

@ -1,65 +0,0 @@
# 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]

View File

@ -1,24 +0,0 @@
# 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'

View File

@ -1,84 +0,0 @@
# 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.')

View File

@ -17,226 +17,8 @@
ML2 Mechanism Driver for Cisco Nexus platforms.
"""
from oslo_config import cfg
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__)
from networking_cisco.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
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

View File

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

View File

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

View File

@ -1,195 +0,0 @@
# 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>
"""

View File

@ -0,0 +1 @@
networking-cisco

View File

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

View File

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

View File

@ -1,216 +0,0 @@
# 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'])

View File

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