# 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.
#
--- /dev/null
+# 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')
# @author: Sumit Naiksatam, Cisco Systems, Inc.
-PLUGINS = 'PLUGINS'
-INVENTORY = 'INVENTORY'
-
-PORT_STATE = 'port-state'
-PORT_UP = "ACTIVE"
-PORT_DOWN = "DOWN"
-
-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'
+# Attachment attributes
+INSTANCE_ID = 'instance_id'
+TENANT_ID = 'tenant_id'
+TENANT_NAME = 'tenant_name'
+HOST_NAME = 'host_name'
-TENANT_ID = 'tenant-id'
-TENANT_NETWORKS = 'tenant-networks'
-TENANT_NAME = 'tenant-name'
-TENANT_QOS_LEVELS = 'tenant-qos-levels'
-TENANT_CREDENTIALS = 'tenant-credentials'
+# Network attributes
+NET_ID = 'id'
+NET_NAME = 'name'
+NET_VLAN_ID = 'vlan_id'
+NET_VLAN_NAME = 'vlan_name'
+NET_PORTS = 'ports'
-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'
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'
"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.")
+++ /dev/null
-# 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
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',
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
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).
#
# @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
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)
"""
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.
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
"""
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',
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.
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):
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(),
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):
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):
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__)
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
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
"""
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:
# 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,
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
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)
-
- 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.
-
- 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)
-
+ self._edit_config(nexus_host, target='running', config=confstr)
+
+ 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)
+
+ 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)
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
"""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.
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}
# 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)
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
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
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):
"""
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"))
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'])
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
import logging
import mock
+from oslo.config import cfg
import webob.exc as wexc
from neutron.api.v2 import base
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
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
'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):
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):
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()
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(
- 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.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, second_net_name, second_net_id,
- second_vlan_name, second_vlan_id, self._hostname,
- INSTANCE)
-
- 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)
+ 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(
+ self.providernet, self.attachment1)
+ mock_db.assert_called_once()
+ for attr in NET_ATTRS:
+ self.assertEqual(new_net_dict[attr], self.providernet[attr])
+
+ 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."""
Code Review / openstack-build / neutron-build.git / commitdiff