openstack
/
neutron
Code Issues Proposed changes

Use pyroute2 for SRIOV VF commands 

Recent changes in some versions of iproute2 CLI output (v4.18),
have invalidated the regular expression used to parse the
"ip link" output.

To solve this problem and avoid future ones, pyroute2 is used to
retrieve the virtual functions information and set the VF attributes
(spoofcheck, min_tx_rate, max_tx_rate and link_state).

pyroute2 extended the "ip link" support to retrieve this information,
adding "ext_mask=1" in the get command. If no virtual functions are
present in this particular network interface, the added method,
"get_link_vfs",  will return an empty list.

The set commands can return a "InterfaceOperationNotSupported" in
case the operation is not supported. For min_tx_rate, if the driver
does not support to set a minimum bandwidth, an "InvalidArgument"
(from a pyroute2.NetlinkError(22)) exception will be raised.

Change-Id: I680da4f64bd114f1caecaaeedbf8a4b1915a0849
Closes-Bug: #1878042
This commit is contained in:
Rodolfo Alonso Hernandez 2020-05-11 17:59:48 +00:00
parent 08e9ec1b56
commit c5d8fd6329
18 changed files with 237 additions and 672 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
neutron/agent/linux/ip_lib.py
View File

@ -29,7 +29,6 @@ from pyroute2.netlink import exceptions as netlink_exceptions
from pyroute2.netlink import rtnl
from pyroute2.netlink.rtnl import ifaddrmsg
from pyroute2.netlink.rtnl import ifinfmsg
from pyroute2 import NetlinkError
from pyroute2 import netns
from neutron._i18n import _
@ -100,9 +99,7 @@ class AddressNotReady(exceptions.NeutronException):
"become ready: %(reason)s")
class InvalidArgument(exceptions.NeutronException):
message = _("Invalid value %(value)s for parameter %(parameter)s "
"provided.")
InvalidArgument = privileged.InvalidArgument
class SubProcessBase(object):
@ -442,13 +439,8 @@ class IpLinkCommand(IpDeviceCommandBase):
self.name, self._parent.namespace, ifinfmsg.IFF_ALLMULTI)
def set_mtu(self, mtu_size):
try:
privileged.set_link_attribute(
self.name, self._parent.namespace, mtu=mtu_size)
except NetlinkError as e:
if e.code == errno.EINVAL:
raise InvalidArgument(parameter="MTU", value=mtu_size)
raise
def set_up(self):
privileged.set_link_attribute(
@ -520,6 +512,13 @@ class IpLinkCommand(IpDeviceCommandBase):
def exists(self):
return privileged.interface_exists(self.name, self._parent.namespace)
def get_vfs(self):
return privileged.get_link_vfs(self.name, self._parent.namespace)
def set_vf_feature(self, vf_config):
return privileged.set_link_vf_feature(
self.name, self._parent.namespace, vf_config)
class IpAddrCommand(IpDeviceCommandBase):
COMMAND = 'addr'

108
neutron/agent/linux/ip_link_support.py
View File

@ -1,108 +0,0 @@
# Copyright 2014 Mellanox Technologies, Ltd
#
# 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 re
from neutron_lib import exceptions as n_exc
from oslo_log import log as logging
from neutron._i18n import _
from neutron.agent.linux import utils
LOG = logging.getLogger(__name__)
class IpLinkSupportError(n_exc.NeutronException):
pass
class UnsupportedIpLinkCommand(IpLinkSupportError):
message = _("ip link command is not supported: %(reason)s")
class InvalidIpLinkCapability(IpLinkSupportError):
message = _("ip link capability %(capability)s is not supported")
class IpLinkConstants(object):
IP_LINK_CAPABILITY_STATE = "state"
IP_LINK_CAPABILITY_VLAN = "vlan"
IP_LINK_CAPABILITY_RATE = "rate"
IP_LINK_CAPABILITY_MIN_TX_RATE = "min_tx_rate"
IP_LINK_CAPABILITY_SPOOFCHK = "spoofchk"
IP_LINK_SUB_CAPABILITY_QOS = "qos"
class IpLinkSupport(object):
VF_BLOCK_REGEX = r"\[ vf NUM(?P<vf_block>.*) \] \]"
CAPABILITY_REGEX = r"\[ %s (.*)"
SUB_CAPABILITY_REGEX = r"\[ %(cap)s (.*) \[ %(subcap)s (.*)"
@classmethod
def get_vf_mgmt_section(cls):
"""Parses ip link help output, and gets vf block"""
output = cls._get_ip_link_output()
vf_block_pattern = re.search(cls.VF_BLOCK_REGEX,
output,
re.DOTALL | re.MULTILINE)
if vf_block_pattern:
return vf_block_pattern.group("vf_block")
@classmethod
def vf_mgmt_capability_supported(cls, vf_section, capability,
subcapability=None):
"""Validate vf capability support
Checks if given vf capability (and sub capability
if given) supported
:param vf_section: vf Num block content
:param capability: for example: vlan, rate, spoofchk, state
:param subcapability: for example: qos
"""
if not vf_section:
return False
if subcapability:
regex = cls.SUB_CAPABILITY_REGEX % {"cap": capability,
"subcap": subcapability}
else:
regex = cls.CAPABILITY_REGEX % capability
pattern_match = re.search(regex, vf_section,
re.DOTALL | re.MULTILINE)
return pattern_match is not None
@classmethod
def _get_ip_link_output(cls):
"""Gets the output of the ip link help command
Runs ip link help command and stores its output
Note: ip link help return error and writes its output to stderr
so we get the output from there. however, if this issue
will be solved and the command will write to stdout, we
will get the output from there too.
"""
try:
ip_cmd = ['ip', 'link', 'help']
_stdout, _stderr = utils.execute(
ip_cmd,
check_exit_code=False,
return_stderr=True,
log_fail_as_error=False)
except Exception as e:
LOG.exception("Failed executing ip command")
raise UnsupportedIpLinkCommand(reason=e)
return _stdout or _stderr

36
neutron/cmd/sanity/checks.py
View File

@ -31,7 +31,6 @@ from neutron.agent.l3 import ha_router
from neutron.agent.l3 import namespaces
from neutron.agent.linux import external_process
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ip_link_support
from neutron.agent.linux import keepalived
from neutron.agent.linux import utils as agent_utils
from neutron.cmd import runtime_checks
@ -149,41 +148,6 @@ def icmpv6_header_match_supported():
actions="NORMAL")
def _vf_management_support(required_caps):
is_supported = True
try:
vf_section = ip_link_support.IpLinkSupport.get_vf_mgmt_section()
for cap in required_caps:
if not ip_link_support.IpLinkSupport.vf_mgmt_capability_supported(
vf_section, cap):
is_supported = False
LOG.debug("ip link command does not support "
"vf capability '%(cap)s'", {'cap': cap})
except ip_link_support.UnsupportedIpLinkCommand:
LOG.exception("Unexpected exception while checking supported "
"ip link command")
return False
return is_supported
def vf_management_supported():
required_caps = (
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_SPOOFCHK,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE)
return _vf_management_support(required_caps)
def vf_extended_management_supported():
required_caps = (
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_SPOOFCHK,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE,
)
return _vf_management_support(required_caps)
def netns_read_requires_helper():
nsname = "netnsreadtest-" + uuidutils.generate_uuid()
ip_lib.create_network_namespace(nsname)

23
neutron/cmd/sanity_check.py
View File

@ -192,25 +192,6 @@ def check_icmpv6_header_match():
return result
def check_vf_management():
result = checks.vf_management_supported()
if not result:
LOG.error('Check for VF management support failed. '
'Please ensure that the version of ip link '
'being used has VF support.')
return result
def check_vf_extended_management():
result = checks.vf_extended_management_supported()
if not result:
LOG.error('Check for VF extended management support failed. '
'Please ensure that the version of ip link '
'being used has VF extended support: version '
'"iproute2-ss140804", git tag "v3.16.0"')
return result
def check_ovsdb_native():
result = checks.ovsdb_native_supported()
if not result:
@ -325,10 +306,6 @@ OPTS = [
help=_('Check for ARP header match support')),
BoolOptCallback('icmpv6_header_match', check_icmpv6_header_match,
help=_('Check for ICMPv6 header match support')),
BoolOptCallback('vf_management', check_vf_management,
help=_('Check for VF management support')),
BoolOptCallback('vf_extended_management', check_vf_extended_management,
help=_('Check for VF extended management support')),
BoolOptCallback('read_netns', check_read_netns,
help=_('Check netns permission settings')),
BoolOptCallback('dnsmasq_local_service_supported',

8
neutron/plugins/ml2/drivers/mech_sriov/agent/common/exceptions.py
View File

@ -26,13 +26,5 @@ class InvalidDeviceError(SriovNicError):
message = _("Invalid Device %(dev_name)s: %(reason)s")
class IpCommandError(SriovNicError):
message = _("ip command failed on device %(dev_name)s: %(reason)s")
class IpCommandOperationNotSupportedError(SriovNicError):
message = _("Operation not supported on device %(dev_name)s")
class InvalidPciSlotError(SriovNicError):
message = _("Invalid pci slot %(pci_slot)s")

35
neutron/plugins/ml2/drivers/mech_sriov/agent/eswitch_manager.py
View File

@ -21,7 +21,6 @@ from neutron_lib.utils import helpers
from oslo_log import log as logging
from neutron._i18n import _
from neutron.agent.linux import ip_link_support
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.plugins.ml2.drivers.mech_sriov.agent import pci_lib
@ -29,6 +28,14 @@ from neutron.plugins.ml2.drivers.mech_sriov.agent import pci_lib
LOG = logging.getLogger(__name__)
IP_LINK_CAPABILITY_STATE = 'state'
IP_LINK_CAPABILITY_VLAN = 'vlan'
IP_LINK_CAPABILITY_RATE = 'max_tx_rate'
IP_LINK_CAPABILITY_MIN_TX_RATE = 'min_tx_rate'
IP_LINK_CAPABILITY_SPOOFCHK = 'spoofchk'
IP_LINK_SUB_CAPABILITY_QOS = 'qos'
class PciOsWrapper(object):
"""OS wrapper for checking virtual functions"""
@ -200,11 +207,11 @@ class EmbSwitch(object):
auto=propagate_uplink_state)
def set_device_rate(self, pci_slot, rate_type, rate_kbps):
"""Set device rate: rate (max_tx_rate), min_tx_rate
"""Set device rate: max_tx_rate, min_tx_rate
@param pci_slot: Virtual Function address
@param rate_type: device rate name type. Could be 'rate' and
'min_tx_rate'.
@param rate_type: device rate name type. Could be 'max_tx_rate' and
'min_tx_rate' ('rate' is not supported anymore).
@param rate_kbps: device rate in kbps
"""
vf_index = self._get_vf_index(pci_slot)
@ -349,7 +356,7 @@ class ESwitchManager(object):
embedded_switch = self._get_emb_eswitch(device_mac, pci_slot)
if embedded_switch:
return embedded_switch.get_device_state(pci_slot)
return pci_lib.LinkState.DISABLE
return pci_lib.LinkState.disable.name
def set_device_max_rate(self, device_mac, pci_slot, max_kbps):
"""Set device max rate
@ -362,9 +369,7 @@ class ESwitchManager(object):
embedded_switch = self._get_emb_eswitch(device_mac, pci_slot)
if embedded_switch:
embedded_switch.set_device_rate(
pci_slot,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE,
max_kbps)
pci_slot, IP_LINK_CAPABILITY_RATE, max_kbps)
def set_device_min_tx_rate(self, device_mac, pci_slot, min_kbps):
"""Set device min_tx_rate
@ -377,9 +382,7 @@ class ESwitchManager(object):
embedded_switch = self._get_emb_eswitch(device_mac, pci_slot)
if embedded_switch:
embedded_switch.set_device_rate(
pci_slot,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE,
min_kbps)
pci_slot, IP_LINK_CAPABILITY_MIN_TX_RATE, min_kbps)
def set_device_state(self, device_mac, pci_slot, admin_state_up,
propagate_uplink_state):
@ -498,9 +501,7 @@ class ESwitchManager(object):
Clear the "rate" configuration from VF by setting it to 0.
@param pci_slot: VF PCI slot
"""
self._clear_rate(
pci_slot,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE)
self._clear_rate(pci_slot, IP_LINK_CAPABILITY_RATE)
def clear_min_tx_rate(self, pci_slot):
"""Clear the VF "min_tx_rate" parameter
@ -508,16 +509,14 @@ class ESwitchManager(object):
Clear the "min_tx_rate" configuration from VF by setting it to 0.
@param pci_slot: VF PCI slot
"""
self._clear_rate(
pci_slot,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE)
self._clear_rate(pci_slot, IP_LINK_CAPABILITY_MIN_TX_RATE)
def _clear_rate(self, pci_slot, rate_type):
"""Clear the VF rate parameter specified in rate_type
Clear the rate configuration from VF by setting it to 0.
@param pci_slot: VF PCI slot
@param rate_type: rate to clear ('rate', 'min_tx_rate')
@param rate_type: rate to clear ('max_tx_rate', 'min_tx_rate')
"""
# NOTE(Moshe Levi): we don't use the self._get_emb_eswitch here,
# because when clearing the VF it may be not assigned. This happens

15
neutron/plugins/ml2/drivers/mech_sriov/agent/extension_drivers/qos_driver.py
View File

@ -15,9 +15,8 @@
from oslo_log import log as logging
from neutron.agent.l2.extensions import qos_linux as qos
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import (
exceptions as exc)
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
from neutron.privileged.agent.linux import ip_lib as priv_ip_lib
from neutron.services.qos.drivers.sriov import driver
LOG = logging.getLogger(__name__)
@ -55,9 +54,9 @@ class QosSRIOVAgentDriver(qos.QosLinuxAgentDriver):
try:
self.eswitch_mgr.set_device_max_rate(
device, pci_slot, max_kbps)
except exc.SriovNicError:
LOG.exception(
"Failed to set device %s max rate", device)
except (priv_ip_lib.InterfaceOperationNotSupported,
priv_ip_lib.InvalidArgument):
LOG.exception("Failed to set device %s max rate", device)
else:
LOG.info("No device with MAC %s defined on agent.", device)
@ -97,8 +96,8 @@ class QosSRIOVAgentDriver(qos.QosLinuxAgentDriver):
try:
self.eswitch_mgr.set_device_min_tx_rate(
device, pci_slot, min_tx_kbps)
except exc.SriovNicError:
LOG.exception(
"Failed to set device %s min_tx_rate", device)
except (priv_ip_lib.InterfaceOperationNotSupported,
priv_ip_lib.InvalidArgument):
LOG.exception("Failed to set device %s min_tx_rate", device)
else:
LOG.info("No device with MAC %s defined on agent.", device)

158
neutron/plugins/ml2/drivers/mech_sriov/agent/pci_lib.py
View File

@ -13,86 +13,42 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import re
import enum
from oslo_log import log as logging
from neutron.agent.linux import ip_lib
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
LOG = logging.getLogger(__name__)
class LinkState(object):
ENABLE = "enable"
DISABLE = "disable"
AUTO = "auto"
class LinkState(enum.Enum):
auto = 0
enable = 1
disable = 2
class PciDeviceIPWrapper(ip_lib.IPWrapper):
"""Wrapper class for ip link commands.
wrapper for getting/setting pci device details using ip link...
"""
VF_PATTERN = r"^vf\s+(?P<vf_index>\d+)\s+"
MAC_PATTERN = r"MAC\s+(?P<mac>[a-fA-F0-9:]+),"
STATE_PATTERN = r"\s+link-state\s+(?P<state>\w+)"
ANY_PATTERN = ".*,"
VF_LINE_FORMAT = VF_PATTERN + MAC_PATTERN + ANY_PATTERN + STATE_PATTERN
VF_DETAILS_REG_EX = re.compile(VF_LINE_FORMAT)
IP_LINK_OP_NOT_SUPPORTED = 'RTNETLINK answers: Operation not supported'
"""Wrapper class for ip link commands related to virtual functions."""
def __init__(self, dev_name):
super(PciDeviceIPWrapper, self).__init__()
self.dev_name = dev_name
def _set_feature(self, vf_index, feature, value):
"""Sets vf feature
Checks if the feature is not supported or there's some
general error during ip link invocation and raises
exception accordingly.
:param vf_index: vf index
:param feature: name of a feature to be passed to ip link,
such as 'state' or 'spoofchk'
:param value: value of the feature setting
"""
try:
self._as_root([], "link", ("set", self.dev_name, "vf",
str(vf_index), feature, value))
except Exception as e:
if self.IP_LINK_OP_NOT_SUPPORTED in str(e):
raise exc.IpCommandOperationNotSupportedError(
dev_name=self.dev_name)
else:
raise exc.IpCommandError(dev_name=self.dev_name,
reason=str(e))
def get_assigned_macs(self, vf_list):
"""Get assigned mac addresses for vf list.
@param vf_list: list of vf indexes
@return: dict mapping of vf to mac
"""
try:
out = self._as_root([], "link", ("show", self.dev_name))
except Exception as e:
LOG.exception("Failed executing ip command")
raise exc.IpCommandError(dev_name=self.dev_name,
reason=e)
ip = self.device(self.dev_name)
vfs = ip.link.get_vfs()
vf_to_mac_mapping = {}
vf_lines = self._get_vf_link_show(vf_list, out)
if vf_lines:
for vf_line in vf_lines:
vf_details = self._parse_vf_link_show(vf_line)
if vf_details:
vf_num = vf_details.get('vf')
vf_mac = vf_details.get("MAC")
vf_to_mac_mapping[vf_num] = vf_mac
for vf_num in vf_list:
if vfs.get(vf_num):
vf_to_mac_mapping[vf_num] = vfs[vf_num]['mac']
return vf_to_mac_mapping
def get_vf_state(self, vf_index):
@ -100,88 +56,48 @@ class PciDeviceIPWrapper(ip_lib.IPWrapper):
@param vf_index: vf index
"""
try:
out = self._as_root([], "link", ("show", self.dev_name))
except Exception as e:
LOG.exception("Failed executing ip command")
raise exc.IpCommandError(dev_name=self.dev_name,
reason=e)
vf_lines = self._get_vf_link_show([vf_index], out)
if vf_lines:
vf_details = self._parse_vf_link_show(vf_lines[0])
if vf_details:
state = vf_details.get("link-state",
LinkState.DISABLE)
if state in (LinkState.AUTO, LinkState.ENABLE):
return state
return LinkState.DISABLE
ip = self.device(self.dev_name)
vfs = ip.link.get_vfs()
vf = vfs.get(vf_index)
if vf:
return LinkState(int(vf['link_state'])).name
return LinkState.disable.name
def set_vf_state(self, vf_index, state, auto=False):
"""sets vf state.
@param vf_index: vf index
@param state: required state {True/False}
@param state: required state {True: enable (1)
False: disable (2)}
@param auto: set link_state to auto (0)
"""
ip = self.device(self.dev_name)
if auto:
status_str = LinkState.AUTO
link_state = 0
else:
status_str = LinkState.ENABLE if state else \
LinkState.DISABLE
self._set_feature(vf_index, "state", status_str)
link_state = 1 if state else 2
vf_config = {'vf': vf_index, 'link_state': link_state}
ip.link.set_vf_feature(vf_config)
def set_vf_spoofcheck(self, vf_index, enabled):
"""sets vf spoofcheck
@param vf_index: vf index
@param enabled: True to enable spoof checking,
False to disable
@param enabled: True to enable (1) spoof checking,
False to disable (0)
"""
setting = "on" if enabled else "off"
self._set_feature(vf_index, "spoofchk", setting)
ip = self.device(self.dev_name)
vf_config = {'vf': vf_index, 'spoofchk': int(enabled)}
ip.link.set_vf_feature(vf_config)
def set_vf_rate(self, vf_index, rate_type, rate_value):
"""sets vf rate.
@param vf_index: vf index
@param rate_type: vf rate type ('rate', 'min_tx_rate')
@param rate_type: vf rate type ('max_tx_rate', 'min_tx_rate')
@param rate_value: vf rate in Mbps
"""
self._set_feature(vf_index, rate_type, str(rate_value))
def _get_vf_link_show(self, vf_list, link_show_out):
"""Get link show output for VFs
get vf link show command output filtered by given vf list
@param vf_list: list of vf indexes
@param link_show_out: link show command output
@return: list of output rows regarding given vf_list
"""
vf_lines = []
for line in link_show_out.split("\n"):
line = line.strip()
if line.startswith("vf"):
details = line.split()
index = int(details[1])
if index in vf_list:
vf_lines.append(line)
if not vf_lines:
LOG.warning("Cannot find vfs %(vfs)s in device %(dev_name)s",
{'vfs': vf_list, 'dev_name': self.dev_name})
return vf_lines
def _parse_vf_link_show(self, vf_line):
"""Parses vf link show command output line.
@param vf_line: link show vf line
"""
vf_details = {}
pattern_match = self.VF_DETAILS_REG_EX.match(vf_line)
if pattern_match:
vf_details["vf"] = int(pattern_match.group("vf_index"))
vf_details["MAC"] = pattern_match.group("mac")
vf_details["link-state"] = pattern_match.group("state")
else:
LOG.warning("failed to parse vf link show line %(line)s: "
"for %(device)s",
{'line': vf_line, 'device': self.dev_name})
return vf_details
ip = self.device(self.dev_name)
vf_config = {'vf': vf_index, 'rate': {rate_type: int(rate_value)}}
ip.link.set_vf_feature(vf_config)

9
neutron/plugins/ml2/drivers/mech_sriov/agent/sriov_nic_agent.py
View File

@ -31,6 +31,7 @@ from oslo_log import log as logging
import oslo_messaging
from oslo_service import loopingcall
from osprofiler import profiler
import pyroute2
import six
from neutron._i18n import _
@ -47,6 +48,7 @@ from neutron.plugins.ml2.drivers.mech_sriov.agent.common import config
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
from neutron.privileged.agent.linux import ip_lib as priv_ip_lib
LOG = logging.getLogger(__name__)
@ -291,10 +293,9 @@ class SriovNicSwitchAgent(object):
self.eswitch_mgr.set_device_state(device, pci_slot,
admin_state_up,
propagate_uplink_state)
except exc.IpCommandOperationNotSupportedError:
LOG.warning("Device %s does not support state change",
device)
except exc.SriovNicError:
except priv_ip_lib.InterfaceOperationNotSupported:
LOG.warning("Device %s does not support state change", device)
except pyroute2.NetlinkError:
LOG.warning("Failed to set device %s state", device)
return False
else:

42
neutron/privileged/agent/linux/ip_lib.py
View File

@ -64,6 +64,8 @@ def _get_scope_name(scope):
return rtnl.rt_scope.get(scope, scope)
# TODO(ralonsoh): move those exceptions out of priv_ip_lib to avoid other
# modules to import this one.
class NetworkNamespaceNotFound(RuntimeError):
message = _("Network namespace %(netns_name)s could not be found.")
@ -102,6 +104,21 @@ class InterfaceOperationNotSupported(RuntimeError):
super(InterfaceOperationNotSupported, self).__init__(message)
class InvalidArgument(RuntimeError):
message = _("Invalid parameter/value used on interface %(device)s, "
"namespace %(namespace)s.")
def __init__(self, message=None, device=None, namespace=None):
# NOTE(slaweq): 'message' can be passed as an optional argument
# because of how privsep daemon works. If exception is raised in
# function called by privsep daemon, it will then try to reraise it
# and will call it always with passing only message from originally
# raised exception.
message = message or self.message % {'device': device,
'namespace': namespace}
super(InvalidArgument, self).__init__(message)
class IpAddressAlreadyExists(RuntimeError):
message = _("IP address %(ip)s already configured on %(device)s.")
@ -234,6 +251,8 @@ def _translate_ip_device_exception(e, device=None, namespace=None):
if e.code == errno.EOPNOTSUPP:
raise InterfaceOperationNotSupported(device=device,
namespace=namespace)
if e.code == errno.EINVAL:
raise InvalidArgument(device=device, namespace=namespace)
def get_link_id(device, namespace, raise_exception=True):
@ -391,6 +410,11 @@ def set_link_attribute(device, namespace, **attributes):
return _run_iproute_link("set", device, namespace, **attributes)
@privileged.default.entrypoint
def set_link_vf_feature(device, namespace, vf_config):
return _run_iproute_link("set", device, namespace=namespace, vf=vf_config)
@privileged.default.entrypoint
def get_link_attributes(device, namespace):
link = _run_iproute_link("get", device, namespace)[0]
@ -407,6 +431,24 @@ def get_link_attributes(device, namespace):
}
@privileged.default.entrypoint
def get_link_vfs(device, namespace):
link = _run_iproute_link('get', device, namespace=namespace, ext_mask=1)[0]
num_vfs = link.get_attr('IFLA_NUM_VF')
vfs = {}
if not num_vfs:
return vfs
vfinfo_list = link.get_attr('IFLA_VFINFO_LIST')
for vinfo in vfinfo_list.get_attrs('IFLA_VF_INFO'):
mac = vinfo.get_attr('IFLA_VF_MAC')
link_state = vinfo.get_attr('IFLA_VF_LINK_STATE')
vfs[mac['vf']] = {'mac': mac['mac'],
'link_state': link_state['link_state']}
return vfs
@privileged.default.entrypoint
def add_neigh_entry(ip_version, ip_address, mac_address, device, namespace,
**kwargs):

6
neutron/tests/functional/sanity/test_sanity.py
View File

@ -77,12 +77,6 @@ class SanityTestCaseRoot(base.BaseSudoTestCase):
def test_icmpv6_header_match_runs(self):
checks.icmpv6_header_match_supported()
def test_vf_management_runs(self):
checks.vf_management_supported()
def test_vf_extended_management_runs(self):
checks.vf_extended_management_supported()
def test_namespace_root_read_detection_runs(self):
checks.netns_read_requires_helper()

181
neutron/tests/unit/agent/linux/test_ip_link_support.py
View File

@ -1,181 +0,0 @@
# Copyright 2014 Mellanox Technologies, Ltd
#
# 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 unittest import mock
from neutron.agent.linux import ip_link_support as ip_link
from neutron.tests import base
class TestIpLinkSupport(base.BaseTestCase):
IP_LINK_HELP = """Usage: ip link add [link DEV] [ name ] NAME
[ txqueuelen PACKETS ]
[ address LLADDR ]
[ broadcast LLADDR ]
[ mtu MTU ] [index IDX ]
[ numtxqueues QUEUE_COUNT ]
[ numrxqueues QUEUE_COUNT ]
type TYPE [ ARGS ]
ip link delete DEV type TYPE [ ARGS ]
ip link set { dev DEVICE | group DEVGROUP } [ { up | down } ]
[ arp { on | off } ]
[ dynamic { on | off } ]
[ multicast { on | off } ]
[ allmulticast { on | off } ]
[ promisc { on | off } ]
[ trailers { on | off } ]
[ txqueuelen PACKETS ]
[ name NEWNAME ]
[ address LLADDR ]
[ broadcast LLADDR ]
[ mtu MTU ]
[ netns PID ]
[ netns NAME ]
[ alias NAME ]
[ vf NUM [ mac LLADDR ]
[ vlan VLANID [ qos VLAN-QOS ] ]
[ rate TXRATE ] ]
[ spoofchk { on | off} ] ]
[ state { auto | enable | disable} ] ]
[ master DEVICE ]
[ nomaster ]
ip link show [ DEVICE | group GROUP ] [up]
TYPE := { vlan | veth | vcan | dummy | ifb | macvlan | macvtap |
can | bridge | bond | ipoib | ip6tnl | ipip | sit |
vxlan | gre | gretap | ip6gre | ip6gretap | vti }
"""
IP_LINK_HELP_NO_STATE = """Usage: ip link add link DEV [ name ] NAME
[ txqueuelen PACKETS ]
[ address LLADDR ]
[ broadcast LLADDR ]
[ mtu MTU ]
type TYPE [ ARGS ]
ip link delete DEV type TYPE [ ARGS ]
ip link set DEVICE [ { up | down } ]
[ arp { on | off } ]
[ dynamic { on | off } ]
[ multicast { on | off } ]
[ allmulticast { on | off } ]
[ promisc { on | off } ]
[ trailers { on | off } ]
[ txqueuelen PACKETS ]
[ name NEWNAME ]
[ address LLADDR ]
[ broadcast LLADDR ]
[ mtu MTU ]
[ netns PID ]
[ alias NAME ]
[ vf NUM [ mac LLADDR ]
[ vlan VLANID [ qos VLAN-QOS ] ]
[ rate TXRATE ] ]
ip link show [ DEVICE ]
TYPE := { vlan | veth | vcan | dummy | ifb | macvlan | can }
"""
IP_LINK_HELP_NO_SPOOFCHK = IP_LINK_HELP_NO_STATE
IP_LINK_HELP_NO_VF = """Usage: ip link set DEVICE { up | down |
arp { on | off } |
dynamic { on | off } |
multicast { on | off } |
allmulticast { on | off } |
promisc { on | off } |
trailers { on | off } |
txqueuelen PACKETS |
name NEWNAME |
address LLADDR | broadcast LLADDR |
mtu MTU }
ip link show [ DEVICE ]
"""
def _test_capability(self, capability, subcapability=None,
expected=True, stdout="", stderr=""):
with mock.patch("neutron.agent.linux.utils.execute") as mock_exec:
mock_exec.return_value = (stdout, stderr)
vf_section = ip_link.IpLinkSupport.get_vf_mgmt_section()
capable = ip_link.IpLinkSupport.vf_mgmt_capability_supported(
vf_section, capability, subcapability)
self.assertEqual(expected, capable)
mock_exec.assert_called_once_with(['ip', 'link', 'help'],
check_exit_code=False,
return_stderr=True,
log_fail_as_error=False)
def test_vf_mgmt(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
stderr=self.IP_LINK_HELP)
def test_execute_with_stdout(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
stdout=self.IP_LINK_HELP)
def test_vf_mgmt_no_state(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
expected=False,
stderr=self.IP_LINK_HELP_NO_STATE)
def test_vf_mgmt_no_spoofchk(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_SPOOFCHK,
expected=False,
stderr=self.IP_LINK_HELP_NO_SPOOFCHK)
def test_vf_mgmt_no_vf(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
expected=False,
stderr=self.IP_LINK_HELP_NO_VF)
def test_vf_mgmt_unknown_capability(self):
self._test_capability(
"state1",
expected=False,
stderr=self.IP_LINK_HELP)
def test_vf_mgmt_sub_capability(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_VLAN,
ip_link.IpLinkConstants.IP_LINK_SUB_CAPABILITY_QOS,
stderr=self.IP_LINK_HELP)
def test_vf_mgmt_sub_capability_mismatch(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
ip_link.IpLinkConstants.IP_LINK_SUB_CAPABILITY_QOS,
expected=False,
stderr=self.IP_LINK_HELP)
def test_vf_mgmt_sub_capability_invalid(self):
self._test_capability(
ip_link.IpLinkConstants.IP_LINK_CAPABILITY_VLAN,
"qos1",
expected=False,
stderr=self.IP_LINK_HELP)
def test_vf_mgmt_error(self):
with mock.patch("neutron.agent.linux.utils.execute") as mock_exec:
mock_exec.side_effect = Exception()
self.assertRaises(
ip_link.UnsupportedIpLinkCommand,
ip_link.IpLinkSupport.get_vf_mgmt_section)

2
neutron/tests/unit/plugins/ml2/drivers/linuxbridge/agent/test_linuxbridge_neutron_agent.py
View File

@ -454,7 +454,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
return_value=vxlan_dev) as add_vxlan_fn,\
mock.patch.object(vxlan_dev.link, 'set_mtu',
side_effect=ip_lib.InvalidArgument(
parameter="MTU", value=mtu)),\
device='device_exists', namespace='ns')),\
mock.patch.object(ip_lib, 'get_device_mtu',
return_value=physical_mtu),\
mock.patch.object(vxlan_dev.link, 'delete') as delete_dev:

14
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/extension_drivers/test_qos_driver.py
View File

@ -22,9 +22,9 @@ from oslo_utils import uuidutils
from neutron.objects.qos import policy
from neutron.objects.qos import rule
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import exceptions
from neutron.plugins.ml2.drivers.mech_sriov.agent.extension_drivers import (
qos_driver)
from neutron.privileged.agent.linux import ip_lib as priv_ip_lib
from neutron.tests import base
@ -116,8 +116,16 @@ class QosSRIOVAgentDriverTestCase(base.BaseTestCase):
self.clear_max_rate_mock.assert_called_once_with(self.PCI_SLOT)
def test__set_vf_max_rate_captures_sriov_failure(self):
self.max_rate_mock.side_effect = exceptions.SriovNicError()
self.qos_driver._set_vf_max_rate(self.ASSIGNED_MAC, self.PCI_SLOT)
msg = 'Failed to set device %s max rate'
with mock.patch.object(qos_driver, 'LOG') as mock_log:
for exc in (priv_ip_lib.InterfaceOperationNotSupported(),
priv_ip_lib.InvalidArgument()):
self.max_rate_mock.side_effect = exc
self.qos_driver._set_vf_max_rate(self.ASSIGNED_MAC,
self.PCI_SLOT)
mock_log.exception.assert_called_once_with(msg,
self.ASSIGNED_MAC)
mock_log.exception.reset_mock()
def test__set_vf_max_rate_unknown_device(self):
with mock.patch.object(self.qos_driver.eswitch_mgr, 'device_exists',

38
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/test_eswitch_manager.py
View File

@ -16,7 +16,6 @@
import os
from unittest import mock
from neutron.agent.linux import ip_link_support
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
@ -70,8 +69,8 @@ class TestESwitchManagerApi(base.BaseTestCase):
PCI_SLOT = '0000:06:00.1'
WRONG_MAC = '00:00:00:00:00:67'
WRONG_PCI = "0000:06:00.6"
MAX_RATE = ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE
MIN_RATE = ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE
MAX_RATE = esm.IP_LINK_CAPABILITY_RATE
MIN_RATE = esm.IP_LINK_CAPABILITY_MIN_TX_RATE
def setUp(self):
super(TestESwitchManagerApi, self).setUp()
@ -524,55 +523,48 @@ class TestEmbSwitch(base.BaseTestCase):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2000)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 2000)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2)
0, esm.IP_LINK_CAPABILITY_RATE, 2)
def test_set_device_max_rate_ok2(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 99)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 99)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 1)
0, esm.IP_LINK_CAPABILITY_RATE, 1)
def test_set_device_max_rate_rounded_ok(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2001)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 2001)
pci_lib_mock.assert_called_with(0, esm.IP_LINK_CAPABILITY_RATE, 2)
def test_set_device_max_rate_rounded_ok2(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2499)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 2499)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2)
0, esm.IP_LINK_CAPABILITY_RATE, 2)
def test_set_device_max_rate_rounded_ok3(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 2500)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 2500)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 3)
0, esm.IP_LINK_CAPABILITY_RATE, 3)
def test_set_device_max_rate_disable(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.set_vf_rate") as pci_lib_mock:
self.emb_switch.set_device_rate(
self.PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 0)
self.PCI_SLOT, esm.IP_LINK_CAPABILITY_RATE, 0)
pci_lib_mock.assert_called_with(
0, ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 0)
0, esm.IP_LINK_CAPABILITY_RATE, 0)
def test_set_device_max_rate_fail(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
@ -581,7 +573,7 @@ class TestEmbSwitch(base.BaseTestCase):
exc.InvalidPciSlotError,
self.emb_switch.set_device_rate,
self.WRONG_PCI_SLOT,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE, 1000)
esm.IP_LINK_CAPABILITY_RATE, 1000)
def test_get_pci_device(self):
with mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."

172
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/test_pci_lib.py
View File

@ -15,28 +15,18 @@
from unittest import mock
from neutron.agent.linux import ip_link_support
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.agent.linux import ip_lib
from neutron.plugins.ml2.drivers.mech_sriov.agent import pci_lib
from neutron.tests import base
class TestPciLib(base.BaseTestCase):
DEV_NAME = "p7p1"
VF_INDEX = 1
VF_INDEX_DISABLE = 0
PF_LINK_SHOW = ('122: p7p1: <BROADCAST,MULTICAST> mtu 1500 qdisc noop'
' state DOWN mode DEFAULT group default qlen 1000')
PF_MAC = ' link/ether f4:52:14:2a:3e:c0 brd ff:ff:ff:ff:ff:ff'
VF_0_LINK_SHOW = (' vf 0 MAC fa:16:3e:b4:81:ac, vlan 4095, spoof'
' checking off, link-state disable')
VF_1_LINK_SHOW = (' vf 1 MAC 00:00:00:00:00:11, vlan 4095, spoof'
' checking off, link-state enable')
VF_2_LINK_SHOW = (' vf 2 MAC fa:16:3e:68:4e:79, vlan 4095, spoof'
' checking off, link-state enable')
VF_LINK_SHOW = '\n'.join((PF_LINK_SHOW, PF_MAC, VF_0_LINK_SHOW,
VF_1_LINK_SHOW, VF_2_LINK_SHOW))
VFS_LIST = {0: {'mac': 'fa:16:3e:b4:81:ac', 'link_state': 2},
1: {'mac': '00:00:00:00:00:11', 'link_state': 1},
2: {'mac': 'fa:16:3e:68:4e:79', 'link_state': 0}}
MAC_MAPPING = {
0: "fa:16:3e:b4:81:ac",
@ -44,121 +34,69 @@ class TestPciLib(base.BaseTestCase):
2: "fa:16:3e:68:4e:79",
}
STATE_MAPPING = { # VF index: state (string), according to VFS_LIST
0: pci_lib.LinkState.disable.name,
1: pci_lib.LinkState.enable.name,
2: pci_lib.LinkState.auto.name,
}
def setUp(self):
super(TestPciLib, self).setUp()
self.pci_wrapper = pci_lib.PciDeviceIPWrapper(self.DEV_NAME)
self.mock_ip_device = mock.Mock()
self.mock_ip_device.link.get_vfs.return_value = self.VFS_LIST
mock.patch.object(ip_lib, 'IPDevice',
return_value=self.mock_ip_device).start()
def test_get_assigned_macs(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.return_value = self.VF_LINK_SHOW
result = self.pci_wrapper.get_assigned_macs([self.VF_INDEX])
self.assertEqual(
{self.VF_INDEX: self.MAC_MAPPING[self.VF_INDEX]}, result)
for idx in range(len(self.VFS_LIST)):
result = self.pci_wrapper.get_assigned_macs([idx])
self.assertEqual({idx: self.MAC_MAPPING[idx]}, result)
def test_get_assigned_macs_fail(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandError,
self.pci_wrapper.get_assigned_macs,
[self.VF_INDEX])
def test_get_assigned_macs_not_present(self):
result = self.pci_wrapper.get_assigned_macs([1000])
self.assertEqual({}, result)
def test_get_vf_state_enable(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.return_value = self.VF_LINK_SHOW
result = self.pci_wrapper.get_vf_state(self.VF_INDEX)
self.assertEqual('enable', result)
def test_get_vf_state(self):
for idx in range(len(self.VFS_LIST)):
result = self.pci_wrapper.get_vf_state(idx)
self.assertEqual(self.STATE_MAPPING[idx], result)
def test_get_vf_state_disable(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.return_value = self.VF_LINK_SHOW
result = self.pci_wrapper.get_vf_state(self.VF_INDEX_DISABLE)
self.assertEqual('disable', result)
def test_get_vf_state_fail(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandError,
self.pci_wrapper.get_vf_state,
self.VF_INDEX)
def test_get_vf_state_not_present(self):
result = self.pci_wrapper.get_vf_state(1000)
self.assertEqual(pci_lib.LinkState.disable.name, result)
def test_set_vf_state(self):
with mock.patch.object(self.pci_wrapper, "_as_root"):
result = self.pci_wrapper.set_vf_state(self.VF_INDEX,
True)
self.assertIsNone(result)
self.pci_wrapper.set_vf_state(self.VF_INDEX, True)
vf = {'vf': self.VF_INDEX, 'link_state': 1}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
def test_set_vf_state_fail(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandError,
self.pci_wrapper.set_vf_state,
self.VF_INDEX,
True)
self.mock_ip_device.link.set_vf_feature.reset_mock()
self.pci_wrapper.set_vf_state(self.VF_INDEX, False)
vf = {'vf': self.VF_INDEX, 'link_state': 2}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
self.mock_ip_device.link.set_vf_feature.reset_mock()
self.pci_wrapper.set_vf_state(self.VF_INDEX, False, auto=True)
vf = {'vf': self.VF_INDEX, 'link_state': 0}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
def test_set_vf_spoofcheck(self):
with mock.patch.object(self.pci_wrapper, "_as_root"):
result = self.pci_wrapper.set_vf_spoofcheck(self.VF_INDEX,
True)
self.assertIsNone(result)
self.pci_wrapper.set_vf_spoofcheck(self.VF_INDEX, True)
vf = {'vf': self.VF_INDEX, 'spoofchk': 1}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
def test_set_vf_spoofcheck_fail(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandError,
self.pci_wrapper.set_vf_spoofcheck,
self.VF_INDEX,
True)
self.mock_ip_device.link.set_vf_feature.reset_mock()
self.pci_wrapper.set_vf_spoofcheck(self.VF_INDEX, False)
vf = {'vf': self.VF_INDEX, 'spoofchk': 0}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
def _set_vf_rate(self, rate, passed=True):
if passed:
with mock.patch.object(self.pci_wrapper, "_as_root") \
as mock_as_root:
result = self.pci_wrapper.set_vf_rate(
self.VF_INDEX,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE,
1000)
self.assertIsNone(result)
mock_as_root.assert_called_once_with(
[], "link", ("set", self.DEV_NAME, "vf",
str(self.VF_INDEX), "rate", '1000'))
else:
with mock.patch.object(self.pci_wrapper, "_as_root",
side_effect=Exception()):
self.assertRaises(exc.IpCommandError,
self.pci_wrapper.set_vf_rate,
self.VF_INDEX,
rate,
1000)
def test_set_vf_rate(self):
self.pci_wrapper.set_vf_rate(self.VF_INDEX, 'max_tx_rate', 20)
vf = {'vf': self.VF_INDEX, 'rate': {'max_tx_rate': 20}}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)
def test_set_vf_rate_max_rate(self):
self._set_vf_rate(
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE)
def test_set_vf_rate_max_rate_fail(self):
self._set_vf_rate('rate', passed=False)
def test_set_vf_rate_min_tx_rate(self):
self._set_vf_rate(
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE)
def test_set_vf_rate_min_tx_rate_fail(self):
self._set_vf_rate(
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_MIN_TX_RATE,
passed=False)
def test_set_vf_state_not_supported(self):
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception(
pci_lib.PciDeviceIPWrapper.IP_LINK_OP_NOT_SUPPORTED)
self.assertRaises(exc.IpCommandOperationNotSupportedError,
self.pci_wrapper.set_vf_state,
self.VF_INDEX,
state=True)
self.mock_ip_device.link.set_vf_feature.reset_mock()
self.pci_wrapper.set_vf_rate(self.VF_INDEX, 'min_tx_rate', 10)
vf = {'vf': self.VF_INDEX, 'rate': {'min_tx_rate': 10}}
self.mock_ip_device.link.set_vf_feature.assert_called_once_with(vf)

8
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/test_sriov_nic_agent.py
View File

@ -19,12 +19,13 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from oslo_config import cfg
from oslo_utils import uuidutils
import pyroute2
from neutron.agent.l2 import l2_agent_extensions_manager as l2_ext_manager
from neutron.agent import rpc as agent_rpc
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import config # noqa
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import exceptions
from neutron.plugins.ml2.drivers.mech_sriov.agent import sriov_nic_agent
from neutron.privileged.agent.linux import ip_lib as priv_ip_lib
from neutron.tests import base
DEVICE_MAC = '11:22:33:44:55:66'
@ -423,8 +424,7 @@ class TestSriovAgent(base.BaseTestCase):
agent.eswitch_mgr = mock.Mock()
agent.eswitch_mgr.device_exists.return_value = True
agent.eswitch_mgr.set_device_state.side_effect = (
exceptions.IpCommandOperationNotSupportedError(
dev_name='aa:bb:cc:dd:ee:ff'))
priv_ip_lib.InterfaceOperationNotSupported())
self.assertTrue(agent.treat_device('aa:bb:cc:dd:ee:ff', '1:2:3:0',
admin_state_up=True))
@ -435,7 +435,7 @@ class TestSriovAgent(base.BaseTestCase):
agent.eswitch_mgr = mock.Mock()
agent.eswitch_mgr.device_exists.return_value = True
agent.eswitch_mgr.set_device_state.side_effect = (
exceptions.SriovNicError())
pyroute2.NetlinkError(22))
self.assertFalse(agent.treat_device('aa:bb:cc:dd:ee:ff', '1:2:3:0',
admin_state_up=True))

33
neutron/tests/unit/privileged/agent/linux/test_ip_lib.py
View File

@ -226,6 +226,39 @@ class IpLibTestCase(base.BaseTestCase):
except OSError as e:
self.assertEqual(errno.EINVAL, e.errno)
def _clean(self, client_mode):
priv_lib.privileged.default.client_mode = client_mode
def test_get_link_vfs(self):
# NOTE(ralonsoh): there should be a functional test checking this
# method, but this is not possible due to the lack of SR-IOV capable
# NICs in the CI servers.
vf_info = []
for idx in range(3):
vf_info.append(pyroute2.netlink.nlmsg_base())
mac_info = {'mac': 'mac_%s' % idx, 'vf': idx}
link_state = {'link_state': idx} # see SR-IOV pci_lib.LinkState
vf_info[idx].setvalue(
{'attrs': [('IFLA_VF_MAC', mac_info),
('IFLA_VF_LINK_STATE', link_state)]})
vfinfo_list = pyroute2.netlink.nlmsg_base()
vfinfo_list.setvalue({'attrs': [('IFLA_VF_INFO', vf_info[0]),
('IFLA_VF_INFO', vf_info[1]),
('IFLA_VF_INFO', vf_info[2])]})
value = pyroute2.netlink.nlmsg_base()
value.setvalue({'attrs': [('IFLA_NUM_VF', 3),
('IFLA_VFINFO_LIST', vfinfo_list)]})
client_mode = priv_lib.privileged.default.client_mode
priv_lib.privileged.default.client_mode = False
self.addCleanup(self._clean, client_mode)
with mock.patch.object(priv_lib, '_run_iproute_link') as mock_iplink:
mock_iplink.return_value = [value]
result = priv_lib.get_link_vfs('device', 'namespace')
self.assertEqual({0: {'mac': 'mac_0', 'link_state': 0},
1: {'mac': 'mac_1', 'link_state': 1},
2: {'mac': 'mac_2', 'link_state': 2}},
result)
class MakeSerializableTestCase(base.BaseTestCase):