Merge "Create decorator for retry on assert_fail"

whitebox_neutron_tempest_plugin/common/utils.py

@@ -13,6 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import functools
 import re
 import subprocess
 import time
@@ -227,3 +228,25 @@ def remote_service_action(client, service, action):
         action=action, service=service)
     LOG.debug("Running '{}' on {}".format(cmd, client.host))
     client.exec_command(cmd)
+
+
+def retry_on_assert_fail(max_retries):
+    """Decorator that retries a function up to max_retries times on asser fail
+    :param max_retries: The maximum number of retries before failing.
+    """
+    def decor(f):
+        @functools.wraps(f)
+        def inner(*args, **kwargs):
+            retries = 0
+            while retries < max_retries:
+                try:
+                    return f(*args, **kwargs)
+                except AssertionError as e:
+                    LOG.debug(
+                        f"Assertion failed: {e}. Retrying ({retries + 1}/"
+                        f"{max_retries})..."
+                    )
+                    retries += 1
+            raise AssertionError(f"Assert failed after {max_retries} retries.")
+        return inner
+    return decor

whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py

@@ -255,6 +255,7 @@ class QosBaseTest(test_qos.QoSTestMixin, base.TrafficFlowTest):
             raise self.skipException(
                 "iperf3 is not available on VM instance")
 
+    @utils.retry_on_assert_fail(max_retries=3)
     def _validate_bw_limit(self, client, server, egress=True, ipv6=False,
                            bw_limit=None, protocol=constants.PROTO_NAME_TCP):
         server_port = self.client.list_ports(
@@ -1014,31 +1015,18 @@ class QosTestExternalNetwork(QosBaseTest):
         self._test_both_bwlimit_dscp(
             vms, max_kbps, dscp_mark, CONF.network.public_network_id)
 
-        # run iperf, ping and tcpdump from the undercloud to
-        # validate step 6 from Polarion test RHELOSP-82027
         LOG.debug('testing BW limit from a VM instances connected to the '
-                  'external network (receiver) to the undercloud node')
+                  'external network (receiver) to an Ext VM')
+        # we need to test ingress from the Ext VM PoV because the BW limit
+        # egress rule is associated to the receiver VM instance
+        self._validate_bw_limit({'ssh_client': self.ext_vm_ssh_client},
+                                vms['receiver'], egress=False,
+                                bw_limit=max_kbps * 1000,
+                                protocol=constants.PROTO_NAME_TCP)
+
         server_port = self.client.list_ports(
             device_id=vms['receiver']['id'])['ports'][0]
         server_ip = server_port['fixed_ips'][0]['ip_address']
-        # we need to test ingress from the undercloud PoV because the BW limit
-        # egress rule is associated to the receiver VM instance
-        perf_measures = self._test_ingress_bw(
-            self.ext_vm_ssh_client,
-            vms['receiver']['ssh_client'],
-            server_ip,
-            protocol=constants.PROTO_NAME_TCP)
-        LOG.debug('perf_measures = {}'.format(perf_measures))
-        measured_bw = self._calculate_bw(perf_measures)
-        LOG.debug(
-            'server_ip = {} / measured_bw = {}'.format(server_ip, measured_bw))
-        bw_limit = max_kbps * 1000
-        LOG.debug('bw_limit = {}'.format(bw_limit))
-        # a 20% of upper deviation is allowed
-        self.assertLess(measured_bw, bw_limit * 1.2)
-        # a 20% of lower deviation is allowed
-        self.assertGreater(measured_bw, bw_limit * 0.8)
-
         LOG.debug('testing DSCP mark from the test device (undercloud) to '
                   'a VM instance connected to the external network')
         # undercloud's interface towards the receiver IP