x/vmware-nsx
Code Issues Proposed changes

Merge "Providernet extension support for the Cisco Nexus plugin"

Browse Source
This commit is contained in:
Jenkins 2013-07-24 22:24:45 +00:00 committed by Gerrit Code Review
commit b7a3939a89
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)

182
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:
# 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)
vlan_created = True
if auto_create and auto_trunk:
# Create vlan and trunk vlan on the port
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:
# 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
if auto_trunk:
# Only trunk vlan on the port
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):

151
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])
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)
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(
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."""