tests/topotests/pim_basic/test_pim.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_pim.py
   6 #
   7 # Copyright (c) 2018 Cumulus Networks, Inc.
   8 #                    Donald Sharp
   9 #
  10
  11 """
  12 test_pim.py: Test pim
  13 """
  14
  15 import os
  16 import sys
  17 import pytest
  18 import json
  19 from functools import partial
  20
  21 pytestmark = pytest.mark.pimd
  22
  23 CWD = os.path.dirname(os.path.realpath(__file__))
  24 sys.path.append(os.path.join(CWD, "../"))
  25
  26 # pylint: disable=C0413
  27 from lib import topotest
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29 from lib.topolog import logger
  30
  31
  32 pytestmark = [pytest.mark.pimd]
  33
  34
  35 def build_topo(tgen):
  36     "Build function"
  37
  38     for routern in range(1, 4):
  39         tgen.add_router("r{}".format(routern))
  40
  41     tgen.add_router("rp")
  42
  43     #   rp ------ r1 -------- r2
  44     #              \
  45     #               --------- r3
  46     # r1 -> .1
  47     # r2 -> .2
  48     # rp -> .3
  49     # r3 -> .4
  50     # loopback network is 10.254.0.X/32
  51     #
  52     # r1 <- sw1 -> r2
  53     # r1-eth0 <-> r2-eth0
  54     # 10.0.20.0/24
  55     sw = tgen.add_switch("sw1")
  56     sw.add_link(tgen.gears["r1"])
  57     sw.add_link(tgen.gears["r2"])
  58
  59     # r1 <- sw2 -> rp
  60     # r1-eth1 <-> rp-eth0
  61     # 10.0.30.0/24
  62     sw = tgen.add_switch("sw2")
  63     sw.add_link(tgen.gears["r1"])
  64     sw.add_link(tgen.gears["rp"])
  65
  66     # 10.0.40.0/24
  67     sw = tgen.add_switch("sw3")
  68     sw.add_link(tgen.gears["r1"])
  69     sw.add_link(tgen.gears["r3"])
  70
  71
  72 def setup_module(mod):
  73     "Sets up the pytest environment"
  74     tgen = Topogen(build_topo, mod.__name__)
  75     tgen.start_topology()
  76
  77     # For all registered routers, load the zebra configuration file
  78     for rname, router in tgen.routers().items():
  79         router.load_config(
  80             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  81         )
  82         router.load_config(
  83             TopoRouter.RD_PIM, os.path.join(CWD, "{}/pimd.conf".format(rname))
  84         )
  85         router.load_config(
  86             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  87         )
  88
  89     # After loading the configurations, this function loads configured daemons.
  90     tgen.start_router()
  91     # tgen.mininet_cli()
  92
  93
  94 def teardown_module(mod):
  95     "Teardown the pytest environment"
  96     tgen = get_topogen()
  97
  98     # This function tears down the whole topology.
  99     tgen.stop_topology()
 100
 101
 102 def test_pim_rp_setup():
 103     "Ensure basic routing has come up and the rp has an outgoing interface"
 104     # Ensure rp and r1 establish pim neighbor ship and bgp has come up
 105     # Finally ensure that the rp has an outgoing interface on r1
 106     tgen = get_topogen()
 107
 108     r1 = tgen.gears["r1"]
 109     json_file = "{}/{}/rp-info.json".format(CWD, r1.name)
 110     expected = json.loads(open(json_file).read())
 111
 112     test_func = partial(
 113         topotest.router_json_cmp, r1, "show ip pim rp-info json", expected
 114     )
 115     _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
 116     assertmsg = '"{}" JSON output mismatches'.format(r1.name)
 117     assert result is None, assertmsg
 118     # tgen.mininet_cli()
 119
 120
 121 def test_pim_send_mcast_stream():
 122     "Establish a Multicast stream from r2 -> r1 and then ensure S,G is created as appropriate"
 123     logger.info("Establish a Mcast stream from r2->r1 and then ensure S,G created")
 124
 125     tgen = get_topogen()
 126
 127     if tgen.routers_have_failure():
 128         pytest.skip(tgen.errors)
 129
 130     rp = tgen.gears["rp"]
 131     r3 = tgen.gears["r3"]
 132     r2 = tgen.gears["r2"]
 133     r1 = tgen.gears["r1"]
 134
 135     # Let's establish a S,G stream from r2 -> r1
 136     CWD = os.path.dirname(os.path.realpath(__file__))
 137     r2.run(
 138         "{}/mcast-tx.py --ttl 5 --count 40 --interval 2 229.1.1.1 r2-eth0 > /tmp/bar".format(
 139             CWD
 140         )
 141     )
 142     # And from r3 -> r1
 143     r3.run(
 144         "{}/mcast-tx.py --ttl 5 --count 40 --interval 2 229.1.1.1 r3-eth0 > /tmp/bar".format(
 145             CWD
 146         )
 147     )
 148
 149     # Let's see that it shows up and we have established some basic state
 150     out = r1.vtysh_cmd("show ip pim upstream json", isjson=True)
 151     expected = {
 152         "229.1.1.1": {
 153             "10.0.20.2": {
 154                 "firstHopRouter": 1,
 155                 "joinState": "NotJoined",
 156                 "regState": "RegPrune",
 157                 "inboundInterface": "r1-eth0",
 158             }
 159         }
 160     }
 161
 162     test_func = partial(
 163         topotest.router_json_cmp, r1, "show ip pim upstream json", expected
 164     )
 165     _, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
 166     assert result is None, "failed to converge pim"
 167     # tgen.mininet_cli()
 168
 169
 170 def test_pim_rp_sees_stream():
 171     "Ensure that the RP sees the stream and has acted accordingly"
 172     tgen = get_topogen()
 173
 174     rp = tgen.gears["rp"]
 175     json_file = "{}/{}/upstream.json".format(CWD, rp.name)
 176     expected = json.loads(open(json_file).read())
 177
 178     test_func = partial(
 179         topotest.router_json_cmp, rp, "show ip pim upstream json", expected
 180     )
 181     _, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
 182     assertmsg = '"{}" JSON output mismatches'.format(rp.name)
 183     assert result is None, assertmsg
 184
 185
 186 def test_pim_igmp_report():
 187     "Send a igmp report from r2->r1 and ensure that the *,G state is created on r1"
 188     logger.info("Send a igmp report from r2-r1 and ensure *,G created")
 189
 190     tgen = get_topogen()
 191
 192     if tgen.routers_have_failure():
 193         pytest.skip(tgen.errors)
 194
 195     r2 = tgen.gears["r2"]
 196     r1 = tgen.gears["r1"]
 197
 198     # Let's send a igmp report from r2->r1
 199     cmd = [os.path.join(CWD, "mcast-rx.py"), "229.1.1.2", "r2-eth0"]
 200     p = r2.popen(cmd)
 201     try:
 202         expected = {
 203             "229.1.1.2": {
 204                 "*": {
 205                     "sourceIgmp": 1,
 206                     "joinState": "Joined",
 207                     "regState": "RegNoInfo",
 208                     "sptBit": 0,
 209                 }
 210             }
 211         }
 212         test_func = partial(
 213             topotest.router_json_cmp, r1, "show ip pim upstream json", expected
 214         )
 215         _, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
 216         assertmsg = '"{}" JSON output mismatches'.format(r1.name)
 217         assert result is None, assertmsg
 218     finally:
 219         if p:
 220             p.terminate()
 221             p.wait()
 222
 223
 224 def test_memory_leak():
 225     "Run the memory leak test and report results."
 226     tgen = get_topogen()
 227     if not tgen.is_memleak_enabled():
 228         pytest.skip("Memory leak test/report is disabled")
 229
 230     tgen.report_memory_leaks()
 231
 232
 233 if __name__ == "__main__":
 234     args = ["-s"] + sys.argv[1:]
 235     sys.exit(pytest.main(args))