All changes contained within the Cisco L2-Network plugin and extensions.
Change-Id: I696a2aebeb9027f3b0ce0b6673910e02e304673f
def build_vif(self, vif_data):
"""Return VIF description."""
return dict(vif_desc=vif_data[const.VIF_DESC])
+
+ def build_result(self, result_data):
+ """Return result True/False"""
+ return dict(result=result_data)
parent_resource = dict(member_name="tenant",
collection_name="extensions/csco/tenants")
member_actions = {'schedule_host': "PUT",
- 'associate_port': "PUT"}
+ 'associate_port': "PUT",
+ 'detach_port': "PUT"}
controller = NovatenantsController(QuantumManager.get_plugin())
return [extensions.ResourceExtension('novatenants', controller,
parent=parent_resource,
return result
except qexception.PortNotFound as exp:
return faults.Fault(faults.PortNotFound(exp))
+
+ def detach_port(self, request, tenant_id, id):
+ content_type = request.best_match_content_type()
+ try:
+ req_params = \
+ self._parse_request_params(request,
+ self._schedule_host_ops_param_list)
+ except exc.HTTPError as exp:
+ return faults.Fault(exp)
+
+ instance_id = req_params['instance_id']
+ instance_desc = req_params['instance_desc']
+
+ try:
+ vif = self._plugin. \
+ detach_port(tenant_id, instance_id, instance_desc)
+ builder = novatenant_view.get_view_builder(request)
+ result = builder.build_result(True)
+ return result
+ except qexception.PortNotFound as exp:
+ return faults.Fault(faults.PortNotFound(exp))
quantum/plugins/cisco/conf/ucs_inventory.ini file. You can configure multiple\r
UCSMs per deployment, multiple chassis per UCSM, and multiple blades per\r
chassis. Chassis ID and blade ID can be obtained from the UCSM (they will\r
- typically be numbers like 1, 2, 3, etc.)\r
+ typically be numbers like 1, 2, 3, etc.). Also make sure that you put the exact\r
+ hostname as nova sees it (the host column in the services table of the nova\r
+ DB will give you that information).\r
\r
[ucsm-1]\r
ip_address = <put_ucsm_ip_address_here>\r
Multi NIC support for VMs\r
-------------------------\r
As indicated earlier, if your Nova setup has a project with more than one network,\r
-Nova will try to create a vritual network interface (VIF) on the VM for each of those\r
-networks. That implies that,\r
+Nova will try to create a virtual network interface (VIF) on the VM for each of those\r
+networks. That implies -\r
\r
(1) You should create the same number of networks in Quantum as in your Nova\r
project.\r
Note that when using UCS and the 802.1Qbh features, the association of the\r
VIF-ID (also referred to as interface ID) on the VM's NIC with a port will\r
happen automatically when the VM is instantiated. At this point, doing a\r
- show_port will reveal the VIF-ID associated with the port.\r
+ show_port will reveal the VIF-ID associated with the port. To indicate that\r
+ this VIF-ID is still detached from the network it would eventually be on, you\r
+ will see the suffix "(detached)" on the VIF-ID. This indicates that although\r
+ the VIF-ID and the port have been associated, the VIF still does not have\r
+ connectivity to the network on which the port resides. That connectivity\r
+ will be established only after the plug/attach operation is performed (as\r
+ described in the next step).\r
\r
# PYTHONPATH=. python quantum/plugins/cisco/client/cli.py show_port demo c4a2bea7-a528-4caf-b16e-80397cd1663a 118ac473-294d-480e-8f6d-425acbbe81ae\r
Logical Port ID: 118ac473-294d-480e-8f6d-425acbbe81ae\r
administrative State: ACTIVE\r
-interface: b73e3585-d074-4379-8dde-931c0fc4db0e\r
+interface: b73e3585-d074-4379-8dde-931c0fc4db0e(detached)\r
on Virtual Network: c4a2bea7-a528-4caf-b16e-80397cd1663a\r
for Tenant: demo\r
\r
\r
\r
8. Unplug an interface and port from the network\r
- Note: Before unplugging, make a note of the interface ID (you can use the\r
- show_port CLI as before). While the VM, which has a VIF with this interface\r
- ID still exists, you can only plug that same interface back into this port.\r
- So the subsequent plug interface operation on this port will have to make\r
- use of the same interface ID.\r
\r
# PYTHONPATH=. python quantum/plugins/cisco/client/cli.py unplug_iface demo c4a2bea7-a528-4caf-b16e-80397cd1663a 118ac473-294d-480e-8f6d-425acbbe81ae\r
Unplugged interface from Logical Port: 118ac473-294d-480e-8f6d-425acbbe81ae\r
on Virtual Network: c4a2bea7-a528-4caf-b16e-80397cd1663a\r
for Tenant: demo\r
\r
+ Note: After unplugging, if you check the details of the port, you will\r
+ see the VIF-IF associated with the port (but now suffixed with the state\r
+ "detached"). At this point, it is possible to plug the VIF into the network\r
+ again making use of the same VIF-ID. In general, once associated, the VIF-ID\r
+ cannot be disassociated with the port until the VM is terminated. After the\r
+ VM is terminated, the VIF-ID will be automatically disassociated from the\r
+ port. To summarize, association and disassociation of the VIF-ID with a port\r
+ happens automatically at the time of creating and terminating the VM. The\r
+ connectivity of the VIF to the network is controlled by the user via the\r
+ plug and unplug operations.\r
+\r
\r
How to test the installation\r
----------------------------\r
#
"""
-import logging as LOG
from configobj import ConfigObj
from quantum.plugins.cisco.common import cisco_constants as const
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
-
class CiscoConfigParser(ConfigObj):
"""Config Parser based on the ConfigObj module"""
NETID_LIST = 'net_id_list'
DELIMITERS = "[,;:\b\s]"
+
+UUID_LENGTH = 36
+
+UNPLUGGED = '(detached)'
+
+ASSOCIATION_STATUS = 'association_status'
+
+ATTACHED = 'attached'
+
+DETACHED = 'detached'
#
"""
-import logging as LOG
import os
from quantum.plugins.cisco.common import cisco_configparser as confp
from quantum.plugins.cisco.common import cisco_exceptions as cexc
from quantum.plugins.cisco.db import l2network_db as cdb
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
-
TENANT = const.NETWORK_ADMIN
CREDENTIALS_FILE = "../conf/credentials.ini"
message = _("PortVnic Binding %(port_id) is not present")
+class InvalidAttach(exceptions.QuantumException):
+ message = _("Unable to plug the attachment %(att_id)s into port " \
+ "%(port_id)s for network %(net_id)s. Association of " \
+ "attachment ID with port ID happens implicitly when " \
+ "VM is instantiated; attach operation can be " \
+ "performed subsequently.")
+
+
try:
_("test")
except NameError:
"""
import hashlib
-import logging as LOG
+import logging
import MySQLdb
import traceback
from quantum.plugins.cisco.db import api as db
from quantum.plugins.cisco.db import l2network_db as cdb
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+LOG = logging.getLogger(__name__)
def get16ByteUUID(uuid):
rvlan = session.query(l2network_models.VlanID).\
filter_by(vlan_used=False).\
first()
+ if not rvlan:
+ raise exc.NoResultFound
rvlanid = session.query(l2network_models.VlanID).\
filter_by(vlan_id=rvlan["vlan_id"]).\
one()
raise c_exc.PortVnicNotFound(port_id=port_id)
+def update_portbinding_instance_id(port_id, instance_id):
+ """Updates port binding for the instance ID"""
+ LOG.debug("db update_portbinding_instance_id() called")
+ session = db.get_session()
+ try:
+ port_binding = session.query(ucs_models.PortBinding).\
+ filter_by(port_id=port_id).\
+ one()
+ port_binding.instance_id = instance_id
+ session.merge(port_binding)
+ session.flush()
+ return port_binding
+ except exc.NoResultFound:
+ raise c_exc.PortVnicNotFound(port_id=port_id)
+
+
+def update_portbinding_vif_id(port_id, vif_id):
+ """Updates port binding for the VIF ID"""
+ LOG.debug("db update_portbinding_vif_id() called")
+ session = db.get_session()
+ try:
+ port_binding = session.query(ucs_models.PortBinding).\
+ filter_by(port_id=port_id).\
+ one()
+ port_binding.vif_id = vif_id
+ session.merge(port_binding)
+ session.flush()
+ return port_binding
+ except exc.NoResultFound:
+ raise c_exc.PortVnicNotFound(port_id=port_id)
+
+
def get_portbinding_dn(blade_intf_dn):
"""Lists a port binding"""
LOG.debug("get_portbinding_dn() called")
"""
import inspect
-import logging as LOG
+import logging
import re
from quantum.common import exceptions as exc
from quantum.plugins.cisco.db import api as db
from quantum.plugins.cisco.db import l2network_db as cdb
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+LOG = logging.getLogger(__name__)
class L2Network(QuantumPluginBase):
def plug_interface(self, tenant_id, net_id, port_id,
remote_interface_id):
"""
- Attaches a remote interface to the specified port on the
+ Provides connectivity to a remote interface to the
specified Virtual Network.
"""
LOG.debug("plug_interface() called\n")
network = db.network_get(net_id)
port = db.port_get(net_id, port_id)
attachment_id = port[const.INTERFACEID]
- if attachment_id and remote_interface_id != attachment_id:
+ if attachment_id == None:
+ raise cexc.InvalidAttach(port_id=port_id, net_id=net_id,
+ att_id=remote_interface_id)
+ attachment_id = attachment_id[:const.UUID_LENGTH]
+ remote_interface_id = remote_interface_id[:const.UUID_LENGTH]
+ if remote_interface_id != attachment_id:
+ LOG.debug("Existing attachment_id:%s, remote_interface_id:%s" % \
+ (attachment_id, remote_interface_id))
raise exc.PortInUse(port_id=port_id, net_id=net_id,
att_id=attachment_id)
self._invoke_device_plugins(self._func_name(), [tenant_id,
net_id, port_id,
- remote_interface_id])
- if attachment_id == None:
- db.port_set_attachment(net_id, port_id, remote_interface_id)
+ attachment_id])
+ db.port_unset_attachment(net_id, port_id)
+ db.port_set_attachment(net_id, port_id, attachment_id)
#Note: The remote_interface_id gets associated with the port
# when the VM is instantiated. The plug interface call results
# in putting the port on the VLAN associated with this network
def unplug_interface(self, tenant_id, net_id, port_id):
"""
- Detaches a remote interface from the specified port on the
+ Removes connectivity of a remote interface to the
specified Virtual Network.
"""
LOG.debug("unplug_interface() called\n")
network = db.network_get(net_id)
+ port = db.port_get(net_id, port_id)
+ attachment_id = port[const.INTERFACEID]
+ if attachment_id == None:
+ raise exc.InvalidDetach(port_id=port_id, net_id=net_id,
+ att_id=remote_interface_id)
self._invoke_device_plugins(self._func_name(), [tenant_id, net_id,
port_id])
+ attachment_id = attachment_id[:const.UUID_LENGTH]
+ attachment_id = attachment_id + const.UNPLUGGED
db.port_unset_attachment(net_id, port_id)
+ db.port_set_attachment(net_id, port_id, attachment_id)
"""
Extension API implementation
def associate_port(self, tenant_id, instance_id, instance_desc):
"""
- Get the portprofile name and the device namei for the dynamic vnic
+ Get the portprofile name and the device name for the dynamic vnic
"""
LOG.debug("associate_port() called\n")
return self._invoke_device_plugins(self._func_name(), [tenant_id,
instance_id,
instance_desc])
+ def detach_port(self, tenant_id, instance_id, instance_desc):
+ """
+ Remove the association of the VIF with the dynamic vnic
+ """
+ LOG.debug("detach_port() called\n")
+ return self._invoke_device_plugins(self._func_name(), [tenant_id,
+ instance_id,
+ instance_desc])
+
def create_multiport(self, tenant_id, net_id_list, port_state, ports_desc):
"""
Creates multiple ports on the specified Virtual Network.
from copy import deepcopy
import inspect
-import logging as LOG
+import logging
import platform
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.common import cisco_exceptions as cexc
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(__name__)
+LOG = logging.getLogger(__name__)
class L2NetworkMultiBlade(L2NetworkModelBase):
def associate_port(self, args):
"""
- Get the portprofile name and the device namei for the dynamic vnic
+ Get the portprofile name and the device name for the dynamic vnic
"""
LOG.debug("associate_port() called\n")
return self._invoke_inventory(const.UCS_PLUGIN, self._func_name(),
args)
+ def detach_port(self, args):
+ """
+ Remove the association of the VIF with the dynamic vnic
+ """
+ LOG.debug("detach_port() called\n")
+ return self._invoke_plugin_per_device(const.UCS_PLUGIN,
+ self._func_name(), args)
+
def create_multiport(self, args):
"""Support for extension API call"""
self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(),
from copy import deepcopy
import inspect
-import logging as LOG
+import logging
import platform
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.common import cisco_exceptions as cexc
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(__name__)
+LOG = logging.getLogger(__name__)
class L2NetworkSingleBlade(L2NetworkModelBase):
return self._invoke_inventory(const.UCS_PLUGIN, self._func_name(),
args)
+ def detach_port(self, args):
+ """
+ Remove the association of the VIF with the dynamic vnic
+ """
+ LOG.debug("detach_port() called\n")
+ return self._invoke_plugin_per_device(const.UCS_PLUGIN,
+ self._func_name(), args)
+
def create_multiport(self, args):
"""Support for extension API call"""
self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(),
from quantum.common import utils
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.common import cisco_credentials as cred
+from quantum.plugins.cisco.db import api as db
+from quantum.plugins.cisco.db import l2network_db as cdb
from quantum.plugins.cisco.db import nexus_db as nxos_db
from quantum.plugins.cisco.l2device_plugin_base import L2DevicePluginBase
from quantum.plugins.cisco.nexus import cisco_nexus_configuration as conf
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)
- net = self._networks.get(net_id)
+ net = self._get_network(tenant_id, net_id)
if net:
self._client.delete_vlan(str(vlan_id), self._nexus_ip,
self._nexus_username, self._nexus_password,
self._nexus_first_port, self._nexus_second_port,
self._nexus_ssh_port)
- self._networks.pop(net_id)
return net
# Network not found
raise exc.NetworkNotFound(net_id=net_id)
"""
Gets the NETWORK ID
"""
- network = self._networks.get(network_id)
+ network = db.network_get(network_id)
if not network:
raise exc.NetworkNotFound(net_id=network_id)
- return network
+ vlan = cdb.get_vlan_binding(network_id)
+ return {const.NET_ID: network_id, const.NET_NAME: network.name,
+ const.NET_PORTS: network.ports,
+ const.NET_VLAN_NAME: vlan.vlan_name,
+ const.NET_VLAN_ID: vlan.vlan_id}
'/extensions/csco/tenants/{tenant_id}'
TENANT_ID = 'nova'
CSCO_EXT_NAME = 'Cisco Nova Tenant'
-ACTION = '/associate_port'
+ASSOCIATE_ACTION = '/associate_port'
+DETACH_ACTION = '/detach_port'
class Libvirt802dot1QbhDriver(VIFDriver):
- """VIF driver for Linux bridge."""
+ """VIF driver for 802.1Qbh"""
def __init__(self):
# We have to send a dummy tenant name here since the client
# needs some tenant name, but the tenant name will not be used
% CSCO_EXT_NAME)
raise excp.ServiceUnavailable()
- def _get_configurations(self, instance, network, mapping):
+ def _update_configurations(self, instance, network, mapping, action):
"""Gets the device name and the profile name from Quantum"""
instance_id = instance['id']
client = Client(HOST, PORT, USE_SSL, format='json', tenant=TENANT_ID,
action_prefix=ACTION_PREFIX_CSCO)
- request_url = "/novatenants/" + project_id + ACTION
+ request_url = "/novatenants/" + project_id + action
data = client.do_request('PUT', request_url, body=instance_data_dict)
- device = data['vif_desc']['device']
- portprofile = data['vif_desc']['portprofile']
- LOG.debug(_("Quantum provided the device: %s") % device)
- LOG.debug(_("Quantum provided the portprofile: %s") % portprofile)
- mac_id = mapping['mac'].replace(':', '')
+ if action == ASSOCIATE_ACTION:
+ device = data['vif_desc']['device']
+ portprofile = data['vif_desc']['portprofile']
+ LOG.debug(_("Quantum provided the device: %s") % device)
+ LOG.debug(_("Quantum provided the portprofile: %s") % portprofile)
+ mac_id = mapping['mac'].replace(':', '')
- result = {
- 'id': mac_id,
- 'mac_address': mapping['mac'],
- 'device_name': device,
- 'profile_name': portprofile,
- }
+ result = {
+ 'id': mac_id,
+ 'mac_address': mapping['mac'],
+ 'device_name': device,
+ 'profile_name': portprofile,
+ }
- return result
+ return result
+ else:
+ return data
def plug(self, instance, network, mapping):
- return self._get_configurations(instance, network, mapping)
+ return self._update_configurations(instance, network, mapping,
+ ASSOCIATE_ACTION)
def unplug(self, instance, network, mapping):
- pass
+ self._update_configurations(instance, network, mapping,
+ DETACH_ACTION)
#
"""
-import logging as LOG
+import logging
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.db import l2network_db as cdb
from quantum.plugins.cisco.l2network_segmentation_base \
import L2NetworkSegmentationMgrBase
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+LOG = logging.getLogger(__name__)
class L2NetworkVLANMgr(L2NetworkSegmentationMgrBase):
tenant_id, net_id)
LOG.debug("test_delete_network_not_found - END")
- def test_delete_networkInUse(self, tenant_id='test_network'):
+ def test_delete_networkInUse(
+ self, tenant_id='test_tenant', instance_tenant_id='nova',
+ nova_user_id='novaadmin', instance_id=10,
+ vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
+
"""
Tests deletion of a Virtual Network when Network is in Use.
"""
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
- self._l2network_plugin.plug_interface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], self.remote_interface)
+ instance_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id}
+ host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
+ instance_id,
+ instance_desc)
+ instance_vif_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id,
+ 'vif_id': vif_id}
+ vif_description = self._l2network_plugin.associate_port(
+ instance_tenant_id, instance_id,
+ instance_vif_desc)
+
self.assertRaises(exc.NetworkInUse,
self._l2network_plugin.delete_network, tenant_id,
new_net_dict[const.NET_ID])
self.tearDownNetworkPortInterface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID])
+ tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
+ new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_delete_networkInUse - END")
def test_show_network(self, net_tenant_id=None):
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_delete_portDNE - END")
- def test_delete_portInUse(self, tenant_id='test_tenant'):
+ def test_delete_portInUse(
+ self, tenant_id='test_tenant', instance_tenant_id='nova',
+ nova_user_id='novaadmin', instance_id=10,
+ vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests deletion of Ports when port is in Use.
"""
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.port_state)
- self._l2network_plugin.plug_interface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], self.remote_interface)
+ instance_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id}
+ host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
+ instance_id,
+ instance_desc)
+ instance_vif_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id,
+ 'vif_id': vif_id}
+ vif_description = self._l2network_plugin.associate_port(
+ instance_tenant_id, instance_id,
+ instance_vif_desc)
self.assertRaises(exc.PortInUse,
self._l2network_plugin.delete_port, tenant_id,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
self.tearDownNetworkPortInterface(
- tenant_id, new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
+ tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
+ new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_delete_portInUse - END")
def test_update_port(self, tenant_id='test_tenant',
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_show_portDNE - END")
- def test_plug_interface(self, tenant_id='test_tenant',
- remote_interface='new_interface'):
+ def test_plug_interface(
+ self, tenant_id='test_tenant', instance_tenant_id='nova',
+ nova_user_id='novaadmin', instance_id=10,
+ vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests attachment of interface to the port
"""
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
+ instance_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id}
+ host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
+ instance_id,
+ instance_desc)
+ instance_vif_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id,
+ 'vif_id': vif_id}
+ vif_description = self._l2network_plugin.associate_port(
+ instance_tenant_id, instance_id,
+ instance_vif_desc)
+
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], remote_interface)
+ port_dict[const.PORT_ID], vif_id)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
- self.assertEqual(port[const.INTERFACEID], remote_interface)
+ self.assertEqual(port[const.INTERFACEID], vif_id)
self.tearDownNetworkPortInterface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID])
+ tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
+ new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
+
LOG.debug("test_plug_interface - END")
def test_plug_interface_networkDNE(
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_plug_interface_portDNE - END")
- def test_plug_interface_portInUse(self, tenant_id='test_tenant',
- remote_interface='new_interface'):
+ def test_plug_interface_portInUse(
+ self, tenant_id='test_tenant', instance_tenant_id='nova',
+ nova_user_id='novaadmin', instance_id=10,
+ vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc',
+ remote_interface='new_interface'):
"""
Tests attachment of new interface to the port when there is an
"""
LOG.debug("test_plug_interface_portInUse - START")
- current_interface = "current_interface"
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
- self._l2network_plugin.plug_interface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], current_interface)
+ instance_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id}
+ host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
+ instance_id,
+ instance_desc)
+ instance_vif_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id,
+ 'vif_id': vif_id}
+ vif_description = self._l2network_plugin.associate_port(
+ instance_tenant_id, instance_id,
+ instance_vif_desc)
+
self.assertRaises(exc.PortInUse,
self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], remote_interface)
+ port_dict[const.PORT_ID], remote_interface)
self.tearDownNetworkPortInterface(
- tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID])
+ tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
+ new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
+
LOG.debug("test_plug_interface_portInUse - END")
- def test_unplug_interface(self, tenant_id='test_tenant'):
+ def test_unplug_interface(
+ self, tenant_id='test_tenant', instance_tenant_id='nova',
+ nova_user_id='novaadmin', instance_id=10,
+ vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
+
"""
Tests detaachment of an interface to a port
"""
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.port_state)
+ instance_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id}
+ host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
+ instance_id,
+ instance_desc)
+ instance_vif_desc = {'project_id': tenant_id,
+ 'user_id': nova_user_id,
+ 'vif_id': vif_id}
+ vif_description = self._l2network_plugin.associate_port(
+ instance_tenant_id, instance_id,
+ instance_vif_desc)
+
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID], self.remote_interface)
+ port_dict[const.PORT_ID], vif_id)
self._l2network_plugin.unplug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
- self.assertEqual(port[const.INTERFACEID], None)
- self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
- port_dict[const.PORT_ID])
+ vif_id_unplugged = vif_id + '(detached)'
+ self.assertEqual(port[const.INTERFACEID], vif_id_unplugged)
+ self.tearDownNetworkPortInterface(
+ tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
+ new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
+
LOG.debug("test_unplug_interface - END")
def test_unplug_interface_networkDNE(self, tenant_id='test_tenant',
self._l2network_plugin.delete_port(tenant_id, network_dict_id, port_id)
self.tearDownNetwork(tenant_id, network_dict_id)
- def tearDownNetworkPortInterface(self, tenant_id, network_dict_id,
- port_id):
+ def tearDownNetworkPortInterface(self, tenant_id, instance_tenant_id,
+ instance_id, instance_desc, network_dict_id,
+ port_id):
"""
Tear down Network Port Interface
"""
- self._l2network_plugin.unplug_interface(tenant_id,
- network_dict_id, port_id)
+ self._l2network_plugin.detach_port(instance_tenant_id, instance_id,
+ instance_desc)
self.tearDownNetworkPort(tenant_id, network_dict_id, port_id)
def tearDownPortProfile(self, tenant_id, port_profile_id):
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.db import l2network_db as cdb
from quantum.plugins.cisco.db import api as db
+from quantum.plugins.cisco.common import cisco_credentials as cred
from quantum.plugins.cisco.nexus import cisco_nexus_plugin
LOG = logging.getLogger('quantum.tests.test_nexus')
"""
Set up function
"""
-
self.tenant_id = "test_tenant_cisco1"
self.net_name = "test_network_cisco1"
self.net_id = 000007
self.vlan_id = 267
self.port_id = "9"
cdb.initialize()
+ cred.Store.initialize()
self._cisco_nexus_plugin = cisco_nexus_plugin.NexusPlugin()
def test_create_network(self, net_tenant_id=None, network_name=None,
- network_id=None, net_vlan_name=None,
- net_vlan_id=None):
+ net_vlan_name=None, net_vlan_id=None):
"""
Tests creation of new Virtual Network.
"""
net_name = network_name
else:
net_name = self.net_name
- if network_id:
- net_id = network_id
- else:
- net_id = self.net_id
if net_vlan_name:
vlan_name = net_vlan_name
else:
else:
vlan_id = self.vlan_id
- network_created = self.create_network(tenant_id, net_id)
+ network_created = self.create_network(tenant_id, net_name)
cdb.add_vlan_binding(vlan_id, vlan_name, network_created["net-id"])
new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, net_name, net_id, vlan_name, vlan_id)
- self.assertEqual(new_net_dict[const.NET_ID], self.net_id)
+ tenant_id, net_name, network_created["net-id"],
+ vlan_name, vlan_id)
+ self.assertEqual(new_net_dict[const.NET_ID],
+ network_created["net-id"])
self.assertEqual(new_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_create_network - END")
- def test_delete_network(self, net_tenant_id=None, network_id=None):
+ def test_delete_network(self, net_tenant_id=None, network_name=None):
"""
Tests deletion of a Virtual Network.
"""
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
- if network_id:
- net_id = network_id
+ if network_name:
+ net_name = network_name
else:
- net_id = self.net_id
+ net_name = self.net_name
+ network_created = self.create_network(tenant_id, net_name)
+ cdb.add_vlan_binding(self.vlan_id, self.vlan_name,
+ network_created["net-id"])
new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, self.net_name, net_id, self.vlan_name, self.vlan_id)
+ tenant_id, self.net_name, network_created["net-id"],
+ self.vlan_name, self.vlan_id)
deleted_net_dict = self._cisco_nexus_plugin.delete_network(
tenant_id, new_net_dict[const.NET_ID])
- self.assertEqual(deleted_net_dict[const.NET_ID], net_id)
+ self.assertEqual(deleted_net_dict[const.NET_ID],
+ network_created["net-id"])
LOG.debug("test_delete_network - END")
def test_delete_network_DNE(self, net_tenant_id=None, net_id='0005'):
LOG.debug("test_delete_network_DNE - END")
- def test_get_network_details(self, net_tenant_id=None, network_id=None):
+ def test_get_network_details(self, net_tenant_id=None, network_name=None):
"""
Tests displays details of a Virtual Network .
"""
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
- if network_id:
- net_id = network_id
+ if network_name:
+ net_name = network_name
else:
- net_id = self.net_id
+ net_name = self.net_name
+ network_created = self.create_network(tenant_id, net_name)
+ cdb.add_vlan_binding(self.vlan_id, self.vlan_name,
+ network_created["net-id"])
new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, self.net_name, net_id, self.vlan_name, self.vlan_id)
+ tenant_id, self.net_name, network_created["net-id"],
+ self.vlan_name, self.vlan_id)
check_net_dict = self._cisco_nexus_plugin.get_network_details(
- tenant_id, net_id)
-
- self.assertEqual(check_net_dict[const.NET_ID], net_id)
+ tenant_id, network_created["net-id"])
+ self.assertEqual(check_net_dict[const.NET_ID],
+ network_created["net-id"])
self.assertEqual(check_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(check_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(check_net_dict[const.NET_VLAN_ID], self.vlan_id)
LOG.debug("test_get_network_details_network_does_not_exist - END")
def test_rename_network(self, new_name="new_network_name",
- net_tenant_id=None, network_id=None):
+ net_tenant_id=None, network_name=None):
"""
Tests rename of a Virtual Network .
"""
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
- if network_id:
- net_id = network_id
+ if network_name:
+ net_name = network_name
else:
- net_id = self.net_id
+ net_name = self.net_name
+ network_created = self.create_network(tenant_id, net_name)
+ cdb.add_vlan_binding(self.vlan_id, self.vlan_name,
+ network_created["net-id"])
new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, self.net_name, net_id, self.vlan_name,
- self.vlan_id)
+ tenant_id, self.net_name, network_created["net-id"],
+ self.vlan_name, self.vlan_id)
rename_net_dict = self._cisco_nexus_plugin.rename_network(
tenant_id, new_net_dict[const.NET_ID], new_name)
self.assertEqual(rename_net_dict[const.NET_NAME], new_name)
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
+
+ network_created = self.create_network(tenant_id, self.net_name)
+ cdb.add_vlan_binding(self.vlan_id, self.vlan_name,
+ network_created["net-id"])
new_net_dict1 = self._cisco_nexus_plugin.create_network(
- tenant_id, self.net_name, self.net_id,
- self.vlan_name, self.vlan_id)
+ tenant_id, self.net_name, network_created["net-id"],
+ self.vlan_name, self.vlan_id)
+ network_created2 = self.create_network(tenant_id, 'test_network2')
+ cdb.add_vlan_binding(265, 'second_vlan', network_created2["net-id"])
new_net_dict2 = self._cisco_nexus_plugin.create_network(
- tenant_id, "New_Network2", "0011",
- "second_vlan", "2003")
+ tenant_id, "New_Network2",
+ network_created2["net-id"], "second_vlan", "2003")
list_net_dict = self._cisco_nexus_plugin.get_all_networks(tenant_id)
net_temp_list = [new_net_dict1, new_net_dict2]
- self.assertEqual(len(list_net_dict), 2)
- self.assertTrue(list_net_dict[0] in net_temp_list)
- self.assertTrue(list_net_dict[1] in net_temp_list)
+ self.assertTrue(net_temp_list[0] in list_net_dict)
+ self.assertTrue(net_temp_list[1] in list_net_dict)
self.tearDownNetwork(tenant_id, new_net_dict1[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict2[const.NET_ID])
LOG.debug("test_list_all_networks - END")
def test_get_vlan_id_for_network(self, net_tenant_id=None,
- network_id=None):
+ network_name=None):
"""
Tests retrieval of vlan id for a Virtual Networks .
"""
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
- if network_id:
- net_id = network_id
+ if network_name:
+ net_name = network_name
else:
- net_id = self.net_id
+ net_name = self.net_name
+
+ network_created = self.create_network(tenant_id, net_name)
+ cdb.add_vlan_binding(self.vlan_id, self.vlan_name,
+ network_created["net-id"])
new_net_dict = self._cisco_nexus_plugin.create_network(
- tenant_id, self.net_name, net_id, self.vlan_name,
- self.vlan_id)
+ tenant_id, self.net_name, network_created["net-id"],
+ self.vlan_name, self.vlan_id)
result_vlan_id = self._cisco_nexus_plugin._get_vlan_id_for_network(
- tenant_id, net_id)
+ tenant_id, network_created["net-id"])
self.assertEqual(result_vlan_id, self.vlan_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_get_vlan_id_for_network - END")
Clean up functions after the tests
"""
self._cisco_nexus_plugin.delete_network(tenant_id, network_dict_id)
+
+ def tearDown(self):
+ """Clear the test environment"""
+ # Remove database contents
+ db.clear_db()
class UCSVICTestPlugin(unittest.TestCase):
+ """
+ Unit Tests for the UCS Plugin functions
+ """
def setUp(self):
self._cisco_ucs_plugin = cisco_ucs_plugin.UCSVICPlugin()
self.device_ip = conf.UCSM_IP_ADDRESS
self._ucs_inventory = ucsinv.UCSInventory()
- self.chassis_id = '1'
- self.blade_id = '5'
- self.blade_intf_distinguished_name = 'sys/chassis-1/blade-5/'\
- 'adaptor-1/host-eth-6'
+ self._ucs_inventory._load_inventory()
+ self.chassis_id_list = self._ucs_inventory._inventory[\
+ self.device_ip].keys()
+ self.chassis_id = self.chassis_id_list[0]
+ self.blade_id_list = self._ucs_inventory._inventory[\
+ self.device_ip][self.chassis_id]
+ self.blade_id = self._ucs_inventory._inventory[\
+ self.device_ip][self.chassis_id][0]
def test_create_network(self):
"""
self.assertEqual(new_net_dict[const.NET_ID], new_network[const.UUID])
self.assertEqual(new_net_dict[const.NET_NAME],
new_network[const.NETWORKNAME])
- self.tearDownNetwork(self.tenant_id, new_network[const.UUID])
+ self.tear_down_network(self.tenant_id, new_network[const.UUID])
def test_delete_network(self):
"""
self.assertEqual(new_net_dict[const.NET_ID], new_network[const.UUID])
self.assertEqual(new_net_dict[const.NET_NAME],
new_network[const.NETWORKNAME])
- self.tearDownNetwork(self.tenant_id, new_network[const.UUID])
+ self.tear_down_network(self.tenant_id, new_network[const.UUID])
def test_get_all_networks(self):
"""
self.assertTrue(net_list[0] in net_id_list)
self.assertTrue(net_list[1] in net_id_list)
- self.tearDownNetwork(self.tenant_id, new_network1[const.UUID])
- self.tearDownNetwork(self.tenant_id, new_network2[const.UUID])
+ self.tear_down_network(self.tenant_id, new_network1[const.UUID])
+ self.tear_down_network(self.tenant_id, new_network2[const.UUID])
def test_get_all_ports(self):
"""
self.assertTrue(str(ports_on_net[0]) == str(port_list[1]) or
str(ports_on_net[0]) == str(port_list[0]))
+ blade_intf_details = self._ucs_inventory._get_rsvd_blade_intf_by_port(
+ self.tenant_id, port_dict1[const.PORTID])
self._cisco_ucs_plugin.delete_port(
self.tenant_id, new_net_dict[const.NET_ID],
port_dict1[const.PORTID], device_ip=self.device_ip,
ucs_inventory=self._ucs_inventory,
chassis_id=self.chassis_id, blade_id=self.blade_id,
- blade_intf_distinguished_name=self.\
- blade_intf_distinguished_name,
+ blade_intf_distinguished_name=blade_intf_details[\
+ const.BLADE_INTF_DN],
least_rsvd_blade_dict=self._ucs_inventory.\
_get_least_reserved_blade())
- self.tearDownNetworkPort(
+ self.tear_down_network_port(
self.tenant_id, new_net_dict[const.NET_ID],
port_dict2[const.PORTID])
profile_name = self._cisco_ucs_plugin.\
_get_profile_name(port_dict[const.PORTID])
self.assertTrue(profile_name != None)
- self.tearDownNetworkPort(
+ self.tear_down_network_port(
self.tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORTID])
ucs_inventory=self._ucs_inventory,
least_rsvd_blade_dict=self._ucs_inventory.\
_get_least_reserved_blade())
+
+ blade_intf_details = self._ucs_inventory._get_rsvd_blade_intf_by_port(
+ self.tenant_id, port_dict[const.PORTID])
port_bind = self._cisco_ucs_plugin.delete_port(
self.tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORTID], device_ip=self.device_ip,
ucs_inventory=self._ucs_inventory,
chassis_id=self.chassis_id, blade_id=self.blade_id,
- blade_intf_distinguished_name=self.\
- blade_intf_distinguished_name,
+ blade_intf_distinguished_name=blade_intf_details[\
+ const.BLADE_INTF_DN],
least_rsvd_blade_dict=self._ucs_inventory.\
_get_least_reserved_blade())
self.assertEqual(port_bind[const.PORTID], new_port[const.UUID])
- self.tearDownNetwork(self.tenant_id, new_net_dict[const.NET_ID])
+ self.tear_down_network(self.tenant_id, new_net_dict[const.NET_ID])
def _test_get_port_details(self, port_state):
"""
self.tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORTID], device_ip=self.device_ip)
self.assertEqual(str(port_dict), str(port_detail))
- self.tearDownNetworkPort(
+ self.tear_down_network_port(
self.tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORTID])
def test_get_port_details_state_up(self):
+ """
+ Tests if the port details is retrieved when port is up
+ """
self._test_get_port_details(const.PORT_UP)
def test_show_port_state_down(self):
+ """
+ Tests if the port details is retrieved when port is down
+ """
self._test_get_port_details(const.PORT_DOWN)
def test_create_port_profile(self):
+ """
+ Tests creation of port profile
+ """
LOG.debug("UCSVICTestPlugin:test_create_port_profile() called\n")
new_network = db.network_create(self.tenant_id, self.net_name)
cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name,
profile_name)
def test_delete_port_profile(self):
+ """
+ Tests deletion of port profile
+ """
LOG.debug("UCSVICTestPlugin:test_delete_port_profile() called\n")
new_network = db.network_create(self.tenant_id, self.net_name)
cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name,
device_ip=self.device_ip)
self.assertEqual(port_bind[const.VLANNAME], new_vlan_name)
self.assertEqual(port_bind[const.VLANID], new_vlanid)
- self.tearDownNetworkPortInterface(
+ self.tear_down_network_port_interface(
self.tenant_id, new_net_dict[const.NET_ID],
new_port[const.UUID])
port_dict[const.PORTID], device_ip=self.device_ip)
self.assertEqual(port_bind[const.VLANNAME], self.vlan_name)
self.assertEqual(port_bind[const.VLANID], self.vlan_id)
- self.tearDownNetworkPortInterface(
+ self.tear_down_network_port_interface(
self.tenant_id, new_net_dict[const.NET_ID],
new_port[const.UUID])
def test_get_vlan_name_for_network(self):
+ """
+ Tests retrieval of vlan name
+ """
LOG.debug("UCSVICTestPlugin:test_get_vlan_name_for_network() called\n")
new_network = db.network_create(self.tenant_id, self.net_name)
cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name,
self.assertEqual(vlan_bind_name, self.vlan_name)
def test_get_vlan_id_for_network(self):
+ """
+ Tests retrieval of vlan id
+ """
LOG.debug("UCSVICTestPlugin:test_get_vlan_id_for_network() called\n")
new_network = db.network_create(self.tenant_id, self.net_name)
cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name,
self.tenant_id, new_network[const.UUID])
self.assertEqual(str(vlan_bind_id), self.vlan_id)
- def test_show_network_NetworkNotFound(self):
+ def test_show_network_not_found(self):
+ """
+ Negative Test for show network details when network not found
+ """
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin.get_network_details,
self.tenant_id, self.net_id,
device_ip=self.device_ip)
- def test_delete_network_NetworkNotFound(self):
+ def test_delete_network_not_found(self):
+ """
+ Negative Test for delete network when network not found
+ """
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin.delete_network,
self.tenant_id, self.net_id,
device_ip=self.device_ip)
def test_delete_port_PortNotFound(self):
+ """
+ Negative Test for delete port when port not found
+ """
new_network = db.network_create(self.tenant_id, self.net_name)
cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name,
new_network[const.UUID])
self.port_id, device_ip=self.device_ip,
ucs_inventory=self._ucs_inventory,
chassis_id=self.chassis_id, blade_id=self.blade_id,
- blade_intf_distinguished_name=self.\
- blade_intf_distinguished_name,
+ blade_intf_distinguished_name=None,
least_rsvd_blade_dict=self._ucs_inventory.\
_get_least_reserved_blade())
- self.tearDownNetwork(self.tenant_id, new_net_dict[const.NET_ID])
+ self.tear_down_network(self.tenant_id, new_net_dict[const.NET_ID])
def tearDown(self):
"""Clear the test environment"""
# Remove database contents
db.clear_db()
- def tearDownNetwork(self, tenant_id, net_id):
+ def tear_down_network(self, tenant_id, net_id):
self._cisco_ucs_plugin.delete_network(tenant_id, net_id,
device_ip=self.device_ip)
- def tearDownNetworkPort(self, tenant_id, net_id, port_id):
+ def tear_down_network_port(self, tenant_id, net_id, port_id):
+ blade_intf_details = self._ucs_inventory._get_rsvd_blade_intf_by_port(
+ tenant_id, port_id)
self._cisco_ucs_plugin.delete_port(
tenant_id, net_id, port_id, device_ip=self.device_ip,
ucs_inventory=self._ucs_inventory,
chassis_id=self.chassis_id, blade_id=self.blade_id,
- blade_intf_distinguished_name=self.\
- blade_intf_distinguished_name,
+ blade_intf_distinguished_name=blade_intf_details[\
+ const.BLADE_INTF_DN],
least_rsvd_blade_dict=self._ucs_inventory.\
_get_least_reserved_blade())
- self.tearDownNetwork(tenant_id, net_id)
+ self.tear_down_network(tenant_id, net_id)
- def tearDownNetworkPortInterface(self, tenant_id, net_id, port_id):
+ def tear_down_network_port_interface(self, tenant_id, net_id, port_id):
self._cisco_ucs_plugin.unplug_interface(
tenant_id, net_id, port_id,
device_ip=self.device_ip)
- self.tearDownNetworkPort(tenant_id, net_id, port_id)
+ self.tear_down_network_port(tenant_id, net_id, port_id)
--- /dev/null
+"""
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright 2011 Cisco Systems, Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# @author: Peter Strunk, Cisco Systems, Inc.
+#
+"""
+
+import unittest
+import logging as LOG
+
+from quantum.common import exceptions as exc
+from quantum.plugins.cisco import l2network_plugin_configuration as conf
+from quantum.plugins.cisco.common import cisco_exceptions as c_exc
+from quantum.plugins.cisco.common import cisco_credentials as creds
+from quantum.plugins.cisco.db import api as db
+from quantum.plugins.cisco.db import l2network_db as cdb
+from quantum.plugins.cisco.segmentation.l2network_vlan_mgr \
+ import L2NetworkVLANMgr
+
+LOG.basicConfig(level=LOG.WARN)
+LOG.getLogger(__name__)
+
+
+class Test_L2Network_Vlan_Mgr(unittest.TestCase):
+
+ _plugins = {}
+ _inventory = {}
+
+ def setUp(self):
+ db.configure_db({'sql_connection': 'sqlite:///:memory:'})
+ cdb.initialize()
+ creds.Store.initialize()
+ self.tenant_id = "network_admin"
+ self.net_name = "TestNetwork1"
+ self.vlan_name = "TestVlan1"
+ self.vlan_id = 300
+ self.net_id = 100
+ self.vlan_mgr = L2NetworkVLANMgr()
+ self.plugin_key = "quantum.plugins.cisco.ucs.cisco_ucs_plugin" +\
+ ".UCSVICPlugin"
+
+ def tearDown(self):
+ db.clear_db()
+
+ def test_reserve_segmentation_id(self):
+ LOG.debug("test_reserve_segmentation_id - START")
+ db.network_create(self.tenant_id, self.net_name)
+ vlan_id = self.vlan_mgr.reserve_segmentation_id(self.tenant_id,
+ self.net_name)
+ self.assertEqual(vlan_id, int(conf.VLAN_START))
+ LOG.debug("test_reserve_segmentation_id - END")
+
+ def test_reserve_segmentation_id_NA(self):
+ LOG.debug("test_reserve_segmentation_id - START")
+ db.clear_db()
+ self.assertRaises(c_exc.VlanIDNotAvailable,
+ self.vlan_mgr.reserve_segmentation_id,
+ self.tenant_id,
+ self.net_name)
+ LOG.debug("test_reserve_segmentation_id - END")
+
+ def test_release_segmentation_id(self):
+ LOG.debug("test_release_segmentation_id - START")
+ db.network_create(self.tenant_id, self.net_name)
+ vlan_id = self.vlan_mgr.reserve_segmentation_id(self.tenant_id,
+ self.net_name)
+ cdb.add_vlan_binding(vlan_id, self.vlan_name, self.net_id)
+ release_return = self.vlan_mgr.release_segmentation_id(self.tenant_id,
+ self.net_id)
+ self.assertEqual(release_return, False)
+ LOG.debug("test_release_segmentation_id - END")
+
+ def test_release_segmentation_id_idDNE(self):
+ LOG.debug("test_release_segmentation_idDNE - START")
+ db.network_create(self.tenant_id, self.net_name)
+ self.assertRaises(exc.NetworkNotFound,
+ self.vlan_mgr.release_segmentation_id,
+ self.tenant_id,
+ self.net_id)
+ LOG.debug("test_release_segmentation_idDNE - END")
#
"""
from copy import deepcopy
-import logging as LOG
+import logging
from quantum.common import exceptions as exc
from quantum.plugins.cisco.l2device_inventory_base \
import cisco_ucs_inventory_configuration as conf
from quantum.plugins.cisco.ucs import cisco_ucs_network_driver
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(__name__)
+LOG = logging.getLogger(__name__)
"""
The _inventory data strcuture contains a nested disctioary:
port_binding[const.INSTANCE_ID]
intf_data[const.VIF_ID] = \
port_binding[const.VIF_ID]
-
host_name = self._get_host_name(ucsm_ip, chassis_id, blade_id)
blade_data = {const.BLADE_INTF_DATA: blade_intf_data,
const.BLADE_UNRESERVED_INTF_COUNT: unreserved_counter,
port_id = port_binding[const.PORTID]
udb.update_portbinding(port_id, instance_id=instance_id,
vif_id=vif_id)
- db.port_set_attachment_by_id(port_id, vif_id)
+ db.port_set_attachment_by_id(port_id, vif_id +
+ const.UNPLUGGED)
device_name = intf_data[const.BLADE_INTF_RHEL_DEVICE_NAME]
profile_name = port_binding[const.PORTPROFILENAME]
dynamicnic_details = \
const.UCSPROFILE: profile_name}
LOG.debug("Found reserved dynamic nic: %s" \
"associated with port %s" %
- (blade_intf_data[blade_intf], port_id))
+ (intf_data, port_id))
LOG.debug("Returning dynamic nic details: %s" %
dynamicnic_details)
return dynamicnic_details
tenant_id)
return None
- def _disassociate_vifid_from_port(self, tenant_id, port_id):
+ def _disassociate_vifid_from_port(self, tenant_id, instance_id, vif_id):
"""
- Return the device name for a reserved interface
+ Disassociate a VIF-ID from a port, this happens when a
+ VM is destroyed
"""
for ucsm_ip in self._inventory_state.keys():
ucsm = self._inventory_state[ucsm_ip]
if intf_data[const.BLADE_INTF_RESERVATION] == \
const.BLADE_INTF_RESERVED and \
intf_data[const.TENANTID] == tenant_id and \
- intf_data[const.PORTID] == port_id:
- vif_id = intf_data[const.VIF_ID]
+ blade_intf_data[blade_intf][const.INSTANCE_ID] == \
+ instance_id and \
+ intf_data[const.VIF_ID][:const.UUID_LENGTH] == \
+ vif_id:
intf_data[const.VIF_ID] = None
intf_data[const.INSTANCE_ID] = None
+ port_binding = udb.get_portbinding_dn(blade_intf)
+ port_id = port_binding[const.PORTID]
udb.update_portbinding(port_id, instance_id=None,
vif_id=None)
+ db.port_unset_attachment_by_id(port_id)
LOG.debug("Disassociated VIF-ID: %s " \
"from port: %s" \
"in UCS inventory state for blade: %s" %
(vif_id, port_id, intf_data))
- return
+ device_params = {const.DEVICE_IP: [ucsm_ip],
+ const.PORTID: port_id}
+ return device_params
LOG.warn("Disassociating VIF-ID in UCS inventory failed. " \
"Could not find a reserved dynamic nic for tenant: %s" %
tenant_id)
on which a dynamic vnic was reserved for this port
"""
LOG.debug("unplug_interface() called\n")
- tenant_id = args[0]
- port_id = args[2]
- self._disassociate_vifid_from_port(tenant_id, port_id)
return self._get_blade_for_port(args)
def schedule_host(self, args):
vif_id = args[2][const.VIF_ID]
vif_info = self._get_instance_port(tenant_id, instance_id, vif_id)
vif_desc = {const.VIF_DESC: vif_info}
+
LOG.debug("vif_desc is: %s" % vif_desc)
return vif_desc
+ def detach_port(self, args):
+ """
+ Remove the VIF-ID and instance name association
+ with the port
+ """
+ LOG.debug("detach_port() called\n")
+ instance_id = args[1]
+ tenant_id = args[2][const.PROJECT_ID]
+ vif_id = args[2][const.VIF_ID]
+ device_params = self._disassociate_vifid_from_port(tenant_id,
+ instance_id,
+ vif_id)
+ return device_params
+
def create_multiport(self, args):
"""
Create multiple ports for a VM
"""
import httplib
-import logging as LOG
+import logging
from xml.etree import ElementTree as et
from quantum.plugins.cisco.common import cisco_exceptions as cexc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.ucs import cisco_getvif as gvif
-
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+LOG = logging.getLogger(__name__)
COOKIE_VALUE = "cookie_placeholder"
PROFILE_NAME = "profilename_placeholder"
#
"""
-import logging as LOG
+import logging
from quantum.common import exceptions as exc
from quantum.common import utils
from quantum.plugins.cisco.l2device_plugin_base import L2DevicePluginBase
from quantum.plugins.cisco.ucs import cisco_ucs_configuration as conf
-LOG.basicConfig(level=LOG.WARN)
-LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+LOG = logging.getLogger(__name__)
class UCSVICPlugin(L2DevicePluginBase):
return udb.update_portbinding(port_id, vlan_name=new_vlan_name,
vlan_id=conf.DEFAULT_VLAN_ID)
+ def create_multiport(self, tenant_id, net_id_list, ports_num, port_id_list,
+ **kwargs):
+ """
+ Creates a port on the specified Virtual Network.
+ """
+ LOG.debug("UCSVICPlugin:create_multiport() called\n")
+ self._set_ucsm(kwargs[const.DEVICE_IP])
+ qos = None
+ ucs_inventory = kwargs[const.UCS_INVENTORY]
+ least_rsvd_blade_dict = kwargs[const.LEAST_RSVD_BLADE_DICT]
+ chassis_id = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_CHASSIS]
+ blade_id = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_ID]
+ blade_data_dict = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_DATA]
+ port_binding_list = []
+ for port_id, net_id in zip(port_id_list, net_id_list):
+ new_port_profile = \
+ self._create_port_profile(tenant_id, net_id, port_id,
+ conf.DEFAULT_VLAN_NAME,
+ conf.DEFAULT_VLAN_ID)
+ profile_name = new_port_profile[const.PROFILE_NAME]
+ rsvd_nic_dict = ucs_inventory.\
+ reserve_blade_interface(self._ucsm_ip, chassis_id,
+ blade_id, blade_data_dict,
+ tenant_id, port_id,
+ profile_name)
+ port_binding = udb.update_portbinding(port_id,
+ portprofile_name=profile_name,
+ vlan_name=conf.DEFAULT_VLAN_NAME,
+ vlan_id=conf.DEFAULT_VLAN_ID,
+ qos=qos)
+ port_binding_list.append(port_binding)
+ return port_binding_list
+
+ def detach_port(self, tenant_id, instance_id, instance_desc, **kwargs):
+ """
+ Remove the association of the VIF with the dynamic vnic
+ """
+ LOG.debug("detach_port() called\n")
+ port_id = kwargs[const.PORTID]
+ kwargs.pop(const.PORTID)
+ return self.unplug_interface(tenant_id, None, port_id, **kwargs)
+
def _get_profile_name(self, port_id):
"""Returns the port profile name based on the port UUID"""
profile_name = conf.PROFILE_NAME_PREFIX \
self._ucsm_ip = ucsm_ip
self._ucsm_username = cred.Store.getUsername(conf.UCSM_IP_ADDRESS)
self._ucsm_password = cred.Store.getPassword(conf.UCSM_IP_ADDRESS)
-
- def create_multiport(self, tenant_id, net_id_list, ports_num, port_id_list,
- **kwargs):
- """
- Creates a port on the specified Virtual Network.
- """
- LOG.debug("UCSVICPlugin:create_multiport() called\n")
- self._set_ucsm(kwargs[const.DEVICE_IP])
- qos = None
- ucs_inventory = kwargs[const.UCS_INVENTORY]
- least_rsvd_blade_dict = kwargs[const.LEAST_RSVD_BLADE_DICT]
- chassis_id = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_CHASSIS]
- blade_id = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_ID]
- blade_data_dict = least_rsvd_blade_dict[const.LEAST_RSVD_BLADE_DATA]
- port_binding_list = []
- for port_id, net_id in zip(port_id_list, net_id_list):
- new_port_profile = \
- self._create_port_profile(tenant_id, net_id, port_id,
- conf.DEFAULT_VLAN_NAME,
- conf.DEFAULT_VLAN_ID)
- profile_name = new_port_profile[const.PROFILE_NAME]
- rsvd_nic_dict = ucs_inventory.\
- reserve_blade_interface(self._ucsm_ip, chassis_id,
- blade_id, blade_data_dict,
- tenant_id, port_id,
- profile_name)
- port_binding = udb.update_portbinding(port_id,
- portprofile_name=profile_name,
- vlan_name=conf.DEFAULT_VLAN_NAME,
- vlan_id=conf.DEFAULT_VLAN_ID,
- qos=qos)
- port_binding_list.append(port_binding)
- return port_binding_list
Code Review / openstack-build / neutron-build.git / commitdiff