diff options
Diffstat (limited to 'tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py')
| -rwxr-xr-x | tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py | 120 |
1 files changed, 72 insertions, 48 deletions
diff --git a/tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py b/tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py index 0fae64402a..600d640a70 100755 --- a/tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py +++ b/tests/topotests/ldp-vpls-topo1/test_ldp_vpls_topo1.py @@ -69,7 +69,7 @@ from functools import partial # 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, "../")) # pylint: disable=C0413 # Import topogen and topotest helpers @@ -80,8 +80,10 @@ from lib.topolog import logger # Required to instantiate the topology builder class. from mininet.topo import Topo + class TemplateTopo(Topo): "Test topology builder" + def build(self, *_args, **_opts): "Build function" tgen = get_topogen(self) @@ -89,35 +91,36 @@ class TemplateTopo(Topo): # # Define FRR Routers # - for router in ['ce1', 'ce2', 'ce3', 'r1', 'r2', 'r3']: + for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]: tgen.add_router(router) # # Define connections # - switch = tgen.add_switch('s1') - switch.add_link(tgen.gears['ce1']) - switch.add_link(tgen.gears['r1']) + switch = tgen.add_switch("s1") + switch.add_link(tgen.gears["ce1"]) + switch.add_link(tgen.gears["r1"]) + + switch = tgen.add_switch("s2") + switch.add_link(tgen.gears["ce2"]) + switch.add_link(tgen.gears["r2"]) - switch = tgen.add_switch('s2') - switch.add_link(tgen.gears['ce2']) - switch.add_link(tgen.gears['r2']) + switch = tgen.add_switch("s3") + switch.add_link(tgen.gears["ce3"]) + switch.add_link(tgen.gears["r3"]) - switch = tgen.add_switch('s3') - switch.add_link(tgen.gears['ce3']) - switch.add_link(tgen.gears['r3']) + switch = tgen.add_switch("s4") + switch.add_link(tgen.gears["r1"]) + switch.add_link(tgen.gears["r2"]) - switch = tgen.add_switch('s4') - switch.add_link(tgen.gears['r1']) - switch.add_link(tgen.gears['r2']) + switch = tgen.add_switch("s5") + switch.add_link(tgen.gears["r1"]) + switch.add_link(tgen.gears["r3"]) - switch = tgen.add_switch('s5') - switch.add_link(tgen.gears['r1']) - switch.add_link(tgen.gears['r3']) + switch = tgen.add_switch("s6") + switch.add_link(tgen.gears["r2"]) + switch.add_link(tgen.gears["r3"]) - switch = tgen.add_switch('s6') - switch.add_link(tgen.gears['r2']) - switch.add_link(tgen.gears['r3']) def setup_module(mod): "Sets up the pytest environment" @@ -129,22 +132,20 @@ def setup_module(mod): # For all registered routers, load the zebra configuration file for rname, router in router_list.iteritems(): router.load_config( - TopoRouter.RD_ZEBRA, - os.path.join(CWD, '{}/zebra.conf'.format(rname)) + TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) ) # Don't start ospfd and ldpd in the CE nodes - if router.name[0] == 'r': + if router.name[0] == "r": router.load_config( - TopoRouter.RD_OSPF, - os.path.join(CWD, '{}/ospfd.conf'.format(rname)) + TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname)) ) router.load_config( - TopoRouter.RD_LDP, - os.path.join(CWD, '{}/ldpd.conf'.format(rname)) + TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname)) ) tgen.start_router() + def teardown_module(mod): "Teardown the pytest environment" tgen = get_topogen() @@ -159,16 +160,16 @@ def router_compare_json_output(rname, command, reference): logger.info('Comparing router "%s" "%s" output', rname, command) tgen = get_topogen() - filename = '{}/{}/{}'.format(CWD, rname, reference) + filename = "{}/{}/{}".format(CWD, rname, reference) expected = json.loads(open(filename).read()) # Run test function until we get an result. Wait at most 80 seconds. - test_func = partial(topotest.router_json_cmp, - tgen.gears[rname], command, expected) + test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected) _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5) assertmsg = '"{}" JSON output mismatches the expected result'.format(rname) assert diff is None, assertmsg + def test_ospf_convergence(): logger.info("Test: check OSPF adjacencies") tgen = get_topogen() @@ -177,8 +178,11 @@ def test_ospf_convergence(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json" + ) + def test_rib(): logger.info("Test: verify RIB") @@ -188,9 +192,10 @@ def test_rib(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: + for rname in ["r1", "r2", "r3"]: router_compare_json_output(rname, "show ip route json", "show_ip_route.ref") + def test_ldp_adjacencies(): logger.info("Test: verify LDP adjacencies") tgen = get_topogen() @@ -199,8 +204,11 @@ def test_ldp_adjacencies(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show mpls ldp discovery json", "show_ldp_discovery.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show mpls ldp discovery json", "show_ldp_discovery.ref" + ) + def test_ldp_neighbors(): logger.info("Test: verify LDP neighbors") @@ -210,8 +218,11 @@ def test_ldp_neighbors(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref" + ) + def test_ldp_bindings(): logger.info("Test: verify LDP bindings") @@ -221,8 +232,11 @@ def test_ldp_bindings(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show mpls ldp binding json", "show_ldp_binding.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show mpls ldp binding json", "show_ldp_binding.ref" + ) + def test_ldp_pwid_bindings(): logger.info("Test: verify LDP PW-ID bindings") @@ -232,8 +246,11 @@ def test_ldp_pwid_bindings(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref" + ) + def test_ldp_pseudowires(): logger.info("Test: verify LDP pseudowires") @@ -243,8 +260,11 @@ def test_ldp_pseudowires(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref" + ) + def test_ldp_pseudowires_after_link_down(): logger.info("Test: verify LDP pseudowires after r1-r2 link goes down") @@ -256,22 +276,26 @@ def test_ldp_pseudowires_after_link_down(): # Shut down r1-r2 link */ tgen = get_topogen() - tgen.gears['r1'].peer_link_enable('r1-eth1', False) + tgen.gears["r1"].peer_link_enable("r1-eth1", False) topotest.sleep(5, "Waiting for the network to reconverge") # check if the pseudowire is still up (using an alternate path for nexthop resolution) - for rname in ['r1', 'r2', 'r3']: - router_compare_json_output(rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref") + for rname in ["r1", "r2", "r3"]: + router_compare_json_output( + rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref" + ) + # Memory leak test template def test_memory_leak(): "Run the memory leak test and report results." tgen = get_topogen() if not tgen.is_memleak_enabled(): - pytest.skip('Memory leak test/report is disabled') + pytest.skip("Memory leak test/report is disabled") tgen.report_memory_leaks() -if __name__ == '__main__': + +if __name__ == "__main__": args = ["-s"] + sys.argv[1:] sys.exit(pytest.main(args)) |