tests/topotests/bgp_lu_topo1/test_bgp_lu.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_lu.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by Volta Networks
   9 #
  10
  11 """
  12 test_bgp_lu.py: Test BGP LU label allocation
  13 """
  14
  15 import os
  16 import sys
  17 import json
  18 from functools import partial
  19 import pytest
  20
  21 # Save the Current Working Directory to find configuration files.
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 # Import topogen and topotest helpers
  27 from lib import topotest
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29
  30 # Required to instantiate the topology builder class.
  31
  32
  33 pytestmark = [pytest.mark.bgpd]
  34
  35
  36 # Basic scenario for BGP-LU. Nodes are directly connected.
  37 # Node 3 is advertising many routes to 2, which advertises them
  38 # as BGP-LU to 1; this way we get routes with actual labels, as
  39 # opposed to implicit-null routes in the 2-node case.
  40 #
  41 #  AS1      BGP-LU        AS2         iBGP        AS2
  42 # +-----+                +-----+                 +-----+
  43 # |     |.1            .2|     |.2             .3|     |
  44 # |  1  +----------------+  2  +-----------------+  3  |
  45 # |     |   10.0.0.0/24  |     |   10.0.1.0/24   |     |
  46 # +-----+                +-----+                 +-----+
  47
  48
  49 def build_topo(tgen):
  50     "Build function"
  51
  52     # This function only purpose is to define allocation and relationship
  53     # between routers, switches and hosts.
  54     #
  55     #
  56     # Create routers
  57     tgen.add_router("R1")
  58     tgen.add_router("R2")
  59     tgen.add_router("R3")
  60
  61     # R1-R2
  62     switch = tgen.add_switch("s1")
  63     switch.add_link(tgen.gears["R1"])
  64     switch.add_link(tgen.gears["R2"])
  65
  66     # R2-R3
  67     switch = tgen.add_switch("s2")
  68     switch.add_link(tgen.gears["R2"])
  69     switch.add_link(tgen.gears["R3"])
  70
  71
  72 def setup_module(mod):
  73     "Sets up the pytest environment"
  74     # This function initiates the topology build with Topogen...
  75     tgen = Topogen(build_topo, mod.__name__)
  76     # ... and here it calls Mininet initialization functions.
  77     tgen.start_topology()
  78
  79     # This is a sample of configuration loading.
  80     router_list = tgen.routers()
  81
  82     # For all registered routers, load the zebra configuration file
  83     for rname, router in router_list.items():
  84         router.load_config(
  85             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  86         )
  87         router.load_config(
  88             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  89         )
  90
  91     # After loading the configurations, this function loads configured daemons.
  92     tgen.start_router()
  93
  94
  95 def teardown_module(mod):
  96     "Teardown the pytest environment"
  97     tgen = get_topogen()
  98
  99     # This function tears down the whole topology.
 100     tgen.stop_topology()
 101
 102
 103 def check_labelpool(router):
 104     json_file = "{}/{}/labelpool.summ.json".format(CWD, router.name)
 105     expected = json.loads(open(json_file).read())
 106
 107     test_func = partial(
 108         topotest.router_json_cmp, router, "show bgp labelpool summary json", expected
 109     )
 110     _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
 111     assertmsg = '"{}" JSON output mismatches - Did not converge'.format(router.name)
 112     assert result is None, assertmsg
 113
 114
 115 def test_converge_bgplu():
 116     "Wait for protocol convergence"
 117
 118     tgen = get_topogen()
 119     # Don't run this test if we have any failure.
 120     if tgen.routers_have_failure():
 121         pytest.skip(tgen.errors)
 122
 123     # tgen.mininet_cli();
 124     r1 = tgen.gears["R1"]
 125     r2 = tgen.gears["R2"]
 126
 127     check_labelpool(r1)
 128     check_labelpool(r2)
 129
 130
 131 def test_clear_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     r1.vtysh_cmd("clear bgp 10.0.0.2")
 144     check_labelpool(r1)
 145     check_labelpool(r2)
 146
 147     r2.vtysh_cmd("clear bgp 10.0.1.3")
 148     check_labelpool(r1)
 149     check_labelpool(r2)
 150
 151     r1.vtysh_cmd("clear bgp 10.0.0.2")
 152     r2.vtysh_cmd("clear bgp 10.0.1.3")
 153     check_labelpool(r1)
 154     check_labelpool(r2)
 155
 156
 157 def test_memory_leak():
 158     "Run the memory leak test and report results."
 159     tgen = get_topogen()
 160     if not tgen.is_memleak_enabled():
 161         pytest.skip("Memory leak test/report is disabled")
 162
 163     tgen.report_memory_leaks()
 164
 165
 166 if __name__ == "__main__":
 167     args = ["-s"] + sys.argv[1:]
 168     sys.exit(pytest.main(args))