-[cisco_plugins]
-
-# (StrOpt) Period-separated module path to the plugin class to use for
-# the Cisco Nexus switches.
-#
-# nexus_plugin = neutron.plugins.cisco.nexus.cisco_nexus_plugin_v2.NexusPlugin
-
-# (StrOpt) Period-separated module path to the plugin class to use for
-# the virtual switches on compute nodes.
-#
-# vswitch_plugin = neutron.plugins.openvswitch.ovs_neutron_plugin.OVSNeutronPluginV2
-
-
[cisco]
# (StrOpt) A short prefix to prepend to the VLAN number when creating a
#
# model_class = neutron.plugins.cisco.models.virt_phy_sw_v2.VirtualPhysicalSwitchModelV2
-# (StrOpt) Period-separated module path to the driver class to use for
-# the Cisco Nexus switches.
-#
-# If no value is configured, a fake driver will be used.
-# nexus_driver = neutron.plugins.cisco.test.nexus.fake_nexus_driver.CiscoNEXUSFakeDriver
-# With real hardware, use the CiscoNEXUSDriver class:
-# nexus_driver = neutron.plugins.cisco.nexus.cisco_nexus_network_driver_v2.CiscoNEXUSDriver
-
# (BoolOpt) A flag to enable Layer 3 support on the Nexus switches.
# Note: This feature is not supported on all models/versions of Cisco
# Nexus switches. To use this feature, all of the Nexus switches in the
# Cisco Nexus Switch configurations.
# Each switch to be managed by Openstack Neutron must be configured here.
-#
-# Cisco Nexus Switch Format.
-# [NEXUS_SWITCH:<IP address of switch>]
-# <hostname>=<port> (1)
-# ssh_port=<ssh port> (2)
-# username=<credential username> (3)
-# password=<credential password> (4)
-#
-# (1) For each host connected to a port on the switch, specify the hostname
-# and the Nexus physical port (interface) it is connected to.
-# (2) The TCP port for connecting via SSH to manage the switch. This is
-# port number 22 unless the switch has been configured otherwise.
-# (3) The username for logging into the switch to manage it.
-# (4) The password for logging into the switch to manage it.
-#
-# Example:
-# [NEXUS_SWITCH:1.1.1.1]
-# compute1=1/1
-# compute2=1/2
-# ssh_port=22
-# username=admin
-# password=mySecretPassword
-
#
# N1KV Format.
# [N1KV:<IP address of VSM>]
--- /dev/null
+# Copyright 2014 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.
+#
+
+"""Remove Cisco Nexus Monolithic Plugin
+
+Revision ID: 1680e1f0c4dc
+Revises: 3c346828361e
+Create Date: 2014-08-31 08:58:37.123992
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '1680e1f0c4dc'
+down_revision = '3c346828361e'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade(active_plugins=None, options=None):
+ op.execute('INSERT INTO cisco_ml2_nexusport_bindings (port_id, '
+ 'vlan_id, switch_ip, instance_id) SELECT '
+ 'port_id, vlan_id, switch_ip, instance_id FROM '
+ 'cisco_nexusport_bindings')
+ op.drop_table('cisco_nexusport_bindings')
+
+
+def downgrade(active_plugins=None, options=None):
+ op.create_table(
+ 'cisco_nexusport_bindings',
+ sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+ sa.Column('port_id', sa.String(255)),
+ sa.Column('vlan_id', sa.Integer(), nullable=False),
+ sa.Column('switch_ip', sa.String(255), nullable=False),
+ sa.Column('instance_id', sa.String(255), nullable=False),
+ sa.PrimaryKeyConstraint('id')
+ )
+ op.execute('INSERT INTO cisco_nexusport_bindings (port_id, '
+ 'vlan_id, switch_ip, instance_id) SELECT '
+ 'port_id, vlan_id, switch_ip, instance_id FROM '
+ 'cisco_ml2_nexusport_bindings')
-3c346828361e
+1680e1f0c4dc
from neutron.plugins.cisco.db.l3 import l3_models # noqa
from neutron.plugins.cisco.db import n1kv_models_v2 # noqa
from neutron.plugins.cisco.db import network_models_v2 # noqa
-from neutron.plugins.cisco.db import nexus_models_v2 # noqa
from neutron.plugins.hyperv import model # noqa
from neutron.plugins.linuxbridge.db import l2network_models_v2 # noqa
from neutron.plugins.metaplugin import meta_models_v2 # noqa
LOGGER_COMPONENT_NAME = "cisco_plugin"
-NEXUS_PLUGIN = 'nexus_plugin'
VSWITCH_PLUGIN = 'vswitch_plugin'
DEVICE_IP = 'device_ip'
UUID_LENGTH = 36
-# Nexus vlan and vxlan segment range
-NEXUS_VLAN_RESERVED_MIN = 3968
-NEXUS_VLAN_RESERVED_MAX = 4047
-NEXUS_VXLAN_MIN = 4096
-NEXUS_VXLAN_MAX = 16000000
+# N1KV vlan and vxlan segment range
+N1KV_VLAN_RESERVED_MIN = 3968
+N1KV_VLAN_RESERVED_MAX = 4047
+N1KV_VXLAN_MIN = 4096
+N1KV_VXLAN_MAX = 16000000
# Type and topic for Cisco cfg agent
# ==================================
from neutron.agent.common import config
-cisco_plugins_opts = [
- cfg.StrOpt('vswitch_plugin',
- default='neutron.plugins.openvswitch.ovs_neutron_plugin.'
- 'OVSNeutronPluginV2',
- help=_("Virtual Switch to use")),
- cfg.StrOpt('nexus_plugin',
- default='neutron.plugins.cisco.nexus.cisco_nexus_plugin_v2.'
- 'NexusPlugin',
- help=_("Nexus Switch to use")),
-]
-
cisco_opts = [
cfg.StrOpt('vlan_name_prefix', default='q-',
help=_("VLAN Name prefix")),
default='neutron.plugins.cisco.models.virt_phy_sw_v2.'
'VirtualPhysicalSwitchModelV2',
help=_("Model Class")),
- cfg.StrOpt('nexus_driver',
- default='neutron.plugins.cisco.test.nexus.'
- 'fake_nexus_driver.CiscoNEXUSFakeDriver',
- help=_("Nexus Driver Name")),
]
cisco_n1k_opts = [
cfg.CONF.register_opts(cisco_opts, "CISCO")
cfg.CONF.register_opts(cisco_n1k_opts, "CISCO_N1K")
-cfg.CONF.register_opts(cisco_plugins_opts, "CISCO_PLUGINS")
config.register_root_helper(cfg.CONF)
# shortcuts
CONF = cfg.CONF
CISCO = cfg.CONF.CISCO
CISCO_N1K = cfg.CONF.CISCO_N1K
-CISCO_PLUGINS = cfg.CONF.CISCO_PLUGINS
#
# device_dictionary - Contains all external device configuration.
for parsed_file in multi_parser.parsed:
for parsed_item in parsed_file.keys():
dev_id, sep, dev_ip = parsed_item.partition(':')
- if dev_id.lower() in ['nexus_switch', 'n1kv']:
+ if dev_id.lower() == 'n1kv':
for dev_key, value in parsed_file[parsed_item].items():
if dev_ip and not first_device_ip:
first_device_ip = dev_ip
if segment_type == c_const.NETWORK_TYPE_VLAN:
if not ((seg_min <= seg_max) and
((seg_min in range(constants.MIN_VLAN_TAG,
- c_const.NEXUS_VLAN_RESERVED_MIN) and
+ c_const.N1KV_VLAN_RESERVED_MIN) and
seg_max in range(constants.MIN_VLAN_TAG,
- c_const.NEXUS_VLAN_RESERVED_MIN)) or
- (seg_min in range(c_const.NEXUS_VLAN_RESERVED_MAX + 1,
+ c_const.N1KV_VLAN_RESERVED_MIN)) or
+ (seg_min in range(c_const.N1KV_VLAN_RESERVED_MAX + 1,
constants.MAX_VLAN_TAG) and
- seg_max in range(c_const.NEXUS_VLAN_RESERVED_MAX + 1,
+ seg_max in range(c_const.N1KV_VLAN_RESERVED_MAX + 1,
constants.MAX_VLAN_TAG)))):
msg = (_("Segment range is invalid, select from "
"%(min)s-%(nmin)s, %(nmax)s-%(max)s") %
{"min": constants.MIN_VLAN_TAG,
- "nmin": c_const.NEXUS_VLAN_RESERVED_MIN - 1,
- "nmax": c_const.NEXUS_VLAN_RESERVED_MAX + 1,
+ "nmin": c_const.N1KV_VLAN_RESERVED_MIN - 1,
+ "nmax": c_const.N1KV_VLAN_RESERVED_MAX + 1,
"max": constants.MAX_VLAN_TAG - 1})
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
c_const.NETWORK_TYPE_MULTI_SEGMENT,
c_const.NETWORK_TYPE_TRUNK]:
if (seg_min > seg_max or
- seg_min < c_const.NEXUS_VXLAN_MIN or
- seg_max > c_const.NEXUS_VXLAN_MAX):
+ seg_min < c_const.N1KV_VXLAN_MIN or
+ seg_max > c_const.N1KV_VXLAN_MAX):
msg = (_("segment range is invalid. Valid range is : "
"%(min)s-%(max)s") %
- {"min": c_const.NEXUS_VXLAN_MIN,
- "max": c_const.NEXUS_VXLAN_MAX})
+ {"min": c_const.N1KV_VXLAN_MIN,
+ "max": c_const.N1KV_VXLAN_MAX})
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
profiles = _get_network_profiles(db_session=context.session)
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
LOG = logging.getLogger(__name__)
return True
-def get_ovs_vlans():
- session = db.get_session()
- bindings = (session.query(ovs_models_v2.VlanAllocation.vlan_id).
- filter_by(allocated=True))
- return [binding.vlan_id for binding in bindings]
-
-
class Credential_db_mixin(object):
"""Mixin class for Cisco Credentials as a resource."""
+++ /dev/null
-# Copyright 2012, Cisco Systems, Inc.
-#
-# 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: Rohit Agarwalla, Cisco Systems, Inc.
-# @author: Arvind Somya, Cisco Systems, Inc. (asomya@cisco.com)
-#
-
-import sqlalchemy.orm.exc as sa_exc
-
-import neutron.db.api as db
-from neutron.openstack.common import log as logging
-from neutron.plugins.cisco.common import cisco_exceptions as c_exc
-from neutron.plugins.cisco.db import nexus_models_v2
-
-
-LOG = logging.getLogger(__name__)
-
-
-def get_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Lists a nexusport binding."""
- LOG.debug(_("get_nexusport_binding() called"))
- return _lookup_all_nexus_bindings(port_id=port_id,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- instance_id=instance_id)
-
-
-def get_nexusvlan_binding(vlan_id, switch_ip):
- """Lists a vlan and switch binding."""
- LOG.debug(_("get_nexusvlan_binding() called"))
- return _lookup_all_nexus_bindings(vlan_id=vlan_id, switch_ip=switch_ip)
-
-
-def add_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Adds a nexusport binding."""
- LOG.debug(_("add_nexusport_binding() called"))
- session = db.get_session()
- binding = nexus_models_v2.NexusPortBinding(port_id=port_id,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- instance_id=instance_id)
- session.add(binding)
- session.flush()
- return binding
-
-
-def remove_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
- """Removes a nexusport binding."""
- LOG.debug(_("remove_nexusport_binding() called"))
- session = db.get_session()
- binding = _lookup_all_nexus_bindings(session=session,
- vlan_id=vlan_id,
- switch_ip=switch_ip,
- port_id=port_id,
- instance_id=instance_id)
- for bind in binding:
- session.delete(bind)
- session.flush()
- return binding
-
-
-def update_nexusport_binding(port_id, new_vlan_id):
- """Updates nexusport binding."""
- if not new_vlan_id:
- LOG.warning(_("update_nexusport_binding called with no vlan"))
- return
- LOG.debug(_("update_nexusport_binding called"))
- session = db.get_session()
- binding = _lookup_one_nexus_binding(session=session, port_id=port_id)
- binding.vlan_id = new_vlan_id
- session.merge(binding)
- session.flush()
- return binding
-
-
-def get_nexusvm_bindings(vlan_id, instance_id):
- """Lists nexusvm bindings."""
- LOG.debug(_("get_nexusvm_binding() called"))
-
- return _lookup_all_nexus_bindings(vlan_id=vlan_id,
- instance_id=instance_id)
-
-
-def get_port_vlan_switch_binding(port_id, vlan_id, switch_ip):
- """Lists nexusvm bindings."""
- LOG.debug(_("get_port_vlan_switch_binding() called"))
- return _lookup_all_nexus_bindings(port_id=port_id,
- switch_ip=switch_ip,
- vlan_id=vlan_id)
-
-
-def get_port_switch_bindings(port_id, switch_ip):
- """List all vm/vlan bindings on a Nexus switch port."""
- LOG.debug(_("get_port_switch_bindings() called, "
- "port:'%(port_id)s', switch:'%(switch_ip)s'"),
- {'port_id': port_id, 'switch_ip': switch_ip})
- try:
- return _lookup_all_nexus_bindings(port_id=port_id,
- switch_ip=switch_ip)
- except c_exc.NexusPortBindingNotFound:
- pass
-
-
-def get_nexussvi_bindings():
- """Lists nexus svi bindings."""
- LOG.debug(_("get_nexussvi_bindings() called"))
- return _lookup_all_nexus_bindings(port_id='router')
-
-
-def _lookup_nexus_bindings(query_type, session=None, **bfilter):
- """Look up 'query_type' Nexus bindings matching the filter.
-
- :param query_type: 'all', 'one' or 'first'
- :param session: db session
- :param bfilter: filter for bindings query
- :return: bindings if query gave a result, else
- raise NexusPortBindingNotFound.
- """
- if session is None:
- session = db.get_session()
- query_method = getattr(session.query(
- nexus_models_v2.NexusPortBinding).filter_by(**bfilter), query_type)
- try:
- bindings = query_method()
- if bindings:
- return bindings
- except sa_exc.NoResultFound:
- pass
- raise c_exc.NexusPortBindingNotFound(**bfilter)
-
-
-def _lookup_all_nexus_bindings(session=None, **bfilter):
- return _lookup_nexus_bindings('all', session, **bfilter)
-
-
-def _lookup_one_nexus_binding(session=None, **bfilter):
- return _lookup_nexus_bindings('one', session, **bfilter)
-
-
-def _lookup_first_nexus_binding(session=None, **bfilter):
- return _lookup_nexus_bindings('first', session, **bfilter)
+++ /dev/null
-# Copyright 2012, Cisco Systems, Inc.
-#
-# 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: Rohit Agarwalla, Cisco Systems, Inc.
-
-import sqlalchemy as sa
-
-from neutron.db import model_base
-
-
-class NexusPortBinding(model_base.BASEV2):
- """Represents a binding of VM's to nexus ports."""
-
- __tablename__ = "cisco_nexusport_bindings"
-
- id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
- port_id = sa.Column(sa.String(255))
- vlan_id = sa.Column(sa.Integer, nullable=False)
- switch_ip = sa.Column(sa.String(255), nullable=False)
- instance_id = sa.Column(sa.String(255), nullable=False)
-
- def __repr__(self):
- """Just the binding, without the id key."""
- return ("<NexusPortBinding(%s,%s,%s,%s)>" %
- (self.port_id, self.vlan_id, self.switch_ip, self.instance_id))
-
- def __eq__(self, other):
- """Compare only the binding, without the id key."""
- return (
- self.port_id == other.port_id and
- self.vlan_id == other.vlan_id and
- self.switch_ip == other.switch_ip and
- self.instance_id == other.instance_id
- )
#
import inspect
-import sys
from neutron.api.v2 import attributes
from neutron.extensions import portbindings
from neutron.extensions import providernet as provider
from neutron import neutron_plugin_base_v2
+from neutron.openstack.common import excutils
+from neutron.openstack.common.gettextutils import _LE
from neutron.openstack.common import importutils
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_credentials_v2 as cred
-from neutron.plugins.cisco.common import cisco_exceptions as cexc
from neutron.plugins.cisco.common import config as conf
from neutron.plugins.cisco.db import network_db_v2 as cdb
-from neutron.plugins.openvswitch import ovs_db_v2 as odb
LOG = logging.getLogger(__name__)
class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
"""Virtual Physical Switch Model.
- This implementation works with OVS and Nexus plugin for the
+ This implementation works with n1kv sub-plugin for the
following topology:
- One or more servers to a nexus switch.
+ One or more servers to a n1kv switch.
"""
__native_bulk_support = True
supported_extension_aliases = ["provider", "binding"]
conf.CiscoConfigOptions()
self._plugins = {}
- for key in conf.CISCO_PLUGINS.keys():
- plugin_obj = conf.CISCO_PLUGINS[key]
- if plugin_obj is not None:
- self._plugins[key] = importutils.import_object(plugin_obj)
- LOG.debug(_("Loaded device plugin %s"),
- conf.CISCO_PLUGINS[key])
+ self._plugins['vswitch_plugin'] = importutils.import_object(
+ 'neutron.plugins.cisco.n1kv.n1kv_neutron_plugin.'
+ 'N1kvNeutronPluginV2')
if ((const.VSWITCH_PLUGIN in self._plugins) and
hasattr(self._plugins[const.VSWITCH_PLUGIN],
{'module': __name__,
'name': self.__class__.__name__})
- # Check whether we have a valid Nexus driver loaded
- self.is_nexus_plugin = False
- nexus_driver = conf.CISCO.nexus_driver
- if nexus_driver.endswith('CiscoNEXUSDriver'):
- self.is_nexus_plugin = True
-
def __getattribute__(self, name):
- """Delegate calls to OVS sub-plugin.
+ """Delegate calls to sub-plugin.
- This delegates the calls to the methods implemented only by the OVS
+ This delegates the calls to the methods implemented by the
sub-plugin. Note: Currently, bulking is handled by the caller
(PluginV2), and this model class expects to receive only non-bulking
calls. If, however, a bulking call is made, this will method will
- delegate the call to the OVS plugin.
+ delegate the call to the sub-plugin.
"""
super_getattribute = super(VirtualPhysicalSwitchModelV2,
self).__getattribute__
func = getattr(self._plugins[plugin_key], function_name)
return func(*args, **kwargs)
- def _get_segmentation_id(self, network_id):
- binding_seg_id = odb.get_network_binding(None, network_id)
- if not binding_seg_id:
- raise cexc.NetworkSegmentIDNotFound(net_id=network_id)
- return binding_seg_id.segmentation_id
-
def _get_provider_vlan_id(self, network):
if (all(attributes.is_attr_set(network.get(attr))
for attr in (provider.NETWORK_TYPE,
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)
+ switch_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.
+ # vlan network, save it for the sub-plugin to use later.
if provider_vlan_id:
- network_id = ovs_output[const.NET_ID]
+ network_id = switch_output[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
+ return switch_output
def update_network(self, context, id, network):
"""Update network.
Perform this operation in the context of the configured device
plugins.
-
- Note that the Nexus sub-plugin does not need to be notified
- (and the Nexus switch does not need to be [re]configured)
- for an update network operation because the Nexus sub-plugin
- is agnostic of all network-level attributes except the
- segmentation ID. Furthermore, updating of the segmentation ID
- is not supported by the OVS plugin since it is considered a
- provider attribute, so it is not supported by this method.
"""
LOG.debug(_("update_network() called"))
plugins.
"""
args = [context, id]
- ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
- self._func_name(),
- args)
+ switch_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
+ return switch_output
def get_network(self, context, id, fields=None):
"""Get network. This method is delegated to the vswitch plugin.
"""
pass # pragma no cover
- def _invoke_nexus_for_net_create(self, context, tenant_id, net_id,
- instance_id, host_id):
- if not self.is_nexus_plugin:
- return False
-
- network = self.get_network(context, net_id)
- vlan_id = self._get_segmentation_id(net_id)
- vlan_name = conf.CISCO.vlan_name_prefix + str(vlan_id)
- 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: host_id,
- }
- self._invoke_plugin_per_device(
- const.NEXUS_PLUGIN,
- 'create_network',
- [network, attachment])
-
def _check_valid_port_device_owner(self, port):
"""Check the port for valid device_owner.
- Don't call the nexus plugin for router and dhcp
+ Don't call the sub-plugin for router and dhcp
port owners.
"""
return port['device_owner'].startswith('compute')
"""
LOG.debug(_("create_port() called"))
args = [context, port]
- ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
- self._func_name(),
- args)
- instance_id = port['port']['device_id']
-
- # Only call nexus plugin if there's a valid instance_id, host_id
- # and device_owner
- try:
- host_id = self._get_port_host_id_from_bindings(port['port'])
- if (instance_id and host_id and
- self._check_valid_port_device_owner(port['port'])):
- net_id = port['port']['network_id']
- tenant_id = port['port']['tenant_id']
- self._invoke_nexus_for_net_create(
- context, tenant_id, net_id, instance_id, host_id)
- except Exception:
- # Create network on the Nexus plugin has failed, so we need
- # to rollback the port creation on the VSwitch plugin.
- exc_info = sys.exc_info()
- try:
- id = ovs_output['id']
- args = [context, id]
- ovs_output = self._invoke_plugin_per_device(
- const.VSWITCH_PLUGIN,
- 'delete_port',
- args)
- finally:
- # Re-raise the original exception
- raise exc_info[0], exc_info[1], exc_info[2]
- return ovs_output
+ return self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
+ self._func_name(),
+ args)
def get_port(self, context, id, fields=None):
"""Get port. This method is delegated to the vswitch plugin.
"""
pass # pragma no cover
- def _check_nexus_net_create_needed(self, new_port, old_port):
- """Check if nexus plugin should be invoked for net_create.
-
- In the following cases, the plugin should be invoked:
- -- a port is attached to a VM instance. The old host id is None
- -- VM migration. The old host id has a valid value
-
- When the plugin needs to be invoked, return the old_host_id,
- and a list of calling arguments.
- Otherwise, return '' for old host id and an empty list
- """
- old_device_id = old_port['device_id']
- new_device_id = new_port.get('device_id')
- new_host_id = self._get_port_host_id_from_bindings(new_port)
- tenant_id = old_port['tenant_id']
- net_id = old_port['network_id']
- old_host_id = self._get_port_host_id_from_bindings(old_port)
-
- LOG.debug(_("tenant_id: %(tid)s, net_id: %(nid)s, "
- "old_device_id: %(odi)s, new_device_id: %(ndi)s, "
- "old_host_id: %(ohi)s, new_host_id: %(nhi)s, "
- "old_device_owner: %(odo)s, new_device_owner: %(ndo)s"),
- {'tid': tenant_id, 'nid': net_id,
- 'odi': old_device_id, 'ndi': new_device_id,
- 'ohi': old_host_id, 'nhi': new_host_id,
- 'odo': old_port.get('device_owner'),
- 'ndo': new_port.get('device_owner')})
-
- # A port is attached to an instance
- if (new_device_id and not old_device_id and new_host_id and
- self._check_valid_port_device_owner(new_port)):
- return '', [tenant_id, net_id, new_device_id, new_host_id]
-
- # An instance is being migrated
- if (old_device_id and old_host_id and new_host_id != old_host_id and
- self._check_valid_port_device_owner(old_port)):
- return old_host_id, [tenant_id, net_id, old_device_id, new_host_id]
-
- # no need to invoke the plugin
- return '', []
-
def update_port(self, context, id, port):
"""Update port.
plugins.
"""
LOG.debug(_("update_port() called"))
- old_port = self.get_port(context, id)
args = [context, id, port]
- ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
- self._func_name(),
- args)
- try:
- # Check if the nexus plugin needs to be invoked
- old_host_id, create_args = self._check_nexus_net_create_needed(
- port['port'], old_port)
-
- # In the case of migration, invoke it to remove
- # the previous port binding
- if old_host_id:
- vlan_id = self._get_segmentation_id(old_port['network_id'])
- delete_args = [old_port['device_id'], vlan_id]
- self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
- "delete_port",
- delete_args)
-
- # Invoke the Nexus plugin to create a net and/or new port binding
- if create_args:
- self._invoke_nexus_for_net_create(context, *create_args)
-
- return ovs_output
- except Exception:
- exc_info = sys.exc_info()
- LOG.error(_("Unable to update port '%s' on Nexus switch"),
- old_port['name'], exc_info=exc_info)
- try:
- # Roll back vSwitch plugin to original port attributes.
- args = [context, id, {'port': old_port}]
- self._invoke_plugin_per_device(
- const.VSWITCH_PLUGIN,
- self._func_name(),
- args)
- finally:
- # Re-raise the original exception
- raise exc_info[0], exc_info[1], exc_info[2]
+ return self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
+ self._func_name(),
+ args)
def delete_port(self, context, id, l3_port_check=True):
"""Delete port.
LOG.debug(_("delete_port() called"))
port = self.get_port(context, id)
- host_id = self._get_port_host_id_from_bindings(port)
-
- if (self.is_nexus_plugin and host_id and
- self._check_valid_port_device_owner(port)):
- vlan_id = self._get_segmentation_id(port['network_id'])
- n_args = [port['device_id'], vlan_id]
- self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
- self._func_name(),
- n_args)
try:
args = [context, id]
- ovs_output = self._invoke_plugin_per_device(
+ switch_output = self._invoke_plugin_per_device(
const.VSWITCH_PLUGIN, self._func_name(),
args, l3_port_check=l3_port_check)
- except Exception:
- exc_info = sys.exc_info()
- # Roll back the delete port on the Nexus plugin
- try:
- tenant_id = port['tenant_id']
- net_id = port['network_id']
- instance_id = port['device_id']
- host_id = port[portbindings.HOST_ID]
- self._invoke_nexus_for_net_create(context, tenant_id, net_id,
- instance_id, host_id)
- finally:
- # Raise the original exception.
- raise exc_info[0], exc_info[1], exc_info[2]
-
- return ovs_output
-
- def add_router_interface(self, context, router_id, interface_info):
- """Add a router interface on a subnet.
-
- Only invoke the Nexus plugin to create SVI if L3 support on
- the Nexus switches is enabled and a Nexus plugin is loaded,
- otherwise send it to the vswitch plugin
- """
- if (conf.CISCO.nexus_l3_enable and self.is_nexus_plugin):
- LOG.debug(_("L3 enabled on Nexus plugin, create SVI on switch"))
- if 'subnet_id' not in interface_info:
- raise cexc.SubnetNotSpecified()
- if 'port_id' in interface_info:
- raise cexc.PortIdForNexusSvi()
- subnet = self.get_subnet(context, interface_info['subnet_id'])
- gateway_ip = subnet['gateway_ip']
- # Get gateway IP address and netmask
- cidr = subnet['cidr']
- netmask = cidr.split('/', 1)[1]
- gateway_ip = gateway_ip + '/' + netmask
- network_id = subnet['network_id']
- vlan_id = self._get_segmentation_id(network_id)
- vlan_name = conf.CISCO.vlan_name_prefix + str(vlan_id)
-
- n_args = [vlan_name, vlan_id, subnet['id'], gateway_ip, router_id]
- return self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
- self._func_name(),
- n_args)
- else:
- LOG.debug(_("L3 disabled or not Nexus plugin, send to vswitch"))
- n_args = [context, router_id, interface_info]
- return self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
- self._func_name(),
- n_args)
-
- def remove_router_interface(self, context, router_id, interface_info):
- """Remove a router interface.
-
- Only invoke the Nexus plugin to delete SVI if L3 support on
- the Nexus switches is enabled and a Nexus plugin is loaded,
- otherwise send it to the vswitch plugin
- """
- if (conf.CISCO.nexus_l3_enable and self.is_nexus_plugin):
- LOG.debug(_("L3 enabled on Nexus plugin, delete SVI from switch"))
+ except Exception as e:
+ with excutils.save_and_reraise_exception():
+ LOG.error(_LE("Unable to delete port '%(pname)s' on switch. "
+ "Exception: %(exp)s"), {'pname': port['name'],
+ 'exp': e})
- subnet = self.get_subnet(context, interface_info['subnet_id'])
- network_id = subnet['network_id']
- vlan_id = self._get_segmentation_id(network_id)
- n_args = [vlan_id, router_id]
-
- return self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
- self._func_name(),
- n_args)
- else:
- LOG.debug(_("L3 disabled or not Nexus plugin, send to vswitch"))
- n_args = [context, router_id, interface_info]
- return self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
- self._func_name(),
- n_args)
+ return switch_output
def create_subnet(self, context, subnet):
"""Create subnet. This method is delegated to the vswitch plugin.
+++ /dev/null
-# 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.
-# @author: Edgar Magana, Cisco Systems, Inc.
-"""
-Init module for Nexus Driver
-"""
+++ /dev/null
-# 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: Debojyoti Dutta, Cisco Systems, Inc.
-# @author: Edgar Magana, Cisco Systems Inc.
-#
-"""
-Implements a Nexus-OS NETCONF over SSHv2 API Client
-"""
-
-
-from ncclient import manager
-
-from neutron.openstack.common import excutils
-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_credentials_v2 as cred
-from neutron.plugins.cisco.common import cisco_exceptions as cexc
-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
-
-LOG = logging.getLogger(__name__)
-
-
-class CiscoNEXUSDriver():
- """Nexus Driver Main Class."""
- def __init__(self):
- cisco_switches = conf.get_device_dictionary()
- self.nexus_switches = dict(((key[1], key[2]), val)
- for key, val in cisco_switches.items()
- if key[0] == 'NEXUS_SWITCH')
- self.credentials = {}
- self.connections = {}
-
- def _edit_config(self, nexus_host, target='running', config='',
- allowed_exc_strs=None):
- """Modify switch config for a target config type.
-
- :param nexus_host: IP address of switch to configure
- :param target: Target config type
- :param config: Configuration string in XML format
- :param allowed_exc_strs: Exceptions which have any of these strings
- as a subset of their exception message
- (str(exception)) can be ignored
-
- :raises: NexusConfigFailed
-
- """
- 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:
- for exc_str in allowed_exc_strs:
- if exc_str in str(e):
- break
- else:
- # Raise a Neutron exception. Include a description of
- # the original ncclient exception. No need to preserve T/B
- raise cexc.NexusConfigFailed(config=config, exc=e)
-
- 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 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,
- username=nexus_user,
- password=nexus_password)
- self.connections[nexus_host] = man
- except Exception as e:
- # Raise a Neutron exception. Include a description of
- # the original ncclient exception. No need to preserve T/B.
- raise cexc.NexusConnectFailed(nexus_host=nexus_host, exc=e)
-
- return self.connections[nexus_host]
-
- def create_xml_snippet(self, cutomized_config):
- """Create XML snippet.
-
- Creates the Proper XML structure for the Nexus Switch Configuration.
- """
- conf_xml_snippet = snipp.EXEC_CONF_SNIPPET % (cutomized_config)
- return conf_xml_snippet
-
- def create_vlan(self, nexus_host, vlanid, vlanname):
- """Create a VLAN on Nexus Switch given the VLAN ID and Name."""
- confstr = self.create_xml_snippet(
- snipp.CMD_VLAN_CONF_SNIPPET % (vlanid, vlanname))
- self._edit_config(nexus_host, target='running', config=confstr)
-
- # Enable VLAN active and no-shutdown states. Some versions of
- # Nexus switch do not allow state changes for the extended VLAN
- # range (1006-4094), but these errors can be ignored (default
- # values are appropriate).
- state_config = [snipp.CMD_VLAN_ACTIVE_SNIPPET,
- snipp.CMD_VLAN_NO_SHUTDOWN_SNIPPET]
- for snippet in state_config:
- try:
- confstr = self.create_xml_snippet(snippet % vlanid)
- self._edit_config(
- nexus_host,
- target='running',
- config=confstr,
- allowed_exc_strs=["Can't modify state for extended",
- "Command is only allowed on VLAN"])
- except cexc.NexusConfigFailed:
- with excutils.save_and_reraise_exception():
- self.delete_vlan(nexus_host, vlanid)
-
- def delete_vlan(self, nexus_host, vlanid):
- """Delete a VLAN on Nexus Switch given the VLAN ID."""
- confstr = snipp.CMD_NO_VLAN_CONF_SNIPPET % vlanid
- confstr = self.create_xml_snippet(confstr)
- self._edit_config(nexus_host, target='running', config=confstr)
-
- def enable_vlan_on_trunk_int(self, nexus_host, vlanid, etype, 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('%s:%s' % (etype, interface),
- nexus_host):
- snippet = snipp.CMD_INT_VLAN_ADD_SNIPPET
- else:
- snippet = snipp.CMD_INT_VLAN_SNIPPET
- confstr = snippet % (etype, interface, vlanid, etype)
- confstr = self.create_xml_snippet(confstr)
- LOG.debug(_("NexusDriver: %s"), confstr)
- self._edit_config(nexus_host, target='running', config=confstr)
-
- def disable_vlan_on_trunk_int(self, nexus_host, vlanid, etype, interface):
- """Disable a VLAN on a trunk interface."""
- confstr = snipp.CMD_NO_VLAN_INT_SNIPPET % (etype, interface,
- vlanid, etype)
- confstr = self.create_xml_snippet(confstr)
- LOG.debug(_("NexusDriver: %s"), confstr)
- self._edit_config(nexus_host, target='running', config=confstr)
-
- def create_and_trunk_vlan(self, nexus_host, vlan_id, vlan_name,
- etype, 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,
- etype, nexus_port)
-
- def delete_and_untrunk_vlan(self, nexus_host, vlan_id, etype, 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,
- etype, 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)
- 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)
- self._edit_config(nexus_host, target='running', config=confstr)
+++ /dev/null
-# Copyright 2012 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.
-# @author: Edgar Magana, Cisco Systems, Inc.
-# @author: Arvind Somya, Cisco Systems, Inc. (asomya@cisco.com)
-#
-
-"""
-PlugIn for Nexus OS driver
-"""
-
-
-from neutron.openstack.common import excutils
-from neutron.openstack.common import importutils
-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 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 import l2device_plugin_base
-
-
-LOG = logging.getLogger(__name__)
-
-
-class NexusPlugin(l2device_plugin_base.L2DevicePluginBase):
- """Nexus PlugIn Main Class."""
- _networks = {}
-
- def __init__(self):
- """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_device_dictionary()
-
- 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 IPs and ports for this host
- host_connections = []
- host = attachment['host_name']
- for switch_type, switch_ip, attr in self._nexus_switches:
- if str(attr) == str(host):
- port = self._nexus_switches[switch_type, switch_ip, attr]
- # Get ether type for port, assume an ethernet type
- # if none specified.
- if ':' in port:
- etype, port_id = port.split(':')
- else:
- etype, port_id = 'ethernet', port
- host_connections.append((switch_ip, etype, port_id))
- if not host_connections:
- 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
- for switch_ip, etype, port_id in host_connections:
- vlan_created = False
- vlan_trunked = False
- eport_id = '%s:%s' % (etype, port_id)
- # Check for switch vlan bindings
- try:
- # This vlan has already been created on this switch
- # via another operation, like SVI bindings.
- nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
- vlan_created = True
- auto_create = False
- except cisco_exc.NexusPortBindingNotFound:
- # No changes, proceed as normal
- pass
-
- try:
- nxos_db.get_port_vlan_switch_binding(eport_id, vlan_id,
- switch_ip)
- except cisco_exc.NexusPortBindingNotFound:
- 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, etype, 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
- elif 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, etype, port_id)
- vlan_trunked = True
-
- try:
- instance = attachment[const.INSTANCE_ID]
- nxos_db.add_nexusport_binding(eport_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 and vlan_trunked:
- LOG.debug(_("Nexus: delete & untrunk vlan %s"),
- vlan_name)
- self._client.delete_and_untrunk_vlan(switch_ip,
- vlan_id,
- etype, 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,
- etype,
- port_id)
-
- net_id = network[const.NET_ID]
- new_net_dict = {const.NET_ID: net_id,
- const.NET_NAME: network[const.NET_NAME],
- const.NET_PORTS: {},
- const.NET_VLAN_NAME: vlan_name,
- const.NET_VLAN_ID: vlan_id}
- self._networks[net_id] = new_net_dict
- return new_net_dict
-
- def add_router_interface(self, vlan_name, vlan_id, subnet_id,
- gateway_ip, router_id):
- """Create VLAN SVI on the Nexus switch."""
- # Find a switch to create the SVI on
- switch_ip = self._find_switch_for_svi()
- if not switch_ip:
- 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_and_trunk_vlan(
- switch_ip, vlan_id, vlan_name, etype=None, nexus_port=None)
- # Check if a router interface has already been created
- try:
- nxos_db.get_nexusvm_bindings(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(switch_ip, vlan_id, gateway_ip)
- nxos_db.add_nexusport_binding('router', str(vlan_id),
- switch_ip, router_id)
-
- return True
-
- def remove_router_interface(self, vlan_id, router_id):
- """Remove VLAN SVI from the Nexus Switch."""
- # Grab switch_ip from database
- switch_ip = nxos_db.get_nexusvm_bindings(vlan_id,
- router_id)[0].switch_ip
-
- # Delete the SVI interface from the switch
- self._client.delete_vlan_svi(switch_ip, vlan_id)
-
- # Invoke delete_port to delete this row
- # And delete vlan if required
- return self.delete_port(router_id, vlan_id)
-
- 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 nexus_switches)
- try:
- bindings = nxos_db.get_nexussvi_bindings()
- # Build a switch dictionary with weights
- for binding in bindings:
- switch_ip = binding.switch_ip
- if switch_ip not in switch_dict:
- switch_dict[switch_ip] = 1
- else:
- switch_dict[switch_ip] += 1
- # Search for the lowest value in the dict
- if switch_dict:
- switch_ip = min(switch_dict, key=switch_dict.get)
- return switch_ip
- except cisco_exc.NexusPortBindingNotFound:
- pass
-
- LOG.debug(_("No round robin or zero weights, using first switch"))
- # Return the first switch in the config
- return conf.first_device_ip
-
- def delete_network(self, tenant_id, net_id, **kwargs):
- """Delete network.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:delete_network() called")) # pragma no cover
-
- def update_network(self, tenant_id, net_id, **kwargs):
- """Update the properties of a particular Virtual Network.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:update_network() called")) # pragma no cover
-
- def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
- """Create port.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:create_port() called")) # pragma no cover
-
- def delete_port(self, device_id, vlan_id):
- """Delete port.
-
- Delete port bindings from the database and scan whether the network
- is still required on the interfaces trunked.
- """
- LOG.debug(_("NexusPlugin:delete_port() called"))
- # Delete DB row(s) for this port
- try:
- rows = nxos_db.get_nexusvm_bindings(vlan_id, device_id)
- 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)
-
- instance_id = False
- for row in rows:
- instance_id = row['instance_id']
- switch_ip = row.switch_ip
- etype, nexus_port = '', ''
- if row['port_id'] == 'router':
- etype, nexus_port = 'vlan', row['port_id']
- auto_untrunk = False
- else:
- etype, nexus_port = row['port_id'].split(':')
-
- nxos_db.remove_nexusport_binding(row.port_id, row.vlan_id,
- row.switch_ip,
- row.instance_id)
- # Check whether there are any remaining instances using this
- # vlan on this Nexus port.
- try:
- nxos_db.get_port_vlan_switch_binding(row.port_id,
- row.vlan_id,
- row.switch_ip)
- except cisco_exc.NexusPortBindingNotFound:
- try:
- if nexus_port and auto_untrunk:
- # Untrunk the vlan from this Nexus interface
- self._client.disable_vlan_on_trunk_int(
- switch_ip, row.vlan_id, etype, nexus_port)
-
- # Check whether there are any remaining instances
- # using this vlan on the Nexus switch.
- if auto_delete:
- try:
- nxos_db.get_nexusvlan_binding(row.vlan_id,
- row.switch_ip)
- except cisco_exc.NexusPortBindingNotFound:
- # Delete this vlan from this switch
- 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
- # consistency, roll back the Nexus database to what
- # it was before this request.
- with excutils.save_and_reraise_exception():
- nxos_db.add_nexusport_binding(row.port_id,
- row.vlan_id,
- row.switch_ip,
- row.instance_id)
-
- return instance_id
-
- def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
- """Update port.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:update_port() called")) # pragma no cover
-
- def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
- **kwargs):
- """Plug interfaces.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:plug_interface() called")) # pragma no cover
-
- def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
- """Unplug interface.
-
- Not applicable to Nexus plugin. Defined here to satisfy abstract
- method requirements.
- """
- LOG.debug(_("NexusPlugin:unplug_interface() called")
- ) # pragma no cover
+++ /dev/null
-# 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: Edgar Magana, Cisco Systems, Inc.
-# @author: Arvind Somya (asomya@cisco.com) Cisco Systems, Inc.
-
-"""
-Nexus-OS XML-based configuration snippets
-"""
-
-
-# The following are standard strings, messages used to communicate with Nexus,
-EXEC_CONF_SNIPPET = """
- <config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
- <configure>
- <__XML__MODE__exec_configure>%s
- </__XML__MODE__exec_configure>
- </configure>
- </config>
-"""
-
-CMD_VLAN_CONF_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <name>
- <vlan-name>%s</vlan-name>
- </name>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_VLAN_ACTIVE_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <state>
- <vstate>active</vstate>
- </state>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_VLAN_NO_SHUTDOWN_SNIPPET = """
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- <__XML__MODE_vlan>
- <no>
- <shutdown/>
- </no>
- </__XML__MODE_vlan>
- </vlan-id-create-delete>
- </vlan>
-"""
-
-CMD_NO_VLAN_CONF_SNIPPET = """
- <no>
- <vlan>
- <vlan-id-create-delete>
- <__XML__PARAM_value>%s</__XML__PARAM_value>
- </vlan-id-create-delete>
- </vlan>
- </no>
-"""
-
-CMD_INT_VLAN_HEADER = """
- <interface>
- <%s>
- <interface>%s</interface>
- <__XML__MODE_if-ethernet-switch>
- <switchport>
- <trunk>
- <allowed>
- <vlan>"""
-
-CMD_VLAN_ID = """
- <vlan_id>%s</vlan_id>"""
-
-CMD_VLAN_ADD_ID = """
- <add>%s
- </add>""" % CMD_VLAN_ID
-
-CMD_INT_VLAN_TRAILER = """
- </vlan>
- </allowed>
- </trunk>
- </switchport>
- </__XML__MODE_if-ethernet-switch>
- </%s>
- </interface>
-"""
-
-CMD_INT_VLAN_SNIPPET = (CMD_INT_VLAN_HEADER +
- CMD_VLAN_ID +
- CMD_INT_VLAN_TRAILER)
-
-CMD_INT_VLAN_ADD_SNIPPET = (CMD_INT_VLAN_HEADER +
- CMD_VLAN_ADD_ID +
- CMD_INT_VLAN_TRAILER)
-
-CMD_NO_VLAN_INT_SNIPPET = """
- <interface>
- <%s>
- <interface>%s</interface>
- <__XML__MODE_if-ethernet-switch>
- <switchport></switchport>
- <switchport>
- <trunk>
- <allowed>
- <vlan>
- <remove>
- <vlan>%s</vlan>
- </remove>
- </vlan>
- </allowed>
- </trunk>
- </switchport>
- </__XML__MODE_if-ethernet-switch>
- </%s>
- </interface>
-"""
-
-FILTER_SHOW_VLAN_BRIEF_SNIPPET = """
- <show xmlns="http://www.cisco.com/nxos:1.0:vlan_mgr_cli">
- <vlan>
- <brief/>
- </vlan>
- </show>
-"""
-
-CMD_VLAN_SVI_SNIPPET = """
-<interface>
- <vlan>
- <vlan>%s</vlan>
- <__XML__MODE_vlan>
- <no>
- <shutdown/>
- </no>
- <ip>
- <address>
- <address>%s</address>
- </address>
- </ip>
- </__XML__MODE_vlan>
- </vlan>
-</interface>
-"""
-
-CMD_NO_VLAN_SVI_SNIPPET = """
-<no>
- <interface>
- <vlan>
- <vlan>%s</vlan>
- </vlan>
- </interface>
-</no>
-"""
+++ /dev/null
-# Copyright 2012 OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import __builtin__
-setattr(__builtin__, '_', lambda x: x)
+++ /dev/null
-# Copyright 2012 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.
-# @author: Rohit Agarwalla, Cisco Systems, Inc.
-
-
-class CiscoNEXUSFakeDriver():
- """Nexus Driver Fake Class."""
-
- def __init__(self):
- pass
-
- def nxos_connect(self, nexus_host, nexus_ssh_port, nexus_user,
- nexus_password):
- """Make the fake connection to the Nexus Switch."""
- pass
-
- def create_xml_snippet(self, cutomized_config):
- """Create XML snippet.
-
- Creates the Proper XML structure for the Nexus Switch
- Configuration.
- """
- pass
-
- def enable_vlan(self, mgr, vlanid, vlanname):
- """Create a VLAN on Nexus Switch given the VLAN ID and Name."""
- pass
-
- def disable_vlan(self, mgr, vlanid):
- """Delete a VLAN on Nexus Switch given the VLAN ID."""
- pass
-
- def disable_switch_port(self, mgr, interface):
- """Disable trunk mode an interface on Nexus Switch."""
- pass
-
- def enable_vlan_on_trunk_int(self, mgr, etype, interface, vlanid):
- """Enable vlan on trunk interface.
-
- Enable trunk mode vlan access an interface on Nexus Switch given
- VLANID.
- """
- pass
-
- def disable_vlan_on_trunk_int(self, mgr, interface, vlanid):
- """Disables vlan in trunk interface.
-
- Enables trunk mode vlan access an interface on Nexus Switch given
- VLANID.
- """
- pass
-
- def create_vlan(self, vlan_name, vlan_id, nexus_host, nexus_user,
- nexus_password, nexus_ports, nexus_ssh_port, vlan_ids):
- """Create VLAN and enable it on interface.
-
- Creates a VLAN and Enable on trunk mode an interface on Nexus Switch
- given the VLAN ID and Name and Interface Number.
- """
- pass
-
- 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.
- """
- pass
-
- def build_vlans_cmd(self):
- """Build a string with all the VLANs on the same Switch."""
- pass
-
- def add_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
- nexus_ports, nexus_ssh_port, vlan_ids=None):
- """Add a vlan from interfaces on the Nexus switch given the VLAN ID."""
- pass
-
- def remove_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
- nexus_ports, nexus_ssh_port):
- """Remove vlan from interfaces.
-
- Removes a vlan from interfaces on the Nexus switch given the VLAN ID.
- """
- pass
+++ /dev/null
-# Copyright (c) 2013 Cisco Systems Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import mock
-from oslo.config import cfg
-
-from neutron.plugins.cisco.common import config as cisco_config
-from neutron.tests import base
-
-
-class TestCiscoNexusPluginConfig(base.BaseTestCase):
-
- def setUp(self):
- # Point neutron config file to: neutron/tests/etc/neutron.conf.test
- self.config_parse()
-
- super(TestCiscoNexusPluginConfig, self).setUp()
-
- def test_config_parse_error(self):
- """Check that config error is raised upon config parser failure."""
- with mock.patch.object(cfg, 'MultiConfigParser') as parser:
- parser.return_value.read.return_value = []
- self.assertRaises(cfg.Error, cisco_config.CiscoConfigOptions)
-
- def test_create_device_dictionary(self):
- """Test creation of the device dictionary based on nexus config."""
- test_config = {
- 'NEXUS_SWITCH:1.1.1.1': {
- 'username': ['admin'],
- 'password': ['mySecretPassword'],
- 'ssh_port': [22],
- 'compute1': ['1/1'],
- 'compute2': ['1/2'],
- },
- 'NEXUS_SWITCH:2.2.2.2': {
- 'username': ['admin'],
- 'password': ['mySecretPassword'],
- 'ssh_port': [22],
- 'compute3': ['1/1'],
- 'compute4': ['1/2'],
- },
- }
- expected_dev_dict = {
- ('NEXUS_SWITCH', '1.1.1.1', 'username'): 'admin',
- ('NEXUS_SWITCH', '1.1.1.1', 'password'): 'mySecretPassword',
- ('NEXUS_SWITCH', '1.1.1.1', 'ssh_port'): 22,
- ('NEXUS_SWITCH', '1.1.1.1', 'compute1'): '1/1',
- ('NEXUS_SWITCH', '1.1.1.1', 'compute2'): '1/2',
- ('NEXUS_SWITCH', '2.2.2.2', 'username'): 'admin',
- ('NEXUS_SWITCH', '2.2.2.2', 'password'): 'mySecretPassword',
- ('NEXUS_SWITCH', '2.2.2.2', 'ssh_port'): 22,
- ('NEXUS_SWITCH', '2.2.2.2', 'compute3'): '1/1',
- ('NEXUS_SWITCH', '2.2.2.2', 'compute4'): '1/2',
- }
- with mock.patch.object(cfg, 'MultiConfigParser') as parser:
- parser.return_value.read.return_value = cfg.CONF.config_file
- parser.return_value.parsed = [test_config]
- cisco_config.CiscoConfigOptions()
- self.assertEqual(cisco_config.device_dictionary,
- expected_dev_dict)
+++ /dev/null
-# Copyright (c) 2012 OpenStack Foundation.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import contextlib
-import copy
-import inspect
-import logging as std_logging
-import mock
-
-import six
-import webob.exc as wexc
-
-from neutron.api import extensions
-from neutron.api.v2 import attributes
-from neutron.api.v2 import base
-from neutron.common import exceptions as n_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 portbindings
-from neutron.extensions import providernet as provider
-from neutron import manager
-from neutron.openstack.common import gettextutils
-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.common import config as cisco_config
-from neutron.plugins.cisco.db import network_db_v2
-from neutron.plugins.cisco.db import nexus_db_v2
-from neutron.plugins.cisco.models import virt_phy_sw_v2
-from neutron.plugins.openvswitch.common import config as ovs_config
-from neutron.plugins.openvswitch import ovs_db_v2
-from neutron.tests.unit import _test_extension_portbindings as test_bindings
-from neutron.tests.unit import test_db_plugin
-from neutron.tests.unit import test_extensions
-
-LOG = logging.getLogger(__name__)
-CORE_PLUGIN = 'neutron.plugins.cisco.network_plugin.PluginV2'
-NEXUS_PLUGIN = 'neutron.plugins.cisco.nexus.cisco_nexus_plugin_v2.NexusPlugin'
-NEXUS_DRIVER = ('neutron.plugins.cisco.nexus.'
- 'cisco_nexus_network_driver_v2.CiscoNEXUSDriver')
-PHYS_NET = 'physnet1'
-BRIDGE_NAME = 'br-eth1'
-VLAN_START = 1000
-VLAN_END = 1100
-COMP_HOST_NAME = 'testhost'
-COMP_HOST_NAME_2 = 'testhost_2'
-NEXUS_IP_ADDR = '1.1.1.1'
-NEXUS_DEV_ID = 'NEXUS_SWITCH'
-NEXUS_USERNAME = 'admin'
-NEXUS_PASSWORD = 'mySecretPassword'
-NEXUS_SSH_PORT = 22
-NEXUS_INTERFACE = '1/1'
-NEXUS_INTERFACE_2 = '1/2'
-NEXUS_PORT_1 = 'ethernet:1/1'
-NEXUS_PORT_2 = 'ethernet:1/2'
-NETWORK_NAME = 'test_network'
-CIDR_1 = '10.0.0.0/24'
-CIDR_2 = '10.0.1.0/24'
-DEVICE_ID_1 = '11111111-1111-1111-1111-111111111111'
-DEVICE_ID_2 = '22222222-2222-2222-2222-222222222222'
-DEVICE_OWNER = 'compute:None'
-
-
-class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
-
- def setUp(self):
- """Configure for end-to-end neutron testing using a mock ncclient.
-
- This setup includes:
- - Configure the OVS plugin to use VLANs in the range of
- VLAN_START-VLAN_END.
- - Configure the Cisco plugin model to use the Nexus driver.
- - Configure the Nexus driver to use an imaginary switch
- at NEXUS_IP_ADDR.
-
- """
- # Configure the OVS and Cisco plugins
- phys_bridge = ':'.join([PHYS_NET, BRIDGE_NAME])
- phys_vlan_range = ':'.join([PHYS_NET, str(VLAN_START), str(VLAN_END)])
- config = {
- ovs_config: {
- 'OVS': {'bridge_mappings': phys_bridge,
- 'network_vlan_ranges': [phys_vlan_range],
- 'tenant_network_type': 'vlan'}
- },
- cisco_config: {
- 'CISCO': {'nexus_driver': NEXUS_DRIVER},
- 'CISCO_PLUGINS': {'nexus_plugin': NEXUS_PLUGIN},
- }
- }
- for module in config:
- for group in config[module]:
- for opt, val in config[module][group].items():
- module.cfg.CONF.set_override(opt, val, group)
-
- # Configure the Nexus switch dictionary
- # TODO(Henry): add tests for other devices
- nexus_config = {
- (NEXUS_DEV_ID, NEXUS_IP_ADDR, 'username'): NEXUS_USERNAME,
- (NEXUS_DEV_ID, NEXUS_IP_ADDR, 'password'): NEXUS_PASSWORD,
- (NEXUS_DEV_ID, NEXUS_IP_ADDR, 'ssh_port'): NEXUS_SSH_PORT,
- (NEXUS_DEV_ID, NEXUS_IP_ADDR, COMP_HOST_NAME): NEXUS_INTERFACE,
- (NEXUS_DEV_ID, NEXUS_IP_ADDR, COMP_HOST_NAME_2): NEXUS_INTERFACE_2,
- }
- nexus_patch = mock.patch.dict(cisco_config.device_dictionary,
- nexus_config)
- nexus_patch.start()
- self.addCleanup(nexus_patch.stop)
-
- # Use a mock netconf client
- self.mock_ncclient = mock.Mock()
- ncclient_patch = mock.patch.dict('sys.modules',
- {'ncclient': self.mock_ncclient})
- ncclient_patch.start()
- self.addCleanup(ncclient_patch.stop)
-
- # Call the parent setUp, start the core plugin
- super(CiscoNetworkPluginV2TestCase, self).setUp(CORE_PLUGIN)
- self.port_create_status = 'DOWN'
-
- # Set Cisco config module's first configured Nexus IP address.
- # Used for SVI placement when round-robin placement is disabled.
- mock.patch.object(cisco_config, 'first_device_ip',
- new=NEXUS_IP_ADDR).start()
-
- def _get_plugin_ref(self):
- return getattr(manager.NeutronManager.get_plugin(),
- "_model")._plugins[const.VSWITCH_PLUGIN]
-
- @contextlib.contextmanager
- def _patch_ncclient(self, attr, value):
- """Configure an attribute on the mock ncclient module.
-
- This method can be used to inject errors by setting a side effect
- or a return value for an ncclient method.
-
- :param attr: ncclient attribute (typically method) to be configured.
- :param value: Value to be configured on the attribute.
-
- """
- # Configure attribute.
- config = {attr: value}
- self.mock_ncclient.configure_mock(**config)
- # Continue testing
- yield
- # Unconfigure attribute
- config = {attr: None}
- self.mock_ncclient.configure_mock(**config)
-
- @staticmethod
- def _config_dependent_side_effect(match_config, exc):
- """Generates a config-dependent side effect for ncclient edit_config.
-
- This method generates a mock side-effect function which can be
- configured on the mock ncclient module for the edit_config method.
- This side effect will cause a given exception to be raised whenever
- the XML config string that is passed to edit_config contains all
- words in a given match config string.
-
- :param match_config: String containing keywords to be matched
- :param exc: Exception to be raised when match is found
- :return: Side effect function for the mock ncclient module's
- edit_config method.
-
- """
- keywords = match_config.split()
-
- def _side_effect_function(target, config):
- if all(word in config for word in keywords):
- raise exc
- return _side_effect_function
-
- def _is_in_nexus_cfg(self, words):
- """Check if any config sent to Nexus contains all words in a list."""
- for call in (self.mock_ncclient.manager.connect.return_value.
- edit_config.mock_calls):
- configlet = call[2]['config']
- if all(word in configlet for word in words):
- return True
- return False
-
- def _is_in_last_nexus_cfg(self, words):
- """Check if last config sent to Nexus contains all words in a list."""
- last_cfg = (self.mock_ncclient.manager.connect.return_value.
- edit_config.mock_calls[-1][2]['config'])
- return all(word in last_cfg for word in words)
-
- def _is_vlan_configured(self, vlan_creation_expected=True,
- add_keyword_expected=False):
- vlan_created = self._is_in_nexus_cfg(['vlan', 'vlan-name'])
- add_appears = self._is_in_last_nexus_cfg(['add'])
- return (self._is_in_last_nexus_cfg(['allowed', 'vlan']) and
- vlan_created == vlan_creation_expected and
- add_appears == add_keyword_expected)
-
- def _is_vlan_unconfigured(self, vlan_deletion_expected=True,
- vlan_untrunk_expected=True):
- vlan_deleted = self._is_in_nexus_cfg(
- ['no', 'vlan', 'vlan-id-create-delete'])
- vlan_untrunked = self._is_in_nexus_cfg(['allowed', 'vlan', 'remove'])
- return (vlan_deleted == vlan_deletion_expected and
- vlan_untrunked == vlan_untrunk_expected)
-
- def _assertExpectedHTTP(self, status, exc):
- """Confirm that an HTTP status corresponds to an expected exception.
-
- Confirm that an HTTP status which has been returned for an
- neutron API request matches the HTTP status corresponding
- to an expected exception.
-
- :param status: HTTP status
- :param exc: Expected exception
-
- """
- if exc in base.FAULT_MAP:
- expected_http = base.FAULT_MAP[exc].code
- else:
- expected_http = wexc.HTTPInternalServerError.code
- self.assertEqual(status, expected_http)
-
-
-class TestCiscoGetAttribute(CiscoNetworkPluginV2TestCase):
-
- def test_get_unsupported_attr_in_lazy_gettext_mode(self):
- """Test get of unsupported attribute in lazy gettext mode.
-
- This test also checks that this operation does not cause
- excessive nesting of calls to deepcopy.
- """
- plugin = manager.NeutronManager.get_plugin()
-
- def _lazy_gettext(msg):
- return gettextutils.Message(msg, domain='neutron')
-
- with mock.patch.dict(six.moves.builtins.__dict__,
- {'_': _lazy_gettext}):
- self.nesting_count = 0
-
- def _count_nesting(*args, **kwargs):
- self.nesting_count += 1
-
- with mock.patch.object(copy, 'deepcopy',
- side_effect=_count_nesting,
- wraps=copy.deepcopy):
- self.assertRaises(AttributeError, getattr, plugin,
- 'an_unsupported_attribute')
- # If there were no nested calls to deepcopy, then the total
- # number of calls to deepcopy should be 2 (1 call for
- # each mod'd field in the AttributeError message raised
- # by the plugin).
- self.assertEqual(self.nesting_count, 2)
-
-
-class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase,
- test_db_plugin.TestBasicGet):
- pass
-
-
-class TestCiscoV2HTTPResponse(CiscoNetworkPluginV2TestCase,
- test_db_plugin.TestV2HTTPResponse):
- pass
-
-
-class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
- test_db_plugin.TestPortsV2,
- test_bindings.PortBindingsHostTestCaseMixin):
-
- @contextlib.contextmanager
- def _create_port_res(self, name=NETWORK_NAME, cidr=CIDR_1,
- do_delete=True, host_id=COMP_HOST_NAME):
- """Create a network, subnet, and port and yield the result.
-
- Create a network, subnet, and port, yield the result,
- then delete the port, subnet, and network.
-
- :param name: Name of network to be created
- :param cidr: cidr address of subnetwork to be created
- :param do_delete: If set to True, delete the port at the
- end of testing
- :param host_id: Name of compute host to use for testing
-
- """
- ctx = context.get_admin_context()
- with self.network(name=name) as network:
- with self.subnet(network=network, cidr=cidr) as subnet:
- net_id = subnet['subnet']['network_id']
- args = (portbindings.HOST_ID, 'device_id', 'device_owner')
- port_dict = {portbindings.HOST_ID: host_id,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}
- res = self._create_port(self.fmt, net_id, arg_list=args,
- context=ctx, **port_dict)
- port = self.deserialize(self.fmt, res)
- yield res
- if do_delete:
- self._delete('ports', port['port']['id'])
- self._delete('subnets', subnet['subnet']['id'])
- self._delete('networks', network['network']['id'])
-
- def test_create_ports_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- #ensures the API choose the emulation code path
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_port
- with mock.patch.object(plugin_ref,
- 'create_port') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_port_bulk(self.fmt, 2,
- net['network']['id'],
- 'test',
- True)
- # Expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'ports',
- wexc.HTTPInternalServerError.code)
-
- def test_create_ports_bulk_native(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
-
- def test_create_ports_bulk_emulated(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
-
- def test_create_ports_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk port create")
- ctx = context.get_admin_context()
- with self.network() as net:
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_port
- with mock.patch.object(plugin_ref,
- 'create_port') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- res = self._create_port_bulk(self.fmt, 2,
- net['network']['id'],
- 'test', True, context=ctx)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'ports',
- wexc.HTTPInternalServerError.code)
-
- def test_nexus_enable_vlan_cmd(self):
- """Verify the syntax of the command to enable a vlan on an intf."""
-
- # First vlan should be configured without 'add' keyword
- with self._create_port_res(name='net1', cidr=CIDR_1):
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- add_keyword_expected=False))
- self.mock_ncclient.reset_mock()
-
- # Second vlan should be configured with 'add' keyword
- with self._create_port_res(name='net2', cidr=CIDR_2):
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=True,
- add_keyword_expected=True))
-
- def test_nexus_vlan_config_two_hosts(self):
- """Verify config/unconfig of vlan on two compute hosts."""
-
- @contextlib.contextmanager
- def _create_port_check_vlan(comp_host_name, device_id,
- vlan_creation_expected=True):
- arg_list = (portbindings.HOST_ID,)
- port_dict = {portbindings.HOST_ID: comp_host_name,
- 'device_id': device_id,
- 'device_owner': DEVICE_OWNER}
- with self.port(subnet=subnet, fmt=self.fmt,
- arg_list=arg_list, **port_dict) as port:
- self.assertTrue(self._is_vlan_configured(
- vlan_creation_expected=vlan_creation_expected,
- add_keyword_expected=False))
- self.mock_ncclient.reset_mock()
- yield
- self._delete('ports', port['port']['id'])
-
- # Create network and subnet
- with self.network(name=NETWORK_NAME) as network:
- with self.subnet(network=network, cidr=CIDR_1) as subnet:
-
- # Create an instance on first compute host
- with _create_port_check_vlan(
- COMP_HOST_NAME, DEVICE_ID_1, vlan_creation_expected=True):
-
- # Create an instance on second compute host
- with _create_port_check_vlan(
- COMP_HOST_NAME_2, DEVICE_ID_2,
- vlan_creation_expected=False):
- pass
-
- # Instance on second host is now terminated.
- # Vlan should be untrunked from port, but vlan should
- # still exist on the switch.
- self.assertTrue(self._is_vlan_unconfigured(
- vlan_deletion_expected=False))
- self.mock_ncclient.reset_mock()
-
- # Instance on first host is now terminated.
- # Vlan should be untrunked from port and vlan should have
- # been deleted from the switch.
- self.assertTrue(self._is_vlan_unconfigured(
- vlan_deletion_expected=True))
-
- def test_nexus_connect_fail(self):
- """Test failure to connect to a Nexus switch.
-
- While creating a network, subnet, and port, simulate a connection
- failure to a nexus switch. Confirm that the expected HTTP code
- is returned for the create port operation.
-
- """
- with self._patch_ncclient('manager.connect.side_effect',
- AttributeError):
- with self._create_port_res(do_delete=False) as res:
- self._assertExpectedHTTP(res.status_int,
- c_exc.NexusConnectFailed)
-
- def test_nexus_config_fail(self):
- """Test a Nexus switch configuration failure.
-
- While creating a network, subnet, and port, simulate a nexus
- switch configuration error. Confirm that the expected HTTP code
- is returned for the create port operation.
-
- """
- with self._patch_ncclient(
- 'manager.connect.return_value.edit_config.side_effect',
- AttributeError):
- with self._create_port_res(do_delete=False) as res:
- self._assertExpectedHTTP(res.status_int,
- c_exc.NexusConfigFailed)
-
- def test_nexus_extended_vlan_range_failure(self):
- """Test that extended VLAN range config errors are ignored.
-
- Some versions of Nexus switch do not allow state changes for
- the extended VLAN range (1006-4094), but these errors can be
- ignored (default values are appropriate). Test that such errors
- are ignored by the Nexus plugin.
-
- """
- config_err_strings = {
- "state active": "Can't modify state for extended",
- "no shutdown": "Command is only allowed on VLAN",
- }
- for config, err_string in config_err_strings.items():
- with self._patch_ncclient(
- 'manager.connect.return_value.edit_config.side_effect',
- self._config_dependent_side_effect(config,
- Exception(err_string))):
- with self._create_port_res() as res:
- self.assertEqual(res.status_int, wexc.HTTPCreated.code)
-
- def test_nexus_vlan_config_rollback(self):
- """Test rollback following Nexus VLAN state config failure.
-
- Test that the Cisco Nexus plugin correctly deletes the VLAN
- on the Nexus switch when the 'state active' command fails (for
- a reason other than state configuration change is rejected
- for the extended VLAN range).
-
- """
- vlan_state_configs = ['state active', 'no shutdown']
- for config in vlan_state_configs:
- with self._patch_ncclient(
- 'manager.connect.return_value.edit_config.side_effect',
- self._config_dependent_side_effect(config, ValueError)):
- with self._create_port_res(do_delete=False) as res:
- # Confirm that the last configuration sent to the Nexus
- # switch was deletion of the VLAN.
- self.assertTrue(
- self._is_in_last_nexus_cfg(['<no>', '<vlan>'])
- )
- self._assertExpectedHTTP(res.status_int,
- c_exc.NexusConfigFailed)
-
- def test_get_seg_id_fail(self):
- """Test handling of a NetworkSegmentIDNotFound exception.
-
- Test the Cisco NetworkSegmentIDNotFound exception by simulating
- a return of None by the OVS DB get_network_binding method
- during port creation.
-
- """
- orig = ovs_db_v2.get_network_binding
-
- def _return_none_if_nexus_caller(self, *args, **kwargs):
- def _calling_func_name(offset=0):
- """Get name of the calling function 'offset' frames back."""
- return inspect.stack()[1 + offset][3]
- if (_calling_func_name(1) == '_get_segmentation_id' and
- _calling_func_name(2) == '_invoke_nexus_for_net_create'):
- return None
- else:
- return orig(self, *args, **kwargs)
-
- with mock.patch.object(ovs_db_v2, 'get_network_binding',
- new=_return_none_if_nexus_caller):
- with self._create_port_res(do_delete=False) as res:
- self._assertExpectedHTTP(res.status_int,
- c_exc.NetworkSegmentIDNotFound)
-
- def test_nexus_host_non_configured(self):
- """Test handling of a NexusComputeHostNotConfigured exception.
-
- Test the Cisco NexusComputeHostNotConfigured exception by using
- a fictitious host name during port creation.
-
- """
- with self._create_port_res(do_delete=False,
- host_id='fakehost') as res:
- self._assertExpectedHTTP(res.status_int,
- c_exc.NexusComputeHostNotConfigured)
-
- def _check_rollback_on_bind_failure(self,
- vlan_deletion_expected,
- vlan_untrunk_expected):
- """Test for proper rollback following add Nexus DB binding failure.
-
- Test that the Cisco Nexus plugin correctly rolls back the vlan
- configuration on the Nexus switch when add_nexusport_binding fails
- within the plugin's create_port() method.
-
- """
- inserted_exc = KeyError
- with mock.patch.object(nexus_db_v2, 'add_nexusport_binding',
- side_effect=inserted_exc):
- with self._create_port_res(do_delete=False) as res:
- # Confirm that the configuration sent to the Nexus
- # switch includes deletion of the vlan (if expected)
- # and untrunking of the vlan from the ethernet interface
- # (if expected).
- self.assertTrue(self._is_vlan_unconfigured(
- vlan_deletion_expected=vlan_deletion_expected,
- vlan_untrunk_expected=vlan_untrunk_expected))
- self._assertExpectedHTTP(res.status_int, inserted_exc)
-
- def test_nexus_rollback_on_bind_failure_non_provider_vlan(self):
- """Test rollback upon DB binding failure for non-provider vlan."""
- self._check_rollback_on_bind_failure(vlan_deletion_expected=True,
- vlan_untrunk_expected=True)
-
- def test_nexus_rollback_on_bind_failure_prov_vlan_no_auto_create(self):
- """Test rollback on bind fail for prov vlan w auto-create disabled."""
- with mock.patch.object(network_db_v2, 'is_provider_vlan',
- return_value=True):
- # Disable auto-create. This config change will be cleared based
- # on cleanup scheduled in the CiscoNetworkPluginV2TestCase
- # class' setUp() method.
- cisco_config.CONF.set_override('provider_vlan_auto_create',
- False, 'CISCO')
- self._check_rollback_on_bind_failure(vlan_deletion_expected=False,
- vlan_untrunk_expected=True)
-
- def test_nexus_rollback_on_bind_failure_prov_vlan_no_auto_trunk(self):
- """Test rollback on bind fail for prov vlan w auto-trunk disabled."""
- with mock.patch.object(network_db_v2, 'is_provider_vlan',
- return_value=True):
- # Disable auto-trunk. This config change will be cleared
- # based on post-test cleanup scheduled in the
- # CiscoNetworkPluginV2TestCase class' setUp() method.
- cisco_config.CONF.set_override('provider_vlan_auto_trunk',
- False, 'CISCO')
- self._check_rollback_on_bind_failure(vlan_deletion_expected=True,
- vlan_untrunk_expected=False)
-
- def test_model_update_port_rollback(self):
- """Test for proper rollback for Cisco model layer update port failure.
-
- Test that the vSwitch plugin port configuration is rolled back
- (restored) by the Cisco plugin model layer when there is a
- failure in the Nexus sub-plugin for an update port operation.
-
- The update port operation simulates a port attachment scenario:
- first a port is created with no instance (null device_id),
- and then a port update is requested with a non-null device_id
- to simulate the port attachment.
-
- """
- with self.port(fmt=self.fmt, device_id='',
- device_owner=DEVICE_OWNER) as orig_port:
-
- inserted_exc = ValueError
- with mock.patch.object(
- virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_nexus_for_net_create',
- side_effect=inserted_exc):
-
- # Send an update port request including a non-null device ID
- data = {'port': {'device_id': DEVICE_ID_2,
- 'device_owner': DEVICE_OWNER,
- portbindings.HOST_ID: COMP_HOST_NAME}}
- port_id = orig_port['port']['id']
- req = self.new_update_request('ports', data, port_id)
- res = req.get_response(self.api)
-
- # Sanity check failure result code
- self._assertExpectedHTTP(res.status_int, inserted_exc)
-
- # Check that the port still has the original device ID
- plugin = base_plugin.NeutronDbPluginV2()
- ctx = context.get_admin_context()
- db_port = plugin._get_port(ctx, port_id)
- self.assertEqual(db_port['device_id'],
- orig_port['port']['device_id'])
-
- def test_model_delete_port_rollback(self):
- """Test for proper rollback for OVS plugin delete port failure.
-
- Test that the nexus port configuration is rolled back (restored)
- by the Cisco model plugin when there is a failure in the OVS
- plugin for a delete port operation.
-
- """
- with self._create_port_res() as res:
-
- # After port is created, we should have one binding for this
- # vlan/nexus switch.
- port = self.deserialize(self.fmt, res)
- start_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(len(start_rows), 1)
-
- # Inject an exception in the OVS plugin delete_port
- # processing, and attempt a port deletion.
- inserted_exc = n_exc.Conflict
- expected_http = base.FAULT_MAP[inserted_exc].code
- with mock.patch.object(l3_db.L3_NAT_db_mixin,
- 'disassociate_floatingips',
- side_effect=inserted_exc):
- self._delete('ports', port['port']['id'],
- expected_code=expected_http)
-
- # Confirm that the Cisco model plugin has restored
- # the nexus configuration for this port after deletion failure.
- end_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(start_rows, end_rows)
-
- def test_nexus_delete_port_rollback(self):
- """Test for proper rollback for nexus plugin delete port failure.
-
- Test for rollback (i.e. restoration) of a VLAN entry in the
- nexus database whenever the nexus plugin fails to reconfigure the
- nexus switch during a delete_port operation.
-
- """
- with self._create_port_res() as res:
-
- port = self.deserialize(self.fmt, res)
-
- # Check that there is only one binding in the nexus database
- # for this VLAN/nexus switch.
- start_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(len(start_rows), 1)
-
- # Simulate a Nexus switch configuration error during
- # port deletion.
- with self._patch_ncclient(
- 'manager.connect.return_value.edit_config.side_effect',
- AttributeError):
- self._delete('ports', port['port']['id'],
- base.FAULT_MAP[c_exc.NexusConfigFailed].code)
-
- # Confirm that the binding has been restored (rolled back).
- end_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(start_rows, end_rows)
-
- def test_model_update_port_attach(self):
- """Test the model for update_port in attaching to an instance.
-
- Mock the routines that call into the plugin code, and make sure they
- are called with correct arguments.
-
- """
- with contextlib.nested(
- self.port(),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_plugin_per_device'),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_nexus_for_net_create')
- ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
- data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}}
-
- req = self.new_update_request('ports', data, port['port']['id'])
- # Note, due to mocking out the two model routines, response won't
- # contain any useful data
- req.get_response(self.api)
-
- # Note that call_args_list is used instead of
- # assert_called_once_with which requires exact match of arguments.
- # This is because the mocked routines contain variable number of
- # arguments and/or dynamic objects.
- self.assertEqual(invoke_plugin_per_device.call_count, 1)
- self.assertEqual(
- invoke_plugin_per_device.call_args_list[0][0][0:2],
- (const.VSWITCH_PLUGIN, 'update_port'))
- self.assertEqual(invoke_nexus_for_net_create.call_count, 1)
- self.assertEqual(
- invoke_nexus_for_net_create.call_args_list[0][0][1:],
- (port['port']['tenant_id'], port['port']['network_id'],
- data['port']['device_id'],
- data['port'][portbindings.HOST_ID],))
-
- def test_model_update_port_migrate(self):
- """Test the model for update_port in migrating an instance.
-
- Mock the routines that call into the plugin code, and make sure they
- are called with correct arguments.
-
- """
- arg_list = (portbindings.HOST_ID,)
- data = {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}
-
- with contextlib.nested(
- self.port(arg_list=arg_list, **data),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_plugin_per_device'),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_nexus_for_net_create')
- ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
- data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
- req = self.new_update_request('ports', data, port['port']['id'])
- # Note, due to mocking out the two model routines, response won't
- # contain any useful data
- req.get_response(self.api)
-
- # Note that call_args_list is used instead of
- # assert_called_once_with which requires exact match of arguments.
- # This is because the mocked routines contain variable number of
- # arguments and/or dynamic objects.
- self.assertEqual(invoke_plugin_per_device.call_count, 2)
- self.assertEqual(
- invoke_plugin_per_device.call_args_list[0][0][0:2],
- (const.VSWITCH_PLUGIN, 'update_port'))
- self.assertEqual(
- invoke_plugin_per_device.call_args_list[1][0][0:2],
- (const.NEXUS_PLUGIN, 'delete_port'))
- self.assertEqual(invoke_nexus_for_net_create.call_count, 1)
- self.assertEqual(
- invoke_nexus_for_net_create.call_args_list[0][0][1:],
- (port['port']['tenant_id'], port['port']['network_id'],
- port['port']['device_id'],
- data['port'][portbindings.HOST_ID],))
-
- def test_model_update_port_net_create_not_needed(self):
- """Test the model for update_port when no action is needed.
-
- Mock the routines that call into the plugin code, and make sure that
- VSWITCH plugin is called with correct arguments, while NEXUS plugin is
- not called at all.
-
- """
- arg_list = (portbindings.HOST_ID,)
- data = {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}
-
- with contextlib.nested(
- self.port(arg_list=arg_list, **data),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_plugin_per_device'),
- mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
- '_invoke_nexus_for_net_create')
- ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
- data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}}
- req = self.new_update_request('ports', data, port['port']['id'])
- # Note, due to mocking out the two model routines, response won't
- # contain any useful data
- req.get_response(self.api)
-
- # Note that call_args_list is used instead of
- # assert_called_once_with which requires exact match of arguments.
- # This is because the mocked routines contain variable number of
- # arguments and/or dynamic objects.
- self.assertEqual(invoke_plugin_per_device.call_count, 1)
- self.assertEqual(
- invoke_plugin_per_device.call_args_list[0][0][0:2],
- (const.VSWITCH_PLUGIN, 'update_port'))
- self.assertFalse(invoke_nexus_for_net_create.called)
-
- def verify_portbinding(self, host_id1, host_id2,
- vlan, device_id, binding_port):
- """Verify a port binding entry in the DB is correct."""
- self.assertEqual(host_id1, host_id2)
- pb = nexus_db_v2.get_nexusvm_bindings(vlan, device_id)
- self.assertEqual(len(pb), 1)
- self.assertEqual(pb[0].port_id, binding_port)
- self.assertEqual(pb[0].switch_ip, NEXUS_IP_ADDR)
-
- def test_db_update_port_attach(self):
- """Test DB for update_port in attaching to an instance.
-
- Query DB for the port binding entry corresponding to the search key
- (vlan, device_id), and make sure that it's bound to correct switch port
-
- """
- with self.port() as port:
- data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}}
-
- req = self.new_update_request('ports', data, port['port']['id'])
- res = self.deserialize(self.fmt, req.get_response(self.api))
- ctx = context.get_admin_context()
- net = self._show('networks', res['port']['network_id'],
- neutron_context=ctx)['network']
- self.assertTrue(attributes.is_attr_set(
- net.get(provider.SEGMENTATION_ID)))
- vlan = net[provider.SEGMENTATION_ID]
- self.assertEqual(vlan, VLAN_START)
- self.verify_portbinding(res['port'][portbindings.HOST_ID],
- data['port'][portbindings.HOST_ID],
- vlan,
- data['port']['device_id'],
- NEXUS_PORT_1)
-
- def test_db_update_port_migrate(self):
- """Test DB for update_port in migrating an instance.
-
- Query DB for the port binding entry corresponding to the search key
- (vlan, device_id), and make sure that it's bound to correct switch port
- before and after the migration.
-
- """
- arg_list = (portbindings.HOST_ID,)
- data = {portbindings.HOST_ID: COMP_HOST_NAME,
- 'device_id': DEVICE_ID_1,
- 'device_owner': DEVICE_OWNER}
-
- with self.port(arg_list=arg_list, **data) as port:
- ctx = context.get_admin_context()
- net = self._show('networks', port['port']['network_id'],
- neutron_context=ctx)['network']
- self.assertTrue(attributes.is_attr_set(
- net.get(provider.SEGMENTATION_ID)))
- vlan = net[provider.SEGMENTATION_ID]
- self.assertEqual(vlan, VLAN_START)
- self.verify_portbinding(port['port'][portbindings.HOST_ID],
- data[portbindings.HOST_ID],
- vlan,
- data['device_id'],
- NEXUS_PORT_1)
-
- new_data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
- req = self.new_update_request('ports',
- new_data, port['port']['id'])
- res = self.deserialize(self.fmt, req.get_response(self.api))
- self.verify_portbinding(res['port'][portbindings.HOST_ID],
- new_data['port'][portbindings.HOST_ID],
- vlan,
- data['device_id'],
- NEXUS_PORT_2)
-
- def test_delete_ports_by_device_id_second_call_failure(self):
- plugin_ref = self._get_plugin_ref()
- self._test_delete_ports_by_device_id_second_call_failure(plugin_ref)
-
- def test_delete_ports_ignores_port_not_found(self):
- plugin_ref = self._get_plugin_ref()
- self._test_delete_ports_ignores_port_not_found(plugin_ref)
-
-
-class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
- test_db_plugin.TestNetworksV2):
-
- def test_create_networks_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_network
- #ensures the API choose the emulation code path
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- with mock.patch.object(plugin_ref,
- 'create_network') as patched_plugin:
- def side_effect(*args, **kwargs):
- return self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
- patched_plugin.side_effect = side_effect
- res = self._create_network_bulk(self.fmt, 2, 'test', True)
- LOG.debug('response is %s', res)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'networks',
- wexc.HTTPInternalServerError.code)
-
- def test_create_networks_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk network create")
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_network
- with mock.patch.object(plugin_ref,
- 'create_network') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- return self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- res = self._create_network_bulk(self.fmt, 2, 'test', True)
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'networks',
- wexc.HTTPInternalServerError.code)
-
- @contextlib.contextmanager
- def _provider_vlan_network(self, phys_net, segment_id, net_name):
- provider_attrs = {provider.NETWORK_TYPE: 'vlan',
- provider.PHYSICAL_NETWORK: phys_net,
- provider.SEGMENTATION_ID: segment_id}
- arg_list = tuple(provider_attrs.keys())
- res = self._create_network(self.fmt, net_name, True,
- arg_list=arg_list, **provider_attrs)
- network = self.deserialize(self.fmt, res)['network']
- yield network
- req = self.new_delete_request('networks', network['id'])
- req.get_response(self.api)
-
- def test_create_provider_vlan_network(self):
- with self._provider_vlan_network(PHYS_NET, '1234',
- 'pvnet1') as network:
- expected = [('name', 'pvnet1'),
- ('admin_state_up', True),
- ('status', 'ACTIVE'),
- ('shared', False),
- (provider.NETWORK_TYPE, 'vlan'),
- (provider.PHYSICAL_NETWORK, PHYS_NET),
- (provider.SEGMENTATION_ID, 1234)]
- for k, v in expected:
- self.assertEqual(network[k], v)
- self.assertTrue(network_db_v2.is_provider_network(network['id']))
-
- def test_delete_provider_vlan_network(self):
- with self._provider_vlan_network(PHYS_NET, '1234',
- 'pvnet1') as network:
- network_id = network['id']
- # Provider network should now be deleted
- self.assertFalse(network_db_v2.is_provider_network(network_id))
-
-
-class TestCiscoSubnetsV2(CiscoNetworkPluginV2TestCase,
- test_db_plugin.TestSubnetsV2):
-
- def test_create_subnets_bulk_emulated_plugin_failure(self):
- real_has_attr = hasattr
-
- #ensures the API choose the emulation code path
- def fakehasattr(item, attr):
- if attr.endswith('__native_bulk_support'):
- return False
- return real_has_attr(item, attr)
-
- with mock.patch('__builtin__.hasattr',
- new=fakehasattr):
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_subnet
- with mock.patch.object(plugin_ref,
- 'create_subnet') as patched_plugin:
-
- def side_effect(*args, **kwargs):
- self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_subnet_bulk(self.fmt, 2,
- net['network']['id'],
- 'test')
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'subnets',
- wexc.HTTPInternalServerError.code)
-
- def test_create_subnets_bulk_native_plugin_failure(self):
- if self._skip_native_bulk:
- self.skipTest("Plugin does not support native bulk subnet create")
- plugin_ref = self._get_plugin_ref()
- orig = plugin_ref.create_subnet
- with mock.patch.object(plugin_ref,
- 'create_subnet') as patched_plugin:
- def side_effect(*args, **kwargs):
- return self._do_side_effect(patched_plugin, orig,
- *args, **kwargs)
-
- patched_plugin.side_effect = side_effect
- with self.network() as net:
- res = self._create_subnet_bulk(self.fmt, 2,
- net['network']['id'],
- 'test')
-
- # We expect an internal server error as we injected a fault
- self._validate_behavior_on_bulk_failure(
- res,
- 'subnets',
- wexc.HTTPInternalServerError.code)
-
-
-class TestCiscoRouterInterfacesV2(CiscoNetworkPluginV2TestCase):
-
- def setUp(self):
- """Configure a log exception counter and an API extension manager."""
- self.log_exc_count = 0
-
- def _count_exception_logs(*args, **kwargs):
- self.log_exc_count += 1
-
- mock.patch.object(std_logging.LoggerAdapter, 'exception',
- autospec=True,
- side_effect=_count_exception_logs,
- wraps=std_logging.LoggerAdapter.exception).start()
- super(TestCiscoRouterInterfacesV2, self).setUp()
- ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
- self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
-
- @contextlib.contextmanager
- def _network_subnet_router(self):
- """Context mgr for creating/deleting a net, subnet, and router."""
- with self.network() as network:
- with self.subnet(network=network) as subnet:
- data = {'router': {'tenant_id': 'test_tenant_id'}}
- request = self.new_create_request('routers', data, self.fmt)
- response = request.get_response(self.ext_api)
- router = self.deserialize(self.fmt, response)
- yield network, subnet, router
- self._delete('routers', router['router']['id'])
-
- @contextlib.contextmanager
- def _router_interface(self, router, subnet, **kwargs):
- """Create a router interface, yield the response, then delete it."""
- interface_data = {}
- if subnet:
- interface_data['subnet_id'] = subnet['subnet']['id']
- interface_data.update(kwargs)
- request = self.new_action_request('routers', interface_data,
- router['router']['id'],
- 'add_router_interface')
- response = request.get_response(self.ext_api)
-
- yield response
-
- # If router interface was created successfully, delete it now.
- if response.status_int == wexc.HTTPOk.code:
- request = self.new_action_request('routers', interface_data,
- router['router']['id'],
- 'remove_router_interface')
- request.get_response(self.ext_api)
-
- @contextlib.contextmanager
- def _network_subnet_router_interface(self, **kwargs):
- """Context mgr for create/deleting a net, subnet, router and intf."""
- with self._network_subnet_router() as (network, subnet, router):
- with self._router_interface(router, subnet,
- **kwargs) as response:
- yield response
-
- def test_port_list_filtered_by_router_id(self):
- """Test port list command filtered by router ID."""
- with self._network_subnet_router() as (network, subnet, router):
- with self._router_interface(router, subnet):
- query_params = "device_id=%s" % router['router']['id']
- req = self.new_list_request('ports', self.fmt, query_params)
- res = self.deserialize(self.fmt, req.get_response(self.api))
- self.assertEqual(len(res['ports']), 1)
- self.assertEqual(res['ports'][0]['device_id'],
- router['router']['id'])
- self.assertFalse(self.log_exc_count)
-
- def test_add_remove_router_intf_with_nexus_l3_enabled(self):
- """Verifies proper add/remove intf operation with Nexus L3 enabled.
-
- With 'nexus_l3_enable' configured to True, confirm that a switched
- virtual interface (SVI) is created/deleted on the Nexus switch when
- a virtual router interface is created/deleted.
- """
- cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
- with self._network_subnet_router_interface():
- self.assertTrue(self._is_in_last_nexus_cfg(
- ['interface', 'vlan', 'ip', 'address']))
- # Clear list of calls made to mock ncclient
- self.mock_ncclient.reset()
- # Router interface is now deleted. Confirm that SVI
- # has been deleted from the Nexus switch.
- self.assertTrue(self._is_in_nexus_cfg(['no', 'interface', 'vlan']))
- self.assertTrue(self._is_in_last_nexus_cfg(['no', 'vlan']))
-
- def test_add_remove_router_intf_with_nexus_l3_disabled(self):
- """Verifies proper add/remove intf operation with Nexus L3 disabled.
-
- With 'nexus_l3_enable' configured to False, confirm that no changes
- are made to the Nexus switch running configuration when a virtual
- router interface is created and then deleted.
- """
- cisco_config.CONF.set_override('nexus_l3_enable', False, 'CISCO')
- with self._network_subnet_router_interface():
- self.assertFalse(self.mock_ncclient.manager.connect.
- return_value.edit_config.called)
-
- def test_create_svi_but_subnet_not_specified_exception(self):
- """Tests raising of SubnetNotSpecified exception.
-
- Tests that a SubnetNotSpecified exception is raised when an
- add_router_interface request is made for creating a switch virtual
- interface (SVI), but the request does not specify a subnet.
- """
- cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
- with self._network_subnet_router() as (network, subnet, router):
- with self._router_interface(router, subnet=None) as response:
- self._assertExpectedHTTP(response.status_int,
- c_exc.SubnetNotSpecified)
-
- def test_create_svi_but_port_id_included_exception(self):
- """Tests raising of PortIdForNexusSvi exception.
-
- Tests that a PortIdForNexusSvi exception is raised when an
- add_router_interface request is made for creating a switch virtual
- interface (SVI), but the request includes a virtual port ID.
- """
- cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
- with self._network_subnet_router_interface(
- port_id='my_port_id') as response:
- self._assertExpectedHTTP(response.status_int,
- c_exc.PortIdForNexusSvi)
-
-
-class TestCiscoPortsV2XML(TestCiscoPortsV2):
- fmt = 'xml'
-
-
-class TestCiscoNetworksV2XML(TestCiscoNetworksV2):
- fmt = 'xml'
-
-
-class TestCiscoSubnetsV2XML(TestCiscoSubnetsV2):
- fmt = 'xml'
-
-
-class TestCiscoRouterInterfacesV2XML(TestCiscoRouterInterfacesV2):
- fmt = 'xml'
+++ /dev/null
-# Copyright (c) 2013 OpenStack Foundation
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import collections
-import mock
-import testtools
-
-from neutron.db import api as db
-from neutron.plugins.cisco.common import cisco_exceptions as c_exc
-from neutron.plugins.cisco.common import config
-from neutron.plugins.cisco.db import nexus_db_v2 as nxdb
-from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
-from neutron.tests.unit import testlib_api
-
-
-class CiscoNexusDbTest(testlib_api.SqlTestCase):
-
- """Unit tests for cisco.db.nexus_models_v2.NexusPortBinding model."""
-
- NpbObj = collections.namedtuple('NpbObj', 'port vlan switch instance')
-
- def setUp(self):
- super(CiscoNexusDbTest, self).setUp()
- self.session = db.get_session()
-
- def _npb_test_obj(self, pnum, vnum, switch=None, instance=None):
- """Create a Nexus port binding test object from a pair of numbers."""
- if pnum is 'router':
- port = pnum
- else:
- port = '1/%s' % str(pnum)
- vlan = str(vnum)
- if switch is None:
- switch = '10.9.8.7'
- if instance is None:
- instance = 'instance_%s_%s' % (str(pnum), str(vnum))
- return self.NpbObj(port, vlan, switch, instance)
-
- def _assert_equal(self, npb, npb_obj):
- self.assertEqual(npb.port_id, npb_obj.port)
- self.assertEqual(int(npb.vlan_id), int(npb_obj.vlan))
- self.assertEqual(npb.switch_ip, npb_obj.switch)
- self.assertEqual(npb.instance_id, npb_obj.instance)
-
- def _add_to_db(self, npbs):
- for npb in npbs:
- nxdb.add_nexusport_binding(
- npb.port, npb.vlan, npb.switch, npb.instance)
-
- def test_nexusportbinding_add_remove(self):
- npb11 = self._npb_test_obj(10, 100)
- npb = nxdb.add_nexusport_binding(
- npb11.port, npb11.vlan, npb11.switch, npb11.instance)
- self._assert_equal(npb, npb11)
- npb = nxdb.remove_nexusport_binding(
- npb11.port, npb11.vlan, npb11.switch, npb11.instance)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb11)
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.remove_nexusport_binding(
- npb11.port, npb11.vlan, npb11.switch, npb11.instance)
-
- def test_nexusportbinding_get(self):
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_to_db([npb11, npb21, npb22])
-
- npb = nxdb.get_nexusport_binding(
- npb11.port, npb11.vlan, npb11.switch, npb11.instance)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb11)
- npb = nxdb.get_nexusport_binding(
- npb21.port, npb21.vlan, npb21.switch, npb21.instance)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb21)
- npb = nxdb.get_nexusport_binding(
- npb22.port, npb22.vlan, npb22.switch, npb22.instance)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb22)
-
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.get_nexusport_binding(
- npb21.port, npb21.vlan, npb21.switch, "dummyInstance")
-
- def test_nexusvlanbinding_get(self):
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_to_db([npb11, npb21, npb22])
-
- npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, npb11.switch)
- self.assertEqual(len(npb_all_v100), 2)
- npb_v200 = nxdb.get_nexusvlan_binding(npb22.vlan, npb22.switch)
- self.assertEqual(len(npb_v200), 1)
- self._assert_equal(npb_v200[0], npb22)
-
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.get_nexusvlan_binding(npb21.vlan, "dummySwitch")
-
- def test_nexusvmbinding_get(self):
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- npb22 = self._npb_test_obj(20, 200)
- self._add_to_db([npb11, npb21, npb22])
-
- npb = nxdb.get_nexusvm_bindings(npb21.vlan, npb21.instance)[0]
- self._assert_equal(npb, npb21)
- npb = nxdb.get_nexusvm_bindings(npb22.vlan, npb22.instance)[0]
- self._assert_equal(npb, npb22)
-
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.get_nexusvm_bindings(npb21.vlan, "dummyInstance")
-
- def test_nexusportvlanswitchbinding_get(self):
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100)
- self._add_to_db([npb11, npb21])
-
- npb = nxdb.get_port_vlan_switch_binding(
- npb11.port, npb11.vlan, npb11.switch)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb11)
-
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.get_port_vlan_switch_binding(
- npb21.port, npb21.vlan, "dummySwitch")
-
- def test_nexusportswitchbinding_get(self):
- npb11 = self._npb_test_obj(10, 100)
- npb21 = self._npb_test_obj(20, 100, switch='2.2.2.2')
- npb22 = self._npb_test_obj(20, 200, switch='2.2.2.2')
- self._add_to_db([npb11, npb21, npb22])
-
- npb = nxdb.get_port_switch_bindings(npb11.port, npb11.switch)
- self.assertEqual(len(npb), 1)
- self._assert_equal(npb[0], npb11)
- npb_all_p20 = nxdb.get_port_switch_bindings(npb21.port, npb21.switch)
- self.assertEqual(len(npb_all_p20), 2)
-
- npb = nxdb.get_port_switch_bindings(npb21.port, "dummySwitch")
- self.assertIsNone(npb)
-
- def test_nexussvibinding_get(self):
- npbr1 = self._npb_test_obj('router', 100)
- npb21 = self._npb_test_obj(20, 100)
- self._add_to_db([npbr1, npb21])
-
- npb_svi = nxdb.get_nexussvi_bindings()
- self.assertEqual(len(npb_svi), 1)
- self._assert_equal(npb_svi[0], npbr1)
-
- npbr2 = self._npb_test_obj('router', 200)
- self._add_to_db([npbr2])
- npb_svi = nxdb.get_nexussvi_bindings()
- self.assertEqual(len(npb_svi), 2)
-
- def test_nexussviswitch_find(self):
- """Test Nexus switch selection for SVI placement."""
- # Configure 2 Nexus switches
- nexus_switches = {
- ('1.1.1.1', 'username'): 'admin',
- ('1.1.1.1', 'password'): 'password1',
- ('1.1.1.1', 'host1'): '1/1',
- ('2.2.2.2', 'username'): 'admin',
- ('2.2.2.2', 'password'): 'password2',
- ('2.2.2.2', 'host2'): '1/1',
- }
- nexus_plugin = cisco_nexus_plugin_v2.NexusPlugin()
- nexus_plugin._client = mock.Mock()
- nexus_plugin._client.nexus_switches = nexus_switches
-
- # Set the Cisco config module's first configured device IP address
- # according to the preceding switch config
- with mock.patch.object(config, 'first_device_ip', new='1.1.1.1'):
-
- # Enable round-robin mode with no SVIs configured on any of the
- # Nexus switches (i.e. no entries in the SVI database). The
- # plugin should select the first switch in the configuration.
- config.CONF.set_override('svi_round_robin', True, 'CISCO')
- switch_ip = nexus_plugin._find_switch_for_svi()
- self.assertEqual(switch_ip, '1.1.1.1')
-
- # Keep round-robin mode enabled, and add entries to the SVI
- # database. The plugin should select the switch with the least
- # number of entries in the SVI database.
- vlan = 100
- npbr11 = self._npb_test_obj('router', vlan, switch='1.1.1.1',
- instance='instance11')
- npbr12 = self._npb_test_obj('router', vlan, switch='1.1.1.1',
- instance='instance12')
- npbr21 = self._npb_test_obj('router', vlan, switch='2.2.2.2',
- instance='instance21')
- self._add_to_db([npbr11, npbr12, npbr21])
- switch_ip = nexus_plugin._find_switch_for_svi()
- self.assertEqual(switch_ip, '2.2.2.2')
-
- # Disable round-robin mode. The plugin should select the
- # first switch in the configuration.
- config.CONF.clear_override('svi_round_robin', 'CISCO')
- switch_ip = nexus_plugin._find_switch_for_svi()
- self.assertEqual(switch_ip, '1.1.1.1')
-
- def test_nexusbinding_update(self):
- npb11 = self._npb_test_obj(10, 100, switch='1.1.1.1', instance='test')
- npb21 = self._npb_test_obj(20, 100, switch='1.1.1.1', instance='test')
- self._add_to_db([npb11, npb21])
-
- npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, '1.1.1.1')
- self.assertEqual(len(npb_all_v100), 2)
-
- npb22 = self._npb_test_obj(20, 200, switch='1.1.1.1', instance='test')
- npb = nxdb.update_nexusport_binding(npb21.port, 200)
- self._assert_equal(npb, npb22)
-
- npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, '1.1.1.1')
- self.assertEqual(len(npb_all_v100), 1)
- self._assert_equal(npb_all_v100[0], npb11)
-
- npb = nxdb.update_nexusport_binding(npb21.port, 0)
- self.assertIsNone(npb)
-
- npb33 = self._npb_test_obj(30, 300, switch='1.1.1.1', instance='test')
- with testtools.ExpectedException(c_exc.NexusPortBindingNotFound):
- nxdb.update_nexusport_binding(npb33.port, 200)
+++ /dev/null
-# Copyright (c) 2012 OpenStack Foundation.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import contextlib
-import mock
-
-from oslo.config import cfg
-
-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.common import config as cisco_config
-from neutron.plugins.cisco.db import network_db_v2 as cdb
-from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
-from neutron.tests.unit import testlib_api
-
-NEXUS_IP_ADDRESS = '1.1.1.1'
-HOSTNAME1 = 'testhost1'
-HOSTNAME2 = 'testhost2'
-HOSTNAME3 = 'testhost3'
-INSTANCE1 = 'testvm1'
-INSTANCE2 = 'testvm2'
-INSTANCE3 = 'testvm3'
-NEXUS_PORT1 = '1/10'
-NEXUS_PORT2 = '1/20'
-NEXUS_PC_IP_ADDRESS = '2.2.2.2'
-NEXUS_PORTCHANNELS = 'portchannel:2'
-PC_HOSTNAME = 'testpchost'
-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(testlib_api.SqlTestCase):
-
- def setUp(self):
- """Set up function."""
- super(TestCiscoNexusPlugin, self).setUp()
- self.tenant_id = "test_tenant_cisco1"
- self.net_name = "test_network_cisco1"
- 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._pchostname = PC_HOSTNAME
-
- 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.attachment3 = {
- const.TENANT_ID: self.second_tenant_id,
- const.INSTANCE_ID: INSTANCE3,
- const.HOST_NAME: HOSTNAME3,
- }
- 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.network3 = {
- const.NET_ID: 8,
- const.NET_NAME: 'vpc_net',
- const.NET_VLAN_NAME: 'q-268',
- const.NET_VLAN_ID: '268',
- }
- self.delete_port_args_1 = [
- self.attachment1[const.INSTANCE_ID],
- self.network1[const.NET_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,
- }
-
- def new_nexus_init(self):
- self._client = importutils.import_object(NEXUS_DRIVER)
- 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,
- (NEXUS_PC_IP_ADDRESS, 'ssh_port'): NEXUS_SSH_PORT,
- }
- self._nexus_switches = {
- ('NEXUS_SWITCH', NEXUS_IP_ADDRESS, HOSTNAME1): NEXUS_PORT1,
- ('NEXUS_SWITCH', NEXUS_IP_ADDRESS, HOSTNAME2): NEXUS_PORT2,
- ('NEXUS_SWITCH', NEXUS_PC_IP_ADDRESS, HOSTNAME3):
- NEXUS_PORTCHANNELS,
- ('NEXUS_SWITCH', NEXUS_PC_IP_ADDRESS, 'ssh_port'):
- NEXUS_SSH_PORT,
- ('NEXUS_SWITCH', NEXUS_IP_ADDRESS, HOSTNAME3):
- NEXUS_PORTCHANNELS,
- ('NEXUS_SWITCH', NEXUS_IP_ADDRESS, 'ssh_port'): NEXUS_SSH_PORT,
- }
- self._client.credentials = {
- NEXUS_IP_ADDRESS: {
- 'username': 'admin',
- 'password': 'pass1234'
- },
- NEXUS_PC_IP_ADDRESS: {
- 'username': 'admin',
- 'password': 'password'
- },
- }
-
- # Use a mock netconf client
- self.mock_ncclient = mock.Mock()
-
- with contextlib.nested(
- mock.patch.dict('sys.modules', {'ncclient': self.mock_ncclient}),
- mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
- '__init__', new=new_nexus_init)
- ):
- self._cisco_nexus_plugin = cisco_nexus_plugin_v2.NexusPlugin()
-
- # Set the Cisco config module's first configured device IP address
- # according to the preceding switch config.
- mock.patch.object(cisco_config, 'first_device_ip',
- new=NEXUS_IP_ADDRESS).start()
-
- def test_create_delete_networks(self):
- """Tests creation of two new Virtual Networks."""
- 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])
-
- expected_instance_id = self._cisco_nexus_plugin.delete_port(
- INSTANCE1, self.vlan_id)
-
- self.assertEqual(expected_instance_id, INSTANCE1)
-
- new_net_dict = self._cisco_nexus_plugin.create_network(
- self.network2, self.attachment1)
- for attr in NET_ATTRS:
- self.assertEqual(new_net_dict[attr], self.network2[attr])
-
- expected_instance_id = self._cisco_nexus_plugin.delete_port(
- INSTANCE1, self.second_vlan_id)
-
- self.assertEqual(expected_instance_id, INSTANCE1)
-
- def _create_delete_providernet(self, auto_create, auto_trunk):
- cfg.CONF.set_override(
- 'provider_vlan_auto_create', auto_create, 'CISCO')
- cfg.CONF.set_override(
- 'provider_vlan_auto_trunk', auto_trunk, 'CISCO')
- with mock.patch.object(cdb, 'is_provider_vlan',
- return_value=True) as mock_db:
- # Create a provider network
- new_net_dict = self._cisco_nexus_plugin.create_network(
- self.providernet, self.attachment1)
- self.assertEqual(mock_db.call_count, 1)
- for attr in NET_ATTRS:
- self.assertEqual(new_net_dict[attr], self.providernet[attr])
- # Delete the provider network
- instance_id = self._cisco_nexus_plugin.delete_port(
- self.attachment1[const.INSTANCE_ID],
- self.providernet[const.NET_VLAN_ID])
- self.assertEqual(instance_id,
- self.attachment1[const.INSTANCE_ID])
-
- def test_create_delete_providernet(self):
- self._create_delete_providernet(auto_create=True, auto_trunk=True)
-
- def test_create_delete_provider_vlan_network_cfg_auto_man(self):
- self._create_delete_providernet(auto_create=True, auto_trunk=False)
-
- def test_create_delete_provider_vlan_network_cfg_man_auto(self):
- self._create_delete_providernet(auto_create=False, auto_trunk=True)
-
- def test_create_delete_provider_vlan_network_cfg_man_man(self):
- self._create_delete_providernet(auto_create=False, auto_trunk=False)
-
- def test_create_delete_network_portchannel(self):
- """Tests creation of a network over a portchannel."""
- new_net_dict = self._cisco_nexus_plugin.create_network(
- self.network3, self.attachment3)
- self.assertEqual(new_net_dict[const.NET_ID],
- self.network3[const.NET_ID])
- self.assertEqual(new_net_dict[const.NET_NAME],
- self.network3[const.NET_NAME])
- self.assertEqual(new_net_dict[const.NET_VLAN_NAME],
- self.network3[const.NET_VLAN_NAME])
- self.assertEqual(new_net_dict[const.NET_VLAN_ID],
- self.network3[const.NET_VLAN_ID])
-
- self._cisco_nexus_plugin.delete_port(
- INSTANCE3, self.network3[const.NET_VLAN_ID]
- )
-
- def _add_router_interface(self):
- """Add a router interface using fixed (canned) parameters."""
- vlan_name = self.vlan_name
- vlan_id = self.vlan_id
- gateway_ip = '10.0.0.1/24'
- router_id = '00000R1'
- subnet_id = '00001'
- return self._cisco_nexus_plugin.add_router_interface(
- vlan_name, vlan_id, subnet_id, gateway_ip, router_id)
-
- def _remove_router_interface(self):
- """Remove a router interface created with _add_router_interface."""
- vlan_id = self.vlan_id
- router_id = '00000R1'
- return self._cisco_nexus_plugin.remove_router_interface(vlan_id,
- router_id)
-
- def test_nexus_add_remove_router_interface(self):
- """Tests addition of a router interface."""
- self.assertTrue(self._add_router_interface())
- self.assertEqual(self._remove_router_interface(), '00000R1')
-
- def test_nexus_dup_add_router_interface(self):
- """Tests a duplicate add of a router interface."""
- self._add_router_interface()
- try:
- self.assertRaises(
- cisco_exc.SubnetInterfacePresent,
- self._add_router_interface)
- finally:
- self._remove_router_interface()
-
- def test_nexus_no_svi_switch_exception(self):
- """Tests failure to find a Nexus switch for SVI placement."""
- # Clear the Nexus switches dictionary.
- with mock.patch.dict(self._cisco_nexus_plugin._client.nexus_switches,
- {}, clear=True):
- # Clear the first Nexus IP address discovered in config
- with mock.patch.object(cisco_config, 'first_device_ip',
- new=None):
- self.assertRaises(cisco_exc.NoNexusSviSwitch,
- self._add_router_interface)
-
- def test_nexus_add_port_after_router_interface(self):
- """Tests creating a port after a router interface.
-
- Test creating a port after an SVI router interface has
- been created. Only a trunk call should be invoked and the
- plugin should not attempt to recreate the vlan.
- """
- self._add_router_interface()
- # Create a network on the switch
- self._cisco_nexus_plugin.create_network(
- self.network1, self.attachment1)
-
- # Grab a list of all mock calls from ncclient
- last_cfgs = (self.mock_ncclient.manager.connect.return_value.
- edit_config.mock_calls)
-
- # The last ncclient call should be for trunking and the second
- # to last call should be creating the SVI interface
- last_cfg = last_cfgs[-1][2]['config']
- self.assertIn('allowed', last_cfg)
-
- slast_cfg = last_cfgs[-2][2]['config']
- self.assertIn('10.0.0.1/24', slast_cfg)
+++ /dev/null
-# Copyright 2014 Cisco Systems, Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-# implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import sys
-
-import mock
-
-from neutron import context
-from neutron.plugins.cisco.common import cisco_constants as const
-from neutron.plugins.cisco.common import config as cisco_config
-from neutron.plugins.cisco.models import virt_phy_sw_v2
-from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
-from neutron.tests.unit import testlib_api
-
-
-class TestCiscoPluginModel(testlib_api.SqlTestCase):
-
- def setUp(self):
- # Point config file to: neutron/tests/etc/neutron.conf.test
- self.config_parse()
-
- super(TestCiscoPluginModel, self).setUp()
-
- def test_non_nexus_device_driver(self):
- """Tests handling of an non-Nexus device driver being configured."""
- with mock.patch.dict(sys.modules, {'mock_driver': mock.Mock()}):
- cisco_config.CONF.set_override('nexus_driver',
- 'mock_driver.Non_Nexus_Driver',
- 'CISCO')
- # Plugin model instance should have is_nexus_plugin set to False
- model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
- self.assertFalse(model.is_nexus_plugin)
-
- # Model's _invoke_nexus_for_net_create should just return False
- user_id = 'user_id'
- tenant_id = 'tenant_id'
- ctx = context.Context(user_id, tenant_id)
- self.assertFalse(model._invoke_nexus_for_net_create(
- ctx, tenant_id, net_id='net_id',
- instance_id='instance_id', host_id='host_id'))
-
- def test_nexus_plugin_calls_ignored_if_plugin_not_loaded(self):
- """Verifies Nexus plugin calls are ignored if plugin is not loaded."""
- cisco_config.CONF.set_override(const.NEXUS_PLUGIN,
- None, 'CISCO_PLUGINS')
- with mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
- 'create_network') as mock_create_network:
- model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
- model._invoke_plugin_per_device(model, const.NEXUS_PLUGIN,
- 'create_network')
- self.assertFalse(mock_create_network.called)
Code Review / openstack-build / neutron-build.git / commitdiff