This commit adds intelligent multiple nexus physical switch support for the Cisco plugin.
The plugin also has been modified to scan for the host when an instance is created and
selectively trunk VLAN's for the port for that host only. It also deletes VLANs from nexus
switches when no longer required.
Implements: blueprint cisco-plugin-enhancements
Change-Id: I6275eb1815310d0d5a8123ca2edbc0a0937718e9
password=<put_password_here>
#Provide the Nexus credentials, if you are using Nexus
-[1.1.1.1]
-username=abc
-password=def
+[<put_nexus_switch_ip_address_here>]
+username=<put_user_name_here>
+password=<put_password_here>
+# Provide credentials and endpoint
+# for keystone here
+[keystone]
+auth_url=<put_keystone_endpoint_here>
+username=<put_user_name_here>
+password=<put_password_here>
[SEGMENTATION]
manager_class=quantum.plugins.cisco.segmentation.l2network_vlan_mgr_v2.L2NetworkVLANMgr
+
+# IMPORTANT: Comment the following lines for production deployments
+[TEST]
+host=testhost
[SWITCH]
-# Change the following to reflect the Nexus switch details
-nexus_ip_address=<put_nexus_switch_ip_address_here>
-#Interfaces connected from the Nexus Switch to the compute hosts ports, e.g.: 1/10 and 1/11
-ports=<put_interfaces_names_here_separated_by_commas>
-#Port number where the SSH will be running at the Nexus Switch, e.g.: 22 (Default)
-nexus_ssh_port=22
+# Ip address of the switch
+[[<put_nexus_switch_ip_address_here>]]
+# Hostname of the node
+[[[<put_hostname_here>]]]
+# Port this node is connected to on the nexus switch
+ports=<put_port_id_here>
+# Port number where the SSH will be running at the Nexus Switch, e.g.: 22 (Default)
+[[[ssh_port]]]
+ssh_port=<put_port_number_here>
[DRIVER]
#name=quantum.plugins.cisco.nexus.cisco_nexus_network_driver_v2.CiscoNEXUSDriver
# 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 logging as LOG
+# @author: Arvind Somya, Cisco Systems, Inc. (asomya@cisco.com)
+#
from sqlalchemy.orm import exc
import quantum.db.api as db
-
+from quantum.openstack.common import log as logging
from quantum.plugins.cisco.common import cisco_exceptions as c_exc
from quantum.plugins.cisco.db import nexus_models_v2
+LOG = logging.getLogger(__name__)
+
+
def get_all_nexusport_bindings():
"""Lists all the nexusport bindings"""
LOG.debug("get_all_nexusport_bindings() called")
return []
-def get_nexusport_binding(vlan_id):
+def get_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
"""Lists a nexusport binding"""
LOG.debug("get_nexusport_binding() called")
session = db.get_session()
try:
binding = (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(vlan_id=vlan_id).all())
+ filter_by(vlan_id=vlan_id).filter_by(switch_ip=switch_ip).
+ filter_by(port_id=port_id).
+ filter_by(instance_id=instance_id).all())
return binding
except exc.NoResultFound:
raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
-def add_nexusport_binding(port_id, vlan_id):
+def get_nexusvlan_binding(vlan_id, switch_ip):
+ """Lists a vlan and switch binding"""
+ LOG.debug("get_nexusvlan_binding() called")
+ session = db.get_session()
+ try:
+ binding = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(vlan_id=vlan_id).filter_by(switch_ip=switch_ip).
+ all())
+ return binding
+ except exc.NoResultFound:
+ raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
+
+
+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, vlan_id)
+ binding = nexus_models_v2.NexusPortBinding(
+ port_id, vlan_id, switch_ip, instance_id)
session.add(binding)
session.flush()
return binding
-def remove_nexusport_binding(vlan_id):
+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()
try:
binding = (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(vlan_id=vlan_id).all())
+ filter_by(vlan_id=vlan_id).filter_by(switch_ip=switch_ip).
+ filter_by(port_id=port_id).
+ filter_by(instance_id=instance_id).all())
+
for bind in binding:
session.delete(bind)
session.flush()
return binding
except exc.NoResultFound:
raise c_exc.NexusPortBindingNotFound()
+
+
+def get_nexusvm_binding(vlan_id, instance_id):
+ """Lists nexusvm bindings"""
+ LOG.debug("get_nexusvm_binding() called")
+ session = db.get_session()
+ try:
+ binding = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(instance_id=instance_id).
+ filter_by(vlan_id=vlan_id).first())
+ return binding
+ except exc.NoResultFound:
+ raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
+
+
+def get_port_vlan_switch_binding(port_id, vlan_id, switch_ip):
+ """Lists nexusvm bindings"""
+ LOG.debug("get_port_vlan_switch_binding() called")
+ session = db.get_session()
+ try:
+ binding = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(port_id=port_id).filter_by(switch_ip=switch_ip).
+ filter_by(vlan_id=vlan_id).all())
+ return binding
+ except exc.NoResultFound:
+ raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
class NexusPortBinding(model_base.BASEV2, L2NetworkBase):
- """Represents a binding of nexus port to vlan_id"""
- __tablename__ = 'nexusport_bindings'
+ """Represents a binding of VM's to nexus ports"""
+ __tablename__ = "nexusport_bindings"
id = Column(Integer, primary_key=True, autoincrement=True)
port_id = Column(String(255))
vlan_id = Column(Integer, nullable=False)
+ switch_ip = Column(String(255))
+ instance_id = Column(String(255))
- def __init__(self, port_id, vlan_id):
+ def __init__(self, port_id, vlan_id, switch_ip, instance_id):
self.port_id = port_id
self.vlan_id = vlan_id
+ self.switch_ip = switch_ip
+ self.instance_id = instance_id
def __repr__(self):
- return "<NexusPortBinding (%s,%d)>" % (self.port_id, self.vlan_id)
+ return "<NexusPortBinding (%s,%d, %s, %s)>" % \
+ (self.port_id, self.vlan_id, self.switch_ip, self.instance_id)
SECTION_CONF = CONF_PARSER_OBJ['MODEL']
MODEL_CLASS = SECTION_CONF['model_class']
+if 'TEST' in CONF_PARSER_OBJ.keys():
+ TEST = CONF_PARSER_OBJ['TEST']
+
CONF_FILE = find_config_file({'plugin': 'cisco'}, "cisco_plugins.ini")
SECTION_CONF = CONF_PARSER_OBJ['SEGMENTATION']
CONF_PARSER_OBJ = confp.CiscoConfigParser(CONF_FILE)
-
# Read the config for the device plugins
PLUGINS = CONF_PARSER_OBJ.walk(CONF_PARSER_OBJ.dummy)
import inspect
import logging
+from keystoneclient.v2_0 import client as keystone_client
+from novaclient.v1_1 import client as nova_client
+
+from quantum.db import l3_db
from quantum.manager import QuantumManager
from quantum.openstack.common import importutils
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.openvswitch import ovs_db_v2 as odb
from quantum import quantum_plugin_base_v2
-
LOG = logging.getLogger(__name__)
_plugins = {}
_inventory = {}
_methods_to_delegate = ['get_network', 'get_networks',
- 'create_port', 'create_port_bulk', 'delete_port',
- 'update_port', 'get_port', 'get_ports',
+ 'create_port_bulk', 'update_port',
+ 'get_port', 'get_ports',
'create_subnet', 'create_subnet_bulk',
- 'delete_subnet', 'update_subnet', 'get_subnet',
- 'get_subnets', ]
+ 'delete_subnet', 'update_subnet',
+ 'get_subnet', 'get_subnets', ]
def __init__(self):
"""
else:
return False
+ def _get_instance_host(self, tenant_id, instance_id):
+ keystone = cred._creds_dictionary['keystone']
+ kc = keystone_client.Client(username=keystone['username'],
+ password=keystone['password'],
+ tenant_id=tenant_id,
+ auth_url=keystone['auth_url'])
+ tenant = kc.tenants.get(tenant_id)
+ tenant_name = tenant.name
+
+ nc = nova_client.Client(keystone['username'],
+ keystone['password'],
+ tenant_name,
+ keystone['auth_url'],
+ no_cache=True)
+ serv = nc.servers.get(instance_id)
+ host = serv.__getattr__('OS-EXT-SRV-ATTR:host')
+
+ return host
+
def create_network(self, context, network):
"""
Perform this operation in the context of the configured device
args = [ovs_output[0]['tenant_id'], ovs_output[0]['name'],
ovs_output[0]['id'], vlan_name, vlan_id,
{'vlan_ids': vlanids}]
- nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
- self._func_name(),
- args)
return ovs_output[0]
except:
# TODO (Sumit): Check if we need to perform any rollback here
LOG.debug("ovs_output: %s\n " % ovs_output)
vlanids = self._get_all_segmentation_ids()
ovs_networks = ovs_output
- for ovs_network in ovs_networks:
- vlan_id = self._get_segmentation_id(ovs_network['id'])
- vlan_name = conf.VLAN_NAME_PREFIX + str(vlan_id)
- args = [ovs_network['tenant_id'], ovs_network['name'],
- ovs_network['id'], vlan_name, vlan_id,
- {'vlan_ids': vlanids}]
- nexus_output = self._invoke_plugin_per_device(
- const.NEXUS_PLUGIN, "create_network", args)
return ovs_output
except:
# TODO (Sumit): Check if we need to perform any rollback here
pass
def create_port(self, context, port):
- """For this model this method will be delegated to vswitch plugin"""
- pass
+ """
+ Perform this operation in the context of the configured device
+ plugins.
+ """
+ LOG.debug("create_port() called\n")
+ try:
+ args = [context, port]
+ ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
+ self._func_name(),
+ args)
+ net_id = port['port']['network_id']
+ instance_id = port['port']['device_id']
+ tenant_id = port['port']['tenant_id']
+
+ net_dict = self.get_network(context, net_id)
+ net_name = net_dict['name']
+
+ vlan_id = self._get_segmentation_id(net_id)
+ host = ''
+ if hasattr(conf, 'TEST'):
+ host = conf.TEST['host']
+ else:
+ host = self._get_instance_host(tenant_id, instance_id)
+
+ # Trunk segmentation id for only this host
+ vlan_name = conf.VLAN_NAME_PREFIX + str(vlan_id)
+ n_args = [tenant_id, net_name, net_id,
+ vlan_name, vlan_id, host, instance_id]
+ nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
+ 'create_network',
+ n_args)
+ return ovs_output[0]
+ except:
+ # TODO (asomya): Check if we need to perform any rollback here
+ raise
def get_port(self, context, id, fields=None):
"""For this model this method will be delegated to vswitch plugin"""
"""For this model this method will be delegated to vswitch plugin"""
pass
- def delete_port(self, context, id, kwargs):
- """For this model this method will be delegated to vswitch plugin"""
- pass
+ def delete_port(self, context, id):
+ """
+ Perform this operation in the context of the configured device
+ plugins.
+ """
+ LOG.debug("delete_port() called\n")
+ try:
+ args = [context, id]
+ port = self.get_port(context, id)
+ vlan_id = self._get_segmentation_id(port['network_id'])
+ n_args = [port['device_id'], vlan_id]
+ ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
+ self._func_name(),
+ args)
+ nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
+ self._func_name(),
+ n_args)
+ return ovs_output[0]
+ except:
+ # TODO (asomya): Check if we need to perform any rollback here
+ raise
def create_subnet(self, context, subnet):
"""For this model this method will be delegated to vswitch plugin"""
CP = confp.CiscoConfigParser(find_config_file({'plugin': 'cisco'},
"nexus.ini"))
-SECTION = CP['SWITCH']
-NEXUS_IP_ADDRESS = SECTION['nexus_ip_address']
-NEXUS_PORTS = SECTION['ports']
-NEXUS_SSH_PORT = SECTION['nexus_ssh_port']
+NEXUS_DETAILS = CP['SWITCH']
SECTION = CP['DRIVER']
NEXUS_DRIVER = SECTION['name']
Implements a Nexus-OS NETCONF over SSHv2 API Client
"""
+import eventlet
import logging
from ncclient import manager
Creates a VLAN and Enable on trunk mode an interface on Nexus Switch
given the VLAN ID and Name and Interface Number
"""
- with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
- nexus_password) as man:
- self.enable_vlan(man, vlan_id, vlan_name)
- if vlan_ids is '':
- vlan_ids = self.build_vlans_cmd()
- LOG.debug("NexusDriver VLAN IDs: %s" % vlan_ids)
- for ports in nexus_ports:
- self.enable_vlan_on_trunk_int(man, ports, vlan_ids)
+ man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
+ nexus_user, nexus_password)
+ self.enable_vlan(man, vlan_id, vlan_name)
+ if vlan_ids is '':
+ vlan_ids = self.build_vlans_cmd()
+ LOG.debug("NexusDriver VLAN IDs: %s" % vlan_ids)
+ for ports in nexus_ports:
+ self.enable_vlan_on_trunk_int(man, ports, vlan_ids)
def delete_vlan(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port):
Delete a VLAN and Disables trunk mode an interface on Nexus Switch
given the VLAN ID and Interface Number
"""
- with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
- nexus_password) as man:
- self.disable_vlan(man, vlan_id)
- for ports in nexus_ports:
- self.disable_vlan_on_trunk_int(man, ports, vlan_id)
+ man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
+ nexus_user, nexus_password)
+ self.disable_vlan(man, vlan_id)
+ for ports in nexus_ports:
+ self.disable_vlan_on_trunk_int(man, ports, vlan_id)
def build_vlans_cmd(self):
"""
"""
Adds a vlan from interfaces on the Nexus switch given the VLAN ID
"""
- with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
- nexus_password) as man:
- if not vlan_ids:
- vlan_ids = self.build_vlans_cmd()
- for ports in nexus_ports:
- self.enable_vlan_on_trunk_int(man, ports, vlan_ids)
+ man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
+ nexus_user, nexus_password)
+ if not vlan_ids:
+ vlan_ids = self.build_vlans_cmd()
+ for ports in nexus_ports:
+ self.enable_vlan_on_trunk_int(man, ports, vlan_ids)
def remove_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port):
"""
Removes a vlan from interfaces on the Nexus switch given the VLAN ID
"""
- with self.nxos_connect(nexus_host, int(nexus_ssh_port), nexus_user,
- nexus_password) as man:
- for ports in nexus_ports:
- self.disable_vlan_on_trunk_int(man, ports, vlan_id)
+ man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
+ nexus_user, nexus_password)
+ for ports in nexus_ports:
+ self.disable_vlan_on_trunk_int(man, ports, vlan_id)
#
# @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
"""
self._client = importutils.import_object(conf.NEXUS_DRIVER)
LOG.debug("Loaded driver %s\n" % conf.NEXUS_DRIVER)
- self._nexus_ip = conf.NEXUS_IP_ADDRESS
- self._nexus_username = cred.Store.get_username(conf.NEXUS_IP_ADDRESS)
- self._nexus_password = cred.Store.get_password(conf.NEXUS_IP_ADDRESS)
- self._nexus_ports = conf.NEXUS_PORTS
- self._nexus_ssh_port = conf.NEXUS_SSH_PORT
+ self._nexus_switches = conf.NEXUS_DETAILS
+ self.credentials = {}
+
+ def get_credential(self, nexus_ip):
+ if not nexus_ip in self.credentials.keys():
+ _nexus_username = cred.Store.get_username(nexus_ip)
+ _nexus_password = cred.Store.get_password(nexus_ip)
+ self.credentials[nexus_ip] = {
+ 'username': _nexus_username,
+ 'password': _nexus_password
+ }
+ return self.credentials[nexus_ip]
def get_all_networks(self, tenant_id):
"""
return self._networks.values()
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
- **kwargs):
+ host, instance):
"""
- Create a VLAN in the switch, and configure the appropriate interfaces
+ Create a VLAN in the appropriate switch/port,
+ and configure the appropriate interfaces
for this VLAN
"""
LOG.debug("NexusPlugin:create_network() called\n")
- vlan_ids = ''
- for key in kwargs:
- if key == 'vlan_ids':
- vlan_ids = kwargs['vlan_ids']
- self._client.create_vlan(
- vlan_name, str(vlan_id), self._nexus_ip,
- self._nexus_username, self._nexus_password,
- self._nexus_ports, self._nexus_ssh_port, vlan_ids)
- for ports in self._nexus_ports:
- try:
- nxos_db.add_nexusport_binding(ports, str(vlan_id))
- except:
- raise excep.NexusPortBindingAlreadyExists(port_id=ports)
-
+ # Grab the switch IP and port for this host
+ switch_ip = ''
+ port_id = ''
+ for switch in self._nexus_switches.keys():
+ for hostname in self._nexus_switches[switch].keys():
+ if str(hostname) == str(host):
+ switch_ip = switch
+ port_id = self._nexus_switches[switch][hostname]['ports']
+ # Check if this network is already in the DB
+ binding = nxos_db.get_port_vlan_switch_binding(
+ port_id, vlan_id, switch_ip)
+ if not binding:
+ _nexus_ip = switch_ip
+ _nexus_ports = (port_id,)
+ _nexus_ssh_port = \
+ self._nexus_switches[switch_ip]['ssh_port']['ssh_port']
+ _nexus_creds = self.get_credential(_nexus_ip)
+ _nexus_username = _nexus_creds['username']
+ _nexus_password = _nexus_creds['password']
+ # Check for vlan/switch binding
+ vbinding = nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
+ if not vbinding:
+ # Create vlan and trunk vlan on the port
+ self._client.create_vlan(
+ vlan_name, str(vlan_id), _nexus_ip,
+ _nexus_username, _nexus_password,
+ _nexus_ports, _nexus_ssh_port, vlan_id)
+ else:
+ # Only trunk vlan on the port
+ man = self._client.nxos_connect(_nexus_ip,
+ int(_nexus_ssh_port),
+ _nexus_username,
+ _nexus_password)
+ self._client.enable_vlan_on_trunk_int(man,
+ port_id,
+ vlan_id)
+
+ nxos_db.add_nexusport_binding(port_id, str(vlan_id),
+ switch_ip, instance)
new_net_dict = {const.NET_ID: net_id,
const.NET_NAME: net_name,
const.NET_PORTS: {},
def delete_network(self, tenant_id, net_id, **kwargs):
"""
- Deletes a VLAN in the switch, and removes the VLAN configuration
+ Deletes the VLAN in all switches, and removes the VLAN configuration
from the relevant interfaces
"""
LOG.debug("NexusPlugin:delete_network() called\n")
- vlan_id = None
- for key in kwargs:
- if key == const.CONTEXT:
- context = kwargs[const.CONTEXT]
- elif key == const.BASE_PLUGIN_REF:
- base_plugin_ref = kwargs[const.BASE_PLUGIN_REF]
- elif key == 'vlan_id':
- vlan_id = kwargs['vlan_id']
- if vlan_id is None:
- vlan_id = self._get_vlan_id_for_network(tenant_id, net_id,
- context, base_plugin_ref)
- ports_id = nxos_db.get_nexusport_binding(vlan_id)
- LOG.debug("NexusPlugin: Interfaces to be disassociated: %s" % ports_id)
- nxos_db.remove_nexusport_binding(vlan_id)
- if net_id:
- self._client.delete_vlan(
- str(vlan_id), self._nexus_ip,
- self._nexus_username, self._nexus_password,
- self._nexus_ports, self._nexus_ssh_port)
- return net_id
- # Network not found
- raise exc.NetworkNotFound(net_id=net_id)
def get_network_details(self, tenant_id, net_id, **kwargs):
"""
Virtual Network.
"""
LOG.debug("NexusPlugin:update_network() called\n")
- if 'net_admin_state' in kwargs:
- net_admin_state = kwargs['net_admin_state']
- vlan_id = kwargs['vlan_id']
- vlan_ids = kwargs['vlan_ids']
- if not net_admin_state:
- self._client.remove_vlan_int(
- str(vlan_id), self._nexus_ip,
- self._nexus_username, self._nexus_password,
- self._nexus_ports, self._nexus_ssh_port)
- else:
- self._client.add_vlan_int(
- str(vlan_id), self._nexus_ip,
- self._nexus_username, self._nexus_password,
- self._nexus_ports, self._nexus_ssh_port,
- vlan_ids)
- return net_id
def get_all_ports(self, tenant_id, net_id, **kwargs):
"""
"""
LOG.debug("NexusPlugin:create_port() called\n")
- def delete_port(self, tenant_id, net_id, port_id, **kwargs):
+ def delete_port(self, device_id, vlan_id):
"""
- This is probably not applicable to the Nexus plugin.
- Delete if not required.
+ Delete port bindings from the database and scan
+ whether the network is still required on
+ the interfaces trunked
"""
LOG.debug("NexusPlugin:delete_port() called\n")
+ # Delete DB row for this port
+ row = nxos_db.get_nexusvm_binding(vlan_id, device_id)
+ if row:
+ nxos_db.remove_nexusport_binding(row['port_id'], row['vlan_id'],
+ row['switch_ip'],
+ row['instance_id'])
+ # Check for any other bindings with the same vlan_id and switch_ip
+ bindings = nxos_db.get_nexusvlan_binding(
+ row['vlan_id'], row['switch_ip'])
+
+ if not bindings:
+ # Delete this vlan from this switch
+ _nexus_ip = row['switch_ip']
+ _nexus_ports = (row['port_id'],)
+ _nexus_ssh_port = \
+ self._nexus_switches[_nexus_ip]['ssh_port']['ssh_port']
+ _nexus_creds = self.get_credential(_nexus_ip)
+ _nexus_username = _nexus_creds['username']
+ _nexus_password = _nexus_creds['password']
+ self._client.delete_vlan(
+ str(row['vlan_id']), _nexus_ip,
+ _nexus_username, _nexus_password,
+ _nexus_ports, _nexus_ssh_port)
+
+ return row['instance_id']
def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
"""
NEXUS_IP_ADDRESS = '1.1.1.1'
NEXUS_USERNAME = 'username'
NEXUS_PASSWORD = 'password'
-NEXUS_PORTS = ['1/10']
+HOSTNAME = 'testhost'
+INSTANCE = 'testvm'
+NEXUS_PORTS = '1/10'
NEXUS_SSH_PORT = '22'
NEXUS_DRIVER = ('quantum.plugins.cisco.tests.unit.v2.nexus.'
'fake_nexus_driver.CiscoNEXUSFakeDriver')
self.second_net_id = 000005
self.second_vlan_name = "q-" + str(self.second_net_id) + "vlan"
self.second_vlan_id = 265
+ self._nexus_switches = {
+ NEXUS_IP_ADDRESS: {
+ HOSTNAME: {
+ 'ports': NEXUS_PORTS,
+ },
+ 'ssh_port': {
+ 'ssh_port': NEXUS_SSH_PORT
+ }
+ }
+ }
+ self._hostname = HOSTNAME
def new_cdb_init():
db.configure_db()
self._nexus_password = NEXUS_PASSWORD
self._nexus_ports = NEXUS_PORTS
self._nexus_ssh_port = NEXUS_SSH_PORT
+ self.credentials = {
+ self._nexus_ip: {
+ 'username': self._nexus_username,
+ 'password': self._nexus_password
+ }
+ }
with mock.patch.object(cdb, 'initialize', new=new_cdb_init):
cdb.initialize()
with mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
'__init__', new=new_nexus_init):
self._cisco_nexus_plugin = cisco_nexus_plugin_v2.NexusPlugin()
+ self._cisco_nexus_plugin._nexus_switches = self._nexus_switches
- def test_a_create_delete_network(self):
+ def test_a_create_network(self):
"""
Tests creation of two new Virtual Network.
Tests deletion of one Virtual Network.
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, net_name, net_id,
- vlan_name, vlan_id, vlan_ids=str(vlan_id))
-
+ vlan_name, vlan_id, self._hostname, INSTANCE)
self.assertEqual(new_net_dict[const.NET_ID], net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
- vlan_ids = str(vlan_id) + "," + str(second_vlan_id)
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, second_net_name, second_net_id,
- second_vlan_name, second_vlan_id,
- vlan_ids=vlan_ids)
+ second_vlan_name, second_vlan_id, self._hostname,
+ INSTANCE)
self.assertEqual(new_net_dict[const.NET_ID], second_net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.second_net_name)
self.second_vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.second_vlan_id)
- expected_net_id = self._cisco_nexus_plugin.delete_network(
- tenant_id, net_id, vlan_id=str(vlan_id))
-
- self.assertEqual(expected_net_id, net_id)
-
- def test_b_nexus_clear_vlan(self):
+ def test_b_nexus_delete_port(self):
"""
Test to clean up second vlan of nexus device
created by test_create_delete_network. This
test will fail if it is run individually.
"""
- tenant_id = self.tenant_id
- second_net_id = self.second_net_id
- second_vlan_id = self.second_vlan_id
-
- expected_second_net_id = self._cisco_nexus_plugin.delete_network(
- tenant_id, second_net_id,
- vlan_id=str(second_vlan_id))
-
- self.assertEqual(expected_second_net_id, second_net_id)
-
- def test_c_update_network_False(self):
- """
- Test to update a network state to False
- resulting in disabling a vlan corresponding to
- that network from the configured nexus interfaces
- """
- tenant_id = self.tenant_id
- net_name = self.net_name
- net_id = self.net_id
- vlan_name = self.vlan_name
- vlan_id = self.vlan_id
- second_net_name = self.second_net_name
- second_net_id = self.second_net_id
- second_vlan_name = self.second_vlan_name
- second_vlan_id = self.second_vlan_id
-
- new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, net_name, net_id,
- vlan_name, vlan_id, vlan_ids=str(vlan_id))
-
- vlan_ids = str(vlan_id) + "," + str(second_vlan_id)
- new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, second_net_name, second_net_id,
- second_vlan_name, second_vlan_id,
- vlan_ids=vlan_ids)
-
- expected_net_id = self._cisco_nexus_plugin.update_network(
- tenant_id, net_id, net_admin_state=False,
- vlan_id=vlan_id, vlan_ids=str(vlan_id))
-
- self.assertEqual(expected_net_id, net_id)
-
- def test_d_nexus_clean_vlan_update(self):
- """
- Cleans up vlans on the nexus for the two
- created networks
- """
- tenant_id = self.tenant_id
- net_id = self.net_id
- vlan_id = self.vlan_id
- second_net_id = self.second_net_id
- second_vlan_id = self.second_vlan_id
-
- netid = self._cisco_nexus_plugin.delete_network(
- tenant_id, net_id, vlan_id=str(vlan_id))
-
- self.assertEqual(netid, net_id)
-
- expected_second_net_id = self._cisco_nexus_plugin.delete_network(
- tenant_id, second_net_id,
- vlan_id=str(second_vlan_id))
-
- self.assertEqual(expected_second_net_id, second_net_id)
-
- def test_e_update_network_True(self):
- """
- Test to update a disabled network state to True
- resulting in enabling a vlan corresponding to
- that network to the configured nexus interfaces
- """
- tenant_id = self.tenant_id
- net_name = self.net_name
- net_id = self.net_id
- vlan_name = self.vlan_name
- vlan_id = self.vlan_id
- second_vlan_id = self.second_vlan_id
-
- self.test_c_update_network_False()
-
- vlan_ids = str(vlan_id) + "," + str(second_vlan_id)
- expected_net_id = self._cisco_nexus_plugin.update_network(
- tenant_id, net_id, net_admin_state=True,
- vlan_id=vlan_id, vlan_ids=str(vlan_ids))
-
- self.assertEqual(expected_net_id, net_id)
+ expected_instance_id = self._cisco_nexus_plugin.delete_port(
+ INSTANCE, self.second_vlan_id
+ )
- self.test_d_nexus_clean_vlan_update()
+ self.assertEqual(expected_instance_id, INSTANCE)
def tearDown(self):
"""Clear the test environment"""
+ pass
# Remove database contents
- db.clear_db(network_models_v2.model_base.BASEV2)
+ #db.clear_db(network_models_v2.model_base.BASEV2)
webob>=1.0.8
python-keystoneclient>=0.2.0
alembic>=0.4.1
+
+# Cisco plugin dependencies
+python-novaclient
+# End Cisco dependencies
Code Review / openstack-build / neutron-build.git / commitdiff