proxmox-mirrors/mirror_frr
1
0
Fork 0
mirror of https://git.proxmox.com/git/mirror_frr synced 2025-08-07 15:33:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #4852 from ashish12pant/fix_log

Browse Source 
tests: Enhance execution logs in topojson
This commit is contained in:
Martin Winter 2019-08-29 04:35:37 +02:00 committed by GitHub
commit 4298dfd12e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 485 additions and 403 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/topotests/bgp-path-attributes-topo1/test_bgp_path_attributes.py
View File

@ -219,7 +219,8 @@ def test_next_hop_attribute(request):
dut = "r1" dut = "r1"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
" present in RIB".format(tc_name)
# Configure next-hop-self to bgp neighbor # Configure next-hop-self to bgp neighbor
input_dict_1 = { input_dict_1 = {
@ -484,7 +485,7 @@ def test_localpref_attribute(request):
"neighbor": { "neighbor": {
"r1": { "r1": {
"dest_link": { "dest_link": {
"r3": { "r2": {
"route_maps": [ "route_maps": [
{"name": "RMAP_LOCAL_PREF", {"name": "RMAP_LOCAL_PREF",
"direction": "in"} "direction": "in"}
@ -499,6 +500,7 @@ def test_localpref_attribute(request):
} }
} }
} }
result = create_router_bgp(tgen, topo, input_dict_4) result = create_router_bgp(tgen, topo, input_dict_4)
assert result is True, "Testcase {} : Failed \n Error: {}".format( assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result) tc_name, result)

34
tests/topotests/bgp-prefix-list-topo1/test_prefix_lists.py
View File

@ -386,8 +386,8 @@ def test_ip_prefix_lists_out_permit(request):
tc_name, result) tc_name, result)
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)
@ -497,8 +497,8 @@ def test_ip_prefix_lists_in_deny_and_permit_any(request):
dut = "r3" dut = "r3"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)
@ -542,7 +542,6 @@ def test_delete_prefix_lists(request):
result = verify_prefix_lists(tgen, input_dict_2) result = verify_prefix_lists(tgen, input_dict_2)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result) tc_name, result)
logger.info(result)
# Delete prefix list # Delete prefix list
input_dict_2 = { input_dict_2 = {
@ -714,9 +713,8 @@ def test_ip_prefix_lists_out_deny_and_permit_any(request):
dut = "r4" dut = "r4"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)
@ -859,8 +857,8 @@ def test_modify_prefix_lists_in_permit_to_deny(request):
dut = "r3" dut = "r3"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)
@ -972,8 +970,8 @@ def test_modify_prefix_lists_in_deny_to_permit(request):
dut = "r3" dut = "r3"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
# Modify ip prefix list # Modify ip prefix list
input_dict_1 = { input_dict_1 = {
@ -1152,8 +1150,8 @@ def test_modify_prefix_lists_out_permit_to_deny(request):
dut = "r4" dut = "r4"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)
@ -1265,8 +1263,8 @@ def test_modify_prefix_lists_out_deny_to_permit(request):
dut = "r4" dut = "r4"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
# Modify ip prefix list # Modify ip prefix list
input_dict_1 = { input_dict_1 = {
@ -1439,8 +1437,8 @@ def test_ip_prefix_lists_implicit_deny(request):
dut = "r4" dut = "r4"
protocol = "bgp" protocol = "bgp"
result = verify_rib(tgen, "ipv4", dut, input_dict_1, protocol=protocol, expected=False) result = verify_rib(tgen, "ipv4", dut, input_dict_1, protocol=protocol, expected=False)
assert result is not True, "Testcase {} : Failed \n Error: {}".format( assert result is not True, "Testcase {} : Failed \n Error: Routes still" \
tc_name, result) " present in RIB".format(tc_name)
write_test_footer(tc_name) write_test_footer(tc_name)

221
tests/topotests/lib/bgp.py
View File

@ -32,7 +32,8 @@ from lib.common_config import (create_common_configuration,
load_config_to_router, load_config_to_router,
check_address_types, check_address_types,
generate_ips, generate_ips,
find_interface_with_greater_ip) find_interface_with_greater_ip,
run_frr_cmd, retry)
BGP_CONVERGENCE_TIMEOUT = 10 BGP_CONVERGENCE_TIMEOUT = 10
@ -116,8 +117,8 @@ def create_router_bgp(tgen, topo, input_dict=None, build=False):
logger.debug("Router %s: 'bgp' not present in input_dict", router) logger.debug("Router %s: 'bgp' not present in input_dict", router)
continue continue
result = __create_bgp_global(tgen, input_dict, router, build) data_all_bgp = __create_bgp_global(tgen, input_dict, router, build)
if result is True: if data_all_bgp:
bgp_data = input_dict[router]["bgp"] bgp_data = input_dict[router]["bgp"]
bgp_addr_data = bgp_data.setdefault("address_family", {}) bgp_addr_data = bgp_data.setdefault("address_family", {})
@ -134,8 +135,18 @@ def create_router_bgp(tgen, topo, input_dict=None, build=False):
or ipv6_data.setdefault("unicast", {}) else False or ipv6_data.setdefault("unicast", {}) else False
if neigh_unicast: if neigh_unicast:
result = __create_bgp_unicast_neighbor( data_all_bgp = __create_bgp_unicast_neighbor(
tgen, topo, input_dict, router, build) tgen, topo, input_dict, router,
config_data=data_all_bgp)
try:
result = create_common_configuration(tgen, router, data_all_bgp,
"bgp", build)
except InvalidCLIError:
# Traceback
errormsg = traceback.format_exc()
logger.error(errormsg)
return errormsg
logger.debug("Exiting lib API: create_router_bgp()") logger.debug("Exiting lib API: create_router_bgp()")
return result return result
@ -157,17 +168,14 @@ def __create_bgp_global(tgen, input_dict, router, build=False):
True or False True or False
""" """
result = False
logger.debug("Entering lib API: __create_bgp_global()") logger.debug("Entering lib API: __create_bgp_global()")
try:
bgp_data = input_dict[router]["bgp"] bgp_data = input_dict[router]["bgp"]
del_bgp_action = bgp_data.setdefault("delete", False) del_bgp_action = bgp_data.setdefault("delete", False)
if del_bgp_action: if del_bgp_action:
config_data = ["no router bgp"] config_data = ["no router bgp"]
result = create_common_configuration(tgen, router, config_data,
"bgp", build=build) return config_data
return result
config_data = [] config_data = []
@ -215,19 +223,11 @@ def __create_bgp_global(tgen, input_dict, router, build=False):
config_data.append(cmd) config_data.append(cmd)
result = create_common_configuration(tgen, router, config_data, return config_data
"bgp", build=build)
except InvalidCLIError:
# Traceback
errormsg = traceback.format_exc()
logger.error(errormsg)
return errormsg
logger.debug("Exiting lib API: create_bgp_global()")
return result
def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False): def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router,
config_data=None):
""" """
Helper API to create configuration for address-family unicast Helper API to create configuration for address-family unicast
@ -240,10 +240,11 @@ def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False):
* `build` : Only for initial setup phase this is set as True. * `build` : Only for initial setup phase this is set as True.
""" """
result = False
logger.debug("Entering lib API: __create_bgp_unicast_neighbor()") logger.debug("Entering lib API: __create_bgp_unicast_neighbor()")
try:
config_data = ["router bgp"] add_neigh = True
if "router bgp "in config_data:
add_neigh = False
bgp_data = input_dict[router]["bgp"]["address_family"] bgp_data = input_dict[router]["bgp"]["address_family"]
for addr_type, addr_dict in bgp_data.iteritems(): for addr_type, addr_dict in bgp_data.iteritems():
@ -253,10 +254,11 @@ def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False):
if not check_address_types(addr_type): if not check_address_types(addr_type):
continue continue
addr_data = addr_dict["unicast"]
if addr_data:
config_data.append("address-family {} unicast".format( config_data.append("address-family {} unicast".format(
addr_type addr_type
)) ))
addr_data = addr_dict["unicast"]
advertise_network = addr_data.setdefault("advertise_networks", advertise_network = addr_data.setdefault("advertise_networks",
[]) [])
for advertise_network_dict in advertise_network: for advertise_network_dict in advertise_network:
@ -279,7 +281,7 @@ def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False):
for ip in network_list: for ip in network_list:
ip = str(ipaddr.IPNetwork(unicode(ip)).network) ip = str(ipaddr.IPNetwork(unicode(ip)).network)
cmd = "network {}/{}\n".format(ip, prefix) cmd = "network {}/{}".format(ip, prefix)
if del_action: if del_action:
cmd = "no {}".format(cmd) cmd = "no {}".format(cmd)
@ -330,7 +332,7 @@ def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False):
if "neighbor" in addr_data: if "neighbor" in addr_data:
neigh_data = __create_bgp_neighbor(topo, input_dict, neigh_data = __create_bgp_neighbor(topo, input_dict,
router, addr_type) router, addr_type, add_neigh)
config_data.extend(neigh_data) config_data.extend(neigh_data)
for addr_type, addr_dict in bgp_data.iteritems(): for addr_type, addr_dict in bgp_data.iteritems():
@ -340,24 +342,16 @@ def __create_bgp_unicast_neighbor(tgen, topo, input_dict, router, build=False):
addr_data = addr_dict["unicast"] addr_data = addr_dict["unicast"]
if "neighbor" in addr_data: if "neighbor" in addr_data:
neigh_addr_data = __create_bgp_unicast_address_family( neigh_addr_data = __create_bgp_unicast_address_family(
topo, input_dict, router, addr_type) topo, input_dict, router, addr_type, add_neigh)
config_data.extend(neigh_addr_data) config_data.extend(neigh_addr_data)
result = create_common_configuration(tgen, router, config_data,
None, build=build)
except InvalidCLIError:
# Traceback
errormsg = traceback.format_exc()
logger.error(errormsg)
return errormsg
logger.debug("Exiting lib API: __create_bgp_unicast_neighbor()") logger.debug("Exiting lib API: __create_bgp_unicast_neighbor()")
return result return config_data
def __create_bgp_neighbor(topo, input_dict, router, addr_type): def __create_bgp_neighbor(topo, input_dict, router, addr_type, add_neigh=True):
""" """
Helper API to create neighbor specific configuration Helper API to create neighbor specific configuration
@ -391,6 +385,7 @@ def __create_bgp_neighbor(topo, input_dict, router, addr_type):
neigh_cxt = "neighbor {}".format(ip_addr) neigh_cxt = "neighbor {}".format(ip_addr)
if add_neigh:
config_data.append("{} remote-as {}".format(neigh_cxt, remote_as)) config_data.append("{} remote-as {}".format(neigh_cxt, remote_as))
if addr_type == "ipv6": if addr_type == "ipv6":
config_data.append("address-family ipv6 unicast") config_data.append("address-family ipv6 unicast")
@ -429,7 +424,8 @@ def __create_bgp_neighbor(topo, input_dict, router, addr_type):
return config_data return config_data
def __create_bgp_unicast_address_family(topo, input_dict, router, addr_type): def __create_bgp_unicast_address_family(topo, input_dict, router, addr_type,
add_neigh=True):
""" """
API prints bgp global config to bgp_json file. API prints bgp global config to bgp_json file.
@ -474,9 +470,9 @@ def __create_bgp_unicast_address_family(topo, input_dict, router, addr_type):
dest_link]["ipv6"].split("/")[0] dest_link]["ipv6"].split("/")[0]
neigh_cxt = "neighbor {}".format(ip_addr) neigh_cxt = "neighbor {}".format(ip_addr)
config_data.append("address-family {} unicast".format( #config_data.append("address-family {} unicast".format(
addr_type # addr_type
)) #))
if deactivate: if deactivate:
config_data.append( config_data.append(
"no neighbor {} activate".format(deactivate)) "no neighbor {} activate".format(deactivate))
@ -531,6 +527,7 @@ def __create_bgp_unicast_address_family(topo, input_dict, router, addr_type):
############################################# #############################################
# Verification APIs # Verification APIs
############################################# #############################################
@retry(attempts=3, wait=2, return_is_str=True)
def verify_router_id(tgen, topo, input_dict): def verify_router_id(tgen, topo, input_dict):
""" """
Running command "show ip bgp json" for DUT and reading router-id Running command "show ip bgp json" for DUT and reading router-id
@ -565,7 +562,7 @@ def verify_router_id(tgen, topo, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_router_id()") logger.debug("Entering lib API: verify_router_id()")
for router in input_dict.keys(): for router in input_dict.keys():
if router not in tgen.routers(): if router not in tgen.routers():
continue continue
@ -576,9 +573,9 @@ def verify_router_id(tgen, topo, input_dict):
"del_router_id", False) "del_router_id", False)
logger.info("Checking router %s router-id", router) logger.info("Checking router %s router-id", router)
show_bgp_json = rnode.vtysh_cmd("show ip bgp json", show_bgp_json = run_frr_cmd(rnode, "show bgp summary json",
isjson=True) isjson=True)
router_id_out = show_bgp_json["routerId"] router_id_out = show_bgp_json["ipv4Unicast"]["routerId"]
router_id_out = ipaddr.IPv4Address(unicode(router_id_out)) router_id_out = ipaddr.IPv4Address(unicode(router_id_out))
# Once router-id is deleted, highest interface ip should become # Once router-id is deleted, highest interface ip should become
@ -598,39 +595,35 @@ def verify_router_id(tgen, topo, input_dict):
router_id_out) router_id_out)
return errormsg return errormsg
logger.info("Exiting lib API: verify_router_id()") logger.debug("Exiting lib API: verify_router_id()")
return True return True
@retry(attempts=20, wait=2, return_is_str=True)
def verify_bgp_convergence(tgen, topo): def verify_bgp_convergence(tgen, topo):
""" """
API will verify if BGP is converged with in the given time frame. API will verify if BGP is converged with in the given time frame.
Running "show bgp summary json" command and verify bgp neighbor Running "show bgp summary json" command and verify bgp neighbor
state is established, state is established,
Parameters Parameters
---------- ----------
* `tgen`: topogen object * `tgen`: topogen object
* `topo`: input json file data * `topo`: input json file data
* `addr_type`: ip_type, ipv4/ipv6 * `addr_type`: ip_type, ipv4/ipv6
Usage Usage
----- -----
# To veriry is BGP is converged for all the routers used in # To veriry is BGP is converged for all the routers used in
topology topology
results = verify_bgp_convergence(tgen, topo, "ipv4") results = verify_bgp_convergence(tgen, topo, "ipv4")
Returns Returns
------- -------
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_bgp_confergence()") logger.debug("Entering lib API: verify_bgp_convergence()")
for router, rnode in tgen.routers().iteritems(): for router, rnode in tgen.routers().iteritems():
logger.info("Verifying BGP Convergence on router %s:", router) logger.info("Verifying BGP Convergence on router %s", router)
show_bgp_json = run_frr_cmd(rnode, "show bgp summary json",
for retry in range(1, 11):
show_bgp_json = rnode.vtysh_cmd("show bgp summary json",
isjson=True) isjson=True)
# Verifying output dictionary show_bgp_json is empty or not # Verifying output dictionary show_bgp_json is empty or not
if not bool(show_bgp_json): if not bool(show_bgp_json):
@ -638,12 +631,11 @@ def verify_bgp_convergence(tgen, topo):
return errormsg return errormsg
# To find neighbor ip type # To find neighbor ip type
total_peer = 0
bgp_addr_type = topo["routers"][router]["bgp"]["address_family"] bgp_addr_type = topo["routers"][router]["bgp"]["address_family"]
for addr_type in bgp_addr_type.keys(): for addr_type in bgp_addr_type.keys():
if not check_address_types(addr_type): if not check_address_types(addr_type):
continue continue
total_peer = 0
bgp_neighbors = bgp_addr_type[addr_type]["unicast"]["neighbor"] bgp_neighbors = bgp_addr_type[addr_type]["unicast"]["neighbor"]
@ -651,6 +643,8 @@ def verify_bgp_convergence(tgen, topo):
total_peer += len(bgp_neighbors[bgp_neighbor]["dest_link"]) total_peer += len(bgp_neighbors[bgp_neighbor]["dest_link"])
for addr_type in bgp_addr_type.keys(): for addr_type in bgp_addr_type.keys():
if not check_address_types(addr_type):
continue
bgp_neighbors = bgp_addr_type[addr_type]["unicast"]["neighbor"] bgp_neighbors = bgp_addr_type[addr_type]["unicast"]["neighbor"]
no_of_peer = 0 no_of_peer = 0
@ -673,25 +667,12 @@ def verify_bgp_convergence(tgen, topo):
no_of_peer += 1 no_of_peer += 1
if no_of_peer == total_peer: if no_of_peer == total_peer:
logger.info("BGP is Converged for router %s", router) logger.info("BGP is Converged for router %s", router)
break
else: else:
logger.warning("BGP is not yet Converged for router %s", errormsg = "BGP is not converged for router {}".format(
router) router)
sleeptime = 2 * retry
if sleeptime <= BGP_CONVERGENCE_TIMEOUT:
# Waiting for BGP to converge
logger.info("Waiting for %s sec for BGP to converge on"
" router %s...", sleeptime, router)
sleep(sleeptime)
else:
show_bgp_summary = rnode.vtysh_cmd("show bgp summary")
errormsg = "TIMEOUT!! BGP is not converged in {} " \
"seconds for router {} \n {}".format(
BGP_CONVERGENCE_TIMEOUT, router,
show_bgp_summary)
return errormsg return errormsg
logger.info("Exiting API: verify_bgp_confergence()") logger.debug("Exiting API: verify_bgp_convergence()")
return True return True
@ -723,7 +704,7 @@ def modify_as_number(tgen, topo, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: modify_as_number()") logger.debug("Entering lib API: modify_as_number()")
try: try:
new_topo = deepcopy(topo["routers"]) new_topo = deepcopy(topo["routers"])
@ -757,11 +738,12 @@ def modify_as_number(tgen, topo, input_dict):
logger.error(errormsg) logger.error(errormsg)
return errormsg return errormsg
logger.info("Exiting lib API: modify_as_number()") logger.debug("Exiting lib API: modify_as_number()")
return True return True
@retry(attempts=3, wait=2, return_is_str=True)
def verify_as_numbers(tgen, topo, input_dict): def verify_as_numbers(tgen, topo, input_dict):
""" """
This API is to verify AS numbers for given DUT by running This API is to verify AS numbers for given DUT by running
@ -791,7 +773,7 @@ def verify_as_numbers(tgen, topo, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_as_numbers()") logger.debug("Entering lib API: verify_as_numbers()")
for router in input_dict.keys(): for router in input_dict.keys():
if router not in tgen.routers(): if router not in tgen.routers():
continue continue
@ -800,7 +782,7 @@ def verify_as_numbers(tgen, topo, input_dict):
logger.info("Verifying AS numbers for dut %s:", router) logger.info("Verifying AS numbers for dut %s:", router)
show_ip_bgp_neighbor_json = rnode.vtysh_cmd( show_ip_bgp_neighbor_json = run_frr_cmd(rnode,
"show ip bgp neighbor json", isjson=True) "show ip bgp neighbor json", isjson=True)
local_as = input_dict[router]["bgp"]["local_as"] local_as = input_dict[router]["bgp"]["local_as"]
bgp_addr_type = topo["routers"][router]["bgp"]["address_family"] bgp_addr_type = topo["routers"][router]["bgp"]["address_family"]
@ -846,7 +828,7 @@ def verify_as_numbers(tgen, topo, input_dict):
"neighbor %s, found expected: %s", "neighbor %s, found expected: %s",
router, bgp_neighbor, remote_as) router, bgp_neighbor, remote_as)
logger.info("Exiting lib API: verify_AS_numbers()") logger.debug("Exiting lib API: verify_AS_numbers()")
return True return True
@ -873,7 +855,7 @@ def clear_bgp_and_verify(tgen, topo, router):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: clear_bgp_and_verify()") logger.debug("Entering lib API: clear_bgp_and_verify()")
if router not in tgen.routers(): if router not in tgen.routers():
return False return False
@ -883,20 +865,14 @@ def clear_bgp_and_verify(tgen, topo, router):
peer_uptime_before_clear_bgp = {} peer_uptime_before_clear_bgp = {}
# Verifying BGP convergence before bgp clear command # Verifying BGP convergence before bgp clear command
for retry in range(1, 11): for retry in range(1, 11):
sleeptime = 2 * retry sleeptime = 3
if sleeptime <= BGP_CONVERGENCE_TIMEOUT:
# Waiting for BGP to converge # Waiting for BGP to converge
logger.info("Waiting for %s sec for BGP to converge on router" logger.info("Waiting for %s sec for BGP to converge on router"
" %s...", sleeptime, router) " %s...", sleeptime, router)
sleep(sleeptime) sleep(sleeptime)
else:
errormsg = "TIMEOUT!! BGP is not converged in {} seconds for" \
" router {}".format(BGP_CONVERGENCE_TIMEOUT, router)
return errormsg
show_bgp_json = rnode.vtysh_cmd("show bgp summary json", show_bgp_json = run_frr_cmd(rnode, "show bgp summary json",
isjson=True) isjson=True)
logger.info(show_bgp_json)
# Verifying output dictionary show_bgp_json is empty or not # Verifying output dictionary show_bgp_json is empty or not
if not bool(show_bgp_json): if not bool(show_bgp_json):
errormsg = "BGP is not running" errormsg = "BGP is not running"
@ -950,33 +926,33 @@ def clear_bgp_and_verify(tgen, topo, router):
" clear", router) " clear", router)
break break
else: else:
logger.warning("BGP is not yet Converged for router %s " logger.info("BGP is not yet Converged for router %s "
"before bgp clear", router) "before bgp clear", router)
else:
errormsg = "TIMEOUT!! BGP is not converged in 30 seconds for" \
" router {}".format(router)
return errormsg
logger.info(peer_uptime_before_clear_bgp) logger.info(peer_uptime_before_clear_bgp)
# Clearing BGP # Clearing BGP
logger.info("Clearing BGP neighborship for router %s..", router) logger.info("Clearing BGP neighborship for router %s..", router)
for addr_type in bgp_addr_type.keys(): for addr_type in bgp_addr_type.keys():
if addr_type == "ipv4": if addr_type == "ipv4":
rnode.vtysh_cmd("clear ip bgp *") run_frr_cmd(rnode, "clear ip bgp *")
elif addr_type == "ipv6": elif addr_type == "ipv6":
rnode.vtysh_cmd("clear bgp ipv6 *") run_frr_cmd(rnode, "clear bgp ipv6 *")
peer_uptime_after_clear_bgp = {} peer_uptime_after_clear_bgp = {}
# Verifying BGP convergence after bgp clear command # Verifying BGP convergence after bgp clear command
for retry in range(1, 11): for retry in range(11):
sleeptime = 2 * retry sleeptime = 3
if sleeptime <= BGP_CONVERGENCE_TIMEOUT:
# Waiting for BGP to converge # Waiting for BGP to converge
logger.info("Waiting for %s sec for BGP to converge on router" logger.info("Waiting for %s sec for BGP to converge on router"
" %s...", sleeptime, router) " %s...", sleeptime, router)
sleep(sleeptime) sleep(sleeptime)
else:
errormsg = "TIMEOUT!! BGP is not converged in {} seconds for" \
" router {}".format(BGP_CONVERGENCE_TIMEOUT, router)
return errormsg
show_bgp_json = rnode.vtysh_cmd("show bgp summary json",
show_bgp_json = run_frr_cmd(rnode, "show bgp summary json",
isjson=True) isjson=True)
# Verifying output dictionary show_bgp_json is empty or not # Verifying output dictionary show_bgp_json is empty or not
if not bool(show_bgp_json): if not bool(show_bgp_json):
@ -1028,9 +1004,12 @@ def clear_bgp_and_verify(tgen, topo, router):
router) router)
break break
else: else:
logger.warning("BGP is not yet Converged for router %s after" logger.info("BGP is not yet Converged for router %s after"
" bgp clear", router) " bgp clear", router)
else:
errormsg = "TIMEOUT!! BGP is not converged in 30 seconds for" \
" router {}".format(router)
return errormsg
logger.info(peer_uptime_after_clear_bgp) logger.info(peer_uptime_after_clear_bgp)
# Comparing peerUptimeEstablishedEpoch dictionaries # Comparing peerUptimeEstablishedEpoch dictionaries
if peer_uptime_before_clear_bgp != peer_uptime_after_clear_bgp: if peer_uptime_before_clear_bgp != peer_uptime_after_clear_bgp:
@ -1041,7 +1020,7 @@ def clear_bgp_and_verify(tgen, topo, router):
" {}".format(router) " {}".format(router)
return errormsg return errormsg
logger.info("Exiting lib API: clear_bgp_and_verify()") logger.debug("Exiting lib API: clear_bgp_and_verify()")
return True return True
@ -1077,7 +1056,7 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_bgp_timers_and_functionality()") logger.debug("Entering lib API: verify_bgp_timers_and_functionality()")
sleep(5) sleep(5)
router_list = tgen.routers() router_list = tgen.routers()
for router in input_dict.keys(): for router in input_dict.keys():
@ -1090,7 +1069,7 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
router) router)
show_ip_bgp_neighbor_json = \ show_ip_bgp_neighbor_json = \
rnode.vtysh_cmd("show ip bgp neighbor json", isjson=True) run_frr_cmd(rnode, "show ip bgp neighbor json", isjson=True)
bgp_addr_type = input_dict[router]["bgp"]["address_family"] bgp_addr_type = input_dict[router]["bgp"]["address_family"]
@ -1178,7 +1157,7 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
sleep(keepalivetimer) sleep(keepalivetimer)
sleep(2) sleep(2)
show_bgp_json = \ show_bgp_json = \
rnode.vtysh_cmd("show bgp summary json", run_frr_cmd(rnode, "show bgp summary json",
isjson=True) isjson=True)
if addr_type == "ipv4": if addr_type == "ipv4":
@ -1192,17 +1171,13 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
(holddowntimer - keepalivetimer): (holddowntimer - keepalivetimer):
if nh_state != "Established": if nh_state != "Established":
errormsg = "BGP neighborship has not gone " \ errormsg = "BGP neighborship has not gone " \
"down in {} sec for neighbor {}\n" \ "down in {} sec for neighbor {}" \
"show_bgp_json: \n {} ".format( .format(timer, bgp_neighbor)
timer, bgp_neighbor,
show_bgp_json)
return errormsg return errormsg
else: else:
logger.info("BGP neighborship is intact in %s" logger.info("BGP neighborship is intact in %s"
" sec for neighbor %s \n " " sec for neighbor %s",
"show_bgp_json : \n %s", timer, bgp_neighbor)
timer, bgp_neighbor,
show_bgp_json)
#################### ####################
# Shutting down peer interface and verifying that BGP # Shutting down peer interface and verifying that BGP
@ -1229,7 +1204,7 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
sleep(keepalivetimer) sleep(keepalivetimer)
sleep(2) sleep(2)
show_bgp_json = \ show_bgp_json = \
rnode.vtysh_cmd("show bgp summary json", run_frr_cmd(rnode, "show bgp summary json",
isjson=True) isjson=True)
if addr_type == "ipv4": if addr_type == "ipv4":
@ -1242,22 +1217,19 @@ def verify_bgp_timers_and_functionality(tgen, topo, input_dict):
if timer == holddowntimer: if timer == holddowntimer:
if nh_state == "Established": if nh_state == "Established":
errormsg = "BGP neighborship has not gone " \ errormsg = "BGP neighborship has not gone " \
"down in {} sec for neighbor {}\n" \ "down in {} sec for neighbor {}" \
"show_bgp_json: \n {} ".format( .format(timer, bgp_neighbor)
timer, bgp_neighbor,
show_bgp_json)
return errormsg return errormsg
else: else:
logger.info("BGP neighborship has gone down in" logger.info("BGP neighborship has gone down in"
" %s sec for neighbor %s \n" " %s sec for neighbor %s",
"show_bgp_json : \n %s", timer, bgp_neighbor)
timer, bgp_neighbor,
show_bgp_json)
logger.info("Exiting lib API: verify_bgp_timers_and_functionality()") logger.debug("Exiting lib API: verify_bgp_timers_and_functionality()")
return True return True
@retry(attempts=3, wait=2, return_is_str=True)
def verify_best_path_as_per_bgp_attribute(tgen, addr_type, router, input_dict, def verify_best_path_as_per_bgp_attribute(tgen, addr_type, router, input_dict,
attribute): attribute):
""" """
@ -1319,7 +1291,7 @@ def verify_best_path_as_per_bgp_attribute(tgen, addr_type, router, input_dict,
sleep(2) sleep(2)
logger.info("Verifying router %s RIB for best path:", router) logger.info("Verifying router %s RIB for best path:", router)
sh_ip_bgp_json = rnode.vtysh_cmd(command, isjson=True) sh_ip_bgp_json = run_frr_cmd(rnode, command, isjson=True)
for route_val in input_dict.values(): for route_val in input_dict.values():
net_data = route_val["bgp"]["address_family"]["ipv4"]["unicast"] net_data = route_val["bgp"]["address_family"]["ipv4"]["unicast"]
@ -1380,7 +1352,7 @@ def verify_best_path_as_per_bgp_attribute(tgen, addr_type, router, input_dict,
else: else:
command = "show ipv6 route json" command = "show ipv6 route json"
rib_routes_json = rnode.vtysh_cmd(command, isjson=True) rib_routes_json = run_frr_cmd(rnode, command, isjson=True)
# Verifying output dictionary rib_routes_json is not empty # Verifying output dictionary rib_routes_json is not empty
if not bool(rib_routes_json): if not bool(rib_routes_json):
@ -1417,6 +1389,7 @@ def verify_best_path_as_per_bgp_attribute(tgen, addr_type, router, input_dict,
return True return True
@retry(attempts=3, wait=2, return_is_str=True)
def verify_best_path_as_per_admin_distance(tgen, addr_type, router, input_dict, def verify_best_path_as_per_admin_distance(tgen, addr_type, router, input_dict,
attribute): attribute):
""" """
@ -1451,7 +1424,7 @@ def verify_best_path_as_per_admin_distance(tgen, addr_type, router, input_dict,
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_best_path_as_per_admin_distance()") logger.debug("Entering lib API: verify_best_path_as_per_admin_distance()")
router_list = tgen.routers() router_list = tgen.routers()
if router not in router_list: if router not in router_list:
return False return False
@ -1490,7 +1463,7 @@ def verify_best_path_as_per_admin_distance(tgen, addr_type, router, input_dict,
compare = "LOWEST" compare = "LOWEST"
# Show ip route # Show ip route
rib_routes_json = rnode.vtysh_cmd(command, isjson=True) rib_routes_json = run_frr_cmd(rnode, command, isjson=True)
# Verifying output dictionary rib_routes_json is not empty # Verifying output dictionary rib_routes_json is not empty
if not bool(rib_routes_json): if not bool(rib_routes_json):

228
tests/topotests/lib/common_config.py
View File

@ -23,6 +23,11 @@ from datetime import datetime
from time import sleep from time import sleep
from subprocess import call from subprocess import call
from subprocess import STDOUT as SUB_STDOUT from subprocess import STDOUT as SUB_STDOUT
from subprocess import PIPE as SUB_PIPE
from subprocess import Popen
from functools import wraps
from re import search as re_search
import StringIO import StringIO
import os import os
import ConfigParser import ConfigParser
@ -41,6 +46,7 @@ FRRCFG_FILE = "frr_json.conf"
FRRCFG_BKUP_FILE = "frr_json_initial.conf" FRRCFG_BKUP_FILE = "frr_json_initial.conf"
ERROR_LIST = ["Malformed", "Failure", "Unknown"] ERROR_LIST = ["Malformed", "Failure", "Unknown"]
ROUTER_LIST = []
#### ####
CD = os.path.dirname(os.path.realpath(__file__)) CD = os.path.dirname(os.path.realpath(__file__))
@ -142,6 +148,35 @@ class InvalidCLIError(Exception):
pass pass
def run_frr_cmd(rnode, cmd, isjson=False):
"""
Execute frr show commands in priviledged mode
* `rnode`: router node on which commands needs to executed
* `cmd`: Command to be executed on frr
* `isjson`: If command is to get json data or not
:return str:
"""
if cmd:
ret_data = rnode.vtysh_cmd(cmd, isjson=isjson)
if True:
if isjson:
logger.debug(ret_data)
print_data = rnode.vtysh_cmd(cmd.rstrip("json"), isjson=False)
else:
print_data = ret_data
logger.info('Output for command [ %s] on router %s:\n%s',
cmd.rstrip("json"), rnode.name, print_data)
return ret_data
else:
raise InvalidCLIError('No actual cmd passed')
def create_common_configuration(tgen, router, data, config_type=None, def create_common_configuration(tgen, router, data, config_type=None,
build=False): build=False):
""" """
@ -186,6 +221,7 @@ def create_common_configuration(tgen, router, data, config_type=None,
frr_cfg_fd.write(config_map[config_type]) frr_cfg_fd.write(config_map[config_type])
for line in data: for line in data:
frr_cfg_fd.write("{} \n".format(str(line))) frr_cfg_fd.write("{} \n".format(str(line)))
frr_cfg_fd.write("\n")
except IOError as err: except IOError as err:
logger.error("Unable to open FRR Config File. error(%s): %s" % logger.error("Unable to open FRR Config File. error(%s): %s" %
@ -215,10 +251,13 @@ def reset_config_on_routers(tgen, routerName=None):
logger.debug("Entering API: reset_config_on_routers") logger.debug("Entering API: reset_config_on_routers")
router_list = tgen.routers() router_list = tgen.routers()
for rname, router in router_list.iteritems(): for rname in ROUTER_LIST:
if routerName and routerName != rname: if routerName and routerName != rname:
continue continue
router = router_list[rname]
logger.info("Configuring router %s to initial test configuration",
rname)
cfg = router.run("vtysh -c 'show running'") cfg = router.run("vtysh -c 'show running'")
fname = "{}/{}/frr.sav".format(TMPDIR, rname) fname = "{}/{}/frr.sav".format(TMPDIR, rname)
dname = "{}/{}/delta.conf".format(TMPDIR, rname) dname = "{}/{}/delta.conf".format(TMPDIR, rname)
@ -235,16 +274,35 @@ def reset_config_on_routers(tgen, routerName=None):
f.close() f.close()
command = "/usr/lib/frr/frr-reload.py --input {}/{}/frr.sav" \ run_cfg_file = "{}/{}/frr.sav".format(TMPDIR, rname)
" --test {}/{}/frr_json_initial.conf > {}". \ init_cfg_file = "{}/{}/frr_json_initial.conf".format(TMPDIR, rname)
format(TMPDIR, rname, TMPDIR, rname, dname) command = "/usr/lib/frr/frr-reload.py --input {} --test {} > {}". \
result = call(command, shell=True, stderr=SUB_STDOUT) format(run_cfg_file, init_cfg_file, dname)
result = call(command, shell=True, stderr=SUB_STDOUT,
stdout=SUB_PIPE)
# Assert if command fail # Assert if command fail
if result > 0: if result > 0:
errormsg = ("Command:{} is failed due to non-zero exit" logger.error("Delta file creation failed. Command executed %s",
" code".format(command)) command)
return errormsg with open(run_cfg_file, 'r') as fd:
logger.info('Running configuration saved in %s is:\n%s',
run_cfg_file, fd.read())
with open(init_cfg_file, 'r') as fd:
logger.info('Test configuration saved in %s is:\n%s',
init_cfg_file, fd.read())
err_cmd = ['/usr/bin/vtysh', '-m', '-f', run_cfg_file]
result = Popen(err_cmd, stdout=SUB_PIPE, stderr=SUB_PIPE)
output = result.communicate()
for out_data in output:
temp_data = out_data.decode('utf-8').lower()
for out_err in ERROR_LIST:
if out_err.lower() in temp_data:
logger.error("Found errors while validating data in"
" %s", run_cfg_file)
raise InvalidCLIError(out_data)
raise InvalidCLIError("Unknown error in %s", output)
f = open(dname, "r") f = open(dname, "r")
delta = StringIO.StringIO() delta = StringIO.StringIO()
@ -264,7 +322,7 @@ def reset_config_on_routers(tgen, routerName=None):
delta.write("end\n") delta.write("end\n")
output = router.vtysh_multicmd(delta.getvalue(), output = router.vtysh_multicmd(delta.getvalue(),
pretty_output=False) pretty_output=False)
logger.info("New configuration for router {}:".format(rname))
delta.close() delta.close()
delta = StringIO.StringIO() delta = StringIO.StringIO()
cfg = router.run("vtysh -c 'show running'") cfg = router.run("vtysh -c 'show running'")
@ -276,6 +334,8 @@ def reset_config_on_routers(tgen, routerName=None):
# Router current configuration to log file or console if # Router current configuration to log file or console if
# "show_router_config" is defined in "pytest.ini" # "show_router_config" is defined in "pytest.ini"
if show_router_config: if show_router_config:
logger.info("Configuration on router {} after config reset:".
format(rname))
logger.info(delta.getvalue()) logger.info(delta.getvalue())
delta.close() delta.close()
@ -297,14 +357,19 @@ def load_config_to_router(tgen, routerName, save_bkup=False):
logger.debug("Entering API: load_config_to_router") logger.debug("Entering API: load_config_to_router")
router_list = tgen.routers() router_list = tgen.routers()
for rname, router in router_list.iteritems(): for rname in ROUTER_LIST:
if rname == routerName: if routerName and routerName != rname:
continue
router = router_list[rname]
try: try:
frr_cfg_file = "{}/{}/{}".format(TMPDIR, rname, FRRCFG_FILE) frr_cfg_file = "{}/{}/{}".format(TMPDIR, rname, FRRCFG_FILE)
frr_cfg_bkup = "{}/{}/{}".format(TMPDIR, rname, frr_cfg_bkup = "{}/{}/{}".format(TMPDIR, rname,
FRRCFG_BKUP_FILE) FRRCFG_BKUP_FILE)
with open(frr_cfg_file, "r") as cfg: with open(frr_cfg_file, "r+") as cfg:
data = cfg.read() data = cfg.read()
logger.info("Applying following configuration on router"
" {}:\n{}".format(rname, data))
if save_bkup: if save_bkup:
with open(frr_cfg_bkup, "w") as bkup: with open(frr_cfg_bkup, "w") as bkup:
bkup.write(data) bkup.write(data)
@ -313,17 +378,17 @@ def load_config_to_router(tgen, routerName, save_bkup=False):
for out_err in ERROR_LIST: for out_err in ERROR_LIST:
if out_err.lower() in output.lower(): if out_err.lower() in output.lower():
raise InvalidCLIError("%s" % output) raise InvalidCLIError("%s" % output)
cfg.truncate(0)
except IOError as err: except IOError as err:
errormsg = ("Unable to open config File. error(%s):" errormsg = ("Unable to open config File. error(%s):"
" %s", (err.errno, err.strerror)) " %s", (err.errno, err.strerror))
return errormsg return errormsg
logger.info("New configuration for router {}:".format(rname))
new_config = router.run("vtysh -c 'show running'")
# Router current configuration to log file or console if # Router current configuration to log file or console if
# "show_router_config" is defined in "pytest.ini" # "show_router_config" is defined in "pytest.ini"
if show_router_config: if show_router_config:
new_config = router.run("vtysh -c 'show running'")
logger.info(new_config) logger.info(new_config)
logger.debug("Exting API: load_config_to_router") logger.debug("Exting API: load_config_to_router")
@ -337,21 +402,25 @@ def start_topology(tgen):
* `tgen` : topogen object * `tgen` : topogen object
""" """
global TMPDIR global TMPDIR, ROUTER_LIST
# Starting topology # Starting topology
tgen.start_topology() tgen.start_topology()
# Starting deamons # Starting deamons
router_list = tgen.routers() router_list = tgen.routers()
ROUTER_LIST = sorted(router_list.keys(),
key=lambda x: int(re_search('\d+', x).group(0)))
TMPDIR = os.path.join(LOGDIR, tgen.modname) TMPDIR = os.path.join(LOGDIR, tgen.modname)
for rname, router in router_list.iteritems(): router_list = tgen.routers()
for rname in ROUTER_LIST:
router = router_list[rname]
try: try:
os.chdir(TMPDIR) os.chdir(TMPDIR)
# Creating rouer named dir and empty zebra.conf bgpd.conf files # Creating router named dir and empty zebra.conf bgpd.conf files
# inside the current directory # inside the current directory
if os.path.isdir('{}'.format(rname)): if os.path.isdir('{}'.format(rname)):
os.system("rm -rf {}".format(rname)) os.system("rm -rf {}".format(rname))
os.mkdir('{}'.format(rname)) os.mkdir('{}'.format(rname))
@ -371,13 +440,11 @@ def start_topology(tgen):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, TopoRouter.RD_ZEBRA,
'{}/{}/zebra.conf'.format(TMPDIR, rname) '{}/{}/zebra.conf'.format(TMPDIR, rname)
# os.path.join(tmpdir, '{}/zebra.conf'.format(rname))
) )
# Loading empty bgpd.conf file to router, to start the bgp deamon # Loading empty bgpd.conf file to router, to start the bgp deamon
router.load_config( router.load_config(
TopoRouter.RD_BGP, TopoRouter.RD_BGP,
'{}/{}/bgpd.conf'.format(TMPDIR, rname) '{}/{}/bgpd.conf'.format(TMPDIR, rname)
# os.path.join(tmpdir, '{}/bgpd.conf'.format(rname))
) )
# Starting routers # Starting routers
@ -548,7 +615,7 @@ def write_test_header(tc_name):
""" Display message at beginning of test case""" """ Display message at beginning of test case"""
count = 20 count = 20
logger.info("*"*(len(tc_name)+count)) logger.info("*"*(len(tc_name)+count))
logger.info("START -> Testcase : %s", tc_name) logger.info("START -> Testcase : %s" % tc_name)
logger.info("*"*(len(tc_name)+count)) logger.info("*"*(len(tc_name)+count))
@ -556,10 +623,65 @@ def write_test_footer(tc_name):
""" Display message at end of test case""" """ Display message at end of test case"""
count = 21 count = 21
logger.info("="*(len(tc_name)+count)) logger.info("="*(len(tc_name)+count))
logger.info("PASSED -> Testcase : %s", tc_name) logger.info("Testcase : %s -> PASSED", tc_name)
logger.info("="*(len(tc_name)+count)) logger.info("="*(len(tc_name)+count))
def retry(attempts=3, wait=2, return_is_str=True, initial_wait=0):
"""
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
"""
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")
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)
for i in range(1, _attempts + 1):
try:
_expected = kwargs.setdefault('expected', True)
kwargs.pop('expected')
ret = func(*args, **kwargs)
logger.debug("Function returned %s" % ret)
if return_is_str and isinstance(ret, bool):
return ret
elif return_is_str and _expected is False:
return ret
if _attempts == i:
return ret
except Exception as err:
if _attempts == i:
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)
func_retry._original = func
return func_retry
return _retry
############################################# #############################################
# These APIs, will used by testcase # These APIs, will used by testcase
############################################# #############################################
@ -589,17 +711,17 @@ def create_interfaces_cfg(tgen, topo, build=False):
interface_name = destRouterLink interface_name = destRouterLink
else: else:
interface_name = data["interface"] interface_name = data["interface"]
interface_data.append("interface {}\n".format( interface_data.append("interface {}".format(
str(interface_name) str(interface_name)
)) ))
if "ipv4" in data: if "ipv4" in data:
intf_addr = c_data["links"][destRouterLink]["ipv4"] intf_addr = c_data["links"][destRouterLink]["ipv4"]
interface_data.append("ip address {}\n".format( interface_data.append("ip address {}".format(
intf_addr intf_addr
)) ))
if "ipv6" in data: if "ipv6" in data:
intf_addr = c_data["links"][destRouterLink]["ipv6"] intf_addr = c_data["links"][destRouterLink]["ipv6"]
interface_data.append("ipv6 address {}\n".format( interface_data.append("ipv6 address {}".format(
intf_addr intf_addr
)) ))
result = create_common_configuration(tgen, c_router, result = create_common_configuration(tgen, c_router,
@ -662,7 +784,7 @@ def create_static_routes(tgen, input_dict, build=False):
for router in input_dict.keys(): for router in input_dict.keys():
if "static_routes" not in input_dict[router]: if "static_routes" not in input_dict[router]:
errormsg = "static_routes not present in input_dict" errormsg = "static_routes not present in input_dict"
logger.info(errormsg) logger.debug(errormsg)
continue continue
static_routes_list = [] static_routes_list = []
@ -768,7 +890,7 @@ def create_prefix_lists(tgen, input_dict, build=False):
for router in input_dict.keys(): for router in input_dict.keys():
if "prefix_lists" not in input_dict[router]: if "prefix_lists" not in input_dict[router]:
errormsg = "prefix_lists not present in input_dict" errormsg = "prefix_lists not present in input_dict"
logger.info(errormsg) logger.debug(errormsg)
continue continue
config_data = [] config_data = []
@ -922,7 +1044,7 @@ def create_route_maps(tgen, input_dict, build=False):
for router in input_dict.keys(): for router in input_dict.keys():
if "route_maps" not in input_dict[router]: if "route_maps" not in input_dict[router]:
errormsg = "route_maps not present in input_dict" errormsg = "route_maps not present in input_dict"
logger.info(errormsg) logger.debug(errormsg)
continue continue
rmap_data = [] rmap_data = []
for rmap_name, rmap_value in \ for rmap_name, rmap_value in \
@ -1014,7 +1136,7 @@ def create_route_maps(tgen, input_dict, build=False):
# Weight # Weight
if weight: if weight:
rmap_data.append("set weight {} \n".format( rmap_data.append("set weight {}".format(
weight)) weight))
# Adding MATCH and SET sequence to RMAP if defined # Adding MATCH and SET sequence to RMAP if defined
@ -1092,7 +1214,8 @@ def create_route_maps(tgen, input_dict, build=False):
############################################# #############################################
# Verification APIs # Verification APIs
############################################# #############################################
def _verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None): @retry(attempts=10, return_is_str=True, initial_wait=2)
def verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None):
""" """
Data will be read from input_dict or input JSON file, API will generate Data will be read from input_dict or input JSON file, API will generate
same prefixes, which were redistributed by either create_static_routes() or same prefixes, which were redistributed by either create_static_routes() or
@ -1140,7 +1263,7 @@ def _verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_rib()") logger.debug("Entering lib API: verify_rib()")
router_list = tgen.routers() router_list = tgen.routers()
for routerInput in input_dict.keys(): for routerInput in input_dict.keys():
@ -1160,9 +1283,8 @@ def _verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None):
else: else:
command = "show ipv6 route json" command = "show ipv6 route json"
sleep(10)
logger.info("Checking router %s RIB:", router) logger.info("Checking router %s RIB:", router)
rib_routes_json = rnode.vtysh_cmd(command, isjson=True) rib_routes_json = run_frr_cmd(rnode, command, isjson=True)
# Verifying output dictionary rib_routes_json is not empty # Verifying output dictionary rib_routes_json is not empty
if bool(rib_routes_json) is False: if bool(rib_routes_json) is False:
@ -1257,30 +1379,10 @@ def _verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None):
logger.info("Verified routes in router %s RIB, found routes" logger.info("Verified routes in router %s RIB, found routes"
" are: %s", dut, found_routes) " are: %s", dut, found_routes)
logger.info("Exiting lib API: verify_rib()") logger.debug("Exiting lib API: verify_rib()")
return True return True
def verify_rib(tgen, addr_type, dut, input_dict, next_hop=None, protocol=None, expected=True):
"""
Wrapper function for `_verify_rib` that tries multiple time to get results.
When the expected result is `False` we actually should expect for an string instead.
"""
# Use currying to hide the parameters and create a test function.
test_func = partial(_verify_rib, tgen, addr_type, dut, input_dict, next_hop, protocol)
# Call the test function and expect it to return True, otherwise try it again.
if expected is True:
_, result = topotest.run_and_expect(test_func, True, count=20, wait=6)
else:
_, result = topotest.run_and_expect_type(test_func, str, count=20, wait=6)
# Return as normal.
return result
def verify_admin_distance_for_static_routes(tgen, input_dict): def verify_admin_distance_for_static_routes(tgen, input_dict):
""" """
API to verify admin distance for static routes as defined in input_dict/ API to verify admin distance for static routes as defined in input_dict/
@ -1311,7 +1413,7 @@ def verify_admin_distance_for_static_routes(tgen, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_admin_distance_for_static_routes()") logger.debug("Entering lib API: verify_admin_distance_for_static_routes()")
for router in input_dict.keys(): for router in input_dict.keys():
if router not in tgen.routers(): if router not in tgen.routers():
@ -1326,7 +1428,7 @@ def verify_admin_distance_for_static_routes(tgen, input_dict):
command = "show ip route json" command = "show ip route json"
else: else:
command = "show ipv6 route json" command = "show ipv6 route json"
show_ip_route_json = rnode.vtysh_cmd(command, isjson=True) show_ip_route_json = run_frr_cmd(rnode, command, isjson=True)
logger.info("Verifying admin distance for static route %s" logger.info("Verifying admin distance for static route %s"
" under dut %s:", static_route, router) " under dut %s:", static_route, router)
@ -1356,7 +1458,7 @@ def verify_admin_distance_for_static_routes(tgen, input_dict):
format(network, router)) format(network, router))
return errormsg return errormsg
logger.info("Exiting lib API: verify_admin_distance_for_static_routes()") logger.debug("Exiting lib API: verify_admin_distance_for_static_routes()")
return True return True
@ -1384,7 +1486,7 @@ def verify_prefix_lists(tgen, input_dict):
errormsg(str) or True errormsg(str) or True
""" """
logger.info("Entering lib API: verify_prefix_lists()") logger.debug("Entering lib API: verify_prefix_lists()")
for router in input_dict.keys(): for router in input_dict.keys():
if router not in tgen.routers(): if router not in tgen.routers():
@ -1393,7 +1495,7 @@ def verify_prefix_lists(tgen, input_dict):
rnode = tgen.routers()[router] rnode = tgen.routers()[router]
# Show ip prefix list # Show ip prefix list
show_prefix_list = rnode.vtysh_cmd("show ip prefix-list") show_prefix_list = run_frr_cmd(rnode, "show ip prefix-list")
# Verify Prefix list is deleted # Verify Prefix list is deleted
prefix_lists_addr = input_dict[router]["prefix_lists"] prefix_lists_addr = input_dict[router]["prefix_lists"]
@ -1403,12 +1505,12 @@ def verify_prefix_lists(tgen, input_dict):
for prefix_list in prefix_lists_addr[addr_type].keys(): for prefix_list in prefix_lists_addr[addr_type].keys():
if prefix_list in show_prefix_list: if prefix_list in show_prefix_list:
errormsg = ("Prefix list {} is not deleted from router" errormsg = ("Prefix list {} is/are present in the router"
" {}".format(prefix_list, router)) " {}".format(prefix_list, router))
return errormsg return errormsg
logger.info("Prefix list %s is/are deleted successfully" logger.info("Prefix list %s is/are not present in the router"
" from router %s", prefix_list, router) " from router %s", prefix_list, router)
logger.info("Exiting lib API: verify_prefix_lissts()") logger.debug("Exiting lib API: verify_prefix_lissts()")
return True return True

25
tests/topotests/lib/topojson.py
View File

@ -20,6 +20,7 @@
from collections import OrderedDict from collections import OrderedDict
from json import dumps as json_dumps from json import dumps as json_dumps
from re import search as re_search
import ipaddr import ipaddr
import pytest import pytest
@ -38,6 +39,9 @@ from lib.common_config import (
from lib.bgp import create_router_bgp from lib.bgp import create_router_bgp
ROUTER_LIST = []
def build_topo_from_json(tgen, topo): def build_topo_from_json(tgen, topo):
""" """
Reads configuration from JSON file. Adds routers, creates interface Reads configuration from JSON file. Adds routers, creates interface
@ -48,13 +52,15 @@ def build_topo_from_json(tgen, topo):
* `topo`: json file data * `topo`: json file data
""" """
listRouters = [] ROUTER_LIST = sorted(topo['routers'].keys(),
for routerN in sorted(topo['routers'].iteritems()): key=lambda x: int(re_search('\d+', x).group(0)))
logger.info('Topo: Add router {}'.format(routerN[0]))
tgen.add_router(routerN[0]) listRouters = ROUTER_LIST[:]
listRouters.append(routerN[0]) for routerN in ROUTER_LIST:
logger.info('Topo: Add router {}'.format(routerN))
tgen.add_router(routerN)
listRouters.append(routerN)
listRouters.sort()
if 'ipv4base' in topo: if 'ipv4base' in topo:
ipv4Next = ipaddr.IPv4Address(topo['link_ip_start']['ipv4']) ipv4Next = ipaddr.IPv4Address(topo['link_ip_start']['ipv4'])
ipv4Step = 2 ** (32 - topo['link_ip_start']['v4mask']) ipv4Step = 2 ** (32 - topo['link_ip_start']['v4mask'])
@ -78,7 +84,7 @@ def build_topo_from_json(tgen, topo):
elif 'link' in x: elif 'link' in x:
return int(x.split('-link')[1]) return int(x.split('-link')[1])
else: else:
return int(x.split('r')[1]) return int(re_search('\d+', x).group(0))
for destRouterLink, data in sorted(topo['routers'][curRouter]['links']. \ for destRouterLink, data in sorted(topo['routers'][curRouter]['links']. \
iteritems(), iteritems(),
key=lambda x: link_sort(x[0])): key=lambda x: link_sort(x[0])):
@ -179,12 +185,13 @@ def build_config_from_json(tgen, topo, save_bkup=True):
data = topo["routers"] data = topo["routers"]
for func_type in func_dict.keys(): for func_type in func_dict.keys():
logger.info('Building configuration for {}'.format(func_type)) logger.info('Checking for {} configuration in input data'.format(
func_type))
func_dict.get(func_type)(tgen, data, build=True) func_dict.get(func_type)(tgen, data, build=True)
for router in sorted(topo['routers'].keys()): for router in sorted(topo['routers'].keys()):
logger.info('Configuring router {}...'.format(router)) logger.debug('Configuring router {}...'.format(router))
result = load_config_to_router(tgen, router, save_bkup) result = load_config_to_router(tgen, router, save_bkup)
if not result: if not result:

2
tests/topotests/pytest.ini
View File

@ -15,7 +15,7 @@ norecursedirs = .git example-test lib docker
# Display router current configuration during test execution, # Display router current configuration during test execution,
# by default configuration will not be shown # by default configuration will not be shown
show_router_config = True # show_router_config = True
# Default daemons binaries path. # Default daemons binaries path.
#frrdir = /usr/lib/frr #frrdir = /usr/lib/frr