tests/topotests/ldp_oc_topo1/test_ldp_oc_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_ldp_oc_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by by Volta Networks
   9 #
  10
  11 r"""
  12 test_ldp_oc_topo1.py: Simple FRR LDP Test
  13
  14              +---------+
  15              |    r1   |
  16              | 1.1.1.1 |
  17              +----+----+
  18                   | .1  r1-eth0
  19                   |
  20             ~~~~~~~~~~~~~
  21           ~~     sw0     ~~
  22           ~~ 10.0.1.0/24 ~~
  23             ~~~~~~~~~~~~~
  24                   |10.0.1.0/24
  25                   |
  26                   | .2  r2-eth0
  27              +----+----+
  28              |    r2   |
  29              | 2.2.2.2 |
  30              +--+---+--+
  31     r2-eth2  .2 |   | .2  r2-eth1
  32          ______/     \______
  33         /                   \
  34   ~~~~~~~~~~~~~        ~~~~~~~~~~~~~
  35 ~~     sw2     ~~    ~~     sw1     ~~
  36 ~~ 10.0.3.0/24 ~~    ~~ 10.0.2.0/24 ~~
  37   ~~~~~~~~~~~~~        ~~~~~~~~~~~~~
  38         |                 /    |
  39          \      _________/     |
  40           \    /                \
  41 r3-eth1 .3 |  | .3  r3-eth0      | .4 r4-eth0
  42       +----+--+---+         +----+----+
  43       |     r3    |         |    r4   |
  44       |  3.3.3.3  |         | 4.4.4.4 |
  45       +-----------+         +---------+
  46 """
  47
  48 import os
  49 import sys
  50 import pytest
  51 import json
  52 from functools import partial
  53
  54 # Save the Current Working Directory to find configuration files.
  55 CWD = os.path.dirname(os.path.realpath(__file__))
  56 sys.path.append(os.path.join(CWD, "../"))
  57
  58 # pylint: disable=C0413
  59 # Import topogen and topotest helpers
  60 from lib import topotest
  61 from lib.topogen import Topogen, TopoRouter, get_topogen
  62 from lib.topolog import logger
  63
  64 # Required to instantiate the topology builder class.
  65
  66 pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
  67
  68
  69 def build_topo(tgen):
  70     "Build function"
  71
  72     #
  73     # Define FRR Routers
  74     #
  75     for router in ["r1", "r2", "r3", "r4"]:
  76         tgen.add_router(router)
  77
  78     #
  79     # Define connections
  80     #
  81     switch = tgen.add_switch("s0")
  82     switch.add_link(tgen.gears["r1"])
  83     switch.add_link(tgen.gears["r2"])
  84
  85     switch = tgen.add_switch("s1")
  86     switch.add_link(tgen.gears["r2"])
  87     switch.add_link(tgen.gears["r3"])
  88     switch.add_link(tgen.gears["r4"])
  89
  90     switch = tgen.add_switch("s2")
  91     switch.add_link(tgen.gears["r2"])
  92     switch.add_link(tgen.gears["r3"])
  93
  94
  95 def setup_module(mod):
  96     "Sets up the pytest environment"
  97     tgen = Topogen(build_topo, mod.__name__)
  98     tgen.start_topology()
  99
 100     router_list = tgen.routers()
 101
 102     # For all registered 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         # Don't start ospfd and ldpd in the CE nodes
 108         if router.name[0] == "r":
 109             router.load_config(
 110                 TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 111             )
 112             router.load_config(
 113                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 114             )
 115
 116     tgen.start_router()
 117
 118
 119 def teardown_module(mod):
 120     "Teardown the pytest environment"
 121     tgen = get_topogen()
 122
 123     # This function tears down the whole topology.
 124     tgen.stop_topology()
 125
 126
 127 def router_compare_json_output(rname, command, reference):
 128     "Compare router JSON output"
 129
 130     logger.info('Comparing router "%s" "%s" output', rname, command)
 131
 132     tgen = get_topogen()
 133     filename = "{}/{}/{}".format(CWD, rname, reference)
 134     expected = json.loads(open(filename).read())
 135
 136     # Run test function until we get an result. Wait at most 80 seconds.
 137     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 138     _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
 139
 140     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 141     assert diff is None, assertmsg
 142
 143
 144 def test_ospf_convergence():
 145     logger.info("Test: check OSPF adjacencies")
 146
 147     tgen = get_topogen()
 148
 149     # Skip if previous fatal error condition is raised
 150     if tgen.routers_have_failure():
 151         pytest.skip(tgen.errors)
 152
 153     for rname in ["r1", "r2", "r3", "r4"]:
 154         router_compare_json_output(
 155             rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
 156         )
 157
 158
 159 def test_rib():
 160     logger.info("Test: verify RIB")
 161     tgen = get_topogen()
 162
 163     # Skip if previous fatal error condition is raised
 164     if tgen.routers_have_failure():
 165         pytest.skip(tgen.errors)
 166
 167     for rname in ["r1", "r2", "r3", "r4"]:
 168         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 169
 170
 171 def test_ldp_adjacencies():
 172     logger.info("Test: verify LDP adjacencies")
 173     tgen = get_topogen()
 174
 175     # Skip if previous fatal error condition is raised
 176     if tgen.routers_have_failure():
 177         pytest.skip(tgen.errors)
 178
 179     for rname in ["r1", "r2", "r3", "r4"]:
 180         router_compare_json_output(
 181             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 182         )
 183
 184
 185 def test_ldp_neighbors():
 186     logger.info("Test: verify LDP neighbors")
 187     tgen = get_topogen()
 188
 189     # Skip if previous fatal error condition is raised
 190     if tgen.routers_have_failure():
 191         pytest.skip(tgen.errors)
 192
 193     for rname in ["r1", "r2", "r3", "r4"]:
 194         router_compare_json_output(
 195             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 196         )
 197
 198
 199 def test_ldp_bindings():
 200     logger.info("Test: verify LDP bindings")
 201     tgen = get_topogen()
 202
 203     # Skip if previous fatal error condition is raised
 204     if tgen.routers_have_failure():
 205         pytest.skip(tgen.errors)
 206
 207     for rname in ["r1", "r2", "r3", "r4"]:
 208         router_compare_json_output(
 209             rname, "show mpls ldp binding json", "show_ldp_binding.ref"
 210         )
 211
 212
 213 # Memory leak test template
 214 def test_memory_leak():
 215     "Run the memory leak test and report results."
 216     tgen = get_topogen()
 217     if not tgen.is_memleak_enabled():
 218         pytest.skip("Memory leak test/report is disabled")
 219
 220     tgen.report_memory_leaks()
 221
 222
 223 if __name__ == "__main__":
 224     args = ["-s"] + sys.argv[1:]
 225     sys.exit(pytest.main(args))