tests/topotests/pim_basic_topo2/test_pim_basic_topo2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_pim_basic_topo2.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2021 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_pim_basic_topo2.py: Test the FRR PIM protocol convergence.
  27 """
  28
  29 import os
  30 import sys
  31 from functools import partial
  32 import pytest
  33
  34 # Save the Current Working Directory to find configuration files.
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 # Import topogen and topotest helpers
  40 from lib import topotest
  41 from lib.topogen import Topogen, TopoRouter, get_topogen
  42 from lib.topolog import logger
  43
  44 # Required to instantiate the topology builder class.
  45
  46 pytestmark = [pytest.mark.bfdd, pytest.mark.pimd]
  47
  48
  49 def build_topo(tgen):
  50     "Build function"
  51
  52     # Create 4 routers
  53     for routern in range(1, 5):
  54         tgen.add_router("r{}".format(routern))
  55
  56     switch = tgen.add_switch("s1")
  57     switch.add_link(tgen.gears["r1"])
  58     switch.add_link(tgen.gears["r2"])
  59
  60     switch = tgen.add_switch("s2")
  61     switch.add_link(tgen.gears["r2"])
  62     switch.add_link(tgen.gears["r3"])
  63
  64     switch = tgen.add_switch("s3")
  65     switch.add_link(tgen.gears["r2"])
  66     switch.add_link(tgen.gears["r4"])
  67
  68
  69 def setup_module(mod):
  70     "Sets up the pytest environment"
  71     tgen = Topogen(build_topo, mod.__name__)
  72     tgen.start_topology()
  73
  74     router_list = tgen.routers()
  75     for rname, router in router_list.items():
  76         daemon_file = "{}/{}/bfdd.conf".format(CWD, rname)
  77         if os.path.isfile(daemon_file):
  78             router.load_config(TopoRouter.RD_BFD, daemon_file)
  79
  80         daemon_file = "{}/{}/pimd.conf".format(CWD, rname)
  81         if os.path.isfile(daemon_file):
  82             router.load_config(TopoRouter.RD_PIM, daemon_file)
  83
  84         daemon_file = "{}/{}/zebra.conf".format(CWD, rname)
  85         if os.path.isfile(daemon_file):
  86             router.load_config(TopoRouter.RD_ZEBRA, daemon_file)
  87
  88     # Initialize all routers.
  89     tgen.start_router()
  90
  91
  92 def teardown_module(_mod):
  93     "Teardown the pytest environment"
  94     tgen = get_topogen()
  95     tgen.stop_topology()
  96
  97
  98 def expect_neighbor(router, interface, peer):
  99     "Wait until peer is present on interface."
 100     logger.info("waiting peer {} in {}".format(peer, interface))
 101     tgen = get_topogen()
 102     test_func = partial(
 103         topotest.router_json_cmp,
 104         tgen.gears[router],
 105         "show ip pim neighbor json",
 106         {interface: {peer: {}}},
 107     )
 108     _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
 109     assertmsg = '"{}" PIM convergence failure'.format(router)
 110     assert result is None, assertmsg
 111
 112
 113 def test_wait_pim_convergence():
 114     "Wait for PIM to converge"
 115     tgen = get_topogen()
 116     if tgen.routers_have_failure():
 117         pytest.skip(tgen.errors)
 118
 119     logger.info("waiting for PIM to converge")
 120
 121     expect_neighbor("r1", "r1-eth0", "192.168.1.2")
 122     expect_neighbor("r2", "r2-eth0", "192.168.1.1")
 123
 124     expect_neighbor("r2", "r2-eth1", "192.168.2.3")
 125     expect_neighbor("r2", "r2-eth2", "192.168.3.4")
 126
 127     expect_neighbor("r3", "r3-eth0", "192.168.2.1")
 128     expect_neighbor("r4", "r4-eth0", "192.168.3.1")
 129
 130
 131 def test_bfd_peers():
 132     "Wait for BFD peers to show up."
 133     tgen = get_topogen()
 134     if tgen.routers_have_failure():
 135         pytest.skip(tgen.errors)
 136
 137     logger.info("waiting for BFD to converge")
 138
 139     def expect_bfd_peer(router, peer):
 140         "Wait until peer is present on interface."
 141         logger.info("waiting BFD peer {} in {}".format(peer, router))
 142         test_func = partial(
 143             topotest.router_json_cmp,
 144             tgen.gears[router],
 145             "show bfd peers json",
 146             [{"peer": peer, "status": "up"}],
 147         )
 148         _, result = topotest.run_and_expect(test_func, None, count=10, wait=1)
 149         assertmsg = '"{}" BFD convergence failure'.format(router)
 150         assert result is None, assertmsg
 151
 152     expect_bfd_peer("r1", "192.168.1.2")
 153     expect_bfd_peer("r2", "192.168.1.1")
 154     expect_bfd_peer("r2", "192.168.2.3")
 155     expect_bfd_peer("r2", "192.168.3.4")
 156     expect_bfd_peer("r3", "192.168.2.1")
 157     expect_bfd_peer("r4", "192.168.3.1")
 158
 159
 160 def test_pim_reconvergence():
 161     "Disconnect a peer and expect it to disconnect."
 162     tgen = get_topogen()
 163     if tgen.routers_have_failure():
 164         pytest.skip(tgen.errors)
 165
 166     logger.info("waiting for disconnect convergence")
 167     tgen.gears["r4"].link_enable("r4-eth0", enabled=False)
 168
 169     def expect_neighbor_down(router, interface, peer):
 170         "Wait until peer is present on interface."
 171         logger.info("waiting peer {} in {} to disappear".format(peer, interface))
 172         test_func = partial(
 173             topotest.router_json_cmp,
 174             tgen.gears[router],
 175             "show ip pim neighbor json",
 176             {interface: {peer: None}},
 177         )
 178         _, result = topotest.run_and_expect(test_func, None, count=5, wait=1)
 179         assertmsg = '"{}" PIM convergence failure'.format(router)
 180         assert result is None, assertmsg
 181
 182     expect_neighbor_down("r2", "r2-eth2", "192.168.3.4")
 183
 184     logger.info("waiting for reconvergence")
 185     tgen.gears["r4"].link_enable("r4-eth0", enabled=True)
 186     expect_neighbor("r2", "r2-eth2", "192.168.3.4")
 187
 188
 189 def test_pim_bfd_profile():
 190     "Test that the BFD profile is properly applied in BFD."
 191     tgen = get_topogen()
 192     if tgen.routers_have_failure():
 193         pytest.skip(tgen.errors)
 194
 195     def expect_bfd_peer_settings(router, settings):
 196         "Expect the following BFD configuration"
 197         logger.info("Verifying BFD peer {} in {}".format(settings["peer"], router))
 198         test_func = partial(
 199             topotest.router_json_cmp,
 200             tgen.gears[router],
 201             "show bfd peers json",
 202             [settings],
 203         )
 204         _, result = topotest.run_and_expect(test_func, None, count=5, wait=1)
 205         assertmsg = '"{}" BFD convergence failure'.format(router)
 206         assert result is None, assertmsg
 207
 208     expect_bfd_peer_settings(
 209         "r1",
 210         {
 211             "peer": "192.168.1.2",
 212             "receive-interval": 250,
 213             "transmit-interval": 250,
 214         },
 215     )
 216
 217     expect_bfd_peer_settings(
 218         "r2",
 219         {
 220             "peer": "192.168.1.1",
 221             "remote-receive-interval": 250,
 222             "remote-transmit-interval": 250,
 223         },
 224     )
 225
 226
 227 def test_memory_leak():
 228     "Run the memory leak test and report results."
 229     tgen = get_topogen()
 230     if not tgen.is_memleak_enabled():
 231         pytest.skip("Memory leak test/report is disabled")
 232
 233     tgen.report_memory_leaks()
 234
 235
 236 if __name__ == "__main__":
 237     args = ["-s"] + sys.argv[1:]
 238     sys.exit(pytest.main(args))