From: Assaf Muller Date: Tue, 5 Aug 2014 20:12:55 +0000 (+0300) Subject: Implement ip_lib.device_exists_with_ip_mac X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=c5277abccaa9ec4580495867e55d56fba428eb85;p=openstack-build%2Fneutron-build.git Implement ip_lib.device_exists_with_ip_mac ip_lib.device_exists simply checks that a device is in an (optional) namespace. This patch adds ip_lib.device_exists_with_ip_mac, which also verifies that the device is configured with the given IP and MAC addresses. The new function is used in functional tests in a patch further down the dependency chain. Added functional tests for ip_lib.device_exists, ip_lib.device_exists_with_ip_mac. Implements: blueprint l3-high-availability Change-Id: I1495bcf1f1a88e2eb78df79cccb64121ccf435fb --- diff --git a/neutron/agent/linux/ip_lib.py b/neutron/agent/linux/ip_lib.py index afc27f788..7ad696ee9 100644 --- a/neutron/agent/linux/ip_lib.py +++ b/neutron/agent/linux/ip_lib.py @@ -549,6 +549,7 @@ class IpNetnsCommand(IpCommandBase): def device_exists(device_name, root_helper=None, namespace=None): + """Return True if the device exists in the namespace.""" try: address = IPDevice(device_name, root_helper, namespace).link.address except RuntimeError: @@ -556,6 +557,23 @@ def device_exists(device_name, root_helper=None, namespace=None): return bool(address) +def device_exists_with_ip_mac(device_name, ip_cidr, mac, namespace=None, + root_helper=None): + """Return True if the device with the given IP and MAC addresses + exists in the namespace. + """ + try: + device = IPDevice(device_name, root_helper, namespace) + if mac != device.link.address: + return False + if ip_cidr not in (ip['cidr'] for ip in device.addr.list()): + return False + except RuntimeError: + return False + else: + return True + + def ensure_device_is_ready(device_name, root_helper=None, namespace=None): dev = IPDevice(device_name, root_helper, namespace) try: diff --git a/neutron/tests/functional/agent/linux/base.py b/neutron/tests/functional/agent/linux/base.py index 29198004a..ffc1038d6 100644 --- a/neutron/tests/functional/agent/linux/base.py +++ b/neutron/tests/functional/agent/linux/base.py @@ -33,9 +33,9 @@ class BaseLinuxTestCase(functional_base.BaseSudoTestCase): self.skipTest(skip_msg) raise - def get_rand_name(self, max_length, prefix='test'): + def get_rand_name(self, max_length=None, prefix='test'): name = prefix + str(random.randint(1, 0x7fffffff)) - return name[:max_length] + return name[:max_length] if max_length is not None else name def create_resource(self, name_prefix, creation_func, *args, **kwargs): """Create a new resource that does not already exist. @@ -47,7 +47,8 @@ class BaseLinuxTestCase(functional_base.BaseSudoTestCase): :param *args *kwargs: These will be passed to the create function. """ while True: - name = self.get_rand_name(n_const.DEVICE_NAME_MAX_LEN, name_prefix) + name = self.get_rand_name(max_length=n_const.DEVICE_NAME_MAX_LEN, + prefix=name_prefix) try: return creation_func(name, *args, **kwargs) except RuntimeError: diff --git a/neutron/tests/functional/agent/linux/test_ip_lib.py b/neutron/tests/functional/agent/linux/test_ip_lib.py new file mode 100644 index 000000000..c3bd0853d --- /dev/null +++ b/neutron/tests/functional/agent/linux/test_ip_lib.py @@ -0,0 +1,134 @@ +# Copyright (c) 2014 Red Hat, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import collections + +from oslo.config import cfg + +from neutron.agent.common import config +from neutron.agent.linux import interface +from neutron.agent.linux import ip_lib +from neutron.common import utils +from neutron.openstack.common import importutils +from neutron.openstack.common import log as logging +from neutron.tests.functional.agent.linux import base + +LOG = logging.getLogger(__name__) +Device = collections.namedtuple('Device', 'name ip_cidr mac_address namespace') + + +class IpLibTestFramework(base.BaseLinuxTestCase): + def setUp(self): + super(IpLibTestFramework, self).setUp() + self.check_sudo_enabled() + self._configure() + + def _configure(self): + config.setup_logging(cfg.CONF) + config.register_root_helper(cfg.CONF) + cfg.CONF.set_override('root_helper', self.root_helper, group='AGENT') + config.register_interface_driver_opts_helper(cfg.CONF) + cfg.CONF.set_override( + 'interface_driver', + 'neutron.agent.linux.interface.OVSInterfaceDriver') + cfg.CONF.register_opts(interface.OPTS) + self.driver = importutils.import_object(cfg.CONF.interface_driver, + cfg.CONF) + + def generate_device_details(self, name=None, ip_cidr=None, + mac_address=None, namespace=None): + return Device(name or self.get_rand_name(), + ip_cidr or '240.0.0.1/24', + mac_address or + utils.get_random_mac('fa:16:3e:00:00:00'.split(':')), + namespace or self.get_rand_name()) + + def _safe_delete_device(self, device): + try: + device.link.delete() + except RuntimeError: + LOG.debug('Could not delete %s, was it already deleted?', device) + + def manage_device(self, attr): + """Create a tuntap with the specified attributes. + + The device is cleaned up at the end of the test. + + :param attr: A Device namedtuple + :return: A tuntap ip_lib.IPDevice + """ + ip = ip_lib.IPWrapper(self.root_helper, namespace=attr.namespace) + ip.netns.add(attr.namespace) + self.addCleanup(ip.netns.delete, attr.namespace) + tap_device = ip.add_tuntap(attr.name) + self.addCleanup(self._safe_delete_device, tap_device) + tap_device.link.set_address(attr.mac_address) + self.driver.init_l3(attr.name, [attr.ip_cidr], + namespace=attr.namespace) + tap_device.link.set_up() + return tap_device + + +class IpLibTestCase(IpLibTestFramework): + def test_device_exists(self): + attr = self.generate_device_details() + + self.assertFalse( + ip_lib.device_exists(attr.name, self.root_helper, + attr.namespace)) + + device = self.manage_device(attr) + + self.assertTrue( + ip_lib.device_exists(device.name, self.root_helper, + attr.namespace)) + + device.link.delete() + + self.assertFalse( + ip_lib.device_exists(attr.name, self.root_helper, + attr.namespace)) + + def test_device_exists_with_ip_mac(self): + attr = self.generate_device_details() + device = self.manage_device(attr) + self.assertTrue( + ip_lib.device_exists_with_ip_mac( + *attr, root_helper=self.root_helper)) + + wrong_ip_cidr = '10.0.0.1/8' + wrong_mac_address = 'aa:aa:aa:aa:aa:aa' + + attr = self.generate_device_details(name='wrong_name') + self.assertFalse( + ip_lib.device_exists_with_ip_mac( + *attr, root_helper=self.root_helper)) + + attr = self.generate_device_details(ip_cidr=wrong_ip_cidr) + self.assertFalse( + ip_lib.device_exists_with_ip_mac( + *attr, root_helper=self.root_helper)) + + attr = self.generate_device_details(mac_address=wrong_mac_address) + self.assertFalse( + ip_lib.device_exists_with_ip_mac( + *attr, root_helper=self.root_helper)) + + attr = self.generate_device_details(namespace='wrong_namespace') + self.assertFalse( + ip_lib.device_exists_with_ip_mac( + *attr, root_helper=self.root_helper)) + + device.link.delete()