tests/topotests/bgp_lu_topo1/test_bgp_lu.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_lu.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by Volta Networks
   8 #
   9 # Permission to use, copy, modify, and/or distribute this software
  10 # for any purpose with or without fee is hereby granted, provided
  11 # that the above copyright notice and this permission notice appear
  12 # in all copies.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 """
  25 test_bgp_lu.py: Test BGP LU label allocation
  26 """
  27
  28 import os
  29 import sys
  30 import json
  31 from functools import partial
  32 from time import sleep
  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 from mininet.topo import Topo
  47
  48
  49 pytestmark = [pytest.mark.bgpd]
  50
  51
  52 # Basic scenario for BGP-LU. Nodes are directly connected.
  53 # Node 3 is advertising many routes to 2, which advertises them
  54 # as BGP-LU to 1; this way we get routes with actual labels, as
  55 # opposed to implicit-null routes in the 2-node case.
  56 #
  57 #  AS1      BGP-LU        AS2         iBGP        AS2
  58 # +-----+                +-----+                 +-----+
  59 # |     |.1            .2|     |.2             .3|     |
  60 # |  1  +----------------+  2  +-----------------+  3  |
  61 # |     |   10.0.0.0/24  |     |   10.0.1.0/24   |     |
  62 # +-----+                +-----+                 +-----+
  63
  64
  65 class TemplateTopo(Topo):
  66     "Test topology builder"
  67
  68     def build(self, *_args, **_opts):
  69         "Build function"
  70         tgen = get_topogen(self)
  71
  72         # This function 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
  81         # R1-R2
  82         switch = tgen.add_switch("s1")
  83         switch.add_link(tgen.gears["R1"])
  84         switch.add_link(tgen.gears["R2"])
  85
  86         # R2-R3
  87         switch = tgen.add_switch("s2")
  88         switch.add_link(tgen.gears["R2"])
  89         switch.add_link(tgen.gears["R3"])
  90
  91
  92 def setup_module(mod):
  93     "Sets up the pytest environment"
  94     # This function initiates the topology build with Topogen...
  95     tgen = Topogen(TemplateTopo, mod.__name__)
  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     # For all registred routers, load the zebra configuration file
 103     for rname, router in router_list.items():
 104         router.load_config(
 105             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 106         )
 107         router.load_config(
 108             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 109         )
 110
 111     # After loading the configurations, this function loads configured daemons.
 112     tgen.start_router()
 113
 114
 115 def teardown_module(mod):
 116     "Teardown the pytest environment"
 117     tgen = get_topogen()
 118
 119     # This function tears down the whole topology.
 120     tgen.stop_topology()
 121
 122
 123 def check_labelpool(router):
 124     json_file = "{}/{}/labelpool.summ.json".format(CWD, router.name)
 125     expected = json.loads(open(json_file).read())
 126
 127     test_func = partial(
 128         topotest.router_json_cmp, router, "show bgp labelpool summary json", expected
 129     )
 130     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 131     assertmsg = '"{}" JSON output mismatches - Did not converge'.format(router.name)
 132     assert result is None, assertmsg
 133
 134
 135 def test_converge_bgplu():
 136     "Wait for protocol convergence"
 137
 138     tgen = get_topogen()
 139     # Don't run this test if we have any failure.
 140     if tgen.routers_have_failure():
 141         pytest.skip(tgen.errors)
 142
 143     # tgen.mininet_cli();
 144     r1 = tgen.gears["R1"]
 145     r2 = tgen.gears["R2"]
 146
 147     check_labelpool(r1)
 148     check_labelpool(r2)
 149
 150
 151 def test_clear_bgplu():
 152     "Wait for protocol convergence"
 153
 154     tgen = get_topogen()
 155     # Don't run this test if we have any failure.
 156     if tgen.routers_have_failure():
 157         pytest.skip(tgen.errors)
 158
 159     # tgen.mininet_cli();
 160     r1 = tgen.gears["R1"]
 161     r2 = tgen.gears["R2"]
 162
 163     r1.vtysh_cmd("clear bgp 10.0.0.2")
 164     check_labelpool(r1)
 165     check_labelpool(r2)
 166
 167     r2.vtysh_cmd("clear bgp 10.0.1.3")
 168     check_labelpool(r1)
 169     check_labelpool(r2)
 170
 171     r1.vtysh_cmd("clear bgp 10.0.0.2")
 172     r2.vtysh_cmd("clear bgp 10.0.1.3")
 173     check_labelpool(r1)
 174     check_labelpool(r2)
 175
 176
 177 def test_memory_leak():
 178     "Run the memory leak test and report results."
 179     tgen = get_topogen()
 180     if not tgen.is_memleak_enabled():
 181         pytest.skip("Memory leak test/report is disabled")
 182
 183     tgen.report_memory_leaks()
 184
 185
 186 if __name__ == "__main__":
 187     args = ["-s"] + sys.argv[1:]
 188     sys.exit(pytest.main(args))