tests/topotests/ldp_vpls_topo1/test_ldp_vpls_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_ldp_vpls_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2017 by
   9 # Network Device Education Foundation, Inc. ("NetDEF")
  10 #
  11
  12 """
  13 test_ldp_vpls_topo1.py:
  14
  15                    +---------+                +---------+
  16                    |         |                |         |
  17                    |   CE1   |                |   CE2   |
  18                    |         |                |         |
  19                    +---------+                +---------+
  20 ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
  21                         |                          |
  22                         |                          |
  23                 rt1-eth0|                          |rt2-eth0
  24                    +---------+  10.0.1.0/24   +---------+
  25                    |         |rt1-eth1        |         |
  26                    |   RT1   +----------------+   RT2   |
  27                    | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
  28                    |         |                |         |
  29                    +---------+                +---------+
  30                 rt1-eth2|                          |rt2-eth2
  31                         |                          |
  32                         |                          |
  33              10.0.2.0/24|        +---------+       |10.0.3.0/24
  34                         |        |         |       |
  35                         |        |   RT3   |       |
  36                         +--------+ 3.3.3.3 +-------+
  37                          rt3-eth2|         |rt3-eth1
  38                                  +---------+
  39                                       |rt3-eth0
  40                                       |
  41                                       |
  42               ce3-eth0 (172.16.1.3/24)|
  43                                  +---------+
  44                                  |         |
  45                                  |   CE3   |
  46                                  |         |
  47                                  +---------+
  48 """
  49
  50 import os
  51 import sys
  52 import pytest
  53 import json
  54 from functools import partial
  55
  56 # Save the Current Working Directory to find configuration files.
  57 CWD = os.path.dirname(os.path.realpath(__file__))
  58 sys.path.append(os.path.join(CWD, "../"))
  59
  60 # pylint: disable=C0413
  61 # Import topogen and topotest helpers
  62 from lib import topotest
  63 from lib.topogen import Topogen, TopoRouter, get_topogen
  64 from lib.topolog import logger
  65
  66 # Required to instantiate the topology builder class.
  67
  68 pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
  69
  70
  71 def build_topo(tgen):
  72     "Build function"
  73
  74     #
  75     # Define FRR Routers
  76     #
  77     for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  78         tgen.add_router(router)
  79
  80     #
  81     # Define connections
  82     #
  83     switch = tgen.add_switch("s1")
  84     switch.add_link(tgen.gears["ce1"])
  85     switch.add_link(tgen.gears["r1"])
  86
  87     switch = tgen.add_switch("s2")
  88     switch.add_link(tgen.gears["ce2"])
  89     switch.add_link(tgen.gears["r2"])
  90
  91     switch = tgen.add_switch("s3")
  92     switch.add_link(tgen.gears["ce3"])
  93     switch.add_link(tgen.gears["r3"])
  94
  95     switch = tgen.add_switch("s4")
  96     switch.add_link(tgen.gears["r1"])
  97     switch.add_link(tgen.gears["r2"])
  98
  99     switch = tgen.add_switch("s5")
 100     switch.add_link(tgen.gears["r1"])
 101     switch.add_link(tgen.gears["r3"])
 102
 103     switch = tgen.add_switch("s6")
 104     switch.add_link(tgen.gears["r2"])
 105     switch.add_link(tgen.gears["r3"])
 106
 107
 108 def setup_module(mod):
 109     "Sets up the pytest environment"
 110     tgen = Topogen(build_topo, mod.__name__)
 111     tgen.start_topology()
 112
 113     router_list = tgen.routers()
 114
 115     # For all registered routers, load the zebra configuration file
 116     for rname, router in router_list.items():
 117         router.load_config(
 118             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 119         )
 120         # Don't start ospfd and ldpd in the CE nodes
 121         if router.name[0] == "r":
 122             router.load_config(
 123                 TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 124             )
 125             router.load_config(
 126                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 127             )
 128
 129     tgen.start_router()
 130
 131
 132 def teardown_module(mod):
 133     "Teardown the pytest environment"
 134     tgen = get_topogen()
 135
 136     # This function tears down the whole topology.
 137     tgen.stop_topology()
 138
 139
 140 def router_compare_json_output(rname, command, reference, count=80, wait=1):
 141     "Compare router JSON output"
 142
 143     logger.info('Comparing router "%s" "%s" output', rname, command)
 144
 145     tgen = get_topogen()
 146     filename = "{}/{}/{}".format(CWD, rname, reference)
 147     expected = json.loads(open(filename).read())
 148
 149     # Run test function until we get an result.
 150     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 151     _, diff = topotest.run_and_expect(test_func, None, count, wait)
 152     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 153     assert diff is None, assertmsg
 154
 155
 156 def test_ospf_convergence():
 157     logger.info("Test: check OSPF adjacencies")
 158     tgen = get_topogen()
 159
 160     # Skip if previous fatal error condition is raised
 161     if tgen.routers_have_failure():
 162         pytest.skip(tgen.errors)
 163
 164     for rname in ["r1", "r2", "r3"]:
 165         router_compare_json_output(
 166             rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
 167         )
 168
 169
 170 def test_rib():
 171     logger.info("Test: verify RIB")
 172     tgen = get_topogen()
 173
 174     # Skip if previous fatal error condition is raised
 175     if tgen.routers_have_failure():
 176         pytest.skip(tgen.errors)
 177
 178     for rname in ["r1", "r2", "r3"]:
 179         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 180
 181
 182 def test_ldp_adjacencies():
 183     logger.info("Test: verify LDP adjacencies")
 184     tgen = get_topogen()
 185
 186     # Skip if previous fatal error condition is raised
 187     if tgen.routers_have_failure():
 188         pytest.skip(tgen.errors)
 189
 190     for rname in ["r1", "r2", "r3"]:
 191         router_compare_json_output(
 192             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 193         )
 194
 195
 196 def test_ldp_neighbors():
 197     logger.info("Test: verify LDP neighbors")
 198     tgen = get_topogen()
 199
 200     # Skip if previous fatal error condition is raised
 201     if tgen.routers_have_failure():
 202         pytest.skip(tgen.errors)
 203
 204     for rname in ["r1", "r2", "r3"]:
 205         router_compare_json_output(
 206             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 207         )
 208
 209
 210 def test_ldp_bindings():
 211     logger.info("Test: verify LDP bindings")
 212     tgen = get_topogen()
 213
 214     # Skip if previous fatal error condition is raised
 215     if tgen.routers_have_failure():
 216         pytest.skip(tgen.errors)
 217
 218     for rname in ["r1", "r2", "r3"]:
 219         router_compare_json_output(
 220             rname, "show mpls ldp binding json", "show_ldp_binding.ref"
 221         )
 222
 223
 224 def test_ldp_pwid_bindings():
 225     logger.info("Test: verify LDP PW-ID bindings")
 226     tgen = get_topogen()
 227
 228     # Skip if previous fatal error condition is raised
 229     if tgen.routers_have_failure():
 230         pytest.skip(tgen.errors)
 231
 232     for rname in ["r1", "r2", "r3"]:
 233         router_compare_json_output(
 234             rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref"
 235         )
 236
 237
 238 def test_ldp_pseudowires():
 239     logger.info("Test: verify LDP pseudowires")
 240     tgen = get_topogen()
 241
 242     # Skip if previous fatal error condition is raised
 243     if tgen.routers_have_failure():
 244         pytest.skip(tgen.errors)
 245
 246     for rname in ["r1", "r2", "r3"]:
 247         router_compare_json_output(
 248             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
 249         )
 250
 251
 252 def test_ldp_pseudowires_after_link_down():
 253     logger.info("Test: verify LDP pseudowires after r1-r2 link goes down")
 254     tgen = get_topogen()
 255
 256     # Skip if previous fatal error condition is raised
 257     if tgen.routers_have_failure():
 258         pytest.skip(tgen.errors)
 259
 260     # Shut down r1-r2 link */
 261     tgen = get_topogen()
 262     rname = "r1"
 263     tgen.gears[rname].peer_link_enable("r1-eth1", False)
 264     router_compare_json_output(
 265         rname,
 266         "show ip route json",
 267         "show_ip_route_after_link_down.ref",
 268         count=160,
 269         wait=1,
 270     )
 271     # check if the pseudowire is still up (using an alternate path
 272     # for nexthop resolution). Give some extra wait time.
 273     for rname in ["r1", "r2", "r3"]:
 274         router_compare_json_output(
 275             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref", count=160, wait=1
 276         )
 277
 278
 279 # Memory leak test template
 280 def test_memory_leak():
 281     "Run the memory leak test and report results."
 282     tgen = get_topogen()
 283     if not tgen.is_memleak_enabled():
 284         pytest.skip("Memory leak test/report is disabled")
 285
 286     tgen.report_memory_leaks()
 287
 288
 289 if __name__ == "__main__":
 290     args = ["-s"] + sys.argv[1:]
 291     sys.exit(pytest.main(args))