tests/topotests/bgp_lu_topo2/test_bgp_lu2.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # test_bgp_lu2.py
   5 #
   6 # Part of FRR/NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by Volta Networks
   9 # Copyright (c) 2021 by Nvidia, Inc.
  10 #
  11
  12 """
  13 test_bgp_lu2.py: Test BGP LU label allocation
  14 """
  15
  16 import os
  17 import sys
  18 import json
  19 from functools import partial
  20 import pytest
  21
  22 # Save the Current Working Directory to find configuration files.
  23 CWD = os.path.dirname(os.path.realpath(__file__))
  24 sys.path.append(os.path.join(CWD, "../"))
  25
  26 # pylint: disable=C0413
  27 # Import topogen and topotest helpers
  28 from lib import topotest
  29 from lib.topogen import Topogen, TopoRouter, get_topogen
  30 from lib.topolog import logger
  31
  32 # Required to instantiate the topology builder class.
  33
  34 pytestmark = [pytest.mark.bgpd]
  35
  36 #
  37 # Basic scenario for BGP-LU. Nodes are directly connected.
  38 # Node 3 is advertising routes to 2, which advertises them
  39 # as BGP-LU to 1; this way we get routes with actual labels, as
  40 # opposed to implicit-null routes in the 2-node case.
  41 #
  42 # R2 is an LER, with MPLS towards R1, and IP towards R3. R1 is an LSR, with
  43 # MPLS on both sides.
  44 #
  45 #
  46 #    AS4     BGP-LU      AS1       BGP-LU       AS2         iBGP        AS2
  47 #  +-----+             +-----+                +-----+                 +-----+
  48 #  |     |.4         .1|     |.1            .2|     |.2             .3|     |
  49 #  |  4  +-------------+  1  +----------------+  2  +-----------------+  3  |
  50 #  |     | 10.0.4.0/24 |     |   10.0.0.0/24  |     |   10.0.1.0/24   |     |
  51 #  +-----+             +-----+                +-----+                 +-----+
  52 #
  53 #
  54
  55
  56 def build_topo(tgen):
  57     "Build function"
  58
  59     # This function's only purpose is to define allocation and relationship
  60     # between routers, switches and hosts.
  61     #
  62     #
  63     # Create routers
  64     tgen.add_router("R1")
  65     tgen.add_router("R2")
  66     tgen.add_router("R3")
  67     tgen.add_router("R4")
  68
  69     # R1-R2
  70     switch = tgen.add_switch("s1")
  71     switch.add_link(tgen.gears["R1"])
  72     switch.add_link(tgen.gears["R2"])
  73
  74     # R2-R3
  75     switch = tgen.add_switch("s2")
  76     switch.add_link(tgen.gears["R2"])
  77     switch.add_link(tgen.gears["R3"])
  78
  79     # R1-R4
  80     switch = tgen.add_switch("s3")
  81     switch.add_link(tgen.gears["R1"])
  82     switch.add_link(tgen.gears["R4"])
  83
  84
  85 def setup_module(mod):
  86     "Sets up the pytest environment"
  87     # This function initiates the topology build with Topogen...
  88     tgen = Topogen(build_topo, mod.__name__)
  89
  90     # Skip if no mpls support
  91     if not tgen.hasmpls:
  92         logger.info("MPLS is not available, skipping test")
  93         pytest.skip("MPLS is not available, skipping")
  94         return
  95
  96     # ... and here it calls Mininet initialization functions.
  97     tgen.start_topology()
  98
  99     # This is a sample of configuration loading.
 100     router_list = tgen.routers()
 101
 102     # Enable mpls input for routers, so we can ping
 103     sval = "net.mpls.conf.{}.input"
 104     topotest.sysctl_assure(router_list["R2"], sval.format("R2-eth0"), 1)
 105     topotest.sysctl_assure(router_list["R1"], sval.format("R1-eth0"), 1)
 106     topotest.sysctl_assure(router_list["R1"], sval.format("R1-eth1"), 1)
 107     topotest.sysctl_assure(router_list["R4"], sval.format("R4-eth0"), 1)
 108
 109     # For all registered routers, load the zebra configuration file
 110     for rname, router in router_list.items():
 111         router.load_config(
 112             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 113         )
 114         router.load_config(
 115             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 116         )
 117
 118         # Have static config for R3 too
 119         if router == router_list["R3"]:
 120             router.load_config(
 121                 TopoRouter.RD_STATIC, os.path.join(CWD, "{}/staticd.conf".format(rname))
 122             )
 123
 124     # After loading the configurations, this function loads configured daemons.
 125     tgen.start_router()
 126
 127
 128 def teardown_module(mod):
 129     "Teardown the pytest environment"
 130     tgen = get_topogen()
 131
 132     # This function tears down the whole topology.
 133     tgen.stop_topology()
 134
 135
 136 def check_labelpool(router):
 137     json_file = "{}/{}/labelpool.summ.json".format(CWD, router.name)
 138     expected = json.loads(open(json_file).read())
 139
 140     test_func = partial(
 141         topotest.router_json_cmp, router, "show bgp labelpool summary json", expected
 142     )
 143     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 144     assertmsg = '"{}" JSON output mismatches - Did not converge'.format(router.name)
 145     assert result is None, assertmsg
 146
 147
 148 def test_converge_bgplu():
 149     "Wait for protocol convergence"
 150
 151     tgen = get_topogen()
 152     # Don't run this test if we have any failure.
 153     if tgen.routers_have_failure():
 154         pytest.skip(tgen.errors)
 155
 156     # TODO -- enable for debugging
 157     # tgen.mininet_cli()
 158
 159     r1 = tgen.gears["R1"]
 160     r2 = tgen.gears["R2"]
 161
 162     check_labelpool(r1)
 163     check_labelpool(r2)
 164
 165
 166 def test_ping():
 167     "Simple ping tests"
 168
 169     tgen = get_topogen()
 170
 171     # Don't run this test if we have any failure.
 172     if tgen.routers_have_failure():
 173         pytest.skip(tgen.errors)
 174
 175     #
 176     logger.info("Ping from R2 to R3")
 177     router = tgen.gears["R2"]
 178     output = router.run("ping -c 4 -w 4 {}".format("10.0.1.3"))
 179     assert " 0% packet loss" in output, "Ping R2->R3 FAILED"
 180     logger.info("Ping from R2 to R3 ... success")
 181
 182     #
 183     logger.info("Ping from R4 to R2")
 184     router = tgen.gears["R4"]
 185     output = router.run("ping -c 4 -w 4 {}".format("10.0.0.2"))
 186     assert " 0% packet loss" in output, "Ping R4->R2 FAILED"
 187     logger.info("Ping from R4 to R2 ... success")
 188
 189     #
 190     logger.info("Ping from R4 to R3")
 191     router = tgen.gears["R4"]
 192     output = router.run("ping -c 4 -w 4 {}".format("10.0.1.3"))
 193     assert " 0% packet loss" in output, "Ping R4->R3 FAILED"
 194     logger.info("Ping from R4 to R3 ... success")
 195
 196
 197 def test_memory_leak():
 198     "Run the memory leak test and report results."
 199     tgen = get_topogen()
 200     if not tgen.is_memleak_enabled():
 201         pytest.skip("Memory leak test/report is disabled")
 202
 203     tgen.report_memory_leaks()
 204
 205
 206 if __name__ == "__main__":
 207     args = ["-s"] + sys.argv[1:]
 208     sys.exit(pytest.main(args))