openstack/deb-networking-cisco
Code Issues Proposed changes

Cisco UCS Manager vendor code for ML2 mechanism driver.

Browse Source 
Change-Id: I23e9db27adc6efda415ef3f3d3046a988fbf7e53
This commit is contained in:
Sandhya Dasu
2015-02-26 16:11:24 -05:00
parent f0dff47bd4
commit 7f57b6caff
13 changed files with 1331 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
etc/neutron/plugins/ml2/ml2_conf_cisco.ini
View File

@@ -116,3 +116,22 @@
# encap=vlan-100
# cidr_exposed=10.10.40.2/16
# gateway_ip=10.10.40.1
[ml2_cisco_ucsm]
# Cisco UCS Manager IP address
# ucsm_ip=1.1.1.1
# Username to connect to UCS Manager
# ucsm_username=user
# Password to connect to UCS Manager
# ucsm_password=password
# SR-IOV and VM-FEX vendors supported by this plugin
# xxxx:yyyy represents vendor_id:product_id
# supported_pci_devs=['2222:3333', '4444:5555']
# Hostname to Service profile mapping for UCS Manager
# controlled compute hosts
# ucsm_host_list=Hostname1:Serviceprofile1, Hostname2:Serviceprofile2

0
networking_cisco/plugins/ml2/drivers/cisco/ucsm/__init__.py Normal file
View File

76
networking_cisco/plugins/ml2/drivers/cisco/ucsm/config.py Normal file
View File

@@ -0,0 +1,76 @@
# Copyright 2015 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.
from oslo_config import cfg
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import constants as const
""" Cisco UCS Manager ML2 Mechanism driver specific configuration.
Following are user configurable options for UCS Manager ML2 Mechanism
driver. The ucsm_username, ucsm_password, and ucsm_ip are
required options. Additional configuration knobs are provided to pre-
create UCS Manager port profiles.
"""
ml2_cisco_ucsm_opts = [
cfg.StrOpt('ucsm_ip',
help=_('Cisco UCS Manager IP address. This is a required field '
'to communicate with a Cisco UCS Manager.')),
cfg.StrOpt('ucsm_username',
help=_('Username for UCS Manager. This is a required field '
'to communicate with a Cisco UCS Manager.')),
cfg.StrOpt('ucsm_password',
secret=True, # do not expose value in the logs
help=_('Password for UCS Manager. This is a required field '
'to communicate with a Cisco UCS Manager.')),
cfg.ListOpt('supported_pci_devs',
default=[const.PCI_INFO_CISCO_VIC_1240,
const.PCI_INFO_INTEL_82599],
help=_('List of comma separated vendor_id:product_id of '
'SR_IOV capable devices supported by this MD. This MD '
'supports both VM-FEX and SR-IOV devices.')),
cfg.ListOpt('ucsm_host_list',
help=_('List of comma separated Host:Service Profile tuples '
'providing the Service Profile associated with each '
'Host to be supported by this MD.')),
]
cfg.CONF.register_opts(ml2_cisco_ucsm_opts, "ml2_cisco_ucsm")
def parse_pci_vendor_config():
vendor_list = []
vendor_config_list = cfg.CONF.ml2_cisco_ucsm.supported_pci_devs
for vendor in vendor_config_list:
vendor_product = vendor.split(':')
if len(vendor_product) != 2:
raise cfg.Error(_("UCS Mech Driver: Invalid PCI device "
"config: %s") % vendor)
vendor_list.append(vendor)
return vendor_list
def parse_ucsm_host_config():
host_dict = {}
host_config_list = cfg.CONF.ml2_cisco_ucsm.ucsm_host_list
for host in host_config_list:
host_sp = host.split(':')
if len(host_sp) != 2:
raise cfg.Error(_("UCS Mech Driver: Invalid Host Service "
"Profile config: %s") % host)
key = host_sp[0]
host_dict[key] = host_sp[1]
return host_dict

41
networking_cisco/plugins/ml2/drivers/cisco/ucsm/constants.py Normal file
View File

@@ -0,0 +1,41 @@
# Copyright 2015 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.
# VIF_DETAILS attribute for port profile name used by the Nova VIF driver.
VIF_DETAILS_PROFILEID = 'profileid'
# Supported network interface cards
PCI_INFO_CISCO_VIC_1240 = "1137:0071"
PCI_INFO_INTEL_82599 = "8086:10c9"
VLAN_PATH = "fabric/lan"
VLAN_COMPRESSION_TYPE = "included"
DESCR = "Created by Openstack UCSM Mech Driver"
PORT_PROFILESETDN = "fabric/lan/profiles"
HIGH_PERF = "high-perf-reqd"
NONE = "none"
VNIC_PATH_PREFIX = "/vnic-"
VLAN_PATH_PREFIX = "/if-"
VLAN_PROFILE_PATH_PREFIX = "/net-"
VLAN_PROFILE_NAME_PREFIX = "OS-"
PORT_PROFILE_NAME_PREFIX = "OS-PP-"
CLIENT_PROFILE_NAME_PREFIX = "OS-CL-"
CLIENT_PROFILE_PATH_PREFIX = "/cl-"
SERVICE_PROFILE_PATH_PREFIX = "org-root/ls-"
ETH0 = "/ether-eth0"
ETH1 = "/ether-eth1"
DUPLICATE_EXCEPTION = "object already exists"

33
networking_cisco/plugins/ml2/drivers/cisco/ucsm/exceptions.py Normal file
View File

@@ -0,0 +1,33 @@
# Copyright 2015 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.
"""Exceptions used by Cisco UCSM ML2 mechanism driver."""
from neutron.common import exceptions
class UcsmConnectFailed(exceptions.NeutronException):
message = _("Unable to connect to UCS Manager %(ucsm_ip)s. "
"Reason: %(exc)s.")
class UcsmConfigReadFailed(exceptions.NeutronException):
message = _("Unable to read config from UCS Manager %(ucsm_ip)s. "
"Reason: %(exc)s.")
class UcsmConfigFailed(exceptions.NeutronException):
message = _("Failed to configure %(config)s on UCS Manager %(ucsm_ip)s. "
"Reason: %(exc)s.")

70
networking_cisco/plugins/ml2/drivers/cisco/ucsm/ucsm_db.py Normal file
View File

@@ -0,0 +1,70 @@
# Copyright 2015 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.
from sqlalchemy import orm
from neutron.db import api as db_api
from neutron.plugins.ml2.drivers.cisco.ucsm import ucsm_model
class UcsmDbModel(object):
def __init__(self):
self.session = db_api.get_session()
def is_port_profile_created(self, vlan_id):
"""Indicates if port profile has been created on UCS Manager."""
entry = self.session.query(ucsm_model.PortProfile).filter_by(
vlan_id=vlan_id).first()
return entry and entry.created_on_ucs
def get_port_profile_for_vlan(self, vlan_id):
"""Returns Vlan id associated with the port profile."""
entry = self.session.query(ucsm_model.PortProfile).filter_by(
vlan_id=vlan_id).first()
return entry.profile_id if entry else None
def add_port_profile(self, profile_name, vlan_id):
"""Adds a port profile and its vlan_id to the table."""
if not self.get_port_profile_for_vlan(vlan_id):
port_profile = ucsm_model.PortProfile(profile_id=profile_name,
vlan_id=vlan_id,
created_on_ucs=False)
with self.session.begin(subtransactions=True):
self.session.add(port_profile)
return port_profile
def set_port_profile_created(self, vlan_id, profile_name):
"""Sets created_on_ucs flag to True."""
with self.session.begin(subtransactions=True):
port_profile = self.session.query(
ucsm_model.PortProfile).filter_by(
vlan_id=vlan_id, profile_id=profile_name).first()
if port_profile:
port_profile.created_on_ucs = True
self.session.merge(port_profile)
else:
new_profile = ucsm_model.PortProfile(profile_id=profile_name,
vlan_id=vlan_id,
created_on_ucs=True)
self.session.add(new_profile)
def delete_vlan_entry(self, vlan_id):
"""Deletes entry for a vlan_id if it exists."""
with self.session.begin(subtransactions=True):
try:
self.session.query(ucsm_model.PortProfile).filter_by(
vlan_id=vlan_id).delete()
except orm.exc.NoResultFound:
return

548
networking_cisco/plugins/ml2/drivers/cisco/ucsm/ucsm_network_driver.py Normal file
View File

@@ -0,0 +1,548 @@
# Copyright 2015 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.
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import importutils
from neutron.i18n import _LE, _LI
from neutron.extensions import portbindings
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import config as config
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import constants as const
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import exceptions as cexc
LOG = logging.getLogger(__name__)
class CiscoUcsmDriver(object):
"""UCS Manager Driver Main Class."""
def __init__(self):
LOG.debug("UCS Manager Network driver found")
self.ucsmsdk = None
self.ucsm_ip = cfg.CONF.ml2_cisco_ucsm.ucsm_ip
self.username = cfg.CONF.ml2_cisco_ucsm.ucsm_username
self.password = cfg.CONF.ml2_cisco_ucsm.ucsm_password
LOG.debug("UCS Manager Network driver Ip: %s", self.ucsm_ip)
self.handles = {}
self.supported_sriov_vnic_types = [portbindings.VNIC_DIRECT,
portbindings.VNIC_MACVTAP]
self.supported_pci_devs = config.parse_pci_vendor_config()
self.ucsm_host_dict = config.parse_ucsm_host_config()
def _validate_config(self):
if not cfg.CONF.ml2_cisco_ucsm.get('ucsm_ip'):
msg = _('UCS Manager IP address is not provided')
LOG.error(msg)
if not cfg.CONF.ml2_cisco_ucsm.get('ucsm_username'):
msg = _('UCS Manager username is not provided')
LOG.error(msg)
if not cfg.CONF.ml2_cisco_ucsm.get('ucsm_password'):
msg = _('UCS Manager password is not provided')
LOG.error(msg)
def check_vnic_type_and_vendor_info(self, vnic_type, profile):
"""Checks if this vnic_type and vendor device info are supported.
Returns True if:
1. the port vnic_type is direct or macvtap and
2. the vendor_id and product_id of the port is supported by
this MD
Useful in determining if this MD should bind the current
port.
"""
# Check for vnic_type
if vnic_type not in self.supported_sriov_vnic_types:
LOG.info(_LI('Non SR-IOV vnic_type: %s.'), vnic_type)
return False
if not profile:
LOG.debug("Port binding missing profile info")
return False
# Check for vendor_info
return self._check_for_supported_vendor(profile)
def _check_for_supported_vendor(self, profile):
"""Checks if the port belongs to a supported vendor.
Returns True for supported_pci_devs.
"""
vendor_info = profile.get('pci_vendor_info')
if not vendor_info:
LOG.debug("Port binding missing pci vendor info")
return False
if vendor_info not in self.supported_pci_devs:
LOG.debug("Unsupported vendor and product type %s",
str(vendor_info))
return False
return True
def is_vmfex_port(self, profile):
"""Checks if the port is a VMFEX port.
Returns True only for port that support VM-FEX.
It is important to distinguish between the two since Port Profiles
on the UCS Manager are created only for the VM-FEX ports.
"""
vendor_info = profile.get('pci_vendor_info')
return vendor_info == const.PCI_INFO_CISCO_VIC_1240
def _import_ucsmsdk(self):
"""Imports the Ucsm SDK module.
This module is not installed as part of the normal Neutron
distributions. It is imported dynamically in this module so that
the import can be mocked, allowing unit testing without requiring
the installation of UcsSdk.
"""
return importutils.import_module('UcsSdk')
def ucs_manager_connect(self):
"""Connects to a UCS Manager."""
self._validate_config()
if not self.ucsmsdk:
self.ucsmsdk = self._import_ucsmsdk()
handle = self.ucsmsdk.UcsHandle()
try:
handle.Login(self.ucsm_ip, self.username, self.password)
self.handles[self.ucsm_ip] = handle
except Exception as e:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConnectFailed(ucsm_ip=self.ucsm_ip, exc=e)
return self.handles[self.ucsm_ip]
def _get_all_portprofiles(self, handle):
"""Gets all port profiles from a specific UCS Manager."""
# Get Managed Object VnicProfile
try:
port_profiles = handle.GetManagedObject(
None,
self.ucsmsdk.VnicProfile.ClassId())
except Exception as e:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigReadFailed(ucsm_ip=self.ucsm_ip, exc=e)
return port_profiles
def _create_vlanprofile(self, handle, vlan_id):
"""Creates VLAN profile to be assosiated with the Port Profile."""
vlan_name = self.make_vlan_name(vlan_id)
vlan_profile_dest = (const.VLAN_PATH + const.VLAN_PROFILE_PATH_PREFIX +
vlan_name)
LOG.debug("Creating Vlan Profile: %s", vlan_name)
try:
vp1 = handle.GetManagedObject(
None,
self.ucsmsdk.FabricLanCloud.ClassId(),
{self.ucsmsdk.FabricLanCloud.DN: const.VLAN_PATH})
if not vp1:
LOG.debug("UCS Manager network driver Vlan Profile "
"path at %s missing", const.VLAN_PATH)
return False
#Create a vlan profile with the given vlan_id
vp2 = handle.AddManagedObject(
vp1,
self.ucsmsdk.FabricVlan.ClassId(),
{self.ucsmsdk.FabricVlan.COMPRESSION_TYPE:
const.VLAN_COMPRESSION_TYPE,
self.ucsmsdk.FabricVlan.DN: vlan_profile_dest,
self.ucsmsdk.FabricVlan.SHARING: const.NONE,
self.ucsmsdk.FabricVlan.PUB_NW_NAME: "",
self.ucsmsdk.FabricVlan.ID: str(vlan_id),
self.ucsmsdk.FabricVlan.MCAST_POLICY_NAME: "",
self.ucsmsdk.FabricVlan.NAME: vlan_name,
self.ucsmsdk.FabricVlan.DEFAULT_NET: "no"})
if not vp2:
LOG.debug("UCS Manager network driver could not create Vlan "
"Profile %s", vlan_name)
return False
LOG.debug("UCS Manager network driver created Vlan Profile %s "
"at %s", vlan_name, vlan_profile_dest)
return True
except Exception as e:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigFailed(config=vlan_name,
ucsm_ip=self.ucsm_ip, exc=e)
def _create_port_profile(self, handle, profile_name, vlan_id, vnic_type):
"""Creates a Port Profile on the UCS Manager.
Significant parameters set in the port profile are:
1. Port profile name - Should match what was set in vif_details
2. High performance mode - For VM-FEX to be enabled/configured on
the port using this port profile, this mode should be enabled.
3. Vlan id - Vlan id used by traffic to and from the port.
"""
port_profile_dest = (const.PORT_PROFILESETDN + const.VNIC_PATH_PREFIX +
profile_name)
# Max ports that this port profile can be applied to
max_ports = 64
vlan_name = self.make_vlan_name(vlan_id)
vlan_associate_path = (const.PORT_PROFILESETDN +
const.VNIC_PATH_PREFIX + profile_name +
const.VLAN_PATH_PREFIX + vlan_name)
cl_profile_name = const.CLIENT_PROFILE_NAME_PREFIX + str(vlan_id)
cl_profile_dest = (const.PORT_PROFILESETDN + const.VNIC_PATH_PREFIX +
profile_name + const.CLIENT_PROFILE_PATH_PREFIX +
cl_profile_name)
# Check if direct or macvtap mode
if vnic_type == portbindings.VNIC_DIRECT:
port_mode = const.HIGH_PERF
else:
port_mode = const.NONE
try:
port_profile = handle.GetManagedObject(
None,
self.ucsmsdk.VnicProfileSet.ClassId(),
{self.ucsmsdk.VnicProfileSet.DN: const.PORT_PROFILESETDN})
if not port_profile:
LOG.debug("UCS Manager network driver Port Profile path at "
"%s missing", const.PORT_PROFILESETDN)
return False
# Create a port profile on the UCS Manager
p_profile = handle.AddManagedObject(
port_profile,
self.ucsmsdk.VnicProfile.ClassId(),
{self.ucsmsdk.VnicProfile.NAME: profile_name,
self.ucsmsdk.VnicProfile.POLICY_OWNER: "local",
self.ucsmsdk.VnicProfile.NW_CTRL_POLICY_NAME: "",
self.ucsmsdk.VnicProfile.PIN_TO_GROUP_NAME: "",
self.ucsmsdk.VnicProfile.DN: port_profile_dest,
self.ucsmsdk.VnicProfile.DESCR: const.DESCR,
self.ucsmsdk.VnicProfile.QOS_POLICY_NAME: "",
self.ucsmsdk.VnicProfile.HOST_NW_IOPERF: port_mode,
self.ucsmsdk.VnicProfile.MAX_PORTS: max_ports})
if not p_profile:
LOG.debug("UCS Manager network driver could not create Port "
"Profile %s at %s", profile_name, port_profile_dest)
return False
LOG.debug("UCS Manager network driver associating Vlan Profile "
"with Port Profile at %s", vlan_associate_path)
# Associate port profile with vlan profile
mo = handle.AddManagedObject(
p_profile,
self.ucsmsdk.VnicEtherIf.ClassId(),
{self.ucsmsdk.VnicEtherIf.DN: vlan_associate_path,
self.ucsmsdk.VnicEtherIf.NAME: vlan_name,
self.ucsmsdk.VnicEtherIf.DEFAULT_NET: "yes"}, True)
if not mo:
LOG.debug("UCS Manager network driver cannot associate Vlan "
"Profile %s to Port Profile %s", vlan_name,
profile_name)
return False
LOG.debug("UCS Manager network driver created Port Profile %s "
"at %s", profile_name, port_profile_dest)
cl_profile = handle.AddManagedObject(
p_profile,
self.ucsmsdk.VmVnicProfCl.ClassId(),
{self.ucsmsdk.VmVnicProfCl.ORG_PATH: ".*",
self.ucsmsdk.VmVnicProfCl.DN: cl_profile_dest,
self.ucsmsdk.VmVnicProfCl.NAME: cl_profile_name,
self.ucsmsdk.VmVnicProfCl.POLICY_OWNER: "local",
self.ucsmsdk.VmVnicProfCl.SW_NAME: ".*",
self.ucsmsdk.VmVnicProfCl.DC_NAME: ".*",
self.ucsmsdk.VmVnicProfCl.DESCR: const.DESCR})
if not cl_profile:
LOG.debug("UCS Manager network driver could not create Client "
"Profile %s at %s", cl_profile_name, cl_profile_dest)
return False
LOG.debug("UCS Manager network driver created Client Profile %s "
"at %s", cl_profile_name, cl_profile_dest)
return True
except Exception as e:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigFailed(config=profile_name,
ucsm_ip=self.ucsm_ip, exc=e)
def create_portprofile(self, profile_name, vlan_id, vnic_type):
"""Top level method to create Port Profiles on the UCS Manager.
Calls all the methods responsible for the individual tasks that
ultimately result in the creation of the Port Profile on the UCS
Manager.
"""
# Connect to UCS Manager
handle = self.ucs_manager_connect()
if not handle:
LOG.error(_LE('UCS Manager network driver failed to connect '
'to UCS Manager.'))
return False
try:
handle.StartTransaction()
# Create Vlan Profile
if not self._create_vlanprofile(handle, vlan_id):
LOG.error(_LE('UCS Manager network driver failed to create '
'Vlan Profile for vlan %s'), str(vlan_id))
return False
# Create Port Profile
if not self._create_port_profile(handle, profile_name, vlan_id,
vnic_type):
LOG.error(_LE('UCS Manager network driver failed to create '
'Port Profile %s'), profile_name)
return False
handle.CompleteTransaction()
except Exception as e:
if const.DUPLICATE_EXCEPTION in str(e):
LOG.debug("UCS Manager network driver found an entry that "
"already exists %s", profile_name)
else:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigFailed(config=profile_name,
ucsm_ip=self.ucsm_ip, exc=e)
finally:
# Disconnect from UCS Manager
self.ucs_manager_disconnect()
return True
def _update_service_profile(self, handle, service_profile, vlan_id):
service_profile_path = (const.SERVICE_PROFILE_PATH_PREFIX +
str(service_profile))
eth0 = service_profile_path + const.ETH0
eth1 = service_profile_path + const.ETH1
eth_port_paths = [eth0, eth1]
obj = handle.GetManagedObject(
None,
self.ucsmsdk.LsServer.ClassId(),
{self.ucsmsdk.LsServer.DN: service_profile_path})
if not obj:
LOG.debug("UCS Manager network driver could not find Service "
"Profile %s", service_profile_path)
return False
for eth_port_path in eth_port_paths:
eth = handle.GetManagedObject(
obj, self.ucsmsdk.VnicEther.ClassId(),
{self.ucsmsdk.VnicEther.DN: eth_port_path}, True)
if eth:
vlan_name = self.make_vlan_name(vlan_id)
vlan_path = eth_port_path + const.VLAN_PATH_PREFIX + vlan_name
eth_if = handle.AddManagedObject(eth,
self.ucsmsdk.VnicEtherIf.ClassId(),
{self.ucsmsdk.VnicEtherIf.DN: vlan_path,
self.ucsmsdk.VnicEtherIf.NAME: vlan_name,
self.ucsmsdk.VnicEtherIf.DEFAULT_NET: "no"}, True)
if not eth_if:
LOG.debug("UCS Manager network driver could not update "
"Service Profile %s with %s", service_profile,
str(vlan_id))
return False
else:
LOG.debug("UCS Manager network driver did not find ethernet "
"port at %s", eth_port_path)
return True
def update_serviceprofile(self, host_id, vlan_id):
service_profile = self.ucsm_host_dict[host_id]
if service_profile:
LOG.debug("UCS Manager network driver Service Profile : %s",
service_profile)
else:
LOG.debug("UCS Manager network driver does not support Host_id %s",
str(host_id))
return
# Connect to UCS Manager
handle = self.ucs_manager_connect()
if not handle:
LOG.error(_LE('UCS Manager network driver failed to connect '
'to UCS Manager.'))
return False
try:
handle.StartTransaction()
# Create Vlan Profile
if not self._create_vlanprofile(handle, vlan_id):
LOG.error(_LE('UCS Manager network driver failed to create '
'Vlan Profile for vlan %s'), str(vlan_id))
return False
# Update Service Profile
if not self._update_service_profile(
handle,
service_profile,
vlan_id):
LOG.error(_LE('UCS Manager network driver failed to update '
'Service Profile %s'), service_profile)
return False
# Complete Transaction
handle.CompleteTransaction()
except Exception as e:
if const.DUPLICATE_EXCEPTION in str(e):
LOG.debug("UCS Manager network driver found an entry that "
"already exists %s", service_profile)
else:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigFailed(config=service_profile,
ucsm_ip=self.ucsm_ip, exc=e)
finally:
# Disconnect from UCS Manager
self.ucs_manager_disconnect()
def _delete_vlan_profile(self, handle, vlan_id):
vlan_name = self.make_vlan_name(vlan_id)
vlan_profile_dest = (const.VLAN_PATH + const.VLAN_PROFILE_PATH_PREFIX +
vlan_name)
obj = handle.GetManagedObject(
None,
self.ucsmsdk.FabricVlan.ClassId(),
{self.ucsmsdk.FabricVlan.DN: vlan_profile_dest})
if obj:
handle.RemoveManagedObject(obj)
return True
def _delete_port_profile(self, handle, port_profile):
port_profile_dest = (const.PORT_PROFILESETDN + const.VNIC_PATH_PREFIX +
port_profile)
# Find port profile on the UCS Manager
p_profile = handle.GetManagedObject(
None,
self.ucsmsdk.VnicProfile.ClassId(),
{self.ucsmsdk.VnicProfile.NAME: port_profile,
self.ucsmsdk.VnicProfile.DN: port_profile_dest})
if not p_profile:
LOG.debug("UCS Manager network driver did not find port profile "
"%s to delete", port_profile_dest)
return
handle.RemoveManagedObject(p_profile)
return True
def _remove_vlan_from_all_service_profiles(self, handle, vlan_id):
service_profile_list = []
for host_id, value in self.ucsm_host_dict.iteritems():
if value:
service_profile_list.append(value)
if not service_profile_list:
# Nothing to do
return
for service_profile in service_profile_list:
service_profile_path = (const.SERVICE_PROFILE_PATH_PREFIX +
service_profile)
eth0 = service_profile_path + const.ETH0
eth1 = service_profile_path + const.ETH1
eth_port_paths = [eth0, eth1]
# 1. From the Service Profile config, access the configuration for
# its ports.
# 2. Check if that Vlan has been configured on each of the ports
# 3. If Vlan conifg found, remove it.
obj = handle.GetManagedObject(
None,
self.ucsmsdk.LsServer.ClassId(),
{self.ucsmsdk.LsServer.DN: service_profile_path})
if obj:
# Check if this vlan_id has been configured on the ports
# in this Service profile
for eth_port_path in eth_port_paths:
eth = handle.GetManagedObject(
obj, self.ucsmsdk.VnicEther.ClassId(),
{self.ucsmsdk.VnicEther.DN: eth_port_path}, True)
if eth:
vlan_name = self.make_vlan_name(vlan_id)
vlan_path = eth_port_path + "/if-" + vlan_name
vlan = handle.GetManagedObject(eth,
self.ucsmsdk.VnicEtherIf.ClassId(),
{self.ucsmsdk.VnicEtherIf.DN: vlan_path})
if vlan:
# Found vlan config. Now remove it.
handle.RemoveManagedObject(vlan)
return True
def delete_all_config_for_vlan(self, vlan_id, port_profile):
# Connect to UCS Manager
handle = self.ucs_manager_connect()
if not handle:
LOG.error(_LE('UCS Manager network driver failed to connect '
'to UCS Manager.'))
return False
try:
handle.StartTransaction()
self._delete_port_profile(handle, port_profile)
self._remove_vlan_from_all_service_profiles(handle, vlan_id)
self._delete_vlan_profile(handle, vlan_id)
# Complete Transaction
handle.CompleteTransaction()
except Exception as e:
# Raise a Neutron exception. Include a description of
# the original exception.
raise cexc.UcsmConfigFailed(config=vlan_id,
ucsm_ip=self.ucsm_ip, exc=e)
finally:
# Disconnect from UCS Manager
self.ucs_manager_disconnect()
def ucs_manager_disconnect(self):
"""Disconnects from the UCS Manager.
After the disconnect, the handle associated with this connection
is no longer valid.
"""
handle = self.handles[self.ucsm_ip]
handle.Logout()
@staticmethod
def make_vlan_name(vlan_id):
return const.VLAN_PROFILE_NAME_PREFIX + str(vlan_id)

6
networking_cisco/tests/unit/ml2/drivers/cisco/n1kv/test_n1kv_db.py
View File

@@ -154,12 +154,6 @@ class PolicyProfileTests(testlib_api.SqlTestCase):
self.assertEqual(profile.id, got_profile.id)
self.assertEqual(profile.name, got_profile.name)
def test_get_policy_profiles(self):
profile = _create_test_policy_profile_if_not_there(self.session)
got_profile = pprofile_mixin._get_policy_profiles().one()
self.assertEqual(profile.id, got_profile.id)
self.assertEqual(profile.name, got_profile.name)
class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):

0
networking_cisco/tests/unit/ml2/drivers/cisco/ucsm/__init__.py Normal file
View File

58
networking_cisco/tests/unit/ml2/drivers/cisco/ucsm/test_cisco_ucsm_common.py Normal file
View File

@@ -0,0 +1,58 @@
# Copyright 2015 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.
from neutron.common import config as neutron_config
from neutron.plugins.ml2 import config as ml2_config
from neutron.tests import base
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import config
UCSM_IP_ADDRESS = '1.1.1.1'
UCSM_USERNAME = 'username'
UCSM_PASSWORD = 'password'
UCSM_PHY_NETS = ['test_physnet']
HOST_CONFIG1 = ['Hostname1:Service_profile1']
class ConfigMixin(object):
"""Mock config for UCSM driver."""
mocked_parser = None
def set_up_mocks(self):
# Mock the configuration file
args = ['--config-file', base.etcdir('neutron.conf.test')]
neutron_config.init(args=args)
# Configure the ML2 mechanism drivers and network types
ml2_opts = {
'mechanism_drivers': ['cisco_ucsm'],
'tenant_network_types': ['vlan'],
}
for opt, val in ml2_opts.items():
ml2_config.cfg.CONF.set_override(opt, val, 'ml2')
# Configure the Cisco UCS Manager mechanism driver
ucsm_test_config = {
'ucsm_ip': UCSM_IP_ADDRESS,
'ucsm_username': UCSM_USERNAME,
'ucsm_password': UCSM_PASSWORD,
'ucsm_host_list': HOST_CONFIG1,
}
for opt, val in ucsm_test_config.items():
config.cfg.CONF.set_override(opt, val, 'ml2_cisco_ucsm')

484
networking_cisco/tests/unit/ml2/drivers/cisco/ucsm/test_cisco_ucsm_driver.py Normal file
View File

@@ -0,0 +1,484 @@
# Copyright 2015 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.
import mock
from neutron.common import constants as n_const
from neutron.db import api as db_api
from neutron.extensions import portbindings
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers.cisco.ucsm import mech_cisco_ucsm as md
from neutron.tests.unit import testlib_api
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import constants as const
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import ucsm_db
from networking_cisco.plugins.ml2.drivers.cisco.ucsm import ucsm_network_driver
from networking_cisco.tests.unit.ml2.drivers.cisco.ucsm import (
test_cisco_ucsm_common as mocked)
UCSM_IP_ADDRESS = '1.1.1.1'
UCSM_USERNAME = 'username'
UCSM_PASSWORD = 'password'
VNIC_NORMAL = 'normal'
VNIC_DIRECT = 'direct'
VNIC_MACVTAP = 'macvtap'
VNIC_TYPES = [VNIC_NORMAL, VNIC_DIRECT, VNIC_MACVTAP]
SRIOV_VNIC_TYPES = [VNIC_DIRECT, VNIC_MACVTAP]
SUPPORTED_PCI_DEVS = ["1137:0071", "8086:10c9"]
NETWORK_ID_1 = 1001
VLAN_ID_1 = 100
VLAN_ID_2 = 101
PORT_STATE_ACTIVE = n_const.PORT_STATUS_ACTIVE
NETWORK_TYPE = 'vlan'
NETWORK_ID = 'test-network'
PORT_NAME = 'port1'
PORT_NAME2 = 'port2'
PORT_ID = '100001'
PORT_ID2 = '100002'
HOST1 = "Hostname1"
PCI_INFO_BAD_NIC = '1111:2222'
PCI_INFO_INVALID = '1111'
UCSM_DRIVER = ('neutron.plugins.ml2.drivers.cisco.ucsm.'
'ucsm_network_driver.CiscoUcsmDriver')
VLAN_SEGMENTS = {api.ID: 'vlan_segment_id',
api.NETWORK_TYPE: 'vlan',
api.PHYSICAL_NETWORK: 'test_physnet',
api.SEGMENTATION_ID: VLAN_ID_1}
VXLAN_SEGMENTS = {api.ID: 'vlan_segment_id',
api.NETWORK_TYPE: 'vxlan',
api.PHYSICAL_NETWORK: 'test_physnet',
api.SEGMENTATION_ID: VLAN_ID_1}
VLAN_SEGMENTS_BAD = {api.ID: 'vlan_segment_id',
api.NETWORK_TYPE: 'vlan',
api.PHYSICAL_NETWORK: 'fake_physnet',
api.SEGMENTATION_ID: VLAN_ID_2}
VLAN_SEGMENTS_GOOD = [{api.ID: 'vlan_segment_id',
api.NETWORK_TYPE: 'vlan',
api.PHYSICAL_NETWORK: 'test_physnet',
api.SEGMENTATION_ID: VLAN_ID_2}]
class FakeNetworkContext(api.NetworkContext):
"""Network context for testing purposes only."""
def __init__(self, segments):
self._network_segments = segments
@property
def current(self):
return {'id': NETWORK_ID}
@property
def original(self):
return None
@property
def network_segments(self):
return self._network_segments
class FakePortContext(object):
"""Port context for testing purposes only."""
def __init__(self, name, port_id, vnic_type, profile,
network_context):
self._port = {
'status': None,
'id': port_id,
'name': name,
'host_id': HOST1,
portbindings.VNIC_TYPE: vnic_type,
portbindings.PROFILE: profile
}
self._network = network_context
self._segment = network_context.network_segments
self.session = db_api.get_session()
@property
def current(self):
return self._port
@property
def original(self):
return None
@property
def network(self):
return self._network
@property
def segment(self):
return self._segment
@property
def bottom_bound_segment(self):
return self._segment
def set_binding(self, segment_id, vif_type, vif_details,
status=None):
self._bound_segment_id = segment_id
self._bound_vif_type = vif_type
self._bound_vif_details = vif_details
self._new_port_status = status
class TestCiscoUcsmMechDriver(testlib_api.SqlTestCase,
mocked.ConfigMixin):
"""Unit tests for Cisco ML2 UCS Manager MD."""
def setUp(self):
"""Sets up mock Ucs Sdk."""
super(TestCiscoUcsmMechDriver, self).setUp()
self.set_up_mocks()
def new_ucsm_driver_init(mech_instance):
mech_instance.ucsm_ip = UCSM_IP_ADDRESS
mech_instance.username = UCSM_USERNAME
mech_instance.password = UCSM_PASSWORD
mech_instance.ucsmsdk = None
mech_instance.handles = {}
mech_instance.supported_sriov_vnic_types = SRIOV_VNIC_TYPES
mech_instance.supported_pci_devs = SUPPORTED_PCI_DEVS
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'__init__',
new=new_ucsm_driver_init).start()
self.mech_driver = md.CiscoUcsmMechanismDriver()
self.mech_driver.initialize()
self.vif_type = portbindings.VIF_TYPE_802_QBH
self.db = ucsm_db.UcsmDbModel()
self.ucsm_driver = ucsm_network_driver.CiscoUcsmDriver()
def _create_network_context(self):
segment = {api.SEGMENTATION_ID: "",
api.NETWORK_TYPE: "",
}
segment[api.SEGMENTATION_ID] = VLAN_ID_1
segment[api.NETWORK_TYPE] = 'vlan'
network_context = FakeNetworkContext(VLAN_SEGMENTS)
return network_context
def _create_port_context_vmfex(self):
"""Creates port context with valid VM-FEX vendor info."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = FakeNetworkContext(VLAN_SEGMENTS)
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
return port_context
def _create_port_context_bad(self):
"""Creates port context with badly formed vendor info."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': PCI_INFO_BAD_NIC}
network_context = FakeNetworkContext(VLAN_SEGMENTS)
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
return port_context
def _create_port_context_sriov(self):
"""Creates port context with valid SR-IOV vendor info."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_MACVTAP
profile = {'pci_vendor_info': const.PCI_INFO_INTEL_82599}
network_context = FakeNetworkContext(VLAN_SEGMENTS)
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
return port_context
def _create_port_context_normal(self):
"""Creates port context with Normal vnic type."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_NORMAL
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = FakeNetworkContext(VLAN_SEGMENTS)
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
return port_context
def test_parse_pci_vendor_config(self):
"""Verifies parsing of both good and bad pci vendor config."""
vendor1 = PCI_INFO_INVALID
vendor2 = const.PCI_INFO_INTEL_82599
self.assertNotIn(vendor1, self.ucsm_driver.supported_pci_devs)
self.assertIn(vendor2, self.ucsm_driver.supported_pci_devs)
def test_vmfex_vnic_type_and_vendor_info(self):
"""Verifies VM-FEX port is recognized as a supported vendor."""
port_context = self._create_port_context_vmfex()
vnic_type = port_context.current.get(portbindings.VNIC_TYPE,
portbindings.VNIC_NORMAL)
profile = port_context.current.get(portbindings.PROFILE, {})
supported = self.ucsm_driver.check_vnic_type_and_vendor_info(
vnic_type, profile)
self.assertTrue(supported)
def test_unsupported_vnic_type_and_vendor_info(self):
"""Verifies unsupported pci vendor is rejected."""
port_context = self._create_port_context_bad()
vnic_type = port_context.current.get(portbindings.VNIC_TYPE,
portbindings.VNIC_NORMAL)
profile = port_context.current.get(portbindings.PROFILE, {})
supported = self.ucsm_driver.check_vnic_type_and_vendor_info(
vnic_type, profile)
self.assertFalse(supported)
def test_sriov_vnic_type_and_vendor_info(self):
"""Verifies SR-IOV port and MACVTAP vnic_type are supported."""
port_context = self._create_port_context_sriov()
vnic_type = port_context.current.get(portbindings.VNIC_TYPE,
portbindings.VNIC_NORMAL)
profile = port_context.current.get(portbindings.PROFILE, {})
supported = self.ucsm_driver.check_vnic_type_and_vendor_info(
vnic_type, profile)
self.assertTrue(supported)
def test_normal_vnic_type(self):
"""Verifies NORMAL vnic type is not supported."""
port_context = self._create_port_context_normal()
vnic_type = port_context.current.get(portbindings.VNIC_TYPE,
portbindings.VNIC_NORMAL)
profile = port_context.current.get(portbindings.PROFILE, {})
supported = self.ucsm_driver.check_vnic_type_and_vendor_info(
vnic_type, profile)
self.assertFalse(supported)
def test_validate_vm_fex_port_cisco(self):
"""Verifies port's pci vendor info makes it VM-FEX capable."""
port_context = self._create_port_context_vmfex()
profile = port_context.current.get(portbindings.PROFILE, {})
valid = self.ucsm_driver.is_vmfex_port(profile)
self.assertTrue(valid)
def test_validate_vm_fex_port_bad(self):
"""Verifies unsupported pci vendor is not VM-FEX capable."""
port_context = self._create_port_context_bad()
profile = port_context.current.get(portbindings.PROFILE, {})
valid = self.ucsm_driver.is_vmfex_port(profile)
self.assertFalse(valid)
def test_validate_vm_fex_port_sriov(self):
"""Verifies valid SR-IOV port is not VM-FEX capable."""
port_context = self._create_port_context_sriov()
profile = port_context.current.get(portbindings.PROFILE, {})
valid = self.ucsm_driver.is_vmfex_port(profile)
# For ex: Intel PCI is supported but is not vm-fex.
# so, should return False
self.assertFalse(valid)
def test_check_segment_vlan(self):
"""Verifies VLAN network segments are supported."""
self.assertTrue(self.mech_driver.check_segment(VLAN_SEGMENTS))
def test_check_segment_vxlan(self):
"""Verifies VXLAN network segments are not supported."""
self.assertFalse(self.mech_driver.check_segment(VXLAN_SEGMENTS))
def test_vmfex_update_port_precommit(self):
"""Verifies MD saves relevant info for VM-FEX ports into DB."""
name = PORT_NAME2
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
profile_name = "OS-PP-100"
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
# Port Profile name and Vlan id are written to DB.
self.mech_driver.update_port_precommit(port_context)
# Look for presence of above entry in the DB.
p_profile = self.db.get_port_profile_for_vlan(VLAN_ID_1)
self.assertEqual(profile_name, p_profile)
# Look to see if flag is set for update_port_postcommit to
# create Port Profile on UCS Manager.
self.assertFalse(self.db.is_port_profile_created(VLAN_ID_1))
def test_sriov_update_port_precommit(self):
"""Verifies MD does not create Port Profiles for SR-IOV ports."""
port_context = self._create_port_context_sriov()
self.mech_driver.update_port_precommit(port_context)
p_profile = self.db.get_port_profile_for_vlan(VLAN_ID_1)
self.assertIsNone(p_profile)
def test_update_port_postcommit_success(self):
"""Verifies duplicate Port Profiles are not being created."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
# Port Profile is added to DB and created on UCS Manager.
self.mech_driver.update_port_precommit(port_context)
self.assertFalse(self.db.is_port_profile_created(VLAN_ID_1))
# Call to UCS Manager driver top level method to create Port Profile
# is mocked to a new method here. This method verifies input params
# are correct.
def new_create_portprofile(mech_context, profile_name, vlan_id,
vnic_type):
return True
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'create_portprofile',
new=new_create_portprofile).start()
self.mech_driver.update_port_postcommit(port_context)
self.assertTrue(self.db.is_port_profile_created(VLAN_ID_1))
def test_update_port_postcommit_failure(self):
"""Verifies duplicate Port Profiles are not being created."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
# Port Profile is added to DB and created on UCS Manager.
self.mech_driver.update_port_precommit(port_context)
self.assertFalse(self.db.is_port_profile_created(VLAN_ID_1))
# Call to UCS Manager driver top level method to create Port Profile
# is mocked to a new method here. This method verifies input params
# are correct.
def new_create_portprofile(mech_context, profile_name, vlan_id,
vnic_type):
return False
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'create_portprofile',
new=new_create_portprofile).start()
self.mech_driver.update_port_postcommit(port_context)
self.assertFalse(self.db.is_port_profile_created(VLAN_ID_1))
def test_update_port_postcommit_direct(self):
"""Verifies UCS Manager driver is called with correct parameters."""
name = PORT_NAME
port_id = PORT_ID
vnic_direct = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_direct,
profile, network_context)
self.mech_driver.update_port_precommit(port_context)
# Call to UCS Manager driver top level method to create Port Profile
# is mocked to a new method here. This method verifies input params
# are correct.
def new_create_portprofile(mech_context, profile_name, vlan_id,
vnic_type):
self.assertEqual("OS-PP-100", profile_name)
self.assertEqual(100, vlan_id)
self.assertEqual(VNIC_DIRECT, vnic_type)
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'create_portprofile',
new=new_create_portprofile).start()
self.mech_driver.update_port_postcommit(port_context)
def test_update_port_postcommit_macvtap(self):
"""Verifies UCS Manager driver is called with correct parameters."""
name = PORT_NAME
port_id = PORT_ID
vnic_macvtap = VNIC_MACVTAP
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_macvtap,
profile, network_context)
self.mech_driver.update_port_precommit(port_context)
# Call to UCS Manager driver top level method to create Port Profile
# is mocked to a new method here. This method verifies input params
# are correct.
def new_create_portprofile(mech_context, profile_name, vlan_id,
vnic_type):
self.assertEqual("OS-PP-100", profile_name)
self.assertEqual(100, vlan_id)
self.assertEqual(VNIC_MACVTAP, vnic_type)
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'create_portprofile',
new=new_create_portprofile).start()
self.mech_driver.update_port_postcommit(port_context)
def test_update_port_postcommit_normal(self):
"""Verifies UCS Manager driver is called with correct parameters."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_NORMAL
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = self._create_network_context()
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
self.mech_driver.update_port_precommit(port_context)
# Call to UCS Manager driver top level method to create Port Profile
# is mocked to a new method here. This method verifies input params
# are correct.
def new_update_serviceprofile(mech_context, host_id, vlan_id):
return True
mock.patch.object(ucsm_network_driver.CiscoUcsmDriver,
'update_serviceprofile',
new=new_update_serviceprofile).start()
self.mech_driver.update_port_postcommit(port_context)
def test_bind_port_active(self):
"""Verifies bind_port sets the port status as active."""
name = PORT_NAME
port_id = PORT_ID
vnic_type = VNIC_DIRECT
profile = {'pci_vendor_info': const.PCI_INFO_CISCO_VIC_1240}
network_context = FakeNetworkContext(VLAN_SEGMENTS_GOOD)
port_context = FakePortContext(name, port_id, vnic_type,
profile, network_context)
self.mech_driver.bind_port(port_context)
self.assertEqual(PORT_STATE_ACTIVE, port_context._new_port_status)

1
setup.cfg
View File

@@ -51,6 +51,7 @@ neutron.service_plugins =
cisco_n1kv_profile = networking_cisco.plugins.ml2.drivers.cisco.n1kv.policy_profile_service:PolicyProfilePlugin
neutron.ml2.extension_drivers =
cisco_n1kv_ext = neutron.plugins.ml2.drivers.cisco.n1kv.n1kv_ext_driver:CiscoN1kvExtensionDriver
cisco_ucsm = neutron.plugins.ml2.drivers.cisco.ucsm.mech_cisco_ucsm:CiscoUcsmMechanismDriver
[build_sphinx]
source-dir = doc/source

2
test-patches.txt
View File

@@ -1 +1 @@
refs/changes/43/157243/15
refs/changes/36/155436/21