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 # Basic scenario for BGP-LU. Nodes are directly connected.
  49 # Node 3 is advertising many routes to 2, which advertises them
  50 # as BGP-LU to 1; this way we get routes with actual labels, as
  51 # opposed to implicit-null routes in the 2-node case.
  52 #
  53 #  AS1      BGP-LU        AS2         iBGP        AS2
  54 # +-----+                +-----+                 +-----+
  55 # |     |.1            .2|     |.2             .3|     |
  56 # |  1  +----------------+  2  +-----------------+  3  |
  57 # |     |   10.0.0.0/24  |     |   10.0.1.0/24   |     |
  58 # +-----+                +-----+                 +-----+
  59
  60
  61 class TemplateTopo(Topo):
  62     "Test topology builder"
  63
  64     def build(self, *_args, **_opts):
  65         "Build function"
  66         tgen = get_topogen(self)
  67
  68         # This function only purpose is to define allocation and relationship
  69         # between routers, switches and hosts.
  70         #
  71         #
  72         # Create routers
  73         tgen.add_router("R1")
  74         tgen.add_router("R2")
  75         tgen.add_router("R3")
  76
  77         # R1-R2
  78         switch = tgen.add_switch("s1")
  79         switch.add_link(tgen.gears["R1"])
  80         switch.add_link(tgen.gears["R2"])
  81
  82         # R2-R3
  83         switch = tgen.add_switch("s2")
  84         switch.add_link(tgen.gears["R2"])
  85         switch.add_link(tgen.gears["R3"])
  86
  87
  88 def setup_module(mod):
  89     "Sets up the pytest environment"
  90     # This function initiates the topology build with Topogen...
  91     tgen = Topogen(TemplateTopo, mod.__name__)
  92     # ... and here it calls Mininet initialization functions.
  93     tgen.start_topology()
  94
  95     # This is a sample of configuration loading.
  96     router_list = tgen.routers()
  97
  98     # For all registred routers, load the zebra configuration file
  99     for rname, router in router_list.items():
 100         router.load_config(
 101             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 102         )
 103         router.load_config(
 104             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 105         )
 106
 107     # After loading the configurations, this function loads configured daemons.
 108     tgen.start_router()
 109
 110
 111 def teardown_module(mod):
 112     "Teardown the pytest environment"
 113     tgen = get_topogen()
 114
 115     # This function tears down the whole topology.
 116     tgen.stop_topology()
 117
 118
 119 def check_labelpool(router):
 120     json_file = "{}/{}/labelpool.summ.json".format(CWD, router.name)
 121     expected = json.loads(open(json_file).read())
 122
 123     test_func = partial(
 124         topotest.router_json_cmp, router, "show bgp labelpool summary json", expected
 125     )
 126     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 127     assertmsg = '"{}" JSON output mismatches - Did not converge'.format(router.name)
 128     assert result is None, assertmsg
 129
 130
 131 def test_converge_bgplu():
 132     "Wait for protocol convergence"
 133
 134     tgen = get_topogen()
 135     # Don't run this test if we have any failure.
 136     if tgen.routers_have_failure():
 137         pytest.skip(tgen.errors)
 138
 139     # tgen.mininet_cli();
 140     r1 = tgen.gears["R1"]
 141     r2 = tgen.gears["R2"]
 142
 143     check_labelpool(r1)
 144     check_labelpool(r2)
 145
 146
 147 def test_clear_bgplu():
 148     "Wait for protocol convergence"
 149
 150     tgen = get_topogen()
 151     # Don't run this test if we have any failure.
 152     if tgen.routers_have_failure():
 153         pytest.skip(tgen.errors)
 154
 155     # tgen.mininet_cli();
 156     r1 = tgen.gears["R1"]
 157     r2 = tgen.gears["R2"]
 158
 159     r1.vtysh_cmd("clear bgp 10.0.0.2")
 160     check_labelpool(r1)
 161     check_labelpool(r2)
 162
 163     r2.vtysh_cmd("clear bgp 10.0.1.3")
 164     check_labelpool(r1)
 165     check_labelpool(r2)
 166
 167     r1.vtysh_cmd("clear bgp 10.0.0.2")
 168     r2.vtysh_cmd("clear bgp 10.0.1.3")
 169     check_labelpool(r1)
 170     check_labelpool(r2)
 171
 172
 173 def test_memory_leak():
 174     "Run the memory leak test and report results."
 175     tgen = get_topogen()
 176     if not tgen.is_memleak_enabled():
 177         pytest.skip("Memory leak test/report is disabled")
 178
 179     tgen.report_memory_leaks()
 180
 181
 182 if __name__ == "__main__":
 183     args = ["-s"] + sys.argv[1:]
 184     sys.exit(pytest.main(args))