proxmox-mirrors/mirror_frr
1
0
Fork 0
mirror of https://git.proxmox.com/git/mirror_frr synced 2025-08-05 02:46:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

topotests: label per nexthop test adds add a while loop for mpls table

Browse Source 
The bgp_vpnv4_per_nexthop_label tests only check to see if the mpls labels
are installed one time. Test runs show that all but one label is installed.
More than likely the test has asked for data while zebra is still installing
it. the mpls_label_check functions must check this result multiple times as
that system may be under heavy load.

A loop is introduced in order to let zebra check the mpls table.

Signed-off-by: Philippe Guibert <philippe.guibert@6wind.com>
This commit is contained in:
Philippe Guibert 2023-06-02 10:55:07 +02:00
parent 4230f85bef
commit f2732f531f
1 changed files with 46 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
tests/topotests/bgp_vpnv4_per_nexthop_label/test_bgp_vpnv4_per_nexthop_label.py
View File

@ -232,6 +232,46 @@ def bgp_vpnv4_table_check_all(router, label_list=None, same=False):
bgp_vpnv4_table_check(router, group=group, label_list=label_list)
def check_show_mpls_table(router, blacklist=None, label_list=None, whitelist=None):
nexthop_list = []
if blacklist:
nexthop_list.append(blacklist)
dump = router.vtysh_cmd("show mpls table json", isjson=True)
for in_label, label_info in dump.items():
if label_list is not None:
label_list.add(in_label)
for nh in label_info["nexthops"]:
if nh["installed"] != True or nh["type"] != "BGP":
return "{}, show mpls table, nexthop is not installed".format(
router.name
)
if "nexthop" in nh.keys():
if nh["nexthop"] in nexthop_list:
return "{}, show mpls table, duplicated or blacklisted nexthop address".format(
router.name
)
nexthop_list.append(nh["nexthop"])
elif "interface" in nh.keys():
if nh["interface"] in nexthop_list:
return "{}, show mpls table, duplicated or blacklisted nexthop interface".format(
router.name
)
nexthop_list.append(nh["interface"])
else:
return "{}, show mpls table, entry with neither nexthop nor interface".format(
router.name
)
if whitelist:
for entry in whitelist:
if entry not in nexthop_list:
return "{}, show mpls table, entry with nexthop {} not present in nexthop list".format(
router.name, entry
)
return None
def mpls_table_check(router, blacklist=None, label_list=None, whitelist=None):
"""
Dump and check 'show mpls table json' output. An assert is triggered in case test fails
@ -240,46 +280,13 @@ def mpls_table_check(router, blacklist=None, label_list=None, whitelist=None):
* 'label_list': the list of labels that should be in inLabel value
* 'whitelist': the list of nexthops (IP or interface) that should be on output
"""
nexthop_list = []
if blacklist:
nexthop_list.append(blacklist)
logger.info("Checking MPLS labels on {}".format(router.name))
dump = router.vtysh_cmd("show mpls table json", isjson=True)
for in_label, label_info in dump.items():
if label_list is not None:
label_list.add(in_label)
for nh in label_info["nexthops"]:
assert (
nh["installed"] == True and nh["type"] == "BGP"
), "{}, show mpls table, nexthop is not installed".format(router.name)
if "nexthop" in nh.keys():
assert (
nh["nexthop"] not in nexthop_list
), "{}, show mpls table, duplicated or blacklisted nexthop address".format(
router.name
)
nexthop_list.append(nh["nexthop"])
elif "interface" in nh.keys():
assert (
nh["interface"] not in nexthop_list
), "{}, show mpls table, duplicated or blacklisted nexthop interface".format(
router.name
)
nexthop_list.append(nh["interface"])
else:
assert (
0
), "{}, show mpls table, entry with neither nexthop nor interface".format(
router.name
)
if whitelist:
for entry in whitelist:
assert (
entry in nexthop_list
), "{}, show mpls table, entry with nexthop {} not present in nexthop list".format(
router.name, entry
)
# Check r2 removed 172.31.0.30 vpnv4 update
test_func = functools.partial(
check_show_mpls_table, router, blacklist, label_list, whitelist
)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, MPLS labels check fail: {}".format(router.name, result)
def check_show_bgp_vpn_prefix_not_found(router, ipversion, prefix, rd, label=None):