Merge pull request #164 from stackhpc/upstream/2023.1-2024-07-08

Synchronise 2023.1 with upstream

Author: markgoddard

neutron/tests/functional/plugins/ml2/drivers/macvtap/agent/test_macvtap_neutron_agent.py

@@ -16,6 +16,8 @@
 
 from neutron_lib import constants
 
+from neutron.agent.linux import ip_lib
+from neutron.common import utils as common_utils
 from neutron.plugins.ml2.drivers.macvtap.agent import macvtap_neutron_agent
 from neutron.tests.common import net_helpers
 from neutron.tests.functional import base as functional_base
@@ -27,11 +29,30 @@ def setUp(self):
         self.mgr = macvtap_neutron_agent.MacvtapManager({})
 
     def test_get_all_devices(self):
+        # NOTE(ralonsoh): Clean-up before testing. This test is executed with
+        # concurrency=1. That means no other test is being executed at the same
+        # time. Because the macvtap interface must be created in the root
+        # namespace (``MacvtapManager`` cannot handle namespaces), the test
+        # deletes any previous existing interface.
+        for mac in self.mgr.get_all_devices():
+            devices = ip_lib.IPWrapper().get_devices()
+            for device in (d for d in devices if d.address == mac):
+                device.link.delete()
+
         # Veth is simulating the hosts eth device. In this test it is used as
         # src_dev for the macvtap
         veth1, veth2 = self.useFixture(net_helpers.VethFixture()).ports
         macvtap = self.useFixture(net_helpers.MacvtapFixture(
             src_dev=veth1.name, mode='bridge',
             prefix=constants.MACVTAP_DEVICE_PREFIX)).ip_dev
-        self.assertEqual(set([macvtap.link.address]),
-                         self.mgr.get_all_devices())
+        try:
+            common_utils.wait_until_true(
+                lambda: {macvtap.link.address} == self.mgr.get_all_devices(),
+                timeout=5)
+        except common_utils.WaitTimeout:
+            msg = 'MacVTap address: %s, read devices: %s\n' % (
+                macvtap.link.address, self.mgr.get_all_devices())
+            for device in ip_lib.IPWrapper().get_devices():
+                msg += '  Device %s, MAC: %s' % (device.name,
+                                                 device.link.address)
+            self.fail(msg)