x/whitebox-neutron-tempest-plugin
Code Issues Proposed changes

Create decorator for retry on assert_fail

Browse Source 
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
This commit is contained in:
Eduardo Olivares 2024-07-19 12:20:41 +02:00
parent c1c17025c6
commit b8df157d77
2 changed files with 32 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
whitebox_neutron_tempest_plugin/common/utils.py
View File

@ -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

30
whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py
View File

@ -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