tests/topotests/ldp_sync_ospf_topo1/test_ldp_sync_ospf_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_ldp_ospf_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2020 by Volta Networks
   9 #
  10
  11 """
  12 test_ldp_vpls_topo1.py:
  13
  14                    +---------+                +---------+
  15                    |         |                |         |
  16                    |   CE1   |                |   CE2   |
  17                    |         |                |         |
  18                    +---------+                +---------+
  19 ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
  20                         |                          |
  21                         |                          |
  22                 rt1-eth0|                          |rt2-eth0
  23                    +---------+  10.0.1.0/24   +---------+
  24                    |         |rt1-eth1        |         |
  25                    |   RT1   +----------------+   RT2   |
  26                    | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
  27                    |         |                |         |
  28                    +---------+                +---------+
  29                 rt1-eth2|                          |rt2-eth2
  30                         |                          |
  31                         |                          |
  32              10.0.2.0/24|        +---------+       |10.0.3.0/24
  33                         |        |         |       |
  34                         |        |   RT3   |       |
  35                         +--------+ 3.3.3.3 +-------+
  36                          rt3-eth2|         |rt3-eth1
  37                                  +---------+
  38                                       |rt3-eth0
  39                                       |
  40                                       |
  41               ce3-eth0 (172.16.1.3/24)|
  42                                  +---------+
  43                                  |         |
  44                                  |   CE3   |
  45                                  |         |
  46                                  +---------+
  47 """
  48
  49 import os
  50 import sys
  51 import pytest
  52 import json
  53 from functools import partial
  54
  55 # Save the Current Working Directory to find configuration files.
  56 CWD = os.path.dirname(os.path.realpath(__file__))
  57 sys.path.append(os.path.join(CWD, "../"))
  58
  59 # pylint: disable=C0413
  60 # Import topogen and topotest helpers
  61 from lib import topotest
  62 from lib.topogen import Topogen, TopoRouter, get_topogen
  63 from lib.topolog import logger
  64
  65 # Required to instantiate the topology builder class.
  66
  67 pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
  68
  69
  70 def build_topo(tgen):
  71     "Build function"
  72
  73     #
  74     # Define FRR Routers
  75     #
  76     for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  77         tgen.add_router(router)
  78
  79     #
  80     # Define connections
  81     #
  82     switch = tgen.add_switch("s1")
  83     switch.add_link(tgen.gears["ce1"])
  84     switch.add_link(tgen.gears["r1"])
  85
  86     switch = tgen.add_switch("s2")
  87     switch.add_link(tgen.gears["ce2"])
  88     switch.add_link(tgen.gears["r2"])
  89
  90     switch = tgen.add_switch("s3")
  91     switch.add_link(tgen.gears["ce3"])
  92     switch.add_link(tgen.gears["r3"])
  93
  94     switch = tgen.add_switch("s4")
  95     switch.add_link(tgen.gears["r1"])
  96     switch.add_link(tgen.gears["r2"])
  97
  98     switch = tgen.add_switch("s5")
  99     switch.add_link(tgen.gears["r1"])
 100     switch.add_link(tgen.gears["r3"])
 101
 102     switch = tgen.add_switch("s6")
 103     switch.add_link(tgen.gears["r2"])
 104     switch.add_link(tgen.gears["r3"])
 105
 106
 107 def setup_module(mod):
 108     "Sets up the pytest environment"
 109     tgen = Topogen(build_topo, mod.__name__)
 110     tgen.start_topology()
 111
 112     router_list = tgen.routers()
 113
 114     # For all registered routers, load the zebra configuration file
 115     for rname, router in router_list.items():
 116         router.load_config(
 117             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 118         )
 119         # Don't start ospfd and ldpd in the CE nodes
 120         if router.name[0] == "r":
 121             router.load_config(
 122                 TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 123             )
 124             router.load_config(
 125                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 126             )
 127
 128     tgen.start_router()
 129
 130
 131 def teardown_module(mod):
 132     "Teardown the pytest environment"
 133     tgen = get_topogen()
 134
 135     # This function tears down the whole topology.
 136     tgen.stop_topology()
 137
 138
 139 def router_compare_json_output(rname, command, reference):
 140     "Compare router JSON output"
 141
 142     logger.info('Comparing router "%s" "%s" output', rname, command)
 143
 144     tgen = get_topogen()
 145     filename = "{}/{}/{}".format(CWD, rname, reference)
 146     expected = json.loads(open(filename).read())
 147
 148     # Run test function until we get an result.
 149     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 150     _, diff = topotest.run_and_expect(test_func, None, count=320, wait=0.5)
 151     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 152     assert diff is None, assertmsg
 153
 154
 155 def test_ospf_convergence():
 156     logger.info("Test: check OSPF adjacencies")
 157     tgen = get_topogen()
 158
 159     # Skip if previous fatal error condition is raised
 160     if tgen.routers_have_failure():
 161         pytest.skip(tgen.errors)
 162
 163     for rname in ["r1", "r2", "r3"]:
 164         router_compare_json_output(
 165             rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
 166         )
 167
 168
 169 def test_rib():
 170     logger.info("Test: verify RIB")
 171     tgen = get_topogen()
 172
 173     # Skip if previous fatal error condition is raised
 174     if tgen.routers_have_failure():
 175         pytest.skip(tgen.errors)
 176
 177     for rname in ["r1", "r2", "r3"]:
 178         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 179
 180
 181 def test_ldp_adjacencies():
 182     logger.info("Test: verify LDP adjacencies")
 183     tgen = get_topogen()
 184
 185     # Skip if previous fatal error condition is raised
 186     if tgen.routers_have_failure():
 187         pytest.skip(tgen.errors)
 188
 189     for rname in ["r1", "r2", "r3"]:
 190         router_compare_json_output(
 191             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 192         )
 193
 194
 195 def test_ldp_neighbors():
 196     logger.info("Test: verify LDP neighbors")
 197     tgen = get_topogen()
 198
 199     # Skip if previous fatal error condition is raised
 200     if tgen.routers_have_failure():
 201         pytest.skip(tgen.errors)
 202
 203     for rname in ["r1", "r2", "r3"]:
 204         router_compare_json_output(
 205             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 206         )
 207
 208
 209 def test_ldp_bindings():
 210     logger.info("Test: verify LDP bindings")
 211     tgen = get_topogen()
 212
 213     # Skip if previous fatal error condition is raised
 214     if tgen.routers_have_failure():
 215         pytest.skip(tgen.errors)
 216
 217     for rname in ["r1", "r2", "r3"]:
 218         router_compare_json_output(
 219             rname, "show mpls ldp binding json", "show_ldp_binding.ref"
 220         )
 221
 222
 223 def test_ldp_pwid_bindings():
 224     logger.info("Test: verify LDP PW-ID bindings")
 225     tgen = get_topogen()
 226
 227     # Skip if previous fatal error condition is raised
 228     if tgen.routers_have_failure():
 229         pytest.skip(tgen.errors)
 230
 231     for rname in ["r1", "r2", "r3"]:
 232         router_compare_json_output(
 233             rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref"
 234         )
 235
 236
 237 def test_ldp_pseudowires():
 238     logger.info("Test: verify LDP pseudowires")
 239     tgen = get_topogen()
 240
 241     # Skip if previous fatal error condition is raised
 242     if tgen.routers_have_failure():
 243         pytest.skip(tgen.errors)
 244
 245     for rname in ["r1", "r2", "r3"]:
 246         router_compare_json_output(
 247             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
 248         )
 249
 250
 251 def test_ldp_igp_sync():
 252     logger.info("Test: verify LDP igp-sync")
 253     tgen = get_topogen()
 254
 255     # Skip if previous fatal error condition is raised
 256     if tgen.routers_have_failure():
 257         pytest.skip(tgen.errors)
 258
 259     for rname in ["r1", "r2", "r3"]:
 260         router_compare_json_output(
 261             rname, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
 262         )
 263
 264
 265 def test_ospf_ldp_sync():
 266     logger.info("Test: verify OSPF igp-sync")
 267     tgen = get_topogen()
 268
 269     # Skip if previous fatal error condition is raised
 270     if tgen.routers_have_failure():
 271         pytest.skip(tgen.errors)
 272
 273     for rname in ["r1", "r2", "r3"]:
 274         router_compare_json_output(
 275             rname, "show ip ospf mpls ldp-sync json", "show_ospf_ldp_sync.ref"
 276         )
 277
 278     for rname in ["r1", "r2", "r3"]:
 279         router_compare_json_output(
 280             rname, "show ip ospf interface json", "show_ip_ospf_interface.ref"
 281         )
 282
 283
 284 def test_r1_eth1_shutdown():
 285     logger.info("Test: verify behaviour after r1-eth1 is shutdown")
 286     tgen = get_topogen()
 287
 288     # Skip if previous fatal error condition is raised
 289     if tgen.routers_have_failure():
 290         pytest.skip(tgen.errors)
 291
 292     # Shut down r1-r2 link */
 293     tgen = get_topogen()
 294     tgen.gears["r1"].peer_link_enable("r1-eth1", False)
 295     topotest.sleep(5, "Waiting for the network to reconverge")
 296
 297     # check if the pseudowire is still up (using an alternate path for nexthop resolution)
 298     for rname in ["r1", "r2", "r3"]:
 299         router_compare_json_output(
 300             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
 301         )
 302
 303     for rname in ["r1", "r2", "r3"]:
 304         router_compare_json_output(
 305             rname,
 306             "show mpls ldp igp-sync json",
 307             "show_ldp_igp_sync_r1_eth1_shutdown.ref",
 308         )
 309
 310     for rname in ["r1", "r2", "r3"]:
 311         router_compare_json_output(
 312             rname,
 313             "show ip ospf mpls ldp-sync json",
 314             "show_ospf_ldp_sync_r1_eth1_shutdown.ref",
 315         )
 316
 317     for rname in ["r1", "r2", "r3"]:
 318         router_compare_json_output(
 319             rname,
 320             "show ip ospf interface json",
 321             "show_ip_ospf_interface_r1_eth1_shutdown.ref",
 322         )
 323
 324
 325 def test_r1_eth1_no_shutdown():
 326     logger.info("Test: verify behaviour after r1-eth1 is no shutdown")
 327     tgen = get_topogen()
 328
 329     # Skip if previous fatal error condition is raised
 330     if tgen.routers_have_failure():
 331         pytest.skip(tgen.errors)
 332
 333     # Run no shutdown on r1-eth1 interface */
 334     tgen = get_topogen()
 335     tgen.gears["r1"].peer_link_enable("r1-eth1", True)
 336     topotest.sleep(5, "Waiting for the network to reconverge")
 337
 338     for rname in ["r1", "r2", "r3"]:
 339         router_compare_json_output(
 340             rname, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
 341         )
 342
 343     for rname in ["r1", "r2", "r3"]:
 344         router_compare_json_output(
 345             rname, "show ip ospf mpls ldp-sync json", "show_ospf_ldp_sync.ref"
 346         )
 347
 348     for rname in ["r1", "r2", "r3"]:
 349         router_compare_json_output(
 350             rname, "show ip ospf interface json", "show_ip_ospf_interface.ref"
 351         )
 352
 353
 354 def test_r2_eth1_shutdown():
 355     logger.info("Test: verify behaviour after r2-eth1 is shutdown")
 356     tgen = get_topogen()
 357
 358     # Skip if previous fatal error condition is raised
 359     if tgen.routers_have_failure():
 360         pytest.skip(tgen.errors)
 361
 362     # Shut down r1-r2 link */
 363     tgen = get_topogen()
 364     tgen.gears["r2"].peer_link_enable("r2-eth1", False)
 365     topotest.sleep(5, "Waiting for the network to reconverge")
 366
 367     for rname in ["r1", "r2", "r3"]:
 368         router_compare_json_output(
 369             rname,
 370             "show mpls ldp igp-sync json",
 371             "show_ldp_igp_sync_r1_eth1_shutdown.ref",
 372         )
 373
 374     for rname in ["r1", "r2", "r3"]:
 375         router_compare_json_output(
 376             rname,
 377             "show ip ospf mpls ldp-sync json",
 378             "show_ospf_ldp_sync_r2_eth1_shutdown.ref",
 379         )
 380
 381     for rname in ["r1", "r2", "r3"]:
 382         router_compare_json_output(
 383             rname,
 384             "show ip ospf interface json",
 385             "show_ip_ospf_interface_r2_eth1_shutdown.ref",
 386         )
 387
 388
 389 def test_r2_eth1_no_shutdown():
 390     logger.info("Test: verify behaviour after r2-eth1 is no shutdown")
 391     tgen = get_topogen()
 392
 393     # Skip if previous fatal error condition is raised
 394     if tgen.routers_have_failure():
 395         pytest.skip(tgen.errors)
 396
 397     # Run no shutdown on r2-eth1 interface */
 398     tgen = get_topogen()
 399     tgen.gears["r2"].peer_link_enable("r2-eth1", True)
 400     topotest.sleep(5, "Waiting for the network to reconverge")
 401
 402     for rname in ["r1", "r2", "r3"]:
 403         router_compare_json_output(
 404             rname, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
 405         )
 406
 407     for rname in ["r1", "r2", "r3"]:
 408         router_compare_json_output(
 409             rname, "show ip ospf mpls ldp-sync json", "show_ospf_ldp_sync.ref"
 410         )
 411
 412     for rname in ["r1", "r2", "r3"]:
 413         router_compare_json_output(
 414             rname, "show ip ospf interface json", "show_ip_ospf_interface.ref"
 415         )
 416
 417
 418 # Memory leak test template
 419 def test_memory_leak():
 420     "Run the memory leak test and report results."
 421     tgen = get_topogen()
 422     if not tgen.is_memleak_enabled():
 423         pytest.skip("Memory leak test/report is disabled")
 424
 425     tgen.report_memory_leaks()
 426
 427
 428 if __name__ == "__main__":
 429     args = ["-s"] + sys.argv[1:]
 430     sys.exit(pytest.main(args))