From b8df157d77f7512e1c5181e878d0d05c5120b3e3 Mon Sep 17 00:00:00 2001
From: Eduardo Olivares <eolivare@redhat.com>
Date: Fri, 19 Jul 2024 12:20:41 +0200
Subject: [PATCH] Create decorator for retry on assert_fail

This patch adds a decorator for those functions that we want to
evaluate more than once in case of an assertion failure. Some
scenario tests are conditioned by the stability of the VM or
interfaces after their creation to obtain bandwidth measurements
according to QoS criteria. For these tests, this decorator allows
for multiple evaluations seeking stability in the measurements

Co-Authored-By: froyo@redhat.com
Change-Id: I4c562a97ff958e8c15408f3d06c54c24e9a2f1fb
---
 .../common/utils.py                           | 23 ++++++++++++++
 .../tests/scenario/test_qos.py                | 30 ++++++-------------
 2 files changed, 32 insertions(+), 21 deletions(-)

diff --git a/whitebox_neutron_tempest_plugin/common/utils.py b/whitebox_neutron_tempest_plugin/common/utils.py
index a7aa36a..462d89c 100644
--- a/whitebox_neutron_tempest_plugin/common/utils.py
+++ b/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
diff --git a/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py b/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py
index 7472286..3db7029 100644
--- a/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py
+++ b/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