diff options
Diffstat (limited to 'tests/topotests/pim-basic/test_pim.py')
| -rw-r--r-- | tests/topotests/pim-basic/test_pim.py | 127 |
1 files changed, 70 insertions, 57 deletions
diff --git a/tests/topotests/pim-basic/test_pim.py b/tests/topotests/pim-basic/test_pim.py index 9101d7e035..2abee39176 100644 --- a/tests/topotests/pim-basic/test_pim.py +++ b/tests/topotests/pim-basic/test_pim.py @@ -32,7 +32,7 @@ import json from functools import partial 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 from lib import topotest @@ -41,15 +41,16 @@ from lib.topolog import logger from mininet.topo import Topo + class PIMTopo(Topo): def build(self, *_args, **_opts): "Build function" tgen = get_topogen(self) for routern in range(1, 4): - tgen.add_router('r{}'.format(routern)) + tgen.add_router("r{}".format(routern)) - tgen.add_router('rp') + tgen.add_router("rp") # rp ------ r1 -------- r2 # \ @@ -63,21 +64,22 @@ class PIMTopo(Topo): # r1 <- sw1 -> r2 # r1-eth0 <-> r2-eth0 # 10.0.20.0/24 - sw = tgen.add_switch('sw1') - sw.add_link(tgen.gears['r1']) - sw.add_link(tgen.gears['r2']) + sw = tgen.add_switch("sw1") + sw.add_link(tgen.gears["r1"]) + sw.add_link(tgen.gears["r2"]) # r1 <- sw2 -> rp # r1-eth1 <-> rp-eth0 # 10.0.30.0/24 - sw = tgen.add_switch('sw2') - sw.add_link(tgen.gears['r1']) - sw.add_link(tgen.gears['rp']) + sw = tgen.add_switch("sw2") + sw.add_link(tgen.gears["r1"]) + sw.add_link(tgen.gears["rp"]) # 10.0.40.0/24 - sw = tgen.add_switch('sw3') - sw.add_link(tgen.gears['r1']) - sw.add_link(tgen.gears['r3']) + sw = tgen.add_switch("sw3") + sw.add_link(tgen.gears["r1"]) + sw.add_link(tgen.gears["r3"]) + def setup_module(mod): "Sets up the pytest environment" @@ -87,21 +89,18 @@ def setup_module(mod): # For all registered routers, load the zebra configuration file for rname, router in tgen.routers().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_PIM, - os.path.join(CWD, '{}/pimd.conf'.format(rname)) + TopoRouter.RD_PIM, os.path.join(CWD, "{}/pimd.conf".format(rname)) ) router.load_config( - TopoRouter.RD_BGP, - os.path.join(CWD, '{}/bgpd.conf'.format(rname)) - ) + TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname)) + ) # After loading the configurations, this function loads configured daemons. tgen.start_router() - #tgen.mininet_cli() + # tgen.mininet_cli() def teardown_module(mod): @@ -111,22 +110,25 @@ def teardown_module(mod): # This function tears down the whole topology. tgen.stop_topology() + def test_pim_rp_setup(): "Ensure basic routing has come up and the rp has an outgoing interface" - #Ensure rp and r1 establish pim neighbor ship and bgp has come up - #Finally ensure that the rp has an outgoing interface on r1 + # Ensure rp and r1 establish pim neighbor ship and bgp has come up + # Finally ensure that the rp has an outgoing interface on r1 tgen = get_topogen() - r1 = tgen.gears['r1'] - json_file = '{}/{}/rp-info.json'.format(CWD, r1.name) + r1 = tgen.gears["r1"] + json_file = "{}/{}/rp-info.json".format(CWD, r1.name) expected = json.loads(open(json_file).read()) - test_func = partial(topotest.router_json_cmp, - r1, 'show ip pim rp-info json', expected) + test_func = partial( + topotest.router_json_cmp, r1, "show ip pim rp-info json", expected + ) _, result = topotest.run_and_expect(test_func, None, count=15, wait=5) assertmsg = '"{}" JSON output mismatches'.format(r1.name) assert result is None, assertmsg - #tgen.mininet_cli() + # tgen.mininet_cli() + def test_pim_send_mcast_stream(): "Establish a Multicast stream from r2 -> r1 and then ensure S,G is created as appropriate" @@ -137,47 +139,58 @@ def test_pim_send_mcast_stream(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - rp = tgen.gears['rp'] - r3 = tgen.gears['r3'] - r2 = tgen.gears['r2'] - r1 = tgen.gears['r1'] + rp = tgen.gears["rp"] + r3 = tgen.gears["r3"] + r2 = tgen.gears["r2"] + r1 = tgen.gears["r1"] # Let's establish a S,G stream from r2 -> r1 CWD = os.path.dirname(os.path.realpath(__file__)) - r2.run("{}/mcast-tx.py --ttl 5 --count 5 --interval 10 229.1.1.1 r2-eth0 > /tmp/bar".format(CWD)) + r2.run( + "{}/mcast-tx.py --ttl 5 --count 5 --interval 10 229.1.1.1 r2-eth0 > /tmp/bar".format( + CWD + ) + ) # And from r3 -> r1 - r3.run("{}/mcast-tx.py --ttl 5 --count 5 --interval 10 229.1.1.1 r3-eth0 > /tmp/bar".format(CWD)) + r3.run( + "{}/mcast-tx.py --ttl 5 --count 5 --interval 10 229.1.1.1 r3-eth0 > /tmp/bar".format( + CWD + ) + ) # Let's see that it shows up and we have established some basic state out = r1.vtysh_cmd("show ip pim upstream json", isjson=True) expected = { - '229.1.1.1': { - '10.0.20.2': { - 'firstHopRouter': 1, - 'joinState': 'NotJoined', - 'regState': 'RegPrune', - 'inboundInterface': 'r1-eth0', + "229.1.1.1": { + "10.0.20.2": { + "firstHopRouter": 1, + "joinState": "NotJoined", + "regState": "RegPrune", + "inboundInterface": "r1-eth0", } } } - assert topotest.json_cmp(out, expected) is None, 'failed to converge pim' - #tgen.mininet_cli() + assert topotest.json_cmp(out, expected) is None, "failed to converge pim" + # tgen.mininet_cli() + def test_pim_rp_sees_stream(): "Ensure that the RP sees the stream and has acted accordingly" tgen = get_topogen() - rp = tgen.gears['rp'] - json_file = '{}/{}/upstream.json'.format(CWD, rp.name) + rp = tgen.gears["rp"] + json_file = "{}/{}/upstream.json".format(CWD, rp.name) expected = json.loads(open(json_file).read()) - test_func = partial(topotest.router_json_cmp, - rp, 'show ip pim upstream json', expected) - _, result = topotest.run_and_expect(test_func, None, count=20, wait=.5) + test_func = partial( + topotest.router_json_cmp, rp, "show ip pim upstream json", expected + ) + _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5) assertmsg = '"{}" JSON output mismatches'.format(rp.name) assert result is None, assertmsg + def test_pim_igmp_report(): "Send a igmp report from r2->r1 and ensure that the *,G state is created on r1" logger.info("Send a igmp report from r2-r1 and ensure *,G created") @@ -187,8 +200,8 @@ def test_pim_igmp_report(): if tgen.routers_have_failure(): pytest.skip(tgen.errors) - r2 = tgen.gears['r2'] - r1 = tgen.gears['r1'] + r2 = tgen.gears["r2"] + r1 = tgen.gears["r1"] # Let's send a igmp report from r2->r1 CWD = os.path.dirname(os.path.realpath(__file__)) @@ -196,28 +209,28 @@ def test_pim_igmp_report(): out = r1.vtysh_cmd("show ip pim upstream json", isjson=True) expected = { - '229.1.1.2': { - '*': { - 'sourceIgmp': 1, - 'joinState': 'Joined', - 'regState': 'RegNoInfo', - 'sptBit': 0, + "229.1.1.2": { + "*": { + "sourceIgmp": 1, + "joinState": "Joined", + "regState": "RegNoInfo", + "sptBit": 0, } } } - assert topotest.json_cmp(out, expected) is None, 'failed to converge pim' + assert topotest.json_cmp(out, expected) is None, "failed to converge pim" 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)) |