# @author: Sumit Naiksatam, Cisco Systems, Inc.
# @author: Rohit Agarwalla, Cisco Systems, Inc.
-"""
-Exceptions used by the Cisco plugin
-"""
+"""Exceptions used by the Cisco plugin."""
from quantum.common import exceptions
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
- """Generate a WSGI response based on the
- exception passed to constructor.
+ """Generate a WSGI response.
+
+ Response is generated based on the exception passed to constructor.
"""
# Replace the body with fault details.
code = self.wrapped_exc.status_int
class PortNotFound(webob.exc.HTTPClientError):
- """
+ """PortNotFound exception.
+
subclass of :class:`~HTTPClientError`
This indicates that the server did not find the port specified
class CredentialNotFound(webob.exc.HTTPClientError):
- """
+ """CredentialNotFound exception.
+
subclass of :class:`~HTTPClientError`
This indicates that the server did not find the Credential specified
class QosNotFound(webob.exc.HTTPClientError):
- """
+ """QosNotFound exception.
+
subclass of :class:`~HTTPClientError`
This indicates that the server did not find the QoS specified
class NovatenantNotFound(webob.exc.HTTPClientError):
- """
+ """NovatenantNotFound exception.
+
subclass of :class:`~HTTPClientError`
This indicates that the server did not find the Novatenant specified
class RequestedStateInvalid(webob.exc.HTTPClientError):
- """
+ """RequestedStateInvalid exception.
+
subclass of :class:`~HTTPClientError`
This indicates that the server could not update the port state to
def get16ByteUUID(uuid):
- """
- Return a 16 byte has of the UUID, used when smaller unique
- ID is required.
+ """Return first 16 bytes of UUID.
+
+ Used when smaller unique ID is required.
"""
return hashlib.md5(uuid).hexdigest()[:16]
class CiscoConfigOptions():
"""Cisco Configuration Options Class."""
+
def __init__(self):
self._create_nexus_dictionary()
def _create_nexus_dictionary(self):
- """
- Create the Nexus dictionary from the cisco_plugins.ini
- NEXUS_SWITCH section(s).
+ """Create the Nexus dictionary.
+
+ Reads data from cisco_plugins.ini NEXUS_SWITCH section(s).
"""
for parsed_file in cfg.CONF._cparser.parsed:
for parsed_item in parsed_file.keys():
def configure_db(options):
- """
- Establish the database, create an engine if needed, and
- register the models.
+ """Configure database.
+
+ Establish the database, create an engine if needed, and register the
+ models.
:param options: Mapping of configuration options
"""
class L2NetworkBase(object):
"""Base class for L2Network Models."""
+
__table_args__ = {'mysql_engine': 'InnoDB'}
def __setitem__(self, key, value):
setattr(self, k, v)
def iteritems(self):
- """Make the model object behave like a dict"
+ """Make the model object behave like a dict.
+
Includes attributes from joins.
"""
local = dict(self)
class VlanID(BASE, L2NetworkBase):
"""Represents a vlan_id usage."""
+
__tablename__ = 'vlan_ids'
vlan_id = Column(Integer, primary_key=True)
class VlanBinding(BASE, L2NetworkBase):
"""Represents a binding of vlan_id to network_id."""
+
__tablename__ = 'vlan_bindings'
vlan_id = Column(Integer, primary_key=True)
class QoS(BASE, L2NetworkBase):
"""Represents QoS for a tenant."""
+
__tablename__ = 'qoss'
qos_id = Column(String(255))
class Credential(BASE, L2NetworkBase):
"""Represents credentials for a tenant."""
+
__tablename__ = 'credentials'
credential_id = Column(String(255))
class QuantumBase(object):
"""Base class for Quantum Models."""
+
__table_args__ = {'mysql_engine': 'InnoDB'}
def __setitem__(self, key, value):
def iteritems(self):
"""Make the model object behave like a dict.
+
Includes attributes from joins.
"""
local = dict(self)
class Port(BASE, QuantumBase):
"""Represents a port on a quantum network."""
+
__tablename__ = 'ports'
uuid = Column(String(255), primary_key=True)
class Network(BASE, QuantumBase):
"""Represents a quantum network."""
+
__tablename__ = 'networks'
uuid = Column(String(255), primary_key=True)
class L2NetworkBase(object):
"""Base class for L2Network Models."""
+
#__table_args__ = {'mysql_engine': 'InnoDB'}
def __setitem__(self, key, value):
setattr(self, k, v)
def iteritems(self):
- """Make the model object behave like a dict"
+ """Make the model object behave like a dict.
+
Includes attributes from joins.
"""
local = dict(self)
class QoS(model_base.BASEV2, L2NetworkBase):
"""Represents QoS for a tenant."""
+
__tablename__ = 'qoss'
qos_id = Column(String(255))
class Credential(model_base.BASEV2, L2NetworkBase):
"""Represents credentials for a tenant."""
+
__tablename__ = 'credentials'
credential_id = Column(String(255))
class NexusPortBinding(model_base.BASEV2, L2NetworkBase):
"""Represents a binding of VM's to nexus ports."""
+
__tablename__ = "nexusport_bindings"
id = Column(Integer, primary_key=True, autoincrement=True)
def get_view_builder(req):
- """get view builder."""
base_url = req.application_url
return ViewBuilder(base_url)
class ViewBuilder(object):
- """
- ViewBuilder for Credential,
- derived from quantum.views.networks
- """
+ """ViewBuilder for Credential, derived from quantum.views.networks."""
+
def __init__(self, base_url):
- """
+ """Initialize builder.
+
:param base_url: url of the root wsgi application
"""
self.base_url = base_url
def get_view_builder(req):
- """get view builder."""
base_url = req.application_url
return ViewBuilder(base_url)
class ViewBuilder(object):
- """
- ViewBuilder for QoS,
- derived from quantum.views.networks
- """
+ """ViewBuilder for QoS, derived from quantum.views.networks."""
+
def __init__(self, base_url):
- """
+ """Initialize builder.
+
:param base_url: url of the root wsgi application
"""
self.base_url = base_url
class Credential(extensions.ExtensionDescriptor):
- """extension class Credential."""
+ """Extension class Credential."""
@classmethod
def get_name(cls):
class L2DevicePluginBase(object):
- """
- Base class for a device-specific plugin.
+ """Base class for a device-specific plugin.
+
An example of a device-specific plugin is a Nexus switch plugin.
The network model relies on device-category-specific plugins to perform
the configuration on each device.
@abstractmethod
def get_all_networks(self, tenant_id, **kwargs):
- """
+ """Get newtorks.
+
:returns:
:raises:
"""
@abstractmethod
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
**kwargs):
- """
+ """Create network.
+
:returns:
:raises:
"""
@abstractmethod
def delete_network(self, tenant_id, net_id, **kwargs):
- """
+ """Delete network.
+
:returns:
:raises:
"""
@abstractmethod
def get_network_details(self, tenant_id, net_id, **kwargs):
- """
+ """Get network details.
+
:returns:
:raises:
"""
@abstractmethod
def update_network(self, tenant_id, net_id, name, **kwargs):
- """
+ """Update network.
+
:returns:
:raises:
"""
@abstractmethod
def get_all_ports(self, tenant_id, net_id, **kwargs):
- """
+ """Get ports.
+
:returns:
:raises:
"""
@abstractmethod
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
- """
+ """Create port.
+
:returns:
:raises:
"""
@abstractmethod
def delete_port(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Delete port.
+
:returns:
:raises:
"""
@abstractmethod
def update_port(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Update port.
+
:returns:
:raises:
"""
@abstractmethod
def get_port_details(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Get port details.
+
:returns:
:raises:
"""
@abstractmethod
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs):
- """
+ """Plug interface.
+
:returns:
:raises:
"""
@abstractmethod
def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Unplug interface.
+
:returns:
:raises:
"""
def create_subnet(self, tenant_id, net_id, ip_version,
subnet_cidr, **kwargs):
- """
+ """Create subnet.
+
:returns:
:raises:
"""
pass
def get_subnets(self, tenant_id, net_id, **kwargs):
- """
+ """Get subnets.
+
:returns:
:raises:
"""
pass
def get_subnet(self, tenant_id, net_id, subnet_id, **kwargs):
- """
+ """Get subnet.
+
:returns:
:raises:
"""
pass
def update_subnet(self, tenant_id, net_id, subnet_id, **kwargs):
- """
+ """Update subnet.
+
:returns:
:raises:
"""
pass
def delete_subnet(self, tenant_id, net_id, subnet_id, **kwargs):
- """
+ """Delete subnet.
+
:returns:
:raises:
"""
@classmethod
def __subclasshook__(cls, klass):
- """
+ """Check plugin class.
+
The __subclasshook__ method is a class method
that will be called everytime a class is tested
using issubclass(klass, Plugin).
class VirtualPhysicalSwitchModelV2(quantum_plugin_base_v2.QuantumPluginBaseV2):
- """
+ """Virtual Physical Switch Model.
+
This implementation works with OVS and Nexus plugin for the
following topology:
One or more servers to a nexus switch.
'get_subnet', 'get_subnets', ]
def __init__(self):
- """
- Initialize the segmentation manager, check which device plugins are
- configured, and load the inventories those device plugins for which the
- inventory is configured
+ """Initialize the segmentation manager.
+
+ Checks which device plugins are configured, and load the inventories
+ those device plugins for which the inventory is configured.
"""
conf.CiscoConfigOptions()
'name': self.__class__.__name__})
def __getattribute__(self, name):
- """
+ """Delegate calls to OVS sub-plugin.
+
This delegates the calls to the methods implemented only by the OVS
sub-plugin. Note: Currently, bulking is handled by the caller
(PluginV2), and this model class expects to receive only non-bulking
return func_name
def _invoke_plugin_per_device(self, plugin_key, function_name, args):
- """
+ """Invoke plugin per device.
+
Invokes a device plugin's relevant functions (based on the
plugin implementation) for completing this operation.
"""
device_params)]
def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
- """
+ """Invoke plugin.
+
Invokes the relevant function on a device plugin's
implementation for completing this operation.
"""
return host
def create_network(self, context, network):
- """
+ """Create network.
+
Perform this operation in the context of the configured device
plugins.
"""
raise
def update_network(self, context, id, network):
- """
+ """Update network.
+
Perform this operation in the context of the configured device
plugins.
"""
return ovs_output[0]
def delete_network(self, context, id):
- """
+ """Delete network.
+
Perform this operation in the context of the configured device
plugins.
"""
return nexus_output
def create_port(self, context, port):
- """
+ """Create port.
+
Perform this operation in the context of the configured device
plugins.
"""
pass
def update_port(self, context, id, port):
- """
+ """Update port.
+
Perform this operation in the context of the configured device
plugins.
"""
raise
def delete_port(self, context, id):
- """
+ """Delete port.
+
Perform this operation in the context of the configured device
plugins.
"""
class PluginV2(db_base_plugin_v2.QuantumDbPluginV2):
- """
- Meta-Plugin with v2 API support for multiple sub-plugins.
- """
+ """Meta-Plugin with v2 API support for multiple sub-plugins."""
+
supported_extension_aliases = ["Cisco Credential", "Cisco qos"]
_methods_to_delegate = ['create_network',
'delete_network', 'update_network', 'get_network',
_master = True
def __init__(self):
- """
- Loads the model class.
- """
+ """Load the model class."""
self._model = importutils.import_object(config.CISCO.model_class)
if hasattr(self._model, "MANAGE_STATE") and self._model.MANAGE_STATE:
self._master = False
LOG.debug(_("Plugin initialization complete"))
def __getattribute__(self, name):
- """
+ """Delegate core API calls to the model class.
+
When the configured model class offers to manage the state of the
logical resources, we delegate the core API calls directly to it.
Note: Bulking calls will be handled by this class, and turned into
return object.__getattribute__(self, name)
def __getattr__(self, name):
- """
+ """Delegate calls to the extensions.
+
This delegates the calls to the extensions explicitly implemented by
the model.
"""
Core API implementation
"""
def create_network(self, context, network):
- """
- Creates a new Virtual Network, and assigns it
- a symbolic name.
- """
+ """Create new Virtual Network, and assigns it a symbolic name."""
LOG.debug(_("create_network() called"))
new_network = super(PluginV2, self).create_network(context,
network)
raise
def update_network(self, context, id, network):
- """
- Updates the symbolic name belonging to a particular
- Virtual Network.
+ """Update network.
+
+ Updates the symbolic name belonging to a particular Virtual Network.
"""
LOG.debug(_("update_network() called"))
upd_net_dict = super(PluginV2, self).update_network(context, id,
return upd_net_dict
def delete_network(self, context, id):
- """
+ """Delete network.
+
Deletes the network with the specified network identifier
belonging to the specified tenant.
"""
raise
def get_network(self, context, id, fields=None):
- """
- Gets a particular network
- """
+ """Get a particular network."""
LOG.debug(_("get_network() called"))
return super(PluginV2, self).get_network(context, id, fields)
def get_networks(self, context, filters=None, fields=None):
- """
- Gets all networks
- """
+ """Get all networks."""
LOG.debug(_("get_networks() called"))
return super(PluginV2, self).get_networks(context, filters, fields)
def create_port(self, context, port):
- """
- Creates a port on the specified Virtual Network.
- """
+ """Create a port on the specified Virtual Network."""
LOG.debug(_("create_port() called"))
new_port = super(PluginV2, self).create_port(context, port)
try:
raise
def delete_port(self, context, id):
- """
- Deletes a port
- """
LOG.debug(_("delete_port() called"))
port = self._get_port(context, id)
- """
+ """Delete port.
+
TODO (Sumit): Disabling this check for now, check later
- #Allow deleting a port only if the administrative state is down,
- #and its operation status is also down
- if port['admin_state_up'] or port['status'] == 'ACTIVE':
- raise exc.PortInUse(port_id=id, net_id=port['network_id'],
- att_id=port['device_id'])
+ Allow deleting a port only if the administrative state is down,
+ and its operation status is also down
+ #if port['admin_state_up'] or port['status'] == 'ACTIVE':
+ # raise exc.PortInUse(port_id=id, net_id=port['network_id'],
+ # att_id=port['device_id'])
"""
try:
kwargs = {const.PORT: port}
raise
def update_port(self, context, id, port):
- """
- Updates the state of a port and returns the updated port
- """
+ """Update the state of a port and return the updated port."""
LOG.debug(_("update_port() called"))
try:
self._invoke_device_plugins(self._func_name(), [context, id,
raise
def create_subnet(self, context, subnet):
- """
+ """Create subnet.
+
Create a subnet, which represents a range of IP addresses
that can be allocated to devices.
"""
raise
def update_subnet(self, context, id, subnet):
- """
- Updates the state of a subnet and returns the updated subnet
- """
+ """Updates the state of a subnet and returns the updated subnet."""
LOG.debug(_("update_subnet() called"))
try:
self._invoke_device_plugins(self._func_name(), [context, id,
raise
def delete_subnet(self, context, id):
- """
- Deletes a subnet
- """
LOG.debug(_("delete_subnet() called"))
with context.session.begin():
subnet = self._get_subnet(context, id)
return host_list
def associate_port(self, tenant_id, instance_id, instance_desc):
- """
- Get the portprofile name and the device name for the dynamic vnic
+ """Associate port.
+
+ Get the portprofile name and the device name for the dynamic vnic.
"""
LOG.debug(_("associate_port() called"))
return self._invoke_device_plugins(self._func_name(), [tenant_id,
instance_desc])
def detach_port(self, tenant_id, instance_id, instance_desc):
- """
- Remove the association of the VIF with the dynamic vnic
- """
+ """Remove the association of the VIF with the dynamic vnic."""
LOG.debug(_("detach_port() called"))
return self._invoke_device_plugins(self._func_name(), [tenant_id,
instance_id,
Private functions
"""
def _invoke_device_plugins(self, function_name, args):
- """
- Device-specific calls including core API and extensions are
- delegated to the model.
+ """Device-specific calls.
+
+ Including core API and extensions are delegated to the model.
"""
if hasattr(self._model, function_name):
return getattr(self._model, function_name)(*args)
class CiscoNEXUSDriver():
- """
- Nexus Driver Main Class
- """
+ """Nexus Driver Main Class."""
def __init__(self):
pass
def nxos_connect(self, nexus_host, nexus_ssh_port, nexus_user,
nexus_password):
- """
- Makes the SSH connection to the Nexus Switch
- """
+ """Make SSH connection to the Nexus Switch."""
man = manager.connect(host=nexus_host, port=nexus_ssh_port,
username=nexus_user, password=nexus_password)
return man
def create_xml_snippet(self, cutomized_config):
- """
- Creates the Proper XML structure for the Nexus Switch Configuration
+ """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 enable_vlan(self, mgr, vlanid, vlanname):
- """
- Creates a VLAN on Nexus Switch given the VLAN ID and Name
- """
+ """Creates a VLAN on Nexus Switch given the VLAN ID and Name."""
confstr = snipp.CMD_VLAN_CONF_SNIPPET % (vlanid, vlanname)
confstr = self.create_xml_snippet(confstr)
mgr.edit_config(target='running', config=confstr)
def disable_vlan(self, mgr, vlanid):
- """
- Delete a VLAN on Nexus Switch given the VLAN ID
- """
+ """Delete a VLAN on Nexus Switch given the VLAN ID."""
confstr = snipp.CMD_NO_VLAN_CONF_SNIPPET % vlanid
confstr = self.create_xml_snippet(confstr)
mgr.edit_config(target='running', config=confstr)
def enable_port_trunk(self, mgr, interface):
- """
- Enables trunk mode an interface on Nexus Switch
- """
+ """Enable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_PORT_TRUNK % (interface)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
mgr.edit_config(target='running', config=confstr)
def disable_switch_port(self, mgr, interface):
- """
- Disables trunk mode an interface on Nexus Switch
- """
+ """Disable trunk mode an interface on Nexus Switch."""
confstr = snipp.CMD_NO_SWITCHPORT % (interface)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
mgr.edit_config(target='running', config=confstr)
def enable_vlan_on_trunk_int(self, mgr, interface, vlanid):
- """
+ """Enable vlan in trunk interface.
+
Enables trunk mode vlan access an interface on Nexus Switch given
- VLANID
+ VLANID.
"""
confstr = snipp.CMD_VLAN_INT_SNIPPET % (interface, vlanid)
confstr = self.create_xml_snippet(confstr)
mgr.edit_config(target='running', config=confstr)
def disable_vlan_on_trunk_int(self, mgr, interface, vlanid):
- """
- Enables trunk mode vlan access an interface on Nexus Switch given
- VLANID
+ """Disable VLAN.
+
+ Disables trunk mode vlan access an interface on Nexus Switch given
+ VLANID.
"""
confstr = snipp.CMD_NO_VLAN_INT_SNIPPET % (interface, vlanid)
confstr = self.create_xml_snippet(confstr)
def create_vlan(self, vlan_name, vlan_id, nexus_host, nexus_user,
nexus_password, nexus_ports,
nexus_ssh_port, vlan_ids=None):
- """
+ """Create VLAN and enablt in on the interface.
+
Creates a VLAN and Enable on trunk mode an interface on Nexus Switch
- given the VLAN ID and Name and Interface Number
+ given the VLAN ID and Name and Interface Number.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
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
+ given the VLAN ID and Interface Number.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
self.disable_vlan_on_trunk_int(man, ports, vlan_id)
def build_vlans_cmd(self):
- """
- Builds a string with all the VLANs on the same Switch
- """
+ """Builds a string with all the VLANs on the same Switch."""
assigned_vlan = cdb.get_all_vlanids_used()
vlans = ''
for vlanid in assigned_vlan:
def add_vlan_int(self, vlan_id, nexus_host, nexus_user, nexus_password,
nexus_ports, nexus_ssh_port, vlan_ids=None):
- """
- Adds a vlan from interfaces on the Nexus switch given the VLAN ID
+ """Add vlan.
+
+ Adds a vlan from interfaces on the Nexus switch given the VLAN ID.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
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
+ """Remove vlan.
+
+ Removes a vlan from interfaces on the Nexus switch given the VLAN ID.
"""
man = self.nxos_connect(nexus_host, int(nexus_ssh_port),
nexus_user, nexus_password)
class NexusPlugin(L2DevicePluginBase):
- """
- Nexus PLugIn Main Class
- """
+ """Nexus PlugIn Main Class."""
_networks = {}
def __init__(self):
- """
- Extracts the configuration parameters from the configuration file
- """
+ """Extract configuration parameters from the configuration file."""
self._client = importutils.import_object(conf.CISCO.nexus_driver)
LOG.debug(_("Loaded driver %s"), conf.CISCO.nexus_driver)
self._nexus_switches = conf.get_nexus_dictionary()
return self.credentials[nexus_ip]
def get_all_networks(self, tenant_id):
- """
- Returns a dictionary containing all
- <network_uuid, network_name> for
+ """Get all networks.
+
+ Returns a dictionary containing all <network_uuid, network_name> for
the specified tenant.
"""
LOG.debug(_("NexusPlugin:get_all_networks() called"))
def create_network(self, tenant_id, net_name, net_id, vlan_name, vlan_id,
host, instance):
- """
- Create a VLAN in the appropriate switch/port,
- and configure the appropriate interfaces
- for this VLAN
+ """Create network.
+
+ Create a VLAN in the appropriate switch/port, and configure the
+ appropriate interfaces for this VLAN.
"""
LOG.debug(_("NexusPlugin:create_network() called"))
# Grab the switch IP and port for this host
return new_net_dict
def delete_network(self, tenant_id, net_id, **kwargs):
- """
+ """Delete network.
+
Deletes the VLAN in all switches, and removes the VLAN configuration
- from the relevant interfaces
+ from the relevant interfaces.
"""
LOG.debug(_("NexusPlugin:delete_network() called"))
def get_network_details(self, tenant_id, net_id, **kwargs):
- """
- Returns the details of a particular network
- """
+ """Return the details of a particular network."""
LOG.debug(_("NexusPlugin:get_network_details() called"))
network = self._get_network(tenant_id, net_id)
return network
def update_network(self, tenant_id, net_id, **kwargs):
- """
- Updates the properties of a particular
- Virtual Network.
- """
+ """Update the properties of a particular Virtual Network."""
LOG.debug(_("NexusPlugin:update_network() called"))
def get_all_ports(self, tenant_id, net_id, **kwargs):
- """
+ """Get all ports.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
LOG.debug(_("NexusPlugin:get_all_ports() called"))
def create_port(self, tenant_id, net_id, port_state, port_id, **kwargs):
- """
+ """Create port.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
LOG.debug(_("NexusPlugin:create_port() called"))
def delete_port(self, device_id, vlan_id):
- """
- Delete port bindings from the database and scan
- whether the network is still required on
- the interfaces trunked
+ """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 for this port
return row['instance_id']
def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
- """
+ """Update port.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
LOG.debug(_("NexusPlugin:update_port() called"))
def get_port_details(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Get port details.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
def plug_interface(self, tenant_id, net_id, port_id, remote_interface_id,
**kwargs):
- """
+ """Plug interfaces.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
LOG.debug(_("NexusPlugin:plug_interface() called"))
def unplug_interface(self, tenant_id, net_id, port_id, **kwargs):
- """
+ """Unplug interface.
+
This is probably not applicable to the Nexus plugin.
Delete if not required.
"""
def _get_vlan_id_for_network(self, tenant_id, network_id, context,
base_plugin_ref):
- """
- Obtain the VLAN ID given the Network ID
- """
+ """Obtain the VLAN ID given the Network ID."""
vlan = cdb.get_vlan_binding(network_id)
return vlan.vlan_id
def _get_network(self, tenant_id, network_id, context, base_plugin_ref):
- """
- Gets the NETWORK ID
- """
+ """Get the Network ID."""
network = base_plugin_ref._get_network(context, network_id)
if not network:
raise exc.NetworkNotFound(net_id=network_id)
super(ExtensionsTestApp, self).__init__(mapper)
def create_request(self, path, body, content_type, method='GET'):
-
"""Test create request."""
LOG.debug("test_create_request - START")
return req
def _create_network(self, name=None):
-
"""Test create network."""
LOG.debug("Creating network - START")
return network_data['network']['id']
def _create_port(self, network_id, port_state):
-
"""Test create port."""
LOG.debug("Creating port for network %s - START", network_id)
class QosExtensionTest(base.BaseTestCase):
def setUp(self):
-
"""Set up function."""
super(QosExtensionTest, self).setUp()
self._l2network_plugin = l2network_plugin.L2Network()
def test_create_qos(self):
-
"""Test create qos."""
LOG.debug("test_create_qos - START")
LOG.debug("test_create_qos - END")
def test_create_qosBADRequest(self):
-
"""Test create qos bad request."""
LOG.debug("test_create_qosBADRequest - START")
LOG.debug("test_create_qosBADRequest - END")
def test_list_qoss(self):
-
"""Test list qoss."""
LOG.debug("test_list_qoss - START")
LOG.debug("test_list_qoss - END")
def test_show_qos(self):
-
"""Test show qos."""
LOG.debug("test_show_qos - START")
LOG.debug("test_show_qos - END")
def test_show_qosDNE(self, qos_id='100'):
-
"""Test show qos does not exist."""
LOG.debug("test_show_qosDNE - START")
LOG.debug("test_show_qosDNE - END")
def test_update_qos(self):
-
"""Test update qos."""
LOG.debug("test_update_qos - START")
LOG.debug("test_update_qos - END")
def test_update_qosDNE(self, qos_id='100'):
-
"""Test update qos does not exist."""
LOG.debug("test_update_qosDNE - START")
LOG.debug("test_update_qosDNE - END")
def test_update_qosBADRequest(self):
-
"""Test update qos bad request."""
LOG.debug("test_update_qosBADRequest - START")
LOG.debug("test_update_qosBADRequest - END")
def test_delete_qos(self):
-
"""Test delte qos."""
LOG.debug("test_delete_qos - START")
LOG.debug("test_delete_qos - END")
def test_delete_qosDNE(self, qos_id='100'):
-
"""Test delte qos does not exist."""
LOG.debug("test_delete_qosDNE - START")
LOG.debug("test_delete_qosDNE - END")
def tearDownQos(self, delete_profile_path):
-
"""Tear Down Qos."""
self.test_app.delete(delete_profile_path)
class CredentialExtensionTest(base.BaseTestCase):
def setUp(self):
-
"""Set up function."""
super(CredentialExtensionTest, self).setUp()
self._l2network_plugin = l2network_plugin.L2Network()
def test_list_credentials(self):
-
"""Test list credentials."""
#Create Credential before listing
LOG.debug("test_list_credentials - END")
def test_create_credential(self):
-
"""Test create credential."""
LOG.debug("test_create_credential - START")
LOG.debug("test_create_credential - END")
def test_create_credentialBADRequest(self):
-
"""Test create credential bad request."""
LOG.debug("test_create_credentialBADRequest - START")
LOG.debug("test_create_credentialBADRequest - END")
def test_show_credential(self):
-
"""Test show credential."""
LOG.debug("test_show_credential - START")
LOG.debug("test_show_credential - END")
def test_show_credentialDNE(self, credential_id='100'):
-
"""Test show credential does not exist."""
LOG.debug("test_show_credentialDNE - START")
LOG.debug("test_show_credentialDNE - END")
def test_update_credential(self):
-
"""Test update credential."""
LOG.debug("test_update_credential - START")
LOG.debug("test_update_credential - END")
def test_update_credBADReq(self):
-
"""Test update credential bad request."""
LOG.debug("test_update_credBADReq - START")
LOG.debug("test_update_credBADReq - END")
def test_update_credentialDNE(self, credential_id='100'):
-
"""Test update credential does not exist."""
LOG.debug("test_update_credentialDNE - START")
LOG.debug("test_update_credentialDNE - END")
def test_delete_credential(self):
-
"""Test delete credential."""
LOG.debug("test_delete_credential - START")
LOG.debug("test_delete_credential - END")
def test_delete_credentialDNE(self, credential_id='100'):
-
"""Test delete credential does not exist."""
LOG.debug("test_delete_credentialDNE - START")
class NexusDB(object):
"""Class consisting of methods to call nexus db methods."""
+
def get_all_nexusportbindings(self):
- """get all nexus port bindings."""
+ """Get all nexus port bindings."""
bindings = []
try:
for bind in nexus_db.get_all_nexusport_bindings():
return bindings
def get_nexusportbinding(self, vlan_id):
- """get nexus port binding."""
+ """Get nexus port binding."""
binding = []
try:
for bind in nexus_db.get_nexusport_binding(vlan_id):
return binding
def create_nexusportbinding(self, port_id, vlan_id):
- """create nexus port binding."""
+ """Create nexus port binding."""
bind_dict = {}
try:
res = nexus_db.add_nexusport_binding(port_id, vlan_id)
LOG.error("Failed to create nexus binding: %s" % str(exc))
def delete_nexusportbinding(self, vlan_id):
- """delete nexus port binding."""
+ """Delete nexus port binding."""
bindings = []
try:
bind = nexus_db.remove_nexusport_binding(vlan_id)
% str(exc))
def update_nexusport_binding(self, port_id, new_vlan_id):
- """update nexus port binding."""
+ """Update nexus port binding."""
try:
res = nexus_db.update_nexusport_binding(port_id, new_vlan_id)
LOG.debug("Updating nexus port binding : %s" % res.port_id)
LOG.debug("Setup")
def testa_create_nexusportbinding(self):
- """create nexus port binding."""
+ """Create nexus port binding."""
binding1 = self.dbtest.create_nexusportbinding("port1", 10)
self.assertTrue(binding1["port-id"] == "port1")
self.tearDown_nexusportbinding()
def testb_getall_nexusportbindings(self):
- """get all nexus port binding."""
+ """Get all nexus port bindings."""
self.dbtest.create_nexusportbinding("port1", 10)
self.dbtest.create_nexusportbinding("port2", 10)
bindings = self.dbtest.get_all_nexusportbindings()
self.tearDown_nexusportbinding()
def testc_delete_nexusportbinding(self):
- """delete nexus port binding."""
+ """Delete nexus port binding."""
self.dbtest.create_nexusportbinding("port1", 10)
self.dbtest.delete_nexusportbinding(10)
bindings = self.dbtest.get_all_nexusportbindings()
self.tearDown_nexusportbinding()
def testd_update_nexusportbinding(self):
- """update nexus port binding."""
+ """Update nexus port binding."""
binding1 = self.dbtest.create_nexusportbinding("port1", 10)
binding1 = self.dbtest.update_nexusport_binding(binding1["port-id"],
20)
self.tearDown_nexusportbinding()
def tearDown_nexusportbinding(self):
- """tear down nexusport binding table."""
+ """Tear down nexus port binding table."""
LOG.debug("Tearing Down Nexus port Bindings")
binds = self.dbtest.get_all_nexusportbindings()
for bind in binds:
LOG.debug("Setup")
def testa_create_vlanbinding(self):
- """test add vlan binding."""
+ """Test add vlan binding."""
net1 = self.quantum.create_network("t1", "netid1")
vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", net1["net-id"])
self.assertTrue(vlan1["vlan-id"] == "10")
self.teardown_network()
def testb_getall_vlanbindings(self):
- """test get all vlan binding."""
+ """Test get all vlan bindings."""
net1 = self.quantum.create_network("t1", "netid1")
net2 = self.quantum.create_network("t1", "netid2")
vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", net1["net-id"])
self.teardown_network()
def testc_delete_vlanbinding(self):
- """test delete vlan binding."""
+ """Test delete vlan binding."""
net1 = self.quantum.create_network("t1", "netid1")
vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", net1["net-id"])
self.assertTrue(vlan1["vlan-id"] == "10")
self.teardown_network()
def testd_update_vlanbinding(self):
- """test update vlan binding."""
+ """Test update vlan binding."""
net1 = self.quantum.create_network("t1", "netid1")
vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", net1["net-id"])
self.assertTrue(vlan1["vlan-id"] == "10")
self.teardown_network()
def testm_test_vlanids(self):
- """test vlanid methods."""
+ """Test vlanid methods."""
l2network_db.create_vlanids()
vlanids = l2network_db.get_all_vlanids()
self.assertTrue(len(vlanids) > 0)
LOG.debug("Setup")
def testa_create_network(self):
- """test to create network."""
+ """Test to create network."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["net-name"] == "plugin_test1")
self.teardown_network_port()
def testb_get_networks(self):
- """test to get all networks."""
+ """Test to get all networks."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["net-name"] == "plugin_test1")
net2 = self.dbtest.create_network(self.tenant_id, "plugin_test2")
self.teardown_network_port()
def testc_delete_network(self):
- """test to delete network."""
+ """Test to delete network."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["net-name"] == "plugin_test1")
self.dbtest.delete_network(net1["net-id"])
self.teardown_network_port()
def testd_update_network(self):
- """test to update (rename) network."""
+ """Test to update (rename) network."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
self.assertTrue(net1["net-name"] == "plugin_test1")
net = self.dbtest.update_network(self.tenant_id, net1["net-id"],
self.teardown_network_port()
def teste_create_port(self):
- """test to create port."""
+ """Test to create port."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
port = self.dbtest.create_port(net1["net-id"])
self.assertTrue(port["net-id"] == net1["net-id"])
self.teardown_network_port()
def testf_delete_port(self):
- """test to delete port."""
+ """Test to delete port."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
port = self.dbtest.create_port(net1["net-id"])
self.assertTrue(port["net-id"] == net1["net-id"])
self.teardown_network_port()
def testg_plug_unplug_interface(self):
- """test to plug/unplug interface."""
+ """Test to plug/unplug interface."""
net1 = self.dbtest.create_network(self.tenant_id, "plugin_test1")
port1 = self.dbtest.create_port(net1["net-id"])
self.dbtest.plug_interface(net1["net-id"], port1["port-id"], "vif1.1")
self.teardown_network_port()
def testh_joined_test(self):
- """test to get network and port."""
+ """Test to get network and port."""
net1 = self.dbtest.create_network("t1", "net1")
port1 = self.dbtest.create_port(net1["net-id"])
self.assertTrue(port1["net-id"] == net1["net-id"])
class CiscoNEXUSFakeDriver():
- """
- Nexus Driver Fake Class
- """
+ """Nexus Driver Fake Class."""
+
def __init__(self):
pass
def nxos_connect(self, nexus_host, nexus_ssh_port, nexus_user,
nexus_password):
- """
- Makes the fake connection to the Nexus Switch
- """
+ """Make the fake connection to the Nexus Switch."""
pass
def create_xml_snippet(self, cutomized_config):
- """
- Creates the Proper XML structure for the Nexus Switch Configuration
+ """Create XML snippet.
+
+ Creates the Proper XML structure for the Nexus Switch
+ Configuration.
"""
pass
def enable_vlan(self, mgr, vlanid, vlanname):
- """
- Creates a VLAN on Nexus Switch given the VLAN ID and Name
- """
+ """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
- """
+ """Delete a VLAN on Nexus Switch given the VLAN ID."""
pass
def enable_port_trunk(self, mgr, interface):
- """
- Enables trunk mode an interface on Nexus Switch
- """
+ """Enable trunk mode an interface on Nexus Switch."""
pass
def disable_switch_port(self, mgr, interface):
- """
- Disables trunk mode an interface on Nexus Switch
- """
+ """Disable trunk mode an interface on Nexus Switch."""
pass
def enable_vlan_on_trunk_int(self, mgr, interface, vlanid):
- """
- Enables trunk mode vlan access an interface on Nexus Switch given
- 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
+ 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
+ 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
+ given the VLAN ID and Interface Number.
"""
pass
def build_vlans_cmd(self):
- """
- Builds a string with all the VLANs on the same Switch
- """
+ """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):
- """
- Adds a vlan from interfaces on the Nexus switch given the VLAN ID
- """
+ """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):
- """
- Removes a vlan from interfaces on the Nexus switch given the VLAN ID
+ """Remove vlan from interfaces.
+
+ Removes a vlan from interfaces on the Nexus switch given the VLAN ID.
"""
pass
Code Review / openstack-build / neutron-build.git / commitdiff