This python fixture was way too complex for what is needed.
Eliminate gratuitous options/over-engineering:
- Change from non-deterministic `wait` and `attempts` to a single
`retry_timeout` value. This is both more deterministic, as well as
what the user should actually be thinking about.
- Use a fixed 2 second pause between executing the wrapped function
rather than a bunch of arbitrary choices of 2, 3 and 4 seconds
spread all over the test code.
- Get rid of the multiple variables for determining what "Positive" and
"Negative" results are. Instead just implement what all the user code
already wants, i.e., boolean False or a str (errormsg) means
"Negative" result otherwise it's a "Positive" result.
- As part of the above the inversion logic is much more comprehensible
in the fixture code (and more correct to boot).
Signed-off-by: Christian Hopps <chopps@labn.net>
#############################################
# Verification APIs
#############################################
-@retry(attempts=4, wait=2, return_is_str=True)
+@retry(retry_timeout=8)
def verify_router_id(tgen, topo, input_dict, expected=True):
"""
Running command "show ip bgp json" for DUT and reading router-id
return True
-@retry(attempts=50, wait=3, return_is_str=True)
+@retry(retry_timeout=150)
def verify_bgp_convergence(tgen, topo, dut=None, expected=True):
"""
API will verify if BGP is converged with in the given time frame.
return True
-@retry(attempts=4, wait=4, return_is_str=True)
+@retry(retry_timeout=16)
def verify_bgp_community(
tgen, addr_type, router, network, input_dict=None, vrf=None, bestpath=False, expected=True
):
return True
-@retry(attempts=4, wait=2, return_is_str=True)
+@retry(retry_timeout=8)
def verify_as_numbers(tgen, topo, input_dict, expected=True):
"""
This API is to verify AS numbers for given DUT by running
return True
-@retry(attempts=50, wait=3, return_is_str=True)
+@retry(retry_timeout=150)
def verify_bgp_convergence_from_running_config(tgen, dut=None, expected=True):
"""
API to verify BGP convergence b/w loopback and physical interface.
return True
-@retry(attempts=4, wait=4, return_is_str=True)
+@retry(retry_timeout=16)
def verify_bgp_attributes(
tgen,
addr_type,
return True
-@retry(attempts=4, wait=2, return_is_str=True)
+@retry(retry_timeout=8)
def verify_best_path_as_per_bgp_attribute(
tgen, addr_type, router, input_dict, attribute, expected=True
):
return True
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_best_path_as_per_admin_distance(
tgen, addr_type, router, input_dict, attribute, expected=True
):
return True
-@retry(attempts=5, wait=2, return_is_str=True, initial_wait=2)
+@retry(retry_timeout=10, initial_wait=2)
def verify_bgp_rib(
tgen, addr_type, dut, input_dict, next_hop=None, aspath=None, multi_nh=None, expected=True
):
return True
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_graceful_restart(tgen, topo, addr_type, input_dict, dut, peer, expected=True):
"""
This API is to verify verify_graceful_restart configuration of DUT and
return True
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_r_bit(tgen, topo, addr_type, input_dict, dut, peer, expected=True):
"""
This API is to verify r_bit in the BGP gr capability advertised
return True
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_eor(tgen, topo, addr_type, input_dict, dut, peer, expected=True):
"""
This API is to verify EOR
return True
-@retry(attempts=4, wait=2, return_is_str=True)
+@retry(retry_timeout=8)
def verify_f_bit(tgen, topo, addr_type, input_dict, dut, peer, expected=True):
"""
This API is to verify f_bit in the BGP gr capability advertised
return True
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_graceful_restart_timers(tgen, topo, addr_type, input_dict, dut, peer):
"""
This API is to verify graceful restart timers, configured and recieved
return True
-@retry(attempts=4, wait=2, return_is_str=True)
+@retry(retry_timeout=8)
def verify_gr_address_family(tgen, topo, addr_type, addr_family, dut, expected=True):
"""
This API is to verify gr_address_family in the BGP gr capability advertised
logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_attributes_for_evpn_routes(
tgen,
topo,
return False
-@retry(attempts=5, wait=2, return_is_str=True)
+@retry(retry_timeout=10)
def verify_evpn_routes(
tgen, topo, dut, input_dict, routeType=5, EthTag=0, next_hop=None, expected=True
):
#
from collections import OrderedDict
-from datetime import datetime
+from datetime import datetime, timedelta
from time import sleep
from copy import deepcopy
from subprocess import call
import ipaddress
import platform
-if sys.version_info[0] > 2:
- import io
- import configparser
-else:
- import StringIO
+try:
+ # Imports from python2
+ from StringIO import StringIO
import ConfigParser as configparser
+except ImportError:
+ # Imports from python3
+ from io import StringIO
+ import configparser
from lib.topolog import logger, logger_config
from lib.topogen import TopoRouter, get_topogen
],
}
+def is_string(value):
+ try:
+ return isinstance(value, basestring)
+ except NameError:
+ return isinstance(value, str)
+
if config.has_option("topogen", "verbosity"):
loglevel = config.get("topogen", "verbosity")
loglevel = loglevel.upper()
return True
-def getStrIO():
- """
- Return a StringIO object appropriate for the current python version.
- """
- if sys.version_info[0] > 2:
- return io.StringIO()
- else:
- return StringIO.StringIO()
-
-
def reset_config_on_routers(tgen, routerName=None):
"""
Resets configuration on routers to the snapshot created using input JSON
raise InvalidCLIError("Unknown error in %s", output)
f = open(dname, "r")
- delta = getStrIO()
+ delta = StringIO()
delta.write("configure terminal\n")
t_delta = f.read()
output = router.vtysh_multicmd(delta.getvalue(), pretty_output=False)
delta.close()
- delta = getStrIO()
+ delta = StringIO()
cfg = router.run("vtysh -c 'show running'")
for line in cfg.split("\n"):
line = line.strip()
return True
-def retry(attempts=3, wait=2, return_is_str=True, initial_wait=0, return_is_dict=False):
+def retry(retry_timeout, initial_wait=0, expected=True, diag_pct=0.75):
"""
- Retries function execution, if return is an errormsg or exception
-
- * `attempts`: Number of attempts to make
- * `wait`: Number of seconds to wait between each attempt
- * `return_is_str`: Return val is an errormsg in case of failure
- * `initial_wait`: Sleeps for this much seconds before executing function
-
+ Fixture: Retries function while it's return value is an errormsg (str), False, or it raises an exception.
+
+ * `retry_timeout`: Retry for at least this many seconds; after waiting initial_wait seconds
+ * `initial_wait`: Sleeps for this many seconds before first executing function
+ * `expected`: if False then the return logic is inverted, except for exceptions,
+ (i.e., a False or errmsg (str) function return ends the retry loop,
+ and returns that False or str value)
+ * `diag_pct`: Percentage of `retry_timeout` to keep testing after negative result would have
+ been returned in order to see if a positive result comes after. This is an
+ important diagnostic tool, and normally should not be disabled. Calls to wrapped
+ functions though, can override the `diag_pct` value to make it larger in case more
+ diagnostic retrying is appropriate.
"""
def _retry(func):
@wraps(func)
def func_retry(*args, **kwargs):
- _wait = kwargs.pop("wait", wait)
- _attempts = kwargs.pop("attempts", attempts)
- _attempts = int(_attempts)
- if _attempts < 0:
- raise ValueError("attempts must be 0 or greater")
+ # We will continue to retry diag_pct of the timeout value to see if test would have passed with a
+ # longer retry timeout value.
+ saved_failure = None
+
+ retry_sleep = 2
+
+ # Allow the wrapped function's args to override the fixtures
+ _retry_timeout = kwargs.pop("retry_timeout", retry_timeout)
+ _expected = kwargs.pop("expected", expected)
+ _initial_wait = kwargs.pop("initial_wait", initial_wait)
+ _diag_pct = kwargs.pop("diag_pct", diag_pct)
+
+ start_time = datetime.now()
+ retry_until = datetime.now() + timedelta(seconds=_retry_timeout + _initial_wait)
if initial_wait > 0:
logger.info("Waiting for [%s]s as initial delay", initial_wait)
sleep(initial_wait)
- _return_is_str = kwargs.pop("return_is_str", return_is_str)
- _return_is_dict = kwargs.pop("return_is_str", return_is_dict)
- _expected = kwargs.setdefault("expected", True)
- kwargs.pop("expected")
- for i in range(1, _attempts + 1):
+ invert_logic = not _expected
+ while True:
+ seconds_left = (retry_until - datetime.now()).total_seconds()
try:
ret = func(*args, **kwargs)
logger.debug("Function returned %s", ret)
- if _return_is_str and isinstance(ret, bool) and _expected:
- return ret
- if (
- isinstance(ret, str) or isinstance(ret, unicode)
- ) and _expected is False:
- return ret
- if _return_is_dict and isinstance(ret, dict):
- return ret
-
- if _attempts == i:
- generate_support_bundle()
- return ret
- except Exception as err:
- if _attempts == i:
- generate_support_bundle()
- logger.info("Max number of attempts (%r) reached", _attempts)
- raise
- else:
- logger.info("Function returned %s", err)
- if i < _attempts:
- logger.info("Retry [#%r] after sleeping for %ss" % (i, _wait))
- sleep(_wait)
+
+ negative_result = ret is False or is_string(ret)
+ if negative_result == invert_logic:
+ # Simple case, successful result in time
+ if not saved_failure:
+ return ret
+
+ # Positive result, but happened after timeout failure, very important to
+ # note for fixing tests.
+ logger.warning("RETRY DIAGNOSTIC: SUCCEED after FAILED with requested timeout of %.1fs; however, succeeded in %.1fs, investigate timeout timing",
+ _retry_timeout, (datetime.now() - start_time).total_seconds())
+ if isinstance(saved_failure, Exception):
+ raise saved_failure # pylint: disable=E0702
+ return saved_failure
+
+ except Exception as error:
+ logger.info("Function raised exception: %s", str(error))
+ ret = error
+
+ if seconds_left < 0 and saved_failure:
+ logger.info("RETRY DIAGNOSTIC: Retry timeout reached, still failing")
+ if isinstance(saved_failure, Exception):
+ raise saved_failure # pylint: disable=E0702
+ return saved_failure
+
+ if seconds_left < 0:
+ logger.info("Retry timeout of %ds reached", _retry_timeout)
+
+ saved_failure = ret
+ retry_extra_delta = timedelta(seconds=seconds_left + _retry_timeout * _diag_pct)
+ retry_until = datetime.now() + retry_extra_delta
+ seconds_left = retry_extra_delta.total_seconds()
+
+ # Generate bundle after setting remaining diagnostic retry time
+ generate_support_bundle()
+
+ # If user has disabled diagnostic retries return now
+ if not _diag_pct:
+ if isinstance(saved_failure, Exception):
+ raise saved_failure
+ return saved_failure
+
+ if saved_failure:
+ logger.info("RETRY DIAG: [failure] Sleeping %ds until next retry with %.1f retry time left - too see if timeout was too short",
+ retry_sleep, seconds_left)
+ else:
+ logger.info("Sleeping %ds until next retry with %.1f retry time left",
+ retry_sleep, seconds_left)
+ sleep(retry_sleep)
func_retry._original = func
return func_retry
#############################################
# Verification APIs
#############################################
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_rib(
tgen,
addr_type,
return True
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_fib_routes(tgen, addr_type, dut, input_dict, next_hop=None):
"""
Data will be read from input_dict or input JSON file, API will generate
return True
-@retry(attempts=3, wait=4, return_is_str=True)
+@retry(retry_timeout=12)
def verify_route_maps(tgen, input_dict):
"""
Running "show route-map" command and verifying given route-map
return True
-@retry(attempts=4, wait=4, return_is_str=True)
+@retry(retry_timeout=16)
def verify_bgp_community(tgen, addr_type, router, network, input_dict=None):
"""
API to veiryf BGP large community is attached in route for any given
return True
-@retry(attempts=3, wait=4, return_is_str=True)
+@retry(retry_timeout=12)
def verify_evpn_vni(tgen, input_dict):
"""
API to verify evpn vni details using "show evpn vni detail json"
return False
-@retry(attempts=3, wait=4, return_is_str=True)
+@retry(retry_timeout=12)
def verify_vrf_vni(tgen, input_dict):
"""
API to verify vrf vni details using "show vrf vni json"
################################
# Verification procs
################################
-@retry(attempts=40, wait=2, return_is_str=True)
+@retry(retry_timeout=80)
def verify_ospf_neighbor(tgen, topo, dut=None, input_dict=None, lan=False, expected=True):
"""
This API is to verify ospf neighborship by running
################################
# Verification procs
################################
-@retry(attempts=10, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def verify_ospf6_neighbor(tgen, topo, dut=None, input_dict=None, lan=False):
"""
This API is to verify ospf neighborship by running
return result
-@retry(attempts=21, wait=2, return_is_str=True)
+@retry(retry_timeout=40)
def verify_ospf_rib(
tgen, dut, input_dict, next_hop=None, tag=None, metric=None, fib=None, expected=True
):
return result
-@retry(attempts=10, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def verify_ospf_interface(tgen, topo, dut=None, lan=False, input_dict=None, expected=True):
"""
This API is to verify ospf routes by running
return result
-@retry(attempts=11, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def verify_ospf_database(tgen, topo, dut, input_dict, expected=True):
"""
This API is to verify ospf lsa's by running
return result
-@retry(attempts=10, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def verify_ospf_summary(tgen, topo, dut, input_dict, expected=True):
"""
This API is to verify ospf routes by running
-@retry(attempts=10, wait=3, return_is_str=True)
+@retry(retry_timeout=30)
def verify_ospf6_rib(tgen, dut, input_dict, next_hop=None,
tag=None, metric=None, fib=None):
"""
return result
-@retry(attempts=3, wait=2, return_is_str=True)
+@retry(retry_timeout=6)
def verify_ospf6_interface(tgen, topo, dut=None,lan=False, input_dict=None):
"""
This API is to verify ospf routes by running
return result
-@retry(attempts=11, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def verify_ospf6_database(tgen, topo, dut, input_dict):
"""
This API is to verify ospf lsa's by running
config_data = []
for lnk in input_dict[router]['links'].keys():
if "ospf6" not in input_dict[router]['links'][lnk]:
- logger.debug("Router %s: ospf6 configs is not present in"
- "input_dict, passed input_dict", router,
- input_dict)
+ logger.debug("Router %s: ospf6 config is not present in"
+ "input_dict, passed input_dict %s", router,
+ str(input_dict))
continue
ospf_data = input_dict[router]['links'][lnk]['ospf6']
data_ospf_area = ospf_data.setdefault("area", None)
#############################################
# Verification APIs
#############################################
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_pim_neighbors(tgen, topo, dut=None, iface=None, nbr_ip=None, expected=True):
"""
Verify all PIM neighbors are up and running, config is verified
return True
-@retry(attempts=21, wait=2, return_is_str=True)
+@retry(retry_timeout=40)
def verify_igmp_groups(tgen, dut, interface, group_addresses, expected=True):
"""
Verify IGMP groups are received from an intended interface
return True
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_upstream_iif(
tgen, dut, iif, src_address, group_addresses, joinState=None, refCount=1, expected=True
):
return True
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_join_state_and_timer(tgen, dut, iif, src_address, group_addresses, expected=True):
"""
Verify join state is updated correctly and join timer is
return True
-@retry(attempts=41, wait=2, return_is_dict=True)
+@retry(retry_timeout=80)
def verify_ip_mroutes(
tgen, dut, src_address, group_addresses, iif, oil, return_uptime=False, mwait=0, expected=True
):
return True if return_uptime == False else uptime_dict
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_pim_rp_info(
tgen, topo, dut, group_addresses, oif=None, rp=None, source=None, iamrp=None, expected=True
):
return True
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_pim_state(
tgen, dut, iif, oil, group_addresses, src_address=None, installed_fl=None, expected=True
):
return output_dict
-@retry(attempts=21, wait=2, return_is_str=True)
+@retry(retry_timeout=40)
def verify_pim_interface(tgen, topo, dut, interface=None, interface_ip=None, expected=True):
"""
Verify all PIM interface are up and running, config is verified
return True
-@retry(attempts=10, wait=2, return_is_str=True)
+@retry(retry_timeout=20)
def clear_ip_mroute_verify(tgen, dut, expected=True):
"""
Clear ip mroute by running "clear ip mroute" cli and verify
return rp_details
-@retry(attempts=6, wait=2, return_is_str=True)
+@retry(retry_timeout=12)
def verify_pim_grp_rp_source(tgen, topo, dut, grp_addr, rp_source, rpadd=None, expected=True):
"""
Verify pim rp info by running "show ip pim rp-info" cli
return errormsg
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_pim_bsr(tgen, topo, dut, bsr_ip, expected=True):
"""
Verify all PIM interface are up and running, config is verified
return True
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_ip_pim_upstream_rpf(tgen, topo, dut, interface, group_addresses, rp=None, expected=True):
"""
Verify IP PIM upstream rpf, config is verified
return result
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_ip_pim_join(tgen, topo, dut, interface, group_addresses, src_address=None, expected=True):
"""
Verify ip pim join by running "show ip pim join" cli
return True
-@retry(attempts=31, wait=2, return_is_dict=True)
+@retry(retry_timeout=60)
def verify_igmp_config(tgen, input_dict, stats_return=False, expected=True):
"""
Verify igmp interface details, verifying following configs:
return True if stats_return == False else igmp_stats
-@retry(attempts=31, wait=2, return_is_str=True)
+@retry(retry_timeout=60)
def verify_pim_config(tgen, input_dict, expected=True):
"""
Verify pim interface details, verifying following configs:
return True
-@retry(attempts=21, wait=2, return_is_dict=True)
+@retry(retry_timeout=40)
def verify_multicast_traffic(tgen, input_dict, return_traffic=False, expected=True):
"""
Verify multicast traffic by running
return refCount
-@retry(attempts=21, wait=2, return_is_str=True)
+@retry(retry_timeout=40)
def verify_multicast_flag_state(tgen, dut, src_address, group_addresses, flag, expected=True):
"""
Verify flag state for mroutes and make sure (*, G)/(S, G) are having
return True
-@retry(attempts=21, wait=2, return_is_str=True)
+@retry(retry_timeout=40)
def verify_igmp_interface(tgen, topo, dut, igmp_iface, interface_ip, expected=True):
"""
Verify all IGMP interface are up and running, config is verified
# Verify mroute not installed
step("Verify mroute not installed in l1")
result = verify_ip_mroutes(
- tgen, dut, src_addr, GROUP_ADDRESS, iif, oil, wait=20, expected=False
+ tgen, dut, src_addr, GROUP_ADDRESS, iif, oil, retry_timeout=20, expected=False
)
assert (
result is not True
), "Testcase {} :Failed \n Error : rp expected {} rp received {}".format(
tc_name,
rp_add1,
+ rp2[group] if group in rp2 else None
)
# Verify if that rp is installed
sleep(6)
dut = "r2"
ospf_covergence = verify_ospf_neighbor(
- tgen, topo, dut=dut, expected=False, attempts=5
+ tgen, topo, dut=dut, expected=False, retry_timeout=10
)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
sleep(6)
dut = "r1"
ospf_covergence = verify_ospf_neighbor(
- tgen, topo, dut=dut, expected=False, attempts=3
+ tgen, topo, dut=dut, expected=False, retry_timeout=6
)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
sleep(6)
dut = "r2"
ospf_covergence = verify_ospf_neighbor(
- tgen, topo, dut=dut, expected=False, attempts=5
+ tgen, topo, dut=dut, expected=False, retry_timeout=10
)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
step("Verify that the neighbour is not FULL between R1 and R2.")
dut = "r1"
ospf_covergence = verify_ospf_neighbor(
- tgen, topo, dut=dut, expected=False, attempts=5
+ tgen, topo, dut=dut, expected=False, retry_timeout=10
)
assert ospf_covergence is not True, "setup_module :Failed \n Error:" " {}".format(
ospf_covergence
step("Verify that route is withdrawn from R2.")
dut = "r1"
result = verify_ospf_rib(
- tgen, dut, input_dict, next_hop=nh, attempts=5, expected=False
+ tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
)
assert (
result is not True
input_dict,
protocol=protocol,
next_hop=nh,
- attempts=5,
+ retry_timeout=10,
expected=False,
)
assert (
step("Verify that route is withdrawn from R2.")
dut = "r1"
result = verify_ospf_rib(
- tgen, dut, input_dict, next_hop=nh, attempts=5, expected=False
+ tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
)
assert (
result is not True
input_dict,
protocol=protocol,
next_hop=nh,
- attempts=5,
+ retry_timeout=10,
expected=False,
)
assert (
step("Verify that all the routes are withdrawn from R0")
dut = "r1"
result = verify_ospf_rib(
- tgen, dut, input_dict, next_hop=nh, attempts=5, expected=False
+ tgen, dut, input_dict, next_hop=nh, retry_timeout=10, expected=False
)
assert (
result is not True
input_dict,
protocol=protocol,
next_hop=nh,
- attempts=5,
+ retry_timeout=10,
expected=False,
)
assert (
dut = "r1"
protocol = "ospf"
- result = verify_ospf_rib(tgen, dut, input_dict, attempts=2, expected=False)
+ result = verify_ospf_rib(tgen, dut, input_dict, retry_timeout=4, expected=False)
assert (
result is not True
), "Testcase {} : Failed \n " "r1: OSPF routes are present \n Error: {}".format(
)
result = verify_rib(
- tgen, "ipv4", dut, input_dict, protocol=protocol, attempts=2, expected=False
+ tgen, "ipv4", dut, input_dict, protocol=protocol, retry_timeout=4, expected=False
)
assert (
result is not True
input_dict,
protocol=protocol,
next_hop=nh,
- attempts=5,
+ retry_timeout=10,
expected=False,
)
assert result is not True, (
next_hop=nh,
protocol=protocol,
fib=True,
+ retry_timeout=6,
expected=False,
- wait=2,
- attempts=3,
)
assert (
result is not True
protocol=protocol,
fib=True,
expected=False,
- wait=2,
- attempts=3,
+ retry_timeout=6,
)
assert (
result is not True
protocol=protocol,
fib=True,
expected=False,
- wait=2,
- attempts=3,
+ retry_timeout=6,
)
assert (
result is not True
protocol=protocol,
fib=True,
expected=False,
- wait=2,
- attempts=3,
+ retry_timeout=6,
)
assert (
result is not True
- ), "Testcase {} : Failed \nError: Routes " " are missing in RIB".format(tc_name)
+ ), "Testcase {} : Failed \nError: Routes " " are present in RIB".format(tc_name)
step(
"Remove the static route configured with nexthop N1 to N8, one"
protocol=protocol,
fib=True,
expected=False,
- wait=2,
- attempts=3,
+ retry_timeout=6,
)
assert (
result is not True
protocol=protocol,
fib=True,
expected=False,
- wait=2,
- attempts=3,
+ retry_timeout=6,
)
assert (
result is not True
projects / mirror / frr.git / commitdiff