Dismiss only specific failure in retry

The use of retry decorator is easy and tempting,
many test methods may find various issues, therefore raising exceptions,
this change forces specific failure to be targeted by retry decorator
in order to avoid dismissing other exceptions which require a fix.

This is done by an obligatory regex, which validates
the exception caught and dismissed has the intended assertion message.

There is optional exception type argument, in order to let other
unexpected exceptions to be raised and fail the test.

Change-Id: Ib9a17d8c6c8778adf546c610dd070c14aff45ecd
This commit is contained in:
Maor Blaustein 2024-07-31 16:07:34 +03:00
parent e11fe76607
commit 7cb8a091d6
3 changed files with 22 additions and 5 deletions

View File

@ -231,9 +231,17 @@ def remote_service_action(client, service, action):
time.sleep(5)
def retry_on_assert_fail(max_retries):
# NOTE(mblue): Please use specific regex to avoid dismissing various issues
def retry_on_assert_fail(max_retries,
assert_regex,
exception_type=AssertionError):
"""Decorator that retries a function up to max_retries times on asser fail
:param max_retries: The maximum number of retries before failing.
In order to avoid dismissing exceptions which lead to bugs,
obligatory regex checked in caught exception message,
also optional specific exception type can be passed.
:param max_retries: Obligatory maximum number of retries before failing.
:param assert_regex: Obligatory regex should be in exception message.
:param exception_type: Optional specific exception related to failure.
"""
def decor(f):
@functools.wraps(f)
@ -242,7 +250,10 @@ def retry_on_assert_fail(max_retries):
while retries < max_retries:
try:
return f(*args, **kwargs)
except AssertionError as e:
except exception_type as e:
if not (re.search(assert_regex, str(e)) or
re.search(assert_regex, repr(e))):
raise
LOG.debug(
f"Assertion failed: {e}. Retrying ({retries + 1}/"
f"{max_retries})..."

View File

@ -255,7 +255,10 @@ 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)
# retry only when noticed measuring issue, as reported in BZ#2274465
@utils.retry_on_assert_fail(
max_retries=2,
assert_regex='not .* than')
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(

View File

@ -509,7 +509,10 @@ class BaseSecGroupLoggingTest(
self.check_log_ssh(
should_log=True, hypervisor_ssh=vm_a['hv_ssh_client'])
@utils.retry_on_assert_fail(max_retries=3)
# retry only when network instability causes packet loss
@utils.retry_on_assert_fail(
max_retries=3,
assert_regex='0% packet loss" not found in output')
def _test_only_accepted_traffic_logged(self):
"""This test verifies that only the log entries of allowed traffic
exist when only the "allow" event is supposed to be logged