openstack/ovn-bgp-agent
Code Issues Proposed changes

Add new privileged method implementations

Browse Source 
This patch add the following new privileged methods implementations:
* create_interface: this method will be used to create dummy, vlan,
  vxlan, bridge and vrf interfaces.
* delete_interface: to delete a interface
* set_link_attribute: to set UP the interface
* add_ip_address: to add an IP address on a interface
* delete_ip_address: to delete an IP address on a interface
* set_master_for_device: defines a master device for a second one

These new method use pyroute2 IPRoute class, replacing the use of the
NDB class.

This patch is creating the functional test framework, in order to
test these new methods.

This patch is also adding a new CI job to the gate (to be defined).

Partial-Bug: #2022357
Change-Id: I40d70829bfccb2df98b822afacbdab7da5a5ab7f
This commit is contained in:
Rodolfo Alonso Hernandez 2023-06-03 00:00:59 +00:00 committed by Luis Tomas Bolivar
parent a5d8436049
commit 1cbfe7823c
16 changed files with 890 additions and 320 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
ovn_bgp_agent/exceptions.py
View File

@ -67,3 +67,11 @@ class PatchPortNotFound(OVNBGPAgentException):
"""
message = _("Patch port not found for localnet: %(localnet)s.")
class IpAddressAlreadyExists(RuntimeError):
message = _("IP address %(ip)s already configured on %(device)s.")
def __init__(self, message=None, ip=None, device=None):
message = message or self.message % {'ip': ip, 'device': device}
super(IpAddressAlreadyExists, self).__init__(message)

404
ovn_bgp_agent/privileged/linux_net.py
View File

@ -12,92 +12,130 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import errno
import ipaddress
import os
import socket
import netaddr
from socket import AF_INET6
from oslo_concurrency import processutils
from oslo_log import log as logging
import pyroute2
from pyroute2 import iproute
from pyroute2 import netlink as pyroute_netlink
from pyroute2.netlink import exceptions as netlink_exceptions
from pyroute2.netlink.rtnl import ndmsg
import tenacity
from ovn_bgp_agent import constants
from ovn_bgp_agent import exceptions as agent_exc
from ovn_bgp_agent.utils import linux_net as l_net
import ovn_bgp_agent.privileged.linux_net
LOG = logging.getLogger(__name__)
_IP_VERSION_FAMILY_MAP = {4: socket.AF_INET, 6: socket.AF_INET6}
@tenacity.retry(
retry=tenacity.retry_if_exception_type(
netlink_exceptions.NetlinkDumpInterrupted),
wait=tenacity.wait_exponential(multiplier=0.02, max=1),
stop=tenacity.stop_after_delay(8),
reraise=True)
@ovn_bgp_agent.privileged.default.entrypoint
def set_device_status(device, status, ndb=None):
_ndb = ndb
if ndb is None:
_ndb = pyroute2.NDB()
try:
with _ndb.interfaces[device] as dev:
if dev['state'] != status:
dev['state'] = status
finally:
if ndb is None:
_ndb.close()
class NetworkInterfaceNotFound(RuntimeError):
message = 'Network interface %(device)s not found'
def __init__(self, message=None, device=None):
message = message or self.message % {'device': device}
super(NetworkInterfaceNotFound, self).__init__(message)
class InterfaceAlreadyExists(RuntimeError):
message = "Interface %(device)s already exists."
def __init__(self, message=None, device=None):
message = message or self.message % {'device': device}
super(InterfaceAlreadyExists, self).__init__(message)
class InterfaceOperationNotSupported(RuntimeError):
message = "Operation not supported on interface %(device)s."
def __init__(self, message=None, device=None):
message = message or self.message % {'device': device}
super(InterfaceOperationNotSupported, self).__init__(message)
class InvalidArgument(RuntimeError):
message = "Invalid parameter/value used on interface %(device)s."
def __init__(self, message=None, device=None):
message = message or self.message % {'device': device}
super(InvalidArgument, self).__init__(message)
def set_device_state(device, state):
set_link_attribute(device, state=state)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_vrf(vrf_name, vrf_table):
with pyroute2.NDB() as ndb:
try:
set_device_status(vrf_name, constants.LINK_UP, ndb=ndb)
except KeyError:
ndb.interfaces.create(
kind="vrf", ifname=vrf_name, vrf_table=int(vrf_table)).set(
'state', constants.LINK_UP).commit()
set_device_state(vrf_name, constants.LINK_UP)
except NetworkInterfaceNotFound:
create_interface(vrf_name, 'vrf', vrf_table=vrf_table,
state=constants.LINK_UP)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_bridge(bridge_name):
with pyroute2.NDB() as ndb:
try:
set_device_status(bridge_name, constants.LINK_UP, ndb=ndb)
except KeyError:
ndb.interfaces.create(
kind="bridge", ifname=bridge_name, br_stp_state=0).set(
'state', constants.LINK_UP).commit()
set_device_state(bridge_name, constants.LINK_UP)
except NetworkInterfaceNotFound:
create_interface(bridge_name, 'bridge', br_stp_state=0,
state=constants.LINK_UP)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_vxlan(vxlan_name, vni, local_ip, dstport):
with pyroute2.NDB() as ndb:
try:
set_device_status(vxlan_name, constants.LINK_UP, ndb=ndb)
except KeyError:
set_device_state(vxlan_name, constants.LINK_UP)
except NetworkInterfaceNotFound:
# FIXME: Perhaps we need to set neigh_suppress on
ndb.interfaces.create(
kind="vxlan", ifname=vxlan_name, vxlan_id=int(vni),
vxlan_port=dstport, vxlan_local=local_ip,
vxlan_learning=False).set('state', constants.LINK_UP).commit()
create_interface(vxlan_name, 'vxlan',
vxlan_id=vni,
vxlan_port=dstport,
vxlan_local=local_ip,
vxlan_learning=False,
state=constants.LINK_UP)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_veth(veth_name, veth_peer):
try:
set_device_status(veth_name, constants.LINK_UP)
except KeyError:
with pyroute2.NDB() as ndb:
ndb.interfaces.create(
kind="veth", ifname=veth_name, peer=veth_peer).set(
'state', constants.LINK_UP).commit()
set_device_status(veth_peer, constants.LINK_UP)
set_device_state(veth_name, constants.LINK_UP)
except NetworkInterfaceNotFound:
create_interface(veth_name, 'veth', peer=veth_peer,
state=constants.LINK_UP)
set_device_state(veth_peer, constants.LINK_UP)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_dummy_device(device):
try:
set_device_state(device, constants.LINK_UP)
except NetworkInterfaceNotFound:
create_interface(device, 'dummy', state=constants.LINK_UP)
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_vlan_device_for_network(bridge, vlan_tag):
vlan_device_name = '{}.{}'.format(bridge, vlan_tag)
try:
set_device_state(vlan_device_name, constants.LINK_UP)
except NetworkInterfaceNotFound:
create_interface(vlan_device_name, 'vlan',
physical_interface=bridge,
vlan_id=vlan_tag,
state=constants.LINK_UP)
@tenacity.retry(
@ -108,30 +146,25 @@ def ensure_veth(veth_name, veth_peer):
reraise=True)
@ovn_bgp_agent.privileged.default.entrypoint
def set_master_for_device(device, master):
with pyroute2.NDB() as ndb:
# Check if already associated to the master, and associate it if not
if (ndb.interfaces[device].get('master') !=
ndb.interfaces[master]['index']):
with ndb.interfaces[device] as iface:
iface.set('master', ndb.interfaces[master]['index'])
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_dummy_device(device):
with pyroute2.NDB() as ndb:
try:
set_device_status(device, constants.LINK_UP, ndb=ndb)
except KeyError:
ndb.interfaces.create(kind="dummy", ifname=device).set(
'state', constants.LINK_UP).commit()
with pyroute2.IPRoute() as ipr:
dev_index = ipr.link_lookup(ifname=device)[0]
master_index = ipr.link_lookup(ifname=master)[0]
# Check if already associated to the master,
# and associate it if not
iface = ipr.link('get', index=dev_index)[0]
if iface.get_attr('IFLA_MASTER') != master_index:
ipr.link('set', index=dev_index, master=master_index)
except IndexError:
LOG.debug("No need to set %s on VRF %s, as one of them is deleted",
device, master)
@ovn_bgp_agent.privileged.default.entrypoint
def delete_device(device):
try:
with pyroute2.NDB() as ndb:
ndb.interfaces[device].remove().commit()
except KeyError:
delete_interface(device)
except NetworkInterfaceNotFound:
LOG.debug("Interfaces %s already deleted.", device)
@ -154,20 +187,6 @@ def route_delete(route):
LOG.debug("Route already deleted: {}".format(route))
@ovn_bgp_agent.privileged.default.entrypoint
def ensure_vlan_device_for_network(bridge, vlan_tag):
vlan_device_name = '{}.{}'.format(bridge, vlan_tag)
with pyroute2.NDB() as ndb:
try:
set_device_status(vlan_device_name, constants.LINK_UP, ndb=ndb)
except KeyError:
ndb.interfaces.create(
kind="vlan", ifname=vlan_device_name, vlan_id=vlan_tag,
link=ndb.interfaces[bridge]['index']).set(
'state', constants.LINK_UP).commit()
@ovn_bgp_agent.privileged.default.entrypoint
def set_kernel_flag(flag, value):
command = ["sysctl", "-w", "{}={}".format(flag, value)]
@ -180,16 +199,8 @@ def set_kernel_flag(flag, value):
@ovn_bgp_agent.privileged.default.entrypoint
def delete_exposed_ips(ips, nic):
with pyroute2.NDB() as ndb:
for ip in ips:
address = '{}/32'.format(ip)
if l_net.get_ip_version(ip) == constants.IP_VERSION_6:
address = '{}/128'.format(ip)
try:
ndb.interfaces[nic].ipaddr[address].remove().commit()
except KeyError:
LOG.debug("IP address {} already removed from nic {}.".format(
ip, nic))
for ip_address in ips:
delete_ip_address(ip_address, nic)
@ovn_bgp_agent.privileged.default.entrypoint
@ -284,42 +295,14 @@ def del_ndp_proxy(ip, dev, vlan=None):
raise
@tenacity.retry(
retry=tenacity.retry_if_exception_type(
netlink_exceptions.NetlinkDumpInterrupted),
wait=tenacity.wait_exponential(multiplier=0.02, max=1),
stop=tenacity.stop_after_delay(8),
reraise=True)
@ovn_bgp_agent.privileged.default.entrypoint
def add_ip_to_dev(ip, nic):
address = '{}/32'.format(ip)
if l_net.get_ip_version(ip) == constants.IP_VERSION_6:
address = '{}/128'.format(ip)
try:
with pyroute2.NDB() as ndb:
with ndb.interfaces[nic] as iface:
iface.add_ip(address)
except KeyError: # Already exists
LOG.debug("IP %s already added to interface %s.", address, nic)
add_ip_address(ip, nic)
@tenacity.retry(
retry=tenacity.retry_if_exception_type(
netlink_exceptions.NetlinkDumpInterrupted),
wait=tenacity.wait_exponential(multiplier=0.02, max=1),
stop=tenacity.stop_after_delay(8),
reraise=True)
@ovn_bgp_agent.privileged.default.entrypoint
def del_ip_from_dev(ip, nic):
address = '{}/32'.format(ip)
if l_net.get_ip_version(ip) == constants.IP_VERSION_6:
address = '{}/128'.format(ip)
try:
with pyroute2.NDB() as ndb:
with ndb.interfaces[nic] as iface:
iface.del_ip(address)
except KeyError: # Already deleted
LOG.debug("IP %s already deleted from interface %s.", address, nic)
delete_ip_address(ip, nic)
@ovn_bgp_agent.privileged.default.entrypoint
@ -402,3 +385,192 @@ def add_unreachable_route(vrf_name):
def create_routing_table_for_bridge(table_number, bridge):
with open('/etc/iproute2/rt_tables', 'a') as rt_tables:
rt_tables.write('{} {}\n'.format(table_number, bridge))
def _translate_ip_device_exception(e, device):
if e.code == errno.ENODEV:
raise NetworkInterfaceNotFound(device=device)
if e.code == errno.EOPNOTSUPP:
raise InterfaceOperationNotSupported(device=device)
if e.code == errno.EINVAL:
raise InvalidArgument(device=device)
if e.code == errno.EEXIST:
raise InterfaceAlreadyExists(device=device)
raise e
def _translate_ip_addr_exception(e, ip, device):
if e.code == errno.EEXIST:
raise agent_exc.IpAddressAlreadyExists(ip=ip, device=device)
if e.code == errno.EADDRNOTAVAIL:
LOG.debug('No need to delete IP address %s on dev %s as it does '
'not exist', ip, device)
return
raise e
def get_attr(pyroute2_obj, attr_name):
"""Get an attribute in a pyroute object
pyroute2 object attributes are stored under a key called 'attrs'. This key
contains a tuple of tuples. E.g.:
pyroute2_obj = {'attrs': (('TCA_KIND': 'htb'),
('TCA_OPTIONS': {...}))}
:param pyroute2_obj: (dict) pyroute2 object
:param attr_name: (string) first value of the tuple we are looking for
:return: (object) second value of the tuple, None if the tuple doesn't
exist
"""
rule_attrs = pyroute2_obj.get('attrs', [])
for attr in (attr for attr in rule_attrs if attr[0] == attr_name):
return attr[1]
def make_serializable(value):
"""Make a pyroute2 object serializable
This function converts 'netlink.nla_slot' object (key, value) in a list
of two elements.
"""
def _ensure_string(value):
return value.decode() if isinstance(value, bytes) else value
if isinstance(value, list):
return [make_serializable(item) for item in value]
elif isinstance(value, pyroute_netlink.nla_slot):
return [_ensure_string(value[0]), make_serializable(value[1])]
elif isinstance(value, pyroute_netlink.nla_base):
return make_serializable(value.dump())
elif isinstance(value, dict):
return {_ensure_string(key): make_serializable(data)
for key, data in value.items()}
elif isinstance(value, tuple):
return tuple(make_serializable(item) for item in value)
return _ensure_string(value)
def _get_link_id(ifname, raise_exception=True):
with iproute.IPRoute() as ip:
link_id = ip.link_lookup(ifname=ifname)
if not link_id or len(link_id) < 1:
if raise_exception:
raise NetworkInterfaceNotFound(device=ifname)
LOG.debug('Interface %(dev)s not found', {'dev': ifname})
return None
return link_id[0]
@ovn_bgp_agent.privileged.default.entrypoint
def get_link_id(device):
return _get_link_id(device, raise_exception=False)
def get_link_state(device_name):
device = get_link_device(device_name)
return device['state'] if device else None
def get_link_device(device_name):
for device in get_link_devices():
if get_attr(device, 'IFLA_IFNAME') == device_name:
return device
@tenacity.retry(
retry=tenacity.retry_if_exception_type(
netlink_exceptions.NetlinkDumpInterrupted),
wait=tenacity.wait_exponential(multiplier=0.02, max=1),
stop=tenacity.stop_after_delay(8),
reraise=True)
@ovn_bgp_agent.privileged.default.entrypoint
def get_link_devices(**kwargs):
"""List interfaces in a namespace
:return: (list) interfaces in a namespace
"""
index = kwargs.pop('index') if 'index' in kwargs else 'all'
try:
with iproute.IPRoute() as ip:
return make_serializable(ip.get_links(index, **kwargs))
except OSError:
raise
def _run_iproute_link(command, ifname, **kwargs):
try:
with iproute.IPRoute() as ip:
idx = _get_link_id(ifname)
return ip.link(command, index=idx, **kwargs)
except netlink_exceptions.NetlinkError as e:
_translate_ip_device_exception(e, ifname)
def _run_iproute_addr(command, device, **kwargs):
try:
with iproute.IPRoute() as ip:
idx = _get_link_id(device)
return ip.addr(command, index=idx, **kwargs)
except netlink_exceptions.NetlinkError as e:
_translate_ip_addr_exception(e, ip=kwargs['address'], device=device)
@ovn_bgp_agent.privileged.default.entrypoint
def create_interface(ifname, kind, **kwargs):
ifname = ifname[:15]
try:
with iproute.IPRoute() as ip:
physical_interface = kwargs.pop('physical_interface', None)
if physical_interface:
link_key = 'vxlan_link' if kind == 'vxlan' else 'link'
kwargs[link_key] = _get_link_id(physical_interface)
ip.link("add", ifname=ifname, kind=kind, **kwargs)
except netlink_exceptions.NetlinkError as e:
_translate_ip_device_exception(e, ifname)
def delete_interface(ifname, **kwargs):
_run_iproute_link('del', ifname, **kwargs)
@ovn_bgp_agent.privileged.default.entrypoint
def set_link_attribute(ifname, **kwargs):
_run_iproute_link("set", ifname, **kwargs)
@ovn_bgp_agent.privileged.default.entrypoint
def add_ip_address(ip_address, ifname):
net = netaddr.IPNetwork(ip_address)
ip_version = l_net.get_ip_version(ip_address)
address = str(net.ip)
prefixlen = 32 if ip_version == 4 else 128
family = _IP_VERSION_FAMILY_MAP[ip_version]
_run_iproute_addr('add',
ifname,
address=address,
mask=prefixlen,
family=family)
@ovn_bgp_agent.privileged.default.entrypoint
def delete_ip_address(ip_address, ifname):
net = netaddr.IPNetwork(ip_address)
ip_version = l_net.get_ip_version(ip_address)
address = str(net.ip)
prefixlen = 32 if ip_version == 4 else 128
family = _IP_VERSION_FAMILY_MAP[ip_version]
_run_iproute_addr("delete",
ifname,
address=address,
mask=prefixlen,
family=family)
@ovn_bgp_agent.privileged.default.entrypoint
def get_ip_addresses(**kwargs):
"""List of IP addresses in a namespace
:return: (tuple) IP addresses in a namespace
"""
with iproute.IPRoute() as ip:
return make_serializable(ip.get_addr(**kwargs))

11
ovn_bgp_agent/tests/base.py
View File

@ -19,6 +19,8 @@ from unittest import mock
from oslotest import base
from ovn_bgp_agent import privileged
class TestCase(base.BaseTestCase):
@ -26,4 +28,13 @@ class TestCase(base.BaseTestCase):
def setUp(self):
super(TestCase, self).setUp()
privileged.default.client_mode = False
privileged.ovs_vsctl_cmd.client_mode = False
privileged.vtysh_cmd.client_mode = False
self.addCleanup(self._clean_up)
self.addCleanup(mock.patch.stopall)
def _clean_up(self):
privileged.default.client_mode = True
privileged.ovs_vsctl_cmd.client_mode = True
privileged.vtysh_cmd.client_mode = True

0
ovn_bgp_agent/tests/functional/__init__.py Normal file
View File

121
ovn_bgp_agent/tests/functional/base.py Normal file
View File

@ -0,0 +1,121 @@
# Derived from: neutron/tests/functional/base.py
# neutron/tests/base.py
#
# 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 abc
import functools
import inspect
import os
import sys
import eventlet.timeout
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import fileutils
from oslotest import base
import ovn_bgp_agent
from ovn_bgp_agent import config
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
def _get_test_log_path():
return os.environ.get('OS_LOG_PATH', '/tmp')
# This is the directory from which infra fetches log files for functional tests
DEFAULT_LOG_DIR = os.path.join(_get_test_log_path(), 'functional-logs')
class _CatchTimeoutMetaclass(abc.ABCMeta):
def __init__(cls, name, bases, dct):
super(_CatchTimeoutMetaclass, cls).__init__(name, bases, dct)
for name, method in inspect.getmembers(
# NOTE(ihrachys): we should use isroutine because it will catch
# both unbound methods (python2) and functions (python3)
cls, predicate=inspect.isroutine):
if name.startswith('test_'):
setattr(cls, name, cls._catch_timeout(method))
@staticmethod
def _catch_timeout(f):
@functools.wraps(f)
def func(self, *args, **kwargs):
try:
return f(self, *args, **kwargs)
except eventlet.Timeout as e:
self.fail('Execution of this test timed out: %s' % e)
return func
def setup_logging(component_name):
"""Sets up the logging options for a log with supplied name."""
logging.setup(cfg.CONF, component_name)
LOG.info("Logging enabled!")
LOG.info("%(prog)s version %(version)s",
{'prog': sys.argv[0], 'version': ovn_bgp_agent.__version__})
LOG.debug("command line: %s", " ".join(sys.argv))
def sanitize_log_path(path):
"""Sanitize the string so that its log path is shell friendly"""
return path.replace(' ', '-').replace('(', '_').replace(')', '_')
# Test worker cannot survive eventlet's Timeout exception, which effectively
# kills the whole worker, with all test cases scheduled to it. This metaclass
# makes all test cases convert Timeout exceptions into unittest friendly
# failure mode (self.fail).
class BaseFunctionalTestCase(base.BaseTestCase,
metaclass=_CatchTimeoutMetaclass):
"""Base class for functional tests."""
COMPONENT_NAME = 'ovn_bgp_agent'
PRIVILEGED_GROUP = 'privsep'
def setUp(self):
super(BaseFunctionalTestCase, self).setUp()
logging.register_options(CONF)
setup_logging(self.COMPONENT_NAME)
fileutils.ensure_tree(DEFAULT_LOG_DIR, mode=0o755)
log_file = sanitize_log_path(
os.path.join(DEFAULT_LOG_DIR, "%s.txt" % self.id()))
self.flags(log_file=log_file)
config.register_opts()
config.setup_privsep()
privsep_helper = os.path.join(
os.getenv('VIRTUAL_ENV', os.path.dirname(sys.executable)[:-4]),
'bin', 'privsep-helper')
self.flags(
helper_command=' '.join(['sudo', '-E', privsep_helper]),
group=self.PRIVILEGED_GROUP)
def flags(self, **kw):
"""Override some configuration values.
The keyword arguments are the names of configuration options to
override and their values.
If a group argument is supplied, the overrides are applied to
the specified configuration option group.
All overrides are automatically cleared at the end of the current
test by the fixtures cleanup process.
"""
group = kw.pop('group', None)
for k, v in kw.items():
CONF.set_override(k, v, group)

0
ovn_bgp_agent/tests/functional/privileged/__init__.py Normal file
View File

400
ovn_bgp_agent/tests/functional/privileged/test_linux_net.py Normal file
View File

@ -0,0 +1,400 @@
# Copyright 2023 Red Hat, Inc.
#
# 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 functools
import random
import netaddr
from oslo_utils import uuidutils
from pyroute2.netlink import rtnl
from pyroute2.netlink.rtnl import ifaddrmsg
from ovn_bgp_agent import constants
from ovn_bgp_agent import exceptions as agent_exc
from ovn_bgp_agent.privileged import linux_net
from ovn_bgp_agent.tests.functional import base as base_functional
from ovn_bgp_agent.tests import utils as test_utils
from ovn_bgp_agent.utils import linux_net as l_net
IP_ADDRESS_EVENTS = {'RTM_NEWADDR': 'added',
'RTM_DELADDR': 'removed'}
IP_ADDRESS_SCOPE = {rtnl.rtscopes['RT_SCOPE_UNIVERSE']: 'global',
rtnl.rtscopes['RT_SCOPE_SITE']: 'site',
rtnl.rtscopes['RT_SCOPE_LINK']: 'link',
rtnl.rtscopes['RT_SCOPE_HOST']: 'host'}
def set_up(ifname):
linux_net.set_link_attribute(ifname, state='up')
def ip_to_cidr(ip, prefix=None):
"""Convert an ip with no prefix to cidr notation
:param ip: An ipv4 or ipv6 address. Convertible to netaddr.IPNetwork.
:param prefix: Optional prefix. If None, the default 32 will be used for
ipv4 and 128 for ipv6.
"""
net = netaddr.IPNetwork(ip)
if prefix is not None:
# Can't pass ip and prefix separately. Must concatenate strings.
net = netaddr.IPNetwork(str(net.ip) + '/' + str(prefix))
return str(net)
def _parse_ip_address(pyroute2_address, device_name):
ip = linux_net.get_attr(pyroute2_address, 'IFA_ADDRESS')
ip_length = pyroute2_address['prefixlen']
event = IP_ADDRESS_EVENTS.get(pyroute2_address.get('event'))
cidr = ip_to_cidr(ip, prefix=ip_length)
flags = linux_net.get_attr(pyroute2_address, 'IFA_FLAGS')
dynamic = not bool(flags & ifaddrmsg.IFA_F_PERMANENT)
tentative = bool(flags & ifaddrmsg.IFA_F_TENTATIVE)
dadfailed = bool(flags & ifaddrmsg.IFA_F_DADFAILED)
scope = IP_ADDRESS_SCOPE[pyroute2_address['scope']]
return {'name': device_name,
'cidr': cidr,
'scope': scope,
'broadcast': linux_net.get_attr(pyroute2_address, 'IFA_BROADCAST'),
'dynamic': dynamic,
'tentative': tentative,
'dadfailed': dadfailed,
'event': event}
def get_ip_addresses(ifname):
device = get_devices_info(ifname=ifname)
if not device:
return
ip_addresses = linux_net.get_ip_addresses(
index=list(device.values())[0]['index'])
return [_parse_ip_address(_ip, ifname) for _ip in ip_addresses]
def get_devices_info(**kwargs):
devices = linux_net.get_link_devices(**kwargs)
retval = {}
for device in devices:
ret = {'index': device['index'],
'name': linux_net.get_attr(device, 'IFLA_IFNAME'),
'operstate': linux_net.get_attr(device, 'IFLA_OPERSTATE'),
'state': device['state'],
'linkmode': linux_net.get_attr(device, 'IFLA_LINKMODE'),
'mtu': linux_net.get_attr(device, 'IFLA_MTU'),
'promiscuity': linux_net.get_attr(device, 'IFLA_PROMISCUITY'),
'mac': linux_net.get_attr(device, 'IFLA_ADDRESS'),
'broadcast': linux_net.get_attr(device, 'IFLA_BROADCAST'),
'master': linux_net.get_attr(device, 'IFLA_MASTER'),
}
ifla_link = linux_net.get_attr(device, 'IFLA_LINK')
if ifla_link:
ret['parent_index'] = ifla_link
ifla_linkinfo = linux_net.get_attr(device, 'IFLA_LINKINFO')
if ifla_linkinfo:
ret['kind'] = linux_net.get_attr(ifla_linkinfo, 'IFLA_INFO_KIND')
ret['slave_kind'] = linux_net.get_attr(ifla_linkinfo,
'IFLA_INFO_SLAVE_KIND')
ifla_data = linux_net.get_attr(ifla_linkinfo, 'IFLA_INFO_DATA')
if ret['kind'] == 'vxlan':
ret['vxlan_id'] = linux_net.get_attr(ifla_data,
'IFLA_VXLAN_ID')
ret['vxlan_group'] = linux_net.get_attr(ifla_data,
'IFLA_VXLAN_GROUP')
ret['vxlan_link_index'] = linux_net.get_attr(ifla_data,
'IFLA_VXLAN_LINK')
ret['vxlan_port'] = linux_net.get_attr(ifla_data,
'IFLA_VXLAN_PORT')
ret['vxlan_local'] = linux_net.get_attr(ifla_data,
'IFLA_VXLAN_LOCAL')
ret['vxlan_learning'] = bool(
linux_net.get_attr(ifla_data, 'IFLA_VXLAN_LEARNING'))
elif ret['kind'] == 'vlan':
ret['vlan_id'] = linux_net.get_attr(ifla_data, 'IFLA_VLAN_ID')
elif ret['kind'] == 'bridge':
ret['stp'] = linux_net.get_attr(ifla_data, 'IFLA_BR_STP_STATE')
ret['forward_delay'] = linux_net.get_attr(
ifla_data, 'IFLA_BR_FORWARD_DELAY')
elif ret['kind'] == 'vrf':
ret['vrf_table'] = linux_net.get_attr(ifla_data,
'IFLA_VRF_TABLE')
retval[device['index']] = ret
for device in retval.values():
if device.get('parent_index'):
parent_device = retval.get(device['parent_index'])
if parent_device:
device['parent_name'] = parent_device['name']
elif device.get('vxlan_link_index'):
device['vxlan_link_name'] = (
retval[device['vxlan_link_index']]['name'])
return retval
class LinuxNetTestCase(base_functional.BaseFunctionalTestCase):
def setUp(self):
super().setUp()
self.dev_name = uuidutils.generate_uuid()[:15]
self.dev_name2 = uuidutils.generate_uuid()[:15]
self.addCleanup(self._delete_interface)
def _delete_interface(self):
def delete_device(device_name):
try:
linux_net.delete_interface(device_name)
except Exception:
pass
if self._get_device(self.dev_name):
delete_device(self.dev_name)
if self._get_device(self.dev_name2):
delete_device(self.dev_name2)
def _get_device(self, device_name):
devices = get_devices_info()
for device in devices.values():
if device['name'] == device_name:
return device
def _assert_state(self, device_name, state):
device = self._get_device(device_name)
return state == device['state']
def _check_status(self, device_name):
fn = functools.partial(self._assert_state, device_name,
constants.LINK_DOWN)
test_utils.wait_until_true(fn, timeout=5)
set_up(device_name)
fn = functools.partial(self._assert_state, device_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_create_interface_dummy(self):
linux_net.create_interface(self.dev_name, 'dummy')
device = self._get_device(self.dev_name)
self.assertEqual('dummy', device['kind'])
self._check_status(self.dev_name)
def test_create_interface_vlan(self):
vlan_id = random.randint(2, 4094)
linux_net.create_interface(self.dev_name, 'dummy')
linux_net.create_interface(self.dev_name2, 'vlan',
physical_interface=self.dev_name,
vlan_id=vlan_id)
device = self._get_device(self.dev_name2)
self.assertEqual('vlan', device['kind'])
self.assertEqual(vlan_id, device['vlan_id'])
self._check_status(self.dev_name)
def test_create_interface_vxlan(self):
vxlan_id = random.randint(2, 4094)
vxlan_port = random.randint(10000, 65534)
vxlan_local = '1.2.3.4'
linux_net.create_interface(self.dev_name, 'vxlan',
vxlan_id=vxlan_id,
vxlan_port=vxlan_port,
vxlan_local=vxlan_local,
vxlan_learning=False,
state=constants.LINK_UP)
device = self._get_device(self.dev_name)
self.assertEqual('vxlan', device['kind'])
self.assertEqual(vxlan_id, device['vxlan_id'])
self.assertEqual(vxlan_port, device['vxlan_port'])
self.assertEqual(vxlan_local, device['vxlan_local'])
self.assertEqual(constants.LINK_UP, device['state'])
self.assertFalse(device['vxlan_learning'])
def test_create_interface_veth(self):
linux_net.create_interface(self.dev_name, 'veth', peer=self.dev_name2)
device = self._get_device(self.dev_name)
self.assertEqual('veth', device['kind'])
self.assertEqual(self.dev_name2, device['parent_name'])
device = self._get_device(self.dev_name2)
self.assertEqual('veth', device['kind'])
self.assertEqual(self.dev_name, device['parent_name'])
self._check_status(self.dev_name)
self._check_status(self.dev_name2)
def test_create_interface_bridge(self):
linux_net.create_interface(self.dev_name, 'bridge', br_stp_state=0)
device = self._get_device(self.dev_name)
self.assertEqual('bridge', device['kind'])
self.assertEqual(0, device['stp'])
self._check_status(self.dev_name)
def test_create_interface_vrf(self):
vrf_table = random.randint(10, 2000)
linux_net.create_interface(self.dev_name, 'vrf', vrf_table=vrf_table)
device = self._get_device(self.dev_name)
self.assertEqual('vrf', device['kind'])
self.assertEqual(vrf_table, device['vrf_table'])
self._check_status(self.dev_name)
def test_add_and_delete_ip_address(self):
def check_ip_address(ip_address, device_name, present=True):
ip_addresses = get_ip_addresses(self.dev_name)
if l_net.get_ip_version(ip_address) == constants.IP_VERSION_6:
address = '{}/128'.format(ip_address)
else:
address = '{}/32'.format(ip_address)
for _ip in ip_addresses:
if _ip['cidr'] == address:
if present:
return
else:
self.fail('IP address %s present in device %s' %
(ip_address, device_name))
if present:
self.fail('IP address %s not found in device %s' %
(ip_address, device_name))
ip_addresses = ('240.0.0.1', 'fd00::1')
linux_net.create_interface(self.dev_name, 'dummy')
for ip_address in ip_addresses:
linux_net.add_ip_address(ip_address, self.dev_name)
check_ip_address(ip_address, self.dev_name)
# ensure nothing breaks if same IP gets added
# It should raise exception that is handled in the utils
self.assertRaises(agent_exc.IpAddressAlreadyExists,
linux_net.add_ip_address, ip_address,
self.dev_name)
for ip_address in ip_addresses:
linux_net.delete_ip_address(ip_address, self.dev_name)
check_ip_address(ip_address, self.dev_name, present=False)
# ensure removing a missing IP is ok
linux_net.delete_ip_address(ip_address, self.dev_name)
def test_add_ip_address_no_device(self):
self.assertRaises(linux_net.NetworkInterfaceNotFound,
linux_net.add_ip_address, '240.0.0.1', self.dev_name)
def test_delete_ip_address_no_device(self):
self.assertRaises(linux_net.NetworkInterfaceNotFound,
linux_net.delete_ip_address, '240.0.0.1',
self.dev_name)
def test_delete_ip_address_no_ip_on_device(self):
linux_net.create_interface(self.dev_name, 'dummy')
# No exception is raised.
linux_net.delete_ip_address('192.168.0.1', self.dev_name)
def _check_device_master_vrf(self, device, master=None):
device_info = self._get_device(device)
if not master:
self.assertIsNone(device_info['master'])
self.assertIsNone(device_info['slave_kind'])
else:
master_info = self._get_device(master)
self.assertEqual(master_info['index'], device_info['master'])
self.assertEqual('vrf', device_info['slave_kind'])
def test_set_master_for_device_bridge(self):
vrf_table = random.randint(10, 2000)
linux_net.create_interface(self.dev_name, 'vrf', vrf_table=vrf_table)
linux_net.create_interface(self.dev_name2, 'bridge', br_stp_state=0)
self._check_device_master_vrf(self.dev_name2)
linux_net.set_master_for_device(self.dev_name2, self.dev_name)
self._check_device_master_vrf(self.dev_name2, master=self.dev_name)
def test_set_master_for_device_dummy(self):
vrf_table = random.randint(10, 2000)
linux_net.create_interface(self.dev_name, 'vrf', vrf_table=vrf_table)
linux_net.create_interface(self.dev_name2, 'dummy')
self._check_device_master_vrf(self.dev_name2)
linux_net.set_master_for_device(self.dev_name2, self.dev_name)
self._check_device_master_vrf(self.dev_name2, master=self.dev_name)
def test_set_master_for_device_vlan(self):
vrf_table = random.randint(10, 2000)
linux_net.create_interface(self.dev_name, 'vrf', vrf_table=vrf_table)
vlan_id = random.randint(2, 4094)
dev_name3 = uuidutils.generate_uuid()[:15]
linux_net.create_interface(self.dev_name2, 'dummy')
linux_net.create_interface(dev_name3, 'vlan',
physical_interface=self.dev_name2,
vlan_id=vlan_id)
self._check_device_master_vrf(dev_name3)
linux_net.set_master_for_device(dev_name3, self.dev_name)
self._check_device_master_vrf(dev_name3, master=self.dev_name)
def test_set_master_for_device_veth(self):
vrf_table = random.randint(10, 2000)
linux_net.create_interface(self.dev_name, 'vrf', vrf_table=vrf_table)
dev_name3 = uuidutils.generate_uuid()[:15]
linux_net.create_interface(self.dev_name2, 'veth', peer=dev_name3)
self._check_device_master_vrf(self.dev_name2)
linux_net.set_master_for_device(self.dev_name2, self.dev_name)
self._check_device_master_vrf(self.dev_name2, master=self.dev_name)
def test_ensure_vlan_device_for_network(self):
self.dev_name = uuidutils.generate_uuid()[:8]
linux_net.create_interface(self.dev_name, 'dummy')
linux_net.set_device_state(self.dev_name, constants.LINK_UP)
vlan_id = random.randint(2, 4094)
# Ensure the method call is idempotent.
for _ in range(2):
linux_net.ensure_vlan_device_for_network(self.dev_name, vlan_id)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_ensure_vrf(self):
vrf_table = random.randint(10, 2000)
# Ensure the method call is idempotent.
for _ in range(2):
linux_net.ensure_vrf(self.dev_name, vrf_table)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_ensure_bridge(self):
# Ensure the method call is idempotent.
for _ in range(2):
linux_net.ensure_bridge(self.dev_name)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_ensure_vxlan(self):
vxlan_id = random.randint(2, 4094)
vxlan_port = random.randint(10000, 65534)
vxlan_local = '1.2.3.4'
# Ensure the method call is idempotent.
for _ in range(2):
linux_net.ensure_vxlan(self.dev_name, vxlan_id, vxlan_local,
vxlan_port)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_ensure_veth(self):
# Ensure the method call is idempotent.
for _ in range(2):
linux_net.ensure_veth(self.dev_name, self.dev_name2)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)
def test_ensure_dummy(self):
for _ in range(2):
linux_net.ensure_dummy_device(self.dev_name)
fn = functools.partial(self._assert_state, self.dev_name,
constants.LINK_UP)
test_utils.wait_until_true(fn, timeout=5)

4
ovn_bgp_agent/tests/unit/drivers/openstack/test_ovn_bgp_driver.py
View File

@ -79,8 +79,10 @@ class TestOVNBGPDriver(test_base.TestCase):
self.mock_ndb = mock.patch.object(linux_net.pyroute2, 'NDB').start()
self.fake_ndb = self.mock_ndb().__enter__()
@mock.patch.object(linux_net, 'ensure_ovn_device')
@mock.patch.object(linux_net, 'ensure_vrf')
@mock.patch.object(frr, 'vrf_leak')
def test_start(self, mock_vrf):
def test_start(self, mock_vrf, *args):
self.bgp_driver.start()
mock_vrf.assert_called_once_with(

7
ovn_bgp_agent/tests/unit/drivers/openstack/test_ovn_stretched_l2_bgp_driver.py
View File

@ -137,10 +137,12 @@ class TestOVNBGPStretchedL2Driver(test_base.TestCase):
self.mock_ndb = mock.patch.object(linux_net.pyroute2, "NDB").start()
self.fake_ndb = self.mock_ndb().__enter__()
@mock.patch.object(linux_net, "ensure_vrf")
@mock.patch.object(linux_net, "ensure_ovn_device")
@mock.patch.object(linux_net, "delete_routes_from_table")
@mock.patch.object(frr, "vrf_leak")
def test_start(self, mock_vrf, mock_delete_routes, mock_ensure_ovn_device):
def test_start(self, mock_vrf, mock_delete_routes, mock_ensure_ovn_device,
*args):
CONF.set_override("clear_vrf_routes_on_startup", True)
self.bgp_driver.start()
@ -160,11 +162,12 @@ class TestOVNBGPStretchedL2Driver(test_base.TestCase):
mock_ensure_ovn_device.assert_called_once_with(
CONF.bgp_nic, CONF.bgp_vrf)
@mock.patch.object(linux_net, "ensure_vrf")
@mock.patch.object(linux_net, "ensure_ovn_device")
@mock.patch.object(linux_net, "delete_routes_from_table")
@mock.patch.object(frr, "vrf_leak")
def test_start_clear_routes(
self, mock_vrf, mock_delete_routes, mock_ensure_ovn_device):
self, mock_vrf, mock_delete_routes, mock_ensure_ovn_device, *args):
CONF.set_override("clear_vrf_routes_on_startup", False)
self.bgp_driver.start()

184
ovn_bgp_agent/tests/unit/privileged/test_linux_net.py
View File

@ -13,22 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
import imp
from pyroute2 import netlink as pyroute_netlink
from socket import AF_INET6
from unittest import mock
from oslo_concurrency import processutils
from ovn_bgp_agent import constants
from ovn_bgp_agent.privileged import linux_net as priv_linux_net
from ovn_bgp_agent.tests import base as test_base
from ovn_bgp_agent.utils import linux_net
# Mock the privsep decorator and reload the module
mock.patch('ovn_bgp_agent.privileged.default.entrypoint', lambda x: x).start()
imp.reload(priv_linux_net)
class FakeException(Exception):
stderr = ''
@ -54,104 +47,6 @@ class TestPrivilegedLinuxNet(test_base.TestCase):
self.dev = 'ethfake'
self.mac = 'aa:bb:cc:dd:ee:ff'
def test_set_device_status(self):
state_dict = {'state': constants.LINK_DOWN}
dev = mock.MagicMock()
dev.__enter__.return_value = state_dict
self.mock_ndb().interfaces = {'fake-dev': dev}
priv_linux_net.set_device_status('fake-dev', constants.LINK_UP)
# Assert the method updates the state to "up"
self.assertEqual(constants.LINK_UP, state_dict['state'])
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vrf(self, mock_dev_status):
priv_linux_net.ensure_vrf('fake-vrf', 10)
mock_dev_status.assert_called_once_with(
'fake-vrf', constants.LINK_UP, ndb=self.fake_ndb)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vrf_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = KeyError('Typhoons')
priv_linux_net.ensure_vrf('fake-vrf', 10)
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='vrf', ifname='fake-vrf', vrf_table=10)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_bridge(self, mock_dev_status):
priv_linux_net.ensure_bridge('fake-bridge')
mock_dev_status.assert_called_once_with(
'fake-bridge', constants.LINK_UP, ndb=self.fake_ndb)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_bridge_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = KeyError('Oblivion')
priv_linux_net.ensure_bridge('fake-bridge')
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='bridge', ifname='fake-bridge', br_stp_state=0)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vxlan(self, mock_dev_status):
priv_linux_net.ensure_vxlan('fake-vxlan', 11, self.ip, 7)
mock_dev_status.assert_called_once_with(
'fake-vxlan', constants.LINK_UP, ndb=self.fake_ndb)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vxlan_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = KeyError('Who Needs Friends')
priv_linux_net.ensure_vxlan('fake-vxlan', 11, self.ip, 7)
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='vxlan', ifname='fake-vxlan', vxlan_id=11, vxlan_port=7,
vxlan_local=self.ip, vxlan_learning=False)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_veth(self, mock_dev_status):
priv_linux_net.ensure_veth('fake-veth', 'fake-veth-peer')
calls = [mock.call('fake-veth', constants.LINK_UP),
mock.call('fake-veth-peer', constants.LINK_UP)]
mock_dev_status.assert_has_calls(calls)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_veth_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = (KeyError('Million and One'), None)
priv_linux_net.ensure_veth('fake-veth', 'fake-veth-peer')
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='veth', ifname='fake-veth', peer='fake-veth-peer')
calls = [mock.call('fake-veth', constants.LINK_UP),
mock.call('fake-veth-peer', constants.LINK_UP)]
mock_dev_status.assert_has_calls(calls)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_dummy_device(self, mock_dev_status):
priv_linux_net.ensure_dummy_device('fake-dev')
mock_dev_status.assert_called_once_with(
'fake-dev', constants.LINK_UP, ndb=self.fake_ndb)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_dummy_device_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = KeyError('All We Have Is Now')
priv_linux_net.ensure_dummy_device('fake-dev')
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='dummy', ifname='fake-dev')
def test_delete_device(self):
dev = mock.Mock()
iface_dict = {'fake-dev': dev}
self.fake_ndb.interfaces = iface_dict
priv_linux_net.delete_device('fake-dev')
dev.remove.assert_called_once_with()
def test_delete_device_keyerror(self):
dev = mock.Mock()
iface_dict = {'fake-dev': dev}
self.fake_ndb.interfaces = iface_dict
priv_linux_net.delete_device('fake-dev-2')
dev.remove.assert_not_called()
def test_route_create(self):
fake_route = {'dst': 'default',
'oif': 1,
@ -173,36 +68,6 @@ class TestPrivilegedLinuxNet(test_base.TestCase):
priv_linux_net.route_delete(fake_route)
fake_route.__enter__().remove.assert_not_called()
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vlan_device_for_network(self, mock_dev_status):
priv_linux_net.ensure_vlan_device_for_network('fake-br', 10)
vlan_name = 'fake-br.10'
mock_dev_status.assert_called_once_with(
vlan_name, constants.LINK_UP, ndb=self.fake_ndb)
@mock.patch.object(priv_linux_net, 'set_device_status')
def test_ensure_vlan_device_for_network_keyerror(self, mock_dev_status):
mock_dev_status.side_effect = KeyError('Boilermaker')
priv_linux_net.ensure_vlan_device_for_network('fake-br', 10)
vlan_name = 'fake-br.10'
self.fake_ndb.interfaces.create.assert_called_once_with(
kind='vlan', ifname=vlan_name, vlan_id=10, link=mock.ANY)
def test_delete_exposed_ips(self):
ip0 = mock.Mock(address='10.10.1.16')
ip1 = mock.Mock(address='2002::1234:abcd:ffff:c0a8:101')
iface = mock.Mock()
iface.ipaddr = {'10.10.1.16/32': ip0,
'2002::1234:abcd:ffff:c0a8:101/128': ip1}
self.fake_ndb.interfaces = {self.dev: iface}
ips = ['10.10.1.16', '2002::1234:abcd:ffff:c0a8:101', '10.10.1.17']
priv_linux_net.delete_exposed_ips(ips, self.dev)
ip0.remove.assert_called_once_with()
ip1.remove.assert_called_once_with()
def test_rule_create(self):
fake_rule = mock.MagicMock()
self.fake_ndb.rules.__getitem__.side_effect = KeyError
@ -228,32 +93,6 @@ class TestPrivilegedLinuxNet(test_base.TestCase):
priv_linux_net.rule_delete(fake_rule)
fake_rule.__enter__().remove.assert_not_called()
def test_delete_ip_rules(self):
rule0 = mock.MagicMock()
rule1 = mock.MagicMock()
self.fake_ndb.rules.__getitem__.side_effect = (rule0, rule1)
ip_rules = {'10/128': {'table': 7, 'family': 'fake'},
'6/128': {'table': 10, 'family': 'fake'}}
priv_linux_net.delete_ip_rules(ip_rules)
# Assert remove() was called on rules
rule0.__enter__().remove.assert_called_once_with()
rule1.__enter__().remove.assert_called_once_with()
def test_delete_ip_rules_exceptions(self):
rule0 = mock.MagicMock()
self.fake_ndb.rules.__getitem__.side_effect = (
KeyError('Limbo'),
pyroute_netlink.exceptions.NetlinkError(123))
ip_rules = {'10/128': {'table': 7, 'family': 'fake'},
'6/128': {'table': 10, 'family': 'fake'}}
priv_linux_net.delete_ip_rules(ip_rules)
# Assert remove() was not called due to the exceptions
self.assertFalse(rule0.__enter__().remove.called)
def test_set_kernel_flag(self):
priv_linux_net.set_kernel_flag('net.ipv6.conf.fake', 1)
self.mock_exc.assert_called_once_with(
@ -306,29 +145,6 @@ class TestPrivilegedLinuxNet(test_base.TestCase):
self.mock_exc.side_effect = exp
self.assertIsNone(priv_linux_net.del_ndp_proxy(self.ipv6, self.dev))
def test_add_ips_to_dev(self):
iface = mock.MagicMock(index=7)
self.fake_ndb.interfaces = {self.dev: iface}
priv_linux_net.add_ip_to_dev(self.ip, self.dev)
priv_linux_net.add_ip_to_dev(self.ipv6, self.dev)
# Assert add_ip() was called for each ip
calls = [mock.call('%s/32' % self.ip),
mock.call('%s/128' % self.ipv6)]
iface.__enter__().add_ip.assert_has_calls(calls)
def test_del_ips_from_dev(self):
iface = mock.MagicMock()
self.fake_ndb.interfaces = {self.dev: iface}
priv_linux_net.del_ip_from_dev(self.ip, self.dev)
priv_linux_net.del_ip_from_dev(self.ipv6, self.dev)
calls = [mock.call('%s/32' % self.ip),
mock.call('%s/128' % self.ipv6)]
iface.__enter__().del_ip.assert_has_calls(calls)
def test_add_ip_nei(self):
priv_linux_net.add_ip_nei(self.ip, self.mac, self.dev)

8
ovn_bgp_agent/tests/unit/utils/test_linux_net.py
View File

@ -85,14 +85,6 @@ class TestLinuxNet(test_base.TestCase):
linux_net.ensure_veth('fake-veth', 'fake-veth-peer')
mock_ensure_veth.assert_called_once_with('fake-veth', 'fake-veth-peer')
def test_set_master_for_device(self):
dev = mock.MagicMock()
self.fake_ndb.interfaces = {
'fake-dev': dev, 'fake-master': {'index': 5}}
linux_net.set_master_for_device('fake-dev', 'fake-master')
dev.__enter__().set.assert_called_once_with('master', 5)
def test_set_master_for_device_already_set(self):
dev = mock.MagicMock()
dev.get.return_value = 5

29
ovn_bgp_agent/tests/utils.py
View File

@ -13,6 +13,35 @@
# License for the specific language governing permissions and limitations
# under the License.
import eventlet
class WaitTimeout(Exception):
"""Default exception coming from wait_until_true() function."""
def create_row(**kwargs):
return type('FakeRow', (object,), kwargs)
def wait_until_true(predicate, timeout=60, sleep=1, exception=None):
"""Wait until callable predicate is evaluated as True
Imported from ``neutron.common.utils``.
:param predicate: Callable deciding whether waiting should continue.
Best practice is to instantiate predicate with functools.partial()
:param timeout: Timeout in seconds how long should function wait.
:param sleep: Polling interval for results in seconds.
:param exception: Exception instance to raise on timeout. If None is passed
(default) then WaitTimeout exception is raised.
"""
try:
with eventlet.Timeout(timeout):
while not predicate():
eventlet.sleep(sleep)
except eventlet.Timeout:
if exception is not None:
# pylint: disable=raising-bad-type
raise exception
raise WaitTimeout('Timed out after %d seconds' % timeout)

6
ovn_bgp_agent/utils/linux_net.py
View File

@ -99,6 +99,8 @@ def ensure_arp_ndp_enabled_for_bridge(bridge, offset, vlan_tag=None):
ipv6 = constants.NDP_IPV6_PREFIX + "%x" % offset
try:
ovn_bgp_agent.privileged.linux_net.add_ip_to_dev(ipv4, bridge)
except agent_exc.IpAddressAlreadyExists:
LOG.debug("IP %s already added on bridge %s", ipv4, bridge)
except KeyError as e:
if "object exists" not in str(e):
LOG.error("Unable to add IP on bridge %s to enable arp/ndp. "
@ -106,6 +108,8 @@ def ensure_arp_ndp_enabled_for_bridge(bridge, offset, vlan_tag=None):
raise
try:
ovn_bgp_agent.privileged.linux_net.add_ip_to_dev(ipv6, bridge)
except agent_exc.IpAddressAlreadyExists:
LOG.debug("IP %s already added on bridge %s", ipv6, bridge)
except KeyError as e:
if "object exists" not in str(e):
LOG.error("Unable to add IP on bridge %s to enable arp/ndp. "
@ -528,7 +532,7 @@ def add_ips_to_dev(nic, ips, clear_local_route_at_table=False):
for ip in ips:
try:
ovn_bgp_agent.privileged.linux_net.add_ip_to_dev(ip, nic)
except KeyError:
except agent_exc.IpAddressAlreadyExists:
# NDB raises KeyError: 'object exists'
# if the ip is already added
already_added_ips.append(ip)

1
test-requirements.txt
View File

@ -5,6 +5,7 @@
hacking>=3.0,<3.1 # Apache-2.0
coverage>=4.0,!=4.4 # Apache-2.0
eventlet>=0.26.1 # MIT
python-subunit>=0.0.18 # Apache-2.0/BSD
oslotest>=1.10.0 # Apache-2.0
pyroute2>=0.6.4;sys_platform!='win32' # Apache-2.0 (+ dual licensed GPL2)

14
tox.ini
View File

@ -9,13 +9,14 @@ ignore_basepython_conflict = true
basepython = python3
usedevelop = True
setenv =
VIRTUAL_ENV={envdir}
PYTHONWARNINGS=default::DeprecationWarning
OS_STDOUT_CAPTURE=1
OS_STDERR_CAPTURE=1
OS_TEST_TIMEOUT=60
deps = -c{env:UPPER_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/master}
-r{toxinidir}/test-requirements.txt
commands = stestr run {posargs}
commands = stestr run --exclude-regex ".tests.functional" {posargs}
[testenv:lower-constraints]
deps = -c{toxinidir}/lower-constraints.txt
@ -27,16 +28,23 @@ commands = flake8 {posargs}
[testenv:venv]
commands = {posargs}
[testenv:functional]
envdir = {toxworkdir}/functional
setenv =
{[testenv]setenv}
commands =
stestr run --exclude-regex ".tests.unit" {posargs}
[testenv:cover]
setenv =
VIRTUAL_ENV={envdir}
PYTHON=coverage run --source ovn_bgp_agent --parallel-mode
commands =
stestr run {posargs}
stestr run --exclude-regex ".tests.functional" {posargs}
coverage combine
coverage html -d cover --omit='*tests*'
coverage xml -o cover/coverage.xml --omit='*tests*'
coverage report --fail-under=92 --skip-covered --omit='*tests*'
coverage report --fail-under=85 --skip-covered --omit='*tests*'
[testenv:docs]
deps = -r{toxinidir}/doc/requirements.txt

3
zuul.d/project.yaml
View File

@ -6,3 +6,6 @@
- publish-openstack-docs-pti
vars:
rtd_webhook_id: '224878'
check:
jobs:
- openstack-tox-functional-with-sudo