diff options
Diffstat (limited to 'tests/topotests/ospf-topo1/test_ospf_topo1.py')
| -rwxr-xr-x | tests/topotests/ospf-topo1/test_ospf_topo1.py | 389 |
1 files changed, 204 insertions, 185 deletions
diff --git a/tests/topotests/ospf-topo1/test_ospf_topo1.py b/tests/topotests/ospf-topo1/test_ospf_topo1.py index 638e394153..d734f378e7 100755 --- a/tests/topotests/ospf-topo1/test_ospf_topo1.py +++ b/tests/topotests/ospf-topo1/test_ospf_topo1.py @@ -34,7 +34,7 @@ import pytest # 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 @@ -45,70 +45,71 @@ from lib.topolog import logger # Required to instantiate the topology builder class. from mininet.topo import Topo + class OSPFTopo(Topo): "Test topology builder" + def build(self, *_args, **_opts): "Build function" tgen = get_topogen(self) # Create 4 routers for routern in range(1, 5): - tgen.add_router('r{}'.format(routern)) + tgen.add_router("r{}".format(routern)) # Create a empty network for router 1 - switch = tgen.add_switch('s1') - switch.add_link(tgen.gears['r1']) + switch = tgen.add_switch("s1") + switch.add_link(tgen.gears["r1"]) # Create a empty network for router 2 - switch = tgen.add_switch('s2') - switch.add_link(tgen.gears['r2']) + switch = tgen.add_switch("s2") + switch.add_link(tgen.gears["r2"]) # Interconect router 1, 2 and 3 - switch = tgen.add_switch('s3') - switch.add_link(tgen.gears['r1']) - switch.add_link(tgen.gears['r2']) - switch.add_link(tgen.gears['r3']) + switch = tgen.add_switch("s3") + switch.add_link(tgen.gears["r1"]) + switch.add_link(tgen.gears["r2"]) + switch.add_link(tgen.gears["r3"]) # Create empty netowrk for router3 - switch = tgen.add_switch('s4') - switch.add_link(tgen.gears['r3']) + switch = tgen.add_switch("s4") + switch.add_link(tgen.gears["r3"]) # Interconect router 3 and 4 - switch = tgen.add_switch('s5') - switch.add_link(tgen.gears['r3']) - switch.add_link(tgen.gears['r4']) + switch = tgen.add_switch("s5") + switch.add_link(tgen.gears["r3"]) + switch.add_link(tgen.gears["r4"]) # Create a empty network for router 4 - switch = tgen.add_switch('s6') - switch.add_link(tgen.gears['r4']) + switch = tgen.add_switch("s6") + switch.add_link(tgen.gears["r4"]) + def setup_module(mod): "Sets up the pytest environment" tgen = Topogen(OSPFTopo, mod.__name__) tgen.start_topology() - ospf6_config = 'ospf6d.conf' - if tgen.gears['r1'].has_version('<', '4.0'): - ospf6_config = 'ospf6d.conf-pre-v4' + ospf6_config = "ospf6d.conf" + if tgen.gears["r1"].has_version("<", "4.0"): + ospf6_config = "ospf6d.conf-pre-v4" router_list = tgen.routers() 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)) ) 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_OSPF6, - os.path.join(CWD, '{}/{}'.format(rname, ospf6_config)) + TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/{}".format(rname, ospf6_config)) ) # Initialize all routers. tgen.start_router() + def teardown_module(mod): "Teardown the pytest environment" tgen = get_topogen() @@ -121,46 +122,50 @@ def compare_show_ipv6_ospf6(rname, expected): result with the expected output. """ tgen = get_topogen() - current = tgen.gears[rname].vtysh_cmd('show ipv6 ospf6 route') + current = tgen.gears[rname].vtysh_cmd("show ipv6 ospf6 route") # Remove the link addresses - current = re.sub(r'fe80::[^ ]+', 'fe80::xxxx:xxxx:xxxx:xxxx', current) - expected = re.sub(r'fe80::[^ ]+', 'fe80::xxxx:xxxx:xxxx:xxxx', expected) + current = re.sub(r"fe80::[^ ]+", "fe80::xxxx:xxxx:xxxx:xxxx", current) + expected = re.sub(r"fe80::[^ ]+", "fe80::xxxx:xxxx:xxxx:xxxx", expected) # Remove the time - current = re.sub(r'\d+:\d{2}:\d{2}', '', current) - expected = re.sub(r'\d+:\d{2}:\d{2}', '', expected) + current = re.sub(r"\d+:\d{2}:\d{2}", "", current) + expected = re.sub(r"\d+:\d{2}:\d{2}", "", expected) + + return topotest.difflines( + topotest.normalize_text(current), + topotest.normalize_text(expected), + title1="Current output", + title2="Expected output", + ) - return topotest.difflines(topotest.normalize_text(current), - topotest.normalize_text(expected), - title1="Current output", - title2="Expected output") def test_ospf_convergence(): "Test OSPF daemon convergence" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") for router, rnode in tgen.routers().iteritems(): logger.info('Waiting for router "%s" convergence', router) # Load expected results from the command - reffile = os.path.join(CWD, '{}/ospfroute.txt'.format(router)) + reffile = os.path.join(CWD, "{}/ospfroute.txt".format(router)) expected = open(reffile).read() # Run test function until we get an result. Wait at most 80 seconds. test_func = partial( - topotest.router_output_cmp, rnode, 'show ip ospf route', expected) - result, diff = topotest.run_and_expect(test_func, '', - count=160, wait=0.5) - assert result, 'OSPF did not converge on {}:\n{}'.format(router, diff) + topotest.router_output_cmp, rnode, "show ip ospf route", expected + ) + result, diff = topotest.run_and_expect(test_func, "", count=160, wait=0.5) + assert result, "OSPF did not converge on {}:\n{}".format(router, diff) + def test_ospf_kernel_route(): "Test OSPF kernel route installation" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") rlist = tgen.routers().values() for router in rlist: @@ -168,25 +173,26 @@ def test_ospf_kernel_route(): routes = topotest.ip4_route(router) expected = { - '10.0.1.0/24': {}, - '10.0.2.0/24': {}, - '10.0.3.0/24': {}, - '10.0.10.0/24': {}, - '172.16.0.0/24': {}, - '172.16.1.0/24': {}, + "10.0.1.0/24": {}, + "10.0.2.0/24": {}, + "10.0.3.0/24": {}, + "10.0.10.0/24": {}, + "172.16.0.0/24": {}, + "172.16.1.0/24": {}, } assertmsg = 'OSPF IPv4 route mismatch in router "{}"'.format(router.name) assert topotest.json_cmp(routes, expected) is None, assertmsg + def test_ospf6_convergence(): "Test OSPF6 daemon convergence" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") - ospf6route_file = '{}/ospf6route_ecmp.txt' + ospf6route_file = "{}/ospf6route_ecmp.txt" for rnum in range(1, 5): - router = 'r{}'.format(rnum) + router = "r{}".format(rnum) logger.info('Waiting for router "%s" IPv6 OSPF convergence', router) @@ -196,39 +202,37 @@ def test_ospf6_convergence(): # Run test function until we get an result. Wait at most 60 seconds. test_func = partial(compare_show_ipv6_ospf6, router, expected) - result, diff = topotest.run_and_expect(test_func, '', - count=25, wait=3) + result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3) if (not result) and (rnum == 1): # Didn't match the new ECMP version - try the old pre-ECMP format - ospf6route_file = '{}/ospf6route.txt' + ospf6route_file = "{}/ospf6route.txt" # Load expected results from the command reffile = os.path.join(CWD, ospf6route_file.format(router)) expected = open(reffile).read() test_func = partial(compare_show_ipv6_ospf6, router, expected) - result, diff = topotest.run_and_expect(test_func, '', - count=1, wait=3) + result, diff = topotest.run_and_expect(test_func, "", count=1, wait=3) if not result: # Didn't match the old version - switch back to new ECMP version # and fail - ospf6route_file = '{}/ospf6route_ecmp.txt' + ospf6route_file = "{}/ospf6route_ecmp.txt" # Load expected results from the command reffile = os.path.join(CWD, ospf6route_file.format(router)) expected = open(reffile).read() test_func = partial(compare_show_ipv6_ospf6, router, expected) - result, diff = topotest.run_and_expect(test_func, '', - count=1, wait=3) + result, diff = topotest.run_and_expect(test_func, "", count=1, wait=3) + + assert result, "OSPF6 did not converge on {}:\n{}".format(router, diff) - assert result, 'OSPF6 did not converge on {}:\n{}'.format(router, diff) def test_ospf6_kernel_route(): "Test OSPF kernel route installation" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") rlist = tgen.routers().values() for router in rlist: @@ -236,216 +240,231 @@ def test_ospf6_kernel_route(): routes = topotest.ip6_route(router) expected = { - '2001:db8:1::/64': {}, - '2001:db8:2::/64': {}, - '2001:db8:3::/64': {}, - '2001:db8:100::/64': {}, - '2001:db8:200::/64': {}, - '2001:db8:300::/64': {}, + "2001:db8:1::/64": {}, + "2001:db8:2::/64": {}, + "2001:db8:3::/64": {}, + "2001:db8:100::/64": {}, + "2001:db8:200::/64": {}, + "2001:db8:300::/64": {}, } assertmsg = 'OSPF IPv6 route mismatch in router "{}"'.format(router.name) assert topotest.json_cmp(routes, expected) is None, assertmsg + def test_ospf_json(): "Test 'show ip ospf json' output for coherency." tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") for rnum in range(1, 5): - router = tgen.gears['r{}'.format(rnum)] + router = tgen.gears["r{}".format(rnum)] logger.info('Comparing router "%s" "show ip ospf json" output', router.name) expected = { - 'routerId': '10.0.255.{}'.format(rnum), - 'tosRoutesOnly': True, - 'rfc2328Conform': True, - 'spfScheduleDelayMsecs': 0, - 'holdtimeMinMsecs': 50, - 'holdtimeMaxMsecs': 5000, - 'lsaMinIntervalMsecs': 5000, - 'lsaMinArrivalMsecs': 1000, - 'writeMultiplier': 20, - 'refreshTimerMsecs': 10000, - 'asbrRouter': 'injectingExternalRoutingInformation', - 'attachedAreaCounter': 1, - 'areas': {} + "routerId": "10.0.255.{}".format(rnum), + "tosRoutesOnly": True, + "rfc2328Conform": True, + "spfScheduleDelayMsecs": 0, + "holdtimeMinMsecs": 50, + "holdtimeMaxMsecs": 5000, + "lsaMinIntervalMsecs": 5000, + "lsaMinArrivalMsecs": 1000, + "writeMultiplier": 20, + "refreshTimerMsecs": 10000, + "asbrRouter": "injectingExternalRoutingInformation", + "attachedAreaCounter": 1, + "areas": {}, } # Area specific additional checks - if router.name == 'r1' or router.name == 'r2' or router.name == 'r3': - expected['areas']['0.0.0.0'] = { - 'areaIfActiveCounter': 2, - 'areaIfTotalCounter': 2, - 'authentication': 'authenticationNone', - 'backbone': True, - 'lsaAsbrNumber': 1, - 'lsaNetworkNumber': 1, - 'lsaNssaNumber': 0, - 'lsaNumber': 7, - 'lsaOpaqueAreaNumber': 0, - 'lsaOpaqueLinkNumber': 0, - 'lsaRouterNumber': 3, - 'lsaSummaryNumber': 2, - 'nbrFullAdjacentCounter': 2, + if router.name == "r1" or router.name == "r2" or router.name == "r3": + expected["areas"]["0.0.0.0"] = { + "areaIfActiveCounter": 2, + "areaIfTotalCounter": 2, + "authentication": "authenticationNone", + "backbone": True, + "lsaAsbrNumber": 1, + "lsaNetworkNumber": 1, + "lsaNssaNumber": 0, + "lsaNumber": 7, + "lsaOpaqueAreaNumber": 0, + "lsaOpaqueLinkNumber": 0, + "lsaRouterNumber": 3, + "lsaSummaryNumber": 2, + "nbrFullAdjacentCounter": 2, } - if router.name == 'r3' or router.name == 'r4': - expected['areas']['0.0.0.1'] = { - 'areaIfActiveCounter': 1, - 'areaIfTotalCounter': 1, - 'authentication': 'authenticationNone', - 'lsaAsbrNumber': 2, - 'lsaNetworkNumber': 1, - 'lsaNssaNumber': 0, - 'lsaNumber': 9, - 'lsaOpaqueAreaNumber': 0, - 'lsaOpaqueLinkNumber': 0, - 'lsaRouterNumber': 2, - 'lsaSummaryNumber': 4, - 'nbrFullAdjacentCounter': 1, + if router.name == "r3" or router.name == "r4": + expected["areas"]["0.0.0.1"] = { + "areaIfActiveCounter": 1, + "areaIfTotalCounter": 1, + "authentication": "authenticationNone", + "lsaAsbrNumber": 2, + "lsaNetworkNumber": 1, + "lsaNssaNumber": 0, + "lsaNumber": 9, + "lsaOpaqueAreaNumber": 0, + "lsaOpaqueLinkNumber": 0, + "lsaRouterNumber": 2, + "lsaSummaryNumber": 4, + "nbrFullAdjacentCounter": 1, } # r4 has more interfaces for area 0.0.0.1 - if router.name == 'r4': - expected['areas']['0.0.0.1'].update({ - 'areaIfActiveCounter': 2, - 'areaIfTotalCounter': 2, - }) + if router.name == "r4": + expected["areas"]["0.0.0.1"].update( + {"areaIfActiveCounter": 2, "areaIfTotalCounter": 2,} + ) # router 3 has an additional area - if router.name == 'r3': - expected['attachedAreaCounter'] = 2 + if router.name == "r3": + expected["attachedAreaCounter"] = 2 - output = router.vtysh_cmd('show ip ospf json', isjson=True) + output = router.vtysh_cmd("show ip ospf json", isjson=True) result = topotest.json_cmp(output, expected) - assert result is None, '"{}" JSON output mismatches the expected result'.format(router.name) + assert result is None, '"{}" JSON output mismatches the expected result'.format( + router.name + ) + def test_ospf_link_down(): "Test OSPF convergence after a link goes down" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") # Simulate a network down event on router3 switch3 interface. - router3 = tgen.gears['r3'] - router3.peer_link_enable('r3-eth0', False) + router3 = tgen.gears["r3"] + router3.peer_link_enable("r3-eth0", False) # Expect convergence on all routers for router, rnode in tgen.routers().iteritems(): logger.info('Waiting for router "%s" convergence after link failure', router) # Load expected results from the command - reffile = os.path.join(CWD, '{}/ospfroute_down.txt'.format(router)) + reffile = os.path.join(CWD, "{}/ospfroute_down.txt".format(router)) expected = open(reffile).read() # Run test function until we get an result. Wait at most 80 seconds. test_func = partial( - topotest.router_output_cmp, rnode, 'show ip ospf route', expected) - result, diff = topotest.run_and_expect(test_func, '', - count=140, wait=0.5) - assert result, 'OSPF did not converge on {}:\n{}'.format(router, diff) + topotest.router_output_cmp, rnode, "show ip ospf route", expected + ) + result, diff = topotest.run_and_expect(test_func, "", count=140, wait=0.5) + assert result, "OSPF did not converge on {}:\n{}".format(router, diff) + def test_ospf_link_down_kernel_route(): "Test OSPF kernel route installation" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") rlist = tgen.routers().values() for router in rlist: - logger.info('Checking OSPF IPv4 kernel routes in "%s" after link down', router.name) + logger.info( + 'Checking OSPF IPv4 kernel routes in "%s" after link down', router.name + ) routes = topotest.ip4_route(router) expected = { - '10.0.1.0/24': {}, - '10.0.2.0/24': {}, - '10.0.3.0/24': {}, - '10.0.10.0/24': {}, - '172.16.0.0/24': {}, - '172.16.1.0/24': {}, + "10.0.1.0/24": {}, + "10.0.2.0/24": {}, + "10.0.3.0/24": {}, + "10.0.10.0/24": {}, + "172.16.0.0/24": {}, + "172.16.1.0/24": {}, } - if router.name == 'r1' or router.name == 'r2': - expected.update({ - '10.0.10.0/24': None, - '172.16.0.0/24': None, - '172.16.1.0/24': None, - }) - elif router.name == 'r3' or router.name == 'r4': - expected.update({ - '10.0.1.0/24': None, - '10.0.2.0/24': None, - }) + if router.name == "r1" or router.name == "r2": + expected.update( + {"10.0.10.0/24": None, "172.16.0.0/24": None, "172.16.1.0/24": None,} + ) + elif router.name == "r3" or router.name == "r4": + expected.update( + {"10.0.1.0/24": None, "10.0.2.0/24": None,} + ) # Route '10.0.3.0' is no longer available for r4 since it is down. - if router.name == 'r4': - expected.update({ - '10.0.3.0/24': None, - }) - assertmsg = 'OSPF IPv4 route mismatch in router "{}" after link down'.format(router.name) + if router.name == "r4": + expected.update( + {"10.0.3.0/24": None,} + ) + assertmsg = 'OSPF IPv4 route mismatch in router "{}" after link down'.format( + router.name + ) assert topotest.json_cmp(routes, expected) is None, assertmsg + def test_ospf6_link_down(): "Test OSPF6 daemon convergence after link goes down" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") for rnum in range(1, 5): - router = 'r{}'.format(rnum) + router = "r{}".format(rnum) - logger.info('Waiting for router "%s" IPv6 OSPF convergence after link down', router) + logger.info( + 'Waiting for router "%s" IPv6 OSPF convergence after link down', router + ) # Load expected results from the command - reffile = os.path.join(CWD, '{}/ospf6route_down.txt'.format(router)) + reffile = os.path.join(CWD, "{}/ospf6route_down.txt".format(router)) expected = open(reffile).read() # Run test function until we get an result. Wait at most 60 seconds. test_func = partial(compare_show_ipv6_ospf6, router, expected) - result, diff = topotest.run_and_expect(test_func, '', - count=25, wait=3) - assert result, 'OSPF6 did not converge on {}:\n{}'.format(router, diff) + result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3) + assert result, "OSPF6 did not converge on {}:\n{}".format(router, diff) + def test_ospf6_link_down_kernel_route(): "Test OSPF kernel route installation" tgen = get_topogen() if tgen.routers_have_failure(): - pytest.skip('skipped because of router(s) failure') + pytest.skip("skipped because of router(s) failure") rlist = tgen.routers().values() for router in rlist: - logger.info('Checking OSPF IPv6 kernel routes in "%s" after link down', router.name) + logger.info( + 'Checking OSPF IPv6 kernel routes in "%s" after link down', router.name + ) routes = topotest.ip6_route(router) expected = { - '2001:db8:1::/64': {}, - '2001:db8:2::/64': {}, - '2001:db8:3::/64': {}, - '2001:db8:100::/64': {}, - '2001:db8:200::/64': {}, - '2001:db8:300::/64': {}, + "2001:db8:1::/64": {}, + "2001:db8:2::/64": {}, + "2001:db8:3::/64": {}, + "2001:db8:100::/64": {}, + "2001:db8:200::/64": {}, + "2001:db8:300::/64": {}, } - if router.name == 'r1' or router.name == 'r2': - expected.update({ - '2001:db8:100::/64': None, - '2001:db8:200::/64': None, - '2001:db8:300::/64': None, - }) - elif router.name == 'r3' or router.name == 'r4': - expected.update({ - '2001:db8:1::/64': None, - '2001:db8:2::/64': None, - }) + if router.name == "r1" or router.name == "r2": + expected.update( + { + "2001:db8:100::/64": None, + "2001:db8:200::/64": None, + "2001:db8:300::/64": None, + } + ) + elif router.name == "r3" or router.name == "r4": + expected.update( + {"2001:db8:1::/64": None, "2001:db8:2::/64": None,} + ) # Route '2001:db8:3::/64' is no longer available for r4 since it is down. - if router.name == 'r4': - expected.update({ - '2001:db8:3::/64': None, - }) - assertmsg = 'OSPF IPv6 route mismatch in router "{}" after link down'.format(router.name) + if router.name == "r4": + expected.update( + {"2001:db8:3::/64": None,} + ) + assertmsg = 'OSPF IPv6 route mismatch in router "{}" after link down'.format( + router.name + ) assert topotest.json_cmp(routes, expected) is None, assertmsg + 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)) |