diff options
Diffstat (limited to 'tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py')
| -rwxr-xr-x | tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py | 294 |
1 files changed, 110 insertions, 184 deletions
diff --git a/tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py b/tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py index 43639a81d1..7a74b62980 100755 --- a/tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py +++ b/tests/topotests/bgp-basic-functionality-topo1/test_bgp_basic_functionality.py @@ -48,8 +48,8 @@ from copy import deepcopy # Save the Current Working Directory to find configuration files. CWD = os.path.dirname(os.path.realpath(__file__)) -sys.path.append(os.path.join(CWD, '../')) -sys.path.append(os.path.join(CWD, '../lib/')) +sys.path.append(os.path.join(CWD, "../")) +sys.path.append(os.path.join(CWD, "../lib/")) # Required to instantiate the topology builder class. @@ -59,30 +59,39 @@ from lib.topogen import Topogen, get_topogen from mininet.topo import Topo from lib.common_config import ( - start_topology, write_test_header, - write_test_footer, reset_config_on_routers, create_static_routes, - verify_rib, verify_admin_distance_for_static_routes + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + create_static_routes, + verify_rib, + verify_admin_distance_for_static_routes, ) from lib.topolog import logger from lib.bgp import ( - verify_bgp_convergence, create_router_bgp, verify_router_id, - modify_as_number, verify_as_numbers, clear_bgp_and_verify, - verify_bgp_timers_and_functionality + verify_bgp_convergence, + create_router_bgp, + verify_router_id, + modify_as_number, + verify_as_numbers, + clear_bgp_and_verify, + verify_bgp_timers_and_functionality, ) from lib.topojson import build_topo_from_json, build_config_from_json # Reading the data from JSON File for topology creation jsonFile = "{}/bgp_basic_functionality.json".format(CWD) try: - with open(jsonFile, 'r') as topoJson: + with open(jsonFile, "r") as topoJson: topo = json.load(topoJson) except IOError: assert False, "Could not read file {}".format(jsonFile) -#Global Variable +# Global Variable KEEPALIVETIMER = 2 HOLDDOWNTIMER = 6 + class CreateTopo(Topo): """ Test BasicTopo - topology 1 @@ -124,8 +133,9 @@ def setup_module(mod): global BGP_CONVERGENCE BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo) - assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}". \ - format(BGP_CONVERGENCE) + assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error: {}".format( + BGP_CONVERGENCE + ) logger.info("Running setup_module() done") @@ -140,8 +150,9 @@ def teardown_module(): # Stop toplogy and Remove tmp files tgen.stop_topology() - logger.info("Testsuite end time: {}". - format(time.asctime(time.localtime(time.time())))) + logger.info( + "Testsuite end time: {}".format(time.asctime(time.localtime(time.time()))) + ) logger.info("=" * 40) @@ -157,7 +168,7 @@ def test_modify_and_delete_router_id(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -165,59 +176,31 @@ def test_modify_and_delete_router_id(request): # Modify router id input_dict = { - 'r1': { - "bgp": { - 'router_id': '12.12.12.12' - } - }, - 'r2': { - "bgp": { - 'router_id': '22.22.22.22' - } - }, - 'r3': { - "bgp": { - 'router_id': '33.33.33.33' - } - }, + "r1": {"bgp": {"router_id": "12.12.12.12"}}, + "r2": {"bgp": {"router_id": "22.22.22.22"}}, + "r3": {"bgp": {"router_id": "33.33.33.33"}}, } result = create_router_bgp(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}".\ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Verifying router id once modified result = verify_router_id(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}".\ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Delete router id input_dict = { - 'r1': { - "bgp": { - 'del_router_id': True - } - }, - 'r2': { - "bgp": { - 'del_router_id': True - } - }, - 'r3': { - "bgp": { - 'del_router_id': True - } - }, + "r1": {"bgp": {"del_router_id": True}}, + "r2": {"bgp": {"del_router_id": True}}, + "r3": {"bgp": {"del_router_id": True}}, } result = create_router_bgp(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Verifying router id once deleted # Once router-id is deleted, highest interface ip should become # router-id result = verify_router_id(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -229,41 +212,23 @@ def test_bgp_config_with_4byte_as_number(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name write_test_header(tc_name) input_dict = { - "r1": { - "bgp": { - "local_as": 131079 - } - }, - "r2": { - "bgp": { - "local_as": 131079 - } - }, - "r3": { - "bgp": { - "local_as": 131079 - } - }, - "r4": { - "bgp": { - "local_as": 131080 - } - } + "r1": {"bgp": {"local_as": 131079}}, + "r2": {"bgp": {"local_as": 131079}}, + "r3": {"bgp": {"local_as": 131079}}, + "r4": {"bgp": {"local_as": 131080}}, } result = modify_as_number(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) result = verify_as_numbers(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -275,7 +240,7 @@ def test_bgp_timers_functionality(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -293,10 +258,10 @@ def test_bgp_timers_functionality(request): "unicast": { "neighbor": { "r2": { - "dest_link":{ + "dest_link": { "r1": { "keepalivetimer": KEEPALIVETIMER, - "holddowntimer": HOLDDOWNTIMER + "holddowntimer": HOLDDOWNTIMER, } } } @@ -308,16 +273,14 @@ def test_bgp_timers_functionality(request): } } result = create_router_bgp(tgen, topo, deepcopy(input_dict)) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Api call to clear bgp, so timer modification would take place - clear_bgp_and_verify(tgen, topo, 'r1') + clear_bgp_and_verify(tgen, topo, "r1") # Verifying bgp timers functionality result = verify_bgp_timers_and_functionality(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -327,7 +290,7 @@ def test_static_routes(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -339,17 +302,18 @@ def test_static_routes(request): # Api call to create static routes input_dict = { "r1": { - "static_routes": [{ - "network": "10.0.20.1/32", - "no_of_ip": 9, - "admin_distance": 100, - "next_hop": "10.0.0.2" - }] + "static_routes": [ + { + "network": "10.0.20.1/32", + "no_of_ip": 9, + "admin_distance": 100, + "next_hop": "10.0.0.2", + } + ] } } result = create_static_routes(tgen, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Api call to redistribute static routes input_dict_1 = { @@ -360,7 +324,7 @@ def test_static_routes(request): "unicast": { "redistribute": [ {"redist_type": "static"}, - {"redist_type": "connected"} + {"redist_type": "connected"}, ] } } @@ -370,17 +334,16 @@ def test_static_routes(request): } result = create_router_bgp(tgen, topo, input_dict_1) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Verifying RIB routes - dut = 'r3' - protocol = 'bgp' - next_hop = ['10.0.0.2', '10.0.0.5'] - result = verify_rib(tgen, 'ipv4', dut, input_dict, next_hop=next_hop, - protocol=protocol) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + dut = "r3" + protocol = "bgp" + next_hop = ["10.0.0.2", "10.0.0.5"] + result = verify_rib( + tgen, "ipv4", dut, input_dict, next_hop=next_hop, protocol=protocol + ) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -390,7 +353,7 @@ def test_admin_distance_for_existing_static_routes(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -401,21 +364,21 @@ def test_admin_distance_for_existing_static_routes(request): input_dict = { "r1": { - "static_routes": [{ - "network": "10.0.20.1/32", - "admin_distance": 10, - "next_hop": "10.0.0.2" - }] + "static_routes": [ + { + "network": "10.0.20.1/32", + "admin_distance": 10, + "next_hop": "10.0.0.2", + } + ] } } result = create_static_routes(tgen, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Verifying admin distance once modified result = verify_admin_distance_for_static_routes(tgen, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -425,7 +388,7 @@ def test_advertise_network_using_network_command(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -442,14 +405,8 @@ def test_advertise_network_using_network_command(request): "ipv4": { "unicast": { "advertise_networks": [ - { - "network": "20.0.0.0/32", - "no_of_network": 10 - }, - { - "network": "30.0.0.0/32", - "no_of_network": 10 - } + {"network": "20.0.0.0/32", "no_of_network": 10}, + {"network": "30.0.0.0/32", "no_of_network": 10}, ] } } @@ -459,15 +416,13 @@ def test_advertise_network_using_network_command(request): } result = create_router_bgp(tgen, topo, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Verifying RIB routes - dut = 'r2' + dut = "r2" protocol = "bgp" - result = verify_rib(tgen, 'ipv4', dut, input_dict, protocol=protocol) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + result = verify_rib(tgen, "ipv4", dut, input_dict, protocol=protocol) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -480,7 +435,7 @@ def test_clear_bgp_and_verify(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -490,9 +445,8 @@ def test_clear_bgp_and_verify(request): reset_config_on_routers(tgen) # clear ip bgp - result = clear_bgp_and_verify(tgen, topo, 'r1') - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + result = clear_bgp_and_verify(tgen, topo, "r1") + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) @@ -509,7 +463,7 @@ def test_bgp_with_loopback_interface(request): tgen = get_topogen() if BGP_CONVERGENCE is not True: - pytest.skip('skipped because of BGP Convergence failure') + pytest.skip("skipped because of BGP Convergence failure") # test case name tc_name = request.node.name @@ -518,79 +472,51 @@ def test_bgp_with_loopback_interface(request): # Creating configuration from JSON reset_config_on_routers(tgen) - for routerN in sorted(topo['routers'].keys()): - for bgp_neighbor in \ - topo['routers'][routerN]['bgp']['address_family']['ipv4'][ - 'unicast']['neighbor'].keys(): + for routerN in sorted(topo["routers"].keys()): + for bgp_neighbor in topo["routers"][routerN]["bgp"]["address_family"]["ipv4"][ + "unicast" + ]["neighbor"].keys(): # Adding ['source_link'] = 'lo' key:value pair - topo['routers'][routerN]['bgp']['address_family']['ipv4'][ - 'unicast']['neighbor'][bgp_neighbor]["dest_link"] = { - 'lo': { - "source_link": "lo", - } - } + topo["routers"][routerN]["bgp"]["address_family"]["ipv4"]["unicast"][ + "neighbor" + ][bgp_neighbor]["dest_link"] = {"lo": {"source_link": "lo",}} # Creating configuration from JSON build_config_from_json(tgen, topo) input_dict = { "r1": { - "static_routes": [{ - "network": "1.0.2.17/32", - "next_hop": "10.0.0.2" - }, - { - "network": "1.0.3.17/32", - "next_hop": "10.0.0.6" - } + "static_routes": [ + {"network": "1.0.2.17/32", "next_hop": "10.0.0.2"}, + {"network": "1.0.3.17/32", "next_hop": "10.0.0.6"}, ] }, "r2": { - "static_routes": [{ - "network": "1.0.1.17/32", - "next_hop": "10.0.0.1" - }, - { - "network": "1.0.3.17/32", - "next_hop": "10.0.0.10" - } + "static_routes": [ + {"network": "1.0.1.17/32", "next_hop": "10.0.0.1"}, + {"network": "1.0.3.17/32", "next_hop": "10.0.0.10"}, ] }, "r3": { - "static_routes": [{ - "network": "1.0.1.17/32", - "next_hop": "10.0.0.5" - }, - { - "network": "1.0.2.17/32", - "next_hop": "10.0.0.9" - }, - { - "network": "1.0.4.17/32", - "next_hop": "10.0.0.14" - } + "static_routes": [ + {"network": "1.0.1.17/32", "next_hop": "10.0.0.5"}, + {"network": "1.0.2.17/32", "next_hop": "10.0.0.9"}, + {"network": "1.0.4.17/32", "next_hop": "10.0.0.14"}, ] }, - "r4": { - "static_routes": [{ - "network": "1.0.3.17/32", - "next_hop": "10.0.0.13" - }] - } + "r4": {"static_routes": [{"network": "1.0.3.17/32", "next_hop": "10.0.0.13"}]}, } result = create_static_routes(tgen, input_dict) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) # Api call verify whether BGP is converged result = verify_bgp_convergence(tgen, topo) - assert result is True, "Testcase {} :Failed \n Error: {}". \ - format(tc_name, result) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) write_test_footer(tc_name) -if __name__ == '__main__': +if __name__ == "__main__": args = ["-s"] + sys.argv[1:] sys.exit(pytest.main(args)) |