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 afc27f788ec29429ec8e698c4d4218f8170e3689..7ad696ee9fcaa4db8075f8f8eabb7db9e38b528c 100644 (file)
--- 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 29198004ae281796ff78d9f70723e5e88a15f3f7..ffc1038d6c57cab099f768a06526c5daaea5f57e 100644 (file)
--- 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 (file)
index 0000000..c3bd085
--- /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()