tests/topotests/bgp_lu_topo2/test_bgp_lu2.py

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