x/vmware-nsx
Code Issues Proposed changes

Providernet extension support for the Cisco Nexus plugin

Browse Source 
Implements blueprint provider-network-extensions-cisco

Change-Id: Ia22c21a7a66d22040811a9b43e7749892405e5e7
This commit is contained in:
HenryGessau 2013-07-09 10:49:40 -04:00
parent b5d1c993de
commit 731401e59c
15 changed files with 495 additions and 403 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
etc/neutron/plugins/cisco/cisco_plugins.ini
View File

@ -20,6 +20,29 @@
# vlan_name_prefix = q-
# Example: vlan_name_prefix = vnet-
# (StrOpt) A short prefix to prepend to the VLAN number when creating a
# provider VLAN interface. For example, if an interface is being created
# for provider VLAN 3003 it will be named 'p-3003' using the default prefix.
#
# provider_vlan_name_prefix = p-
# Example: provider_vlan_name_prefix = PV-
# (BoolOpt) A flag indicating whether Openstack networking should manage the
# creation and removal of VLAN interfaces for provider networks on the Nexus
# switches. If the flag is set to False then Openstack will not create or
# remove VLAN interfaces for provider networks, and the administrator needs
# to manage these interfaces manually or by external orchestration.
#
# provider_vlan_auto_create = True
# (BoolOpt) A flag indicating whether Openstack networking should manage
# the adding and removing of provider VLANs from trunk ports on the Nexus
# switches. If the flag is set to False then Openstack will not add or
# remove provider VLANs from trunk ports, and the administrator needs to
# manage these operations manually or by external orchestration.
#
# provider_vlan_auto_trunk = True
# (StrOpt) Period-separated module path to the model class to use for
# the Cisco neutron plugin.
#

60
neutron/db/migration/alembic_migrations/versions/e6b16a30d97_cisco_provider_nets.py Normal file
View File

@ -0,0 +1,60 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 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.
#
"""Add cisco_provider_networks table
Revision ID: e6b16a30d97
Revises: 2032abe8edac
Create Date: 2013-07-18 21:46:12.792504
"""
# revision identifiers, used by Alembic.
revision = 'e6b16a30d97'
down_revision = '2032abe8edac'
# Change to ['*'] if this migration applies to all plugins
migration_for_plugins = [
'neutron.plugins.cisco.network_plugin.PluginV2'
]
from alembic import op
import sqlalchemy as sa
from neutron.db import migration
def upgrade(active_plugin=None, options=None):
if not migration.should_run(active_plugin, migration_for_plugins):
return
op.create_table(
'cisco_provider_networks',
sa.Column('network_id', sa.String(length=36), nullable=False),
sa.Column('network_type', sa.String(length=255), nullable=False),
sa.Column('segmentation_id', sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint('network_id')
)
def downgrade(active_plugin=None, options=None):
if not migration.should_run(active_plugin, migration_for_plugins):
return
op.drop_table('cisco_provider_networks')

95
neutron/plugins/cisco/common/cisco_constants.py
View File

@ -17,48 +17,24 @@
# @author: Sumit Naiksatam, Cisco Systems, Inc.
PLUGINS = 'PLUGINS'
INVENTORY = 'INVENTORY'
# Attachment attributes
INSTANCE_ID = 'instance_id'
TENANT_ID = 'tenant_id'
TENANT_NAME = 'tenant_name'
HOST_NAME = 'host_name'
PORT_STATE = 'port-state'
PORT_UP = "ACTIVE"
PORT_DOWN = "DOWN"
# Network attributes
NET_ID = 'id'
NET_NAME = 'name'
NET_VLAN_ID = 'vlan_id'
NET_VLAN_NAME = 'vlan_name'
NET_PORTS = 'ports'
UUID = 'uuid'
TENANTID = 'tenant_id'
NETWORKID = 'network_id'
NETWORKNAME = 'name'
NETWORKPORTS = 'ports'
INTERFACEID = 'interface_id'
PORTSTATE = 'state'
PORTID = 'port_id'
PPNAME = 'name'
PPVLANID = 'vlan_id'
PPQOS = 'qos'
VLANID = 'vlan_id'
VLANNAME = 'vlan_name'
QOS = 'qos'
ATTACHMENT = 'attachment'
PORT_ID = 'port-id'
NET_ID = 'net-id'
NET_NAME = 'net-name'
NET_PORTS = 'net-ports'
NET_VLAN_NAME = 'net-vlan-name'
NET_VLAN_ID = 'net-vlan-id'
NET_TENANTS = 'net-tenants'
TENANT_ID = 'tenant-id'
TENANT_NETWORKS = 'tenant-networks'
TENANT_NAME = 'tenant-name'
TENANT_QOS_LEVELS = 'tenant-qos-levels'
TENANT_CREDENTIALS = 'tenant-credentials'
QOS_LEVEL_ID = 'qos_id'
QOS_LEVEL_NAME = 'qos_name'
QOS_LEVEL_ASSOCIATIONS = 'qos-level-associations'
QOS_LEVEL_DESCRIPTION = 'qos_desc'
# Network types
NETWORK_TYPE_FLAT = 'flat'
NETWORK_TYPE_LOCAL = 'local'
NETWORK_TYPE_VLAN = 'vlan'
NETWORK_TYPE_NONE = 'none'
CREDENTIAL_ID = 'credential_id'
CREDENTIAL_NAME = 'credential_name'
@ -71,52 +47,13 @@ PASSWORD = 'password'
LOGGER_COMPONENT_NAME = "cisco_plugin"
RESERVED_NIC_HOSTNAME = "reserved-dynamic-nic-hostname"
RESERVED_NIC_NAME = "reserved-dynamic-nic-device-name"
RHEL_DEVICE_NAME_REPFIX = "eth"
NEXUS_PLUGIN = 'nexus_plugin'
VSWITCH_PLUGIN = 'vswitch_plugin'
PLUGIN_OBJ_REF = 'plugin-obj-ref'
PARAM_LIST = 'param-list'
DEVICE_IP = 'device_ip'
NO_VLAN_ID = 0
HOST_LIST = 'host_list'
HOST_1 = 'host_1'
VIF_DESC = 'vif_desc'
DEVICENAME = 'device'
IP_ADDRESS = 'ip_address'
CHASSIS_ID = 'chassis_id'
BLADE_ID = 'blade_id'
HOST_NAME = 'host_name'
INSTANCE_ID = 'instance_id'
VIF_ID = 'vif_id'
PROJECT_ID = 'project_id'
NETWORK_ADMIN = 'network_admin'
NETID_LIST = 'net_id_list'
DELIMITERS = "[,;:\b\s]"
UUID_LENGTH = 36
UNPLUGGED = '(detached)'
ASSOCIATION_STATUS = 'association_status'
ATTACHED = 'attached'
DETACHED = 'detached'
NETWORK = 'network'
PORT = 'port'
BASE_PLUGIN_REF = 'base_plugin_ref'

5
neutron/plugins/cisco/common/cisco_exceptions.py
View File

@ -79,6 +79,11 @@ class CredentialAlreadyExists(exceptions.NeutronException):
"for tenant %(tenant_id)s")
class ProviderNetworkExists(exceptions.NeutronException):
"""Provider network already exists."""
message = _("Provider network %s already exists")
class NexusComputeHostNotConfigured(exceptions.NeutronException):
"""Connection to compute host is not configured."""
message = _("Connection to %(host)s is not configured.")

48
neutron/plugins/cisco/common/cisco_utils.py
View File

@ -1,48 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2011 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.
#
# @author: Sumit Naiksatam, Cisco Systems, Inc.
import hashlib
import logging
from neutron.plugins.cisco.common import cisco_constants as const
LOG = logging.getLogger(__name__)
def get16ByteUUID(uuid):
"""Return first 16 bytes of UUID.
Used when smaller unique ID is required.
"""
return hashlib.md5(uuid).hexdigest()[:16]
def make_net_dict(net_id, net_name, ports):
"""Helper funciton."""
res = {const.NET_ID: net_id, const.NET_NAME: net_name}
res[const.NET_PORTS] = ports
return res
def make_port_dict(port_id, port_state, net_id, attachment):
"""Helper funciton."""
res = {const.PORT_ID: port_id, const.PORT_STATE: port_state}
res[const.NET_ID] = net_id
res[const.ATTACHMENT] = attachment
return res

8
neutron/plugins/cisco/common/config.py
View File

@ -34,6 +34,14 @@ cisco_plugins_opts = [
cisco_opts = [
cfg.StrOpt('vlan_name_prefix', default='q-',
help=_("VLAN Name prefix")),
cfg.StrOpt('provider_vlan_name_prefix', default='p-',
help=_("VLAN Name prefix for provider vlans")),
cfg.BoolOpt('provider_vlan_auto_create', default=True,
help='Provider VLANs are automatically created as needed '
'on the Nexus switch'),
cfg.BoolOpt('provider_vlan_auto_trunk', default=True,
help='Provider VLANs are automatically trunked as needed '
'on the ports of the Nexus switch'),
cfg.BoolOpt('svi_round_robin', default=False,
help=_("Distribute SVI interfaces over all switches")),
cfg.StrOpt('model_class',

50
neutron/plugins/cisco/db/network_db_v2.py
View File

@ -20,6 +20,7 @@ from sqlalchemy.orm import exc
from neutron.db import api as db
from neutron.openstack.common import log as logging
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.db import network_models_v2
from neutron.plugins.openvswitch import ovs_models_v2
@ -257,6 +258,55 @@ def update_credential(tenant_id, credential_id,
tenant_id=tenant_id)
def add_provider_network(network_id, network_type, segmentation_id):
"""Add a network to the provider network table."""
session = db.get_session()
if session.query(network_models_v2.ProviderNetwork).filter_by(
network_id=network_id).first():
raise c_exc.ProviderNetworkExists(network_id)
pnet = network_models_v2.ProviderNetwork(network_id=network_id,
network_type=network_type,
segmentation_id=segmentation_id)
session.add(pnet)
session.flush()
def remove_provider_network(network_id):
"""Remove network_id from the provider network table.
:param network_id: Any network id. If it is not in the table, do nothing.
:return: network_id if it was in the table and successfully removed.
"""
session = db.get_session()
pnet = (session.query(network_models_v2.ProviderNetwork).
filter_by(network_id=network_id).first())
if pnet:
session.delete(pnet)
session.flush()
return network_id
def is_provider_network(network_id):
"""Return True if network_id is in the provider network table."""
session = db.get_session()
if session.query(network_models_v2.ProviderNetwork).filter_by(
network_id=network_id).first():
return True
def is_provider_vlan(vlan_id):
"""Check for a for a vlan provider network with the specified vland_id.
Returns True if the provider network table contains a vlan network
with the specified vlan_id.
"""
session = db.get_session()
if (session.query(network_models_v2.ProviderNetwork).
filter_by(network_type=const.NETWORK_TYPE_VLAN,
segmentation_id=vlan_id).first()):
return True
def get_ovs_vlans():
session = db.get_session()
bindings = (session.query(ovs_models_v2.VlanAllocation.vlan_id).

14
neutron/plugins/cisco/db/network_models_v2.py
View File

@ -16,7 +16,7 @@
#
# @author: Rohit Agarwalla, Cisco Systems, Inc.
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy import Column, ForeignKey, Integer, String, Boolean
from neutron.db import model_base
from neutron.openstack.common import uuidutils
@ -82,3 +82,15 @@ class Credential(model_base.BASEV2):
self.credential_name,
self.user_name,
self.password)
class ProviderNetwork(model_base.BASEV2):
"""Represents networks that were created as provider networks."""
__tablename__ = 'cisco_provider_networks'
network_id = Column(String(36),
ForeignKey('networks.id', ondelete="CASCADE"),
primary_key=True)
network_type = Column(String(255), nullable=False)
segmentation_id = Column(Integer, nullable=False)

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

@ -58,15 +58,6 @@ class L2DevicePluginBase(object):
"""
pass
@abstractmethod
def get_network_details(self, tenant_id, net_id, **kwargs):
"""Get network details.
:returns:
:raises:
"""
pass
@abstractmethod
def update_network(self, tenant_id, net_id, name, **kwargs):
"""Update network.

55
neutron/plugins/cisco/models/virt_phy_sw_v2.py
View File

@ -26,7 +26,9 @@ import sys
from novaclient.v1_1 import client as nova_client
from oslo.config import cfg
from neutron.api.v2 import attributes
from neutron.db import api as db_api
from neutron.extensions import providernet as provider
from neutron import neutron_plugin_base_v2
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
@ -49,7 +51,7 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
"""
MANAGE_STATE = True
__native_bulk_support = True
supported_extension_aliases = []
supported_extension_aliases = ["provider"]
_plugins = {}
_methods_to_delegate = ['create_network_bulk',
'get_network', 'get_networks',
@ -202,6 +204,15 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
return host
def _get_provider_vlan_id(self, network):
if (all(attributes.is_attr_set(network.get(attr))
for attr in (provider.NETWORK_TYPE,
provider.PHYSICAL_NETWORK,
provider.SEGMENTATION_ID))
and
network[provider.NETWORK_TYPE] == const.NETWORK_TYPE_VLAN):
return network[provider.SEGMENTATION_ID]
def create_network(self, context, network):
"""Create network.
@ -209,10 +220,21 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
plugins.
"""
LOG.debug(_("create_network() called"))
provider_vlan_id = self._get_provider_vlan_id(network[const.NETWORK])
args = [context, network]
ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
self._func_name(),
args)
# The vswitch plugin did all the verification. If it's a provider
# vlan network, save it for the nexus plugin to use later.
if provider_vlan_id:
network_id = ovs_output[0][const.NET_ID]
cdb.add_provider_network(network_id,
const.NETWORK_TYPE_VLAN,
provider_vlan_id)
LOG.debug(_("provider network added to DB: %(network_id)s, "
"%(vlan_id)s"), {'network_id': network_id,
'vlan_id': provider_vlan_id})
return ovs_output[0]
def update_network(self, context, id, network):
@ -230,6 +252,13 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
provider attribute, so it is not supported by this method.
"""
LOG.debug(_("update_network() called"))
# We can only support updating of provider attributes if all the
# configured sub-plugins support it. Currently we have no method
# in place for checking whether a sub-plugin supports it,
# so assume not.
provider._raise_if_updates_provider_attributes(network['network'])
args = [context, id, network]
ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
self._func_name(),
@ -246,6 +275,8 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
self._func_name(),
args)
if cdb.remove_provider_network(id):
LOG.debug(_("provider network removed from DB: %s"), id)
return ovs_output[0]
def get_network(self, context, id, fields=None):
@ -261,22 +292,20 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
if not self.config_nexus:
return False
net_dict = self.get_network(context, net_id)
net_name = net_dict['name']
network = self.get_network(context, net_id)
vlan_id = self._get_segmentation_id(net_id)
host = self._get_instance_host(tenant_id, instance_id)
# Trunk segmentation id for only this host
vlan_name = conf.CISCO.vlan_name_prefix + str(vlan_id)
n_args = [tenant_id, net_name, net_id,
vlan_name, vlan_id, host, instance_id]
nexus_output = self._invoke_plugin_per_device(
network[const.NET_VLAN_ID] = vlan_id
network[const.NET_VLAN_NAME] = vlan_name
attachment = {
const.TENANT_ID: tenant_id,
const.INSTANCE_ID: instance_id,
const.HOST_NAME: self._get_instance_host(tenant_id, instance_id),
}
self._invoke_plugin_per_device(
const.NEXUS_PLUGIN,
'create_network',
n_args)
return nexus_output
[network, attachment])
@staticmethod
def _should_call_create_net(device_owner, instance_id):

2
neutron/plugins/cisco/network_plugin.py
View File

@ -29,7 +29,7 @@ from neutron.db import models_v2
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as cexc
from neutron.plugins.cisco.common import config # noqa
from neutron.plugins.cisco.common import config
from neutron.plugins.cisco.db import network_db_v2 as cdb
LOG = logging.getLogger(__name__)

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

@ -26,8 +26,10 @@ import logging
from ncclient import manager
from neutron.openstack.common import excutils
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_credentials_v2 as cred
from neutron.plugins.cisco.common import cisco_exceptions as cexc
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco.common import config as conf
from neutron.plugins.cisco.db import nexus_db_v2
from neutron.plugins.cisco.nexus import cisco_nexus_snippets as snipp
@ -37,13 +39,15 @@ LOG = logging.getLogger(__name__)
class CiscoNEXUSDriver():
"""Nexus Driver Main Class."""
def __init__(self):
self.nexus_switches = conf.get_nexus_dictionary()
self.credentials = {}
self.connections = {}
def _edit_config(self, mgr, target='running', config='',
def _edit_config(self, nexus_host, target='running', config='',
allowed_exc_strs=None):
"""Modify switch config for a target config type.
:param mgr: NetConf client manager
: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
@ -55,6 +59,7 @@ class CiscoNEXUSDriver():
"""
if not allowed_exc_strs:
allowed_exc_strs = []
mgr = self.nxos_connect(nexus_host)
try:
mgr.edit_config(target, config=config)
except Exception as e:
@ -66,12 +71,31 @@ class CiscoNEXUSDriver():
# the original ncclient exception.
raise cexc.NexusConfigFailed(config=config, exc=e)
def nxos_connect(self, nexus_host, nexus_ssh_port, nexus_user,
nexus_password):
def get_credential(self, nexus_ip):
if nexus_ip not in self.credentials:
nexus_username = cred.Store.get_username(nexus_ip)
nexus_password = cred.Store.get_password(nexus_ip)
self.credentials[nexus_ip] = {
const.USERNAME: nexus_username,
const.PASSWORD: nexus_password
}
return self.credentials[nexus_ip]
def get_switch_and_port_id(self, host_name):
for switch_ip, attr in self.nexus_switches:
if str(attr) == host_name:
return switch_ip, self.nexus_switches[switch_ip, attr]
return None, None
def nxos_connect(self, nexus_host):
"""Make SSH connection to the Nexus Switch."""
if getattr(self.connections.get(nexus_host), 'connected', None):
return self.connections[nexus_host]
nexus_ssh_port = int(self.nexus_switches[nexus_host, 'ssh_port'])
nexus_creds = self.get_credential(nexus_host)
nexus_user = nexus_creds[const.USERNAME]
nexus_password = nexus_creds[const.PASSWORD]
try:
man = manager.connect(host=nexus_host,
port=nexus_ssh_port,
@ -93,11 +117,11 @@ class CiscoNEXUSDriver():
conf_xml_snippet = snipp.EXEC_CONF_SNIPPET % (cutomized_config)
return conf_xml_snippet
def enable_vlan(self, mgr, vlanid, vlanname):
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(mgr, target='running', config=confstr)
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
@ -109,143 +133,77 @@ class CiscoNEXUSDriver():
try:
confstr = self.create_xml_snippet(snippet % vlanid)
self._edit_config(
mgr,
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.disable_vlan(mgr, vlanid)
self.delete_vlan(nexus_host, vlanid)
def disable_vlan(self, mgr, 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(mgr, target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def enable_port_trunk(self, mgr, interface):
def enable_port_trunk(self, nexus_host, interface):
"""Enable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_PORT_TRUNK % (interface)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(mgr, target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def disable_switch_port(self, mgr, interface):
def disable_switch_port(self, nexus_host, interface):
"""Disable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_NO_SWITCHPORT % (interface)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(mgr, target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def enable_vlan_on_trunk_int(self, mgr, nexus_switch, interface, vlanid):
"""Enable vlan in trunk interface.
Enables trunk mode vlan access an interface on Nexus Switch given
VLANID.
"""
def enable_vlan_on_trunk_int(self, nexus_host, vlanid, interface):
"""Enable a VLAN on a trunk interface."""
# If one or more VLANs are already configured on this interface,
# include the 'add' keyword.
if nexus_db_v2.get_port_switch_bindings(interface, nexus_switch):
if nexus_db_v2.get_port_switch_bindings(interface, nexus_host):
snippet = snipp.CMD_INT_VLAN_ADD_SNIPPET
else:
snippet = snipp.CMD_INT_VLAN_SNIPPET
confstr = snippet % (interface, vlanid)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(mgr, target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def disable_vlan_on_trunk_int(self, mgr, interface, vlanid):
"""Disable VLAN.
Disables trunk mode vlan access an interface on Nexus Switch given
VLANID.
"""
def disable_vlan_on_trunk_int(self, nexus_host, vlanid, interface):
"""Disable a VLAN on a trunk interface."""
confstr = snipp.CMD_NO_VLAN_INT_SNIPPET % (interface, vlanid)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(mgr, target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def create_vlan(self, vlan_name, vlan_id, nexus_host, nexus_user,
nexus_password, nexus_ports,
nexus_ssh_port, vlan_ids=None):
"""Create VLAN and enablt in on the interface.
def create_and_trunk_vlan(self, nexus_host, vlan_id, vlan_name,
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, nexus_port)
Creates a VLAN and Enable on trunk mode an interface on Nexus Switch
given the VLAN ID and Name and Interface Number.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
self.enable_vlan(man, vlan_id, vlan_name)
if vlan_ids is '':
vlan_ids = self.build_vlans_cmd()
LOG.debug(_("NexusDriver VLAN IDs: %s"), vlan_ids)
for ports in nexus_ports:
self.enable_vlan_on_trunk_int(man, nexus_host, ports, vlan_ids)
def delete_vlan(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port):
"""Delete vlan.
Delete a VLAN and Disables trunk mode an interface on Nexus Switch
given the VLAN ID and Interface Number.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
self.disable_vlan(man, vlan_id)
for ports in nexus_ports:
self.disable_vlan_on_trunk_int(man, ports, vlan_id)
def build_vlans_cmd(self):
"""Builds a string with all the VLANs on the same Switch."""
assigned_vlan = cdb.get_all_vlanids_used()
vlans = ''
for vlanid in assigned_vlan:
vlans = str(vlanid["vlan_id"]) + ',' + vlans
if vlans == '':
vlans = 'none'
return vlans.strip(',')
def add_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port, vlan_ids=None):
"""Add vlan.
Adds a vlan from interfaces on the Nexus switch given the VLAN ID.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
if not vlan_ids:
vlan_ids = self.build_vlans_cmd()
for ports in nexus_ports:
self.enable_vlan_on_trunk_int(man, nexus_host, ports, vlan_ids)
def remove_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port):
"""Remove vlan.
Removes a vlan from interfaces on the Nexus switch given the VLAN ID.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
for ports in nexus_ports:
self.disable_vlan_on_trunk_int(man, ports, vlan_id)
def create_vlan_svi(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ssh_port, gateway_ip):
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
def delete_and_untrunk_vlan(self, nexus_host, vlan_id, nexus_port):
"""Delete VLAN and untrunk it from the specified ports."""
self.delete_vlan(nexus_host, vlan_id)
if nexus_port:
self.disable_vlan_on_trunk_int(nexus_host, vlan_id, nexus_port)
def create_vlan_svi(self, nexus_host, vlan_id, gateway_ip):
confstr = snipp.CMD_VLAN_SVI_SNIPPET % (vlan_id, gateway_ip)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
man.edit_config(target='running', config=confstr)
def delete_vlan_svi(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ssh_port):
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
self._edit_config(nexus_host, target='running', config=confstr)
def delete_vlan_svi(self, nexus_host, vlan_id):
confstr = snipp.CMD_NO_VLAN_SVI_SNIPPET % vlan_id
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
man.edit_config(target='running', config=confstr)
self._edit_config(nexus_host, target='running', config=confstr)

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

@ -30,9 +30,9 @@ from neutron.common import exceptions as exc
from neutron.openstack.common import excutils
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_credentials_v2 as cred
from neutron.plugins.cisco.common import cisco_exceptions as cisco_exc
from neutron.plugins.cisco.common import config as conf
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco.db import nexus_db_v2 as nxos_db
from neutron.plugins.cisco.l2device_plugin_base import L2DevicePluginBase
@ -48,18 +48,6 @@ class NexusPlugin(L2DevicePluginBase):
"""Extract configuration parameters from the configuration file."""
self._client = importutils.import_object(conf.CISCO.nexus_driver)
LOG.debug(_("Loaded driver %s"), conf.CISCO.nexus_driver)
self._nexus_switches = conf.get_nexus_dictionary()
self.credentials = {}
def get_credential(self, nexus_ip):
if nexus_ip not in self.credentials:
_nexus_username = cred.Store.get_username(nexus_ip)
_nexus_password = cred.Store.get_password(nexus_ip)
self.credentials[nexus_ip] = {
'username': _nexus_username,
'password': _nexus_password
}
return self.credentials[nexus_ip]
def get_all_networks(self, tenant_id):
"""Get all networks.
@ -70,76 +58,88 @@ class NexusPlugin(L2DevicePluginBase):
LOG.debug(_("NexusPlugin:get_all_networks() called"))
return self._networks.values()
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
host, instance):
"""Create network.
def create_network(self, network, attachment):
"""Create or update a network when an attachment is changed.
This method is not invoked at the usual plugin create_network() time.
Instead, it is invoked on create/update port.
:param network: Network on which the port operation is happening
:param attachment: Details about the owner of the port
Create a VLAN in the appropriate switch/port, and configure the
appropriate interfaces for this VLAN.
"""
LOG.debug(_("NexusPlugin:create_network() called"))
# Grab the switch IP and port for this host
for switch_ip, attr in self._nexus_switches:
if str(attr) == str(host):
port_id = self._nexus_switches[switch_ip, attr]
break
else:
host = str(attachment[const.HOST_NAME])
switch_ip, port_id = self._client.get_switch_and_port_id(host)
if not switch_ip and not port_id:
raise cisco_exc.NexusComputeHostNotConfigured(host=host)
vlan_id = network[const.NET_VLAN_ID]
vlan_name = network[const.NET_VLAN_NAME]
auto_create = True
auto_trunk = True
if cdb.is_provider_vlan(vlan_id):
vlan_name = ''.join([conf.CISCO.provider_vlan_name_prefix,
str(vlan_id)])
auto_create = conf.CISCO.provider_vlan_auto_create
auto_trunk = conf.CISCO.provider_vlan_auto_trunk
# Check if this network is already in the DB
vlan_created = False
vlan_enabled = False
vlan_trunked = False
try:
nxos_db.get_port_vlan_switch_binding(port_id, vlan_id, switch_ip)
except cisco_exc.NexusPortBindingNotFound:
_nexus_ip = switch_ip
_nexus_ports = (port_id,)
_nexus_ssh_port = \
self._nexus_switches[switch_ip, 'ssh_port']
_nexus_creds = self.get_credential(_nexus_ip)
_nexus_username = _nexus_creds['username']
_nexus_password = _nexus_creds['password']
# Check for vlan/switch binding
try:
nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
except cisco_exc.NexusPortBindingNotFound:
if auto_create and auto_trunk:
# Create vlan and trunk vlan on the port
self._client.create_vlan(
vlan_name, str(vlan_id), _nexus_ip,
_nexus_username, _nexus_password,
_nexus_ports, _nexus_ssh_port, vlan_id)
LOG.debug("Nexus: create & trunk vlan %s" % vlan_name)
self._client.create_and_trunk_vlan(
switch_ip, vlan_id, vlan_name, port_id)
vlan_created = True
vlan_trunked = True
elif auto_create:
# Create vlan but do not trunk it on the port
LOG.debug("Nexus: create vlan %s" % vlan_name)
self._client.create_vlan(switch_ip, vlan_id, vlan_name)
vlan_created = True
else:
if auto_trunk:
# Only trunk vlan on the port
man = self._client.nxos_connect(_nexus_ip,
int(_nexus_ssh_port),
_nexus_username,
_nexus_password)
self._client.enable_vlan_on_trunk_int(man,
_nexus_ip,
port_id,
vlan_id)
vlan_enabled = True
LOG.debug("Nexus: trunk vlan %s" % vlan_name)
self._client.enable_vlan_on_trunk_int(
switch_ip, vlan_id, port_id)
vlan_trunked = True
try:
instance = attachment[const.INSTANCE_ID]
nxos_db.add_nexusport_binding(port_id, str(vlan_id),
switch_ip, instance)
except Exception:
with excutils.save_and_reraise_exception():
# Add binding failed, roll back any vlan creation/enabling
if vlan_created:
self._client.delete_vlan(
str(vlan_id), _nexus_ip,
_nexus_username, _nexus_password,
_nexus_ports, _nexus_ssh_port)
if vlan_enabled:
self._client.disable_vlan_on_trunk_int(man,
port_id,
vlan_id)
if vlan_created and vlan_trunked:
LOG.debug("Nexus: delete & untrunk vlan %s" % vlan_name)
self._client.delete_and_untrunk_vlan(switch_ip, vlan_id,
port_id)
elif vlan_created:
LOG.debug("Nexus: delete vlan %s" % vlan_name)
self._client.delete_vlan(switch_ip, vlan_id)
elif vlan_trunked:
LOG.debug("Nexus: untrunk vlan %s" % vlan_name)
self._client.disable_vlan_on_trunk_int(switch_ip, vlan_id,
port_id)
net_id = network[const.NET_ID]
new_net_dict = {const.NET_ID: net_id,
const.NET_NAME: net_name,
const.NET_NAME: network[const.NET_NAME],
const.NET_PORTS: {},
const.NET_VLAN_NAME: vlan_name,
const.NET_VLAN_ID: vlan_id}
@ -152,33 +152,22 @@ class NexusPlugin(L2DevicePluginBase):
# Find a switch to create the SVI on
switch_ip = self._find_switch_for_svi()
if not switch_ip:
raise cisco_exc.NoNexusSwitch()
_nexus_ip = switch_ip
_nexus_ssh_port = self._nexus_switches[switch_ip, 'ssh_port']
_nexus_creds = self.get_credential(_nexus_ip)
_nexus_username = _nexus_creds['username']
_nexus_password = _nexus_creds['password']
raise cisco_exc.NoNexusSviSwitch()
# Check if this vlan exists on the switch already
try:
nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
except cisco_exc.NexusPortBindingNotFound:
# Create vlan and trunk vlan on the port
self._client.create_vlan(
vlan_name, str(vlan_id), _nexus_ip,
_nexus_username, _nexus_password,
[], _nexus_ssh_port, vlan_id)
self._client.create_and_trunk_vlan(
switch_ip, vlan_id, vlan_name, nexus_port=None)
# Check if a router interface has already been created
try:
nxos_db.get_nexusvm_binding(vlan_id, router_id)
raise cisco_exc.SubnetInterfacePresent(subnet_id=subnet_id,
router_id=router_id)
except cisco_exc.NexusPortBindingNotFound:
self._client.create_vlan_svi(vlan_id, _nexus_ip, _nexus_username,
_nexus_password, _nexus_ssh_port,
gateway_ip)
self._client.create_vlan_svi(switch_ip, vlan_id, gateway_ip)
nxos_db.add_nexusport_binding('router', str(vlan_id),
switch_ip, router_id)
@ -187,17 +176,11 @@ class NexusPlugin(L2DevicePluginBase):
def remove_router_interface(self, vlan_id, router_id):
"""Remove VLAN SVI from the Nexus Switch."""
# Grab switch_ip from database
row = nxos_db.get_nexusvm_binding(vlan_id, router_id)
switch_ip = nxos_db.get_nexusvm_binding(vlan_id,
router_id)['switch_ip']
# Delete the SVI interface from the switch
_nexus_ip = row['switch_ip']
_nexus_ssh_port = self._nexus_switches[_nexus_ip, 'ssh_port']
_nexus_creds = self.get_credential(_nexus_ip)
_nexus_username = _nexus_creds['username']
_nexus_password = _nexus_creds['password']
self._client.delete_vlan_svi(vlan_id, _nexus_ip, _nexus_username,
_nexus_password, _nexus_ssh_port)
self._client.delete_vlan_svi(switch_ip, vlan_id)
# Invoke delete_port to delete this row
# And delete vlan if required
@ -206,10 +189,11 @@ class NexusPlugin(L2DevicePluginBase):
def _find_switch_for_svi(self):
"""Get a switch to create the SVI on."""
LOG.debug(_("Grabbing a switch to create SVI"))
nexus_switches = self._client.nexus_switches
if conf.CISCO.svi_round_robin:
LOG.debug(_("Using round robin to create SVI"))
switch_dict = dict(
(switch_ip, 0) for switch_ip, _ in self._nexus_switches)
(switch_ip, 0) for switch_ip, _ in nexus_switches)
try:
bindings = nxos_db.get_nexussvi_bindings()
# Build a switch dictionary with weights
@ -228,7 +212,7 @@ class NexusPlugin(L2DevicePluginBase):
LOG.debug(_("No round robin or zero weights, using first switch"))
# Return the first switch in the config
for switch_ip, attr in self._nexus_switches:
for switch_ip, attr in nexus_switches:
return switch_ip
def delete_network(self, tenant_id, net_id, **kwargs):
@ -239,12 +223,6 @@ class NexusPlugin(L2DevicePluginBase):
"""
LOG.debug(_("NexusPlugin:delete_network() called"))
def get_network_details(self, tenant_id, net_id, **kwargs):
"""Return the details of a particular network."""
LOG.debug(_("NexusPlugin:get_network_details() called"))
network = self._get_network(tenant_id, net_id)
return network
def update_network(self, tenant_id, net_id, **kwargs):
"""Update the properties of a particular Virtual Network."""
LOG.debug(_("NexusPlugin:update_network() called"))
@ -278,6 +256,18 @@ class NexusPlugin(L2DevicePluginBase):
except cisco_exc.NexusPortBindingNotFound:
return
auto_delete = True
auto_untrunk = True
if cdb.is_provider_vlan(vlan_id):
auto_delete = conf.CISCO.provider_vlan_auto_create
auto_untrunk = conf.CISCO.provider_vlan_auto_trunk
LOG.debug("delete_network(): provider vlan %s" % vlan_id)
switch_ip = row['switch_ip']
nexus_port = None
if row['port_id'] != 'router':
nexus_port = row['port_id']
nxos_db.remove_nexusport_binding(row['port_id'], row['vlan_id'],
row['switch_ip'],
row['instance_id'])
@ -287,19 +277,11 @@ class NexusPlugin(L2DevicePluginBase):
except cisco_exc.NexusPortBindingNotFound:
try:
# Delete this vlan from this switch
_nexus_ip = row['switch_ip']
_nexus_ports = ()
if row['port_id'] != 'router':
_nexus_ports = (row['port_id'],)
_nexus_ssh_port = (self._nexus_switches[_nexus_ip,
'ssh_port'])
_nexus_creds = self.get_credential(_nexus_ip)
_nexus_username = _nexus_creds['username']
_nexus_password = _nexus_creds['password']
self._client.delete_vlan(
str(row['vlan_id']), _nexus_ip,
_nexus_username, _nexus_password,
_nexus_ports, _nexus_ssh_port)
if nexus_port and auto_untrunk:
self._client.disable_vlan_on_trunk_int(
switch_ip, row['vlan_id'], nexus_port)
if auto_delete:
self._client.delete_vlan(switch_ip, row['vlan_id'])
except Exception:
# The delete vlan operation on the Nexus failed,
# so this delete_port request has failed. For

30
neutron/tests/unit/cisco/test_network_plugin.py
View File

@ -18,6 +18,7 @@ import inspect
import logging
import mock
from oslo.config import cfg
import webob.exc as wexc
from neutron.api.v2 import base
@ -25,6 +26,7 @@ from neutron.common import exceptions as q_exc
from neutron import context
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import l3_db
from neutron.extensions import providernet as provider
from neutron.manager import NeutronManager
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
@ -519,6 +521,16 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
test_db_plugin.TestNetworksV2):
def setUp(self):
self.physnet = 'testphys1'
self.vlan_range = '100:199'
phys_vrange = ':'.join([self.physnet, self.vlan_range])
cfg.CONF.set_override('tenant_network_type', 'vlan', 'OVS')
cfg.CONF.set_override('network_vlan_ranges', [phys_vrange], 'OVS')
self.addCleanup(cfg.CONF.reset)
super(TestCiscoNetworksV2, self).setUp()
def test_create_networks_bulk_emulated_plugin_failure(self):
real_has_attr = hasattr
@ -566,6 +578,24 @@ class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
'networks',
wexc.HTTPInternalServerError.code)
def test_create_provider_vlan_network(self):
provider_attrs = {provider.NETWORK_TYPE: 'vlan',
provider.PHYSICAL_NETWORK: self.physnet,
provider.SEGMENTATION_ID: '1234'}
arg_list = tuple(provider_attrs.keys())
res = self._create_network(self.fmt, 'pvnet1', True,
arg_list=arg_list, **provider_attrs)
net = self.deserialize(self.fmt, res)
expected = [('name', 'pvnet1'),
('admin_state_up', True),
('status', 'ACTIVE'),
('shared', False),
(provider.NETWORK_TYPE, 'vlan'),
(provider.PHYSICAL_NETWORK, self.physnet),
(provider.SEGMENTATION_ID, 1234)]
for k, v in expected:
self.assertEqual(net['network'][k], v)
class TestCiscoSubnetsV2(CiscoNetworkPluginV2TestCase,
test_db_plugin.TestSubnetsV2):

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

@ -15,23 +15,32 @@
import mock
from oslo.config import cfg
from neutron.db import api as db
from neutron.extensions import providernet as provider
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as cisco_exc
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
from neutron.tests import base
NEXUS_IP_ADDRESS = '1.1.1.1'
NEXUS_USERNAME = 'username'
NEXUS_PASSWORD = 'password'
HOSTNAME = 'testhost'
INSTANCE = 'testvm'
NEXUS_PORTS = '1/10'
HOSTNAME1 = 'testhost1'
HOSTNAME2 = 'testhost2'
INSTANCE1 = 'testvm1'
INSTANCE2 = 'testvm2'
NEXUS_PORT1 = '1/10'
NEXUS_PORT2 = '1/20'
NEXUS_SSH_PORT = '22'
NEXUS_DRIVER = ('neutron.plugins.cisco.nexus.'
'cisco_nexus_network_driver_v2.CiscoNEXUSDriver')
NET_ATTRS = [const.NET_ID,
const.NET_NAME,
const.NET_VLAN_NAME,
const.NET_VLAN_ID]
class TestCiscoNexusPlugin(base.BaseTestCase):
@ -44,28 +53,56 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
self.net_id = 7
self.vlan_name = "q-" + str(self.net_id) + "vlan"
self.vlan_id = 267
self.second_tenant_id = "test_tenant_2"
self.second_net_name = "test_network_cisco2"
self.second_net_id = 5
self.second_vlan_name = "q-" + str(self.second_net_id) + "vlan"
self.second_vlan_id = 265
self._nexus_switches = {
(NEXUS_IP_ADDRESS, HOSTNAME): NEXUS_PORTS,
(NEXUS_IP_ADDRESS, 'ssh_port'): NEXUS_SSH_PORT,
self.attachment1 = {
const.TENANT_ID: self.tenant_id,
const.INSTANCE_ID: INSTANCE1,
const.HOST_NAME: HOSTNAME1,
}
self.attachment2 = {
const.TENANT_ID: self.second_tenant_id,
const.INSTANCE_ID: INSTANCE2,
const.HOST_NAME: HOSTNAME2,
}
self.network1 = {
const.NET_ID: self.net_id,
const.NET_NAME: self.net_name,
const.NET_VLAN_NAME: self.vlan_name,
const.NET_VLAN_ID: self.vlan_id,
}
self.network2 = {
const.NET_ID: self.second_net_id,
const.NET_NAME: self.second_net_name,
const.NET_VLAN_NAME: self.second_vlan_name,
const.NET_VLAN_ID: self.second_vlan_id,
}
self.providernet = {
const.NET_ID: 9,
const.NET_NAME: 'pnet1',
const.NET_VLAN_NAME: 'p-300',
const.NET_VLAN_ID: 300,
provider.NETWORK_TYPE: 'vlan',
provider.PHYSICAL_NETWORK: self.net_name + '200:299',
provider.SEGMENTATION_ID: 300,
}
self._hostname = HOSTNAME
def new_nexus_init(self):
self._client = importutils.import_object(NEXUS_DRIVER)
self._nexus_ip = NEXUS_IP_ADDRESS
self._nexus_username = NEXUS_USERNAME
self._nexus_password = NEXUS_PASSWORD
self._nexus_ports = NEXUS_PORTS
self._nexus_ssh_port = NEXUS_SSH_PORT
self.credentials = {
self._nexus_ip: {
'username': self._nexus_username,
'password': self._nexus_password
self._client.nexus_switches = {
(NEXUS_IP_ADDRESS, HOSTNAME1): NEXUS_PORT1,
(NEXUS_IP_ADDRESS, 'ssh_port'): NEXUS_SSH_PORT,
(NEXUS_IP_ADDRESS, HOSTNAME2): NEXUS_PORT2,
(NEXUS_IP_ADDRESS, 'ssh_port'): NEXUS_SSH_PORT,
}
self._client.credentials = {
NEXUS_IP_ADDRESS: {
'username': 'admin',
'password': 'pass1234'
},
}
db.configure_db()
@ -78,51 +115,69 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
with mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
'__init__', new=new_nexus_init):
self._cisco_nexus_plugin = cisco_nexus_plugin_v2.NexusPlugin()
self._cisco_nexus_plugin._nexus_switches = self._nexus_switches
self.addCleanup(self.patch_obj.stop)
def test_create_networks(self):
"""Tests creation of two new Virtual Networks."""
tenant_id = self.tenant_id
net_name = self.net_name
net_id = self.net_id
vlan_name = self.vlan_name
vlan_id = self.vlan_id
second_net_name = self.second_net_name
second_net_id = self.second_net_id
second_vlan_name = self.second_vlan_name
second_vlan_id = self.second_vlan_id
new_net_dict = self._cisco_nexus_plugin.create_network(
self.network1, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.network1[attr])
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, net_name, net_id,
vlan_name, vlan_id, self._hostname, INSTANCE)
self.assertEqual(new_net_dict[const.NET_ID], net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.network2, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.network2[attr])
def test_create_providernet(self):
with mock.patch.object(cdb, 'is_provider_vlan',
return_value=True) as mock_db:
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, second_net_name, second_net_id,
second_vlan_name, second_vlan_id, self._hostname,
INSTANCE)
self.providernet, self.attachment1)
mock_db.assert_called_once()
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
self.assertEqual(new_net_dict[const.NET_ID], second_net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.second_net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME],
self.second_vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.second_vlan_id)
def test_create_provider_vlan_network_cfg_auto_man(self):
cfg.CONF.set_override('provider_vlan_auto_create', True, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', False, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_create_provider_vlan_network_cfg_man_auto(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', True, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_create_provider_vlan_network_cfg_man_man(self):
cfg.CONF.set_override('provider_vlan_auto_create', False, 'CISCO')
cfg.CONF.set_override('provider_vlan_auto_trunk', False, 'CISCO')
self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan', return_value=True):
new_net_dict = self._cisco_nexus_plugin.create_network(
self.providernet, self.attachment1)
for attr in NET_ATTRS:
self.assertEqual(new_net_dict[attr], self.providernet[attr])
def test_nexus_delete_port(self):
"""Test deletion of a vlan."""
self._cisco_nexus_plugin.create_network(
self.tenant_id, self.net_name, self.net_id, self.vlan_name,
self.vlan_id, self._hostname, INSTANCE)
self.network1, self.attachment1)
expected_instance_id = self._cisco_nexus_plugin.delete_port(
INSTANCE, self.vlan_id)
INSTANCE1, self.vlan_id)
self.assertEqual(expected_instance_id, INSTANCE)
self.assertEqual(expected_instance_id, INSTANCE1)
def test_nexus_add_remove_router_interface(self):
"""Tests addition of a router interface."""