tests/topotests/ldp_vpls_topo1/test_ldp_vpls_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ldp_vpls_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2017 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_ldp_vpls_topo1.py:
  27
  28                    +---------+                +---------+
  29                    |         |                |         |
  30                    |   CE1   |                |   CE2   |
  31                    |         |                |         |
  32                    +---------+                +---------+
  33 ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
  34                         |                          |
  35                         |                          |
  36                 rt1-eth0|                          |rt2-eth0
  37                    +---------+  10.0.1.0/24   +---------+
  38                    |         |rt1-eth1        |         |
  39                    |   RT1   +----------------+   RT2   |
  40                    | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
  41                    |         |                |         |
  42                    +---------+                +---------+
  43                 rt1-eth2|                          |rt2-eth2
  44                         |                          |
  45                         |                          |
  46              10.0.2.0/24|        +---------+       |10.0.3.0/24
  47                         |        |         |       |
  48                         |        |   RT3   |       |
  49                         +--------+ 3.3.3.3 +-------+
  50                          rt3-eth2|         |rt3-eth1
  51                                  +---------+
  52                                       |rt3-eth0
  53                                       |
  54                                       |
  55               ce3-eth0 (172.16.1.3/24)|
  56                                  +---------+
  57                                  |         |
  58                                  |   CE3   |
  59                                  |         |
  60                                  +---------+
  61 """
  62
  63 import os
  64 import sys
  65 import pytest
  66 import json
  67 from functools import partial
  68
  69 # Save the Current Working Directory to find configuration files.
  70 CWD = os.path.dirname(os.path.realpath(__file__))
  71 sys.path.append(os.path.join(CWD, "../"))
  72
  73 # pylint: disable=C0413
  74 # Import topogen and topotest helpers
  75 from lib import topotest
  76 from lib.topogen import Topogen, TopoRouter, get_topogen
  77 from lib.topolog import logger
  78
  79 # Required to instantiate the topology builder class.
  80
  81 pytestmark = [pytest.mark.ldpd, pytest.mark.ospfd]
  82
  83
  84 def build_topo(tgen):
  85     "Build function"
  86
  87     #
  88     # Define FRR Routers
  89     #
  90     for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  91         tgen.add_router(router)
  92
  93     #
  94     # Define connections
  95     #
  96     switch = tgen.add_switch("s1")
  97     switch.add_link(tgen.gears["ce1"])
  98     switch.add_link(tgen.gears["r1"])
  99
 100     switch = tgen.add_switch("s2")
 101     switch.add_link(tgen.gears["ce2"])
 102     switch.add_link(tgen.gears["r2"])
 103
 104     switch = tgen.add_switch("s3")
 105     switch.add_link(tgen.gears["ce3"])
 106     switch.add_link(tgen.gears["r3"])
 107
 108     switch = tgen.add_switch("s4")
 109     switch.add_link(tgen.gears["r1"])
 110     switch.add_link(tgen.gears["r2"])
 111
 112     switch = tgen.add_switch("s5")
 113     switch.add_link(tgen.gears["r1"])
 114     switch.add_link(tgen.gears["r3"])
 115
 116     switch = tgen.add_switch("s6")
 117     switch.add_link(tgen.gears["r2"])
 118     switch.add_link(tgen.gears["r3"])
 119
 120
 121 def setup_module(mod):
 122     "Sets up the pytest environment"
 123     tgen = Topogen(build_topo, mod.__name__)
 124     tgen.start_topology()
 125
 126     router_list = tgen.routers()
 127
 128     # For all registered routers, load the zebra configuration file
 129     for rname, router in router_list.items():
 130         router.load_config(
 131             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 132         )
 133         # Don't start ospfd and ldpd in the CE nodes
 134         if router.name[0] == "r":
 135             router.load_config(
 136                 TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
 137             )
 138             router.load_config(
 139                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 140             )
 141
 142     tgen.start_router()
 143
 144
 145 def teardown_module(mod):
 146     "Teardown the pytest environment"
 147     tgen = get_topogen()
 148
 149     # This function tears down the whole topology.
 150     tgen.stop_topology()
 151
 152
 153 def router_compare_json_output(rname, command, reference, count=80, wait=1):
 154     "Compare router JSON output"
 155
 156     logger.info('Comparing router "%s" "%s" output', rname, command)
 157
 158     tgen = get_topogen()
 159     filename = "{}/{}/{}".format(CWD, rname, reference)
 160     expected = json.loads(open(filename).read())
 161
 162     # Run test function until we get an result.
 163     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 164     _, diff = topotest.run_and_expect(test_func, None, count, wait)
 165     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 166     assert diff is None, assertmsg
 167
 168
 169 def test_ospf_convergence():
 170     logger.info("Test: check OSPF adjacencies")
 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(
 179             rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
 180         )
 181
 182
 183 def test_rib():
 184     logger.info("Test: verify RIB")
 185     tgen = get_topogen()
 186
 187     # Skip if previous fatal error condition is raised
 188     if tgen.routers_have_failure():
 189         pytest.skip(tgen.errors)
 190
 191     for rname in ["r1", "r2", "r3"]:
 192         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 193
 194
 195 def test_ldp_adjacencies():
 196     logger.info("Test: verify LDP adjacencies")
 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 discovery json", "show_ldp_discovery.ref"
 206         )
 207
 208
 209 def test_ldp_neighbors():
 210     logger.info("Test: verify LDP neighbors")
 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 neighbor json", "show_ldp_neighbor.ref"
 220         )
 221
 222
 223 def test_ldp_bindings():
 224     logger.info("Test: verify LDP 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 mpls ldp binding json", "show_ldp_binding.ref"
 234         )
 235
 236
 237 def test_ldp_pwid_bindings():
 238     logger.info("Test: verify LDP PW-ID bindings")
 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 binding json", "show_l2vpn_binding.ref"
 248         )
 249
 250
 251 def test_ldp_pseudowires():
 252     logger.info("Test: verify LDP pseudowires")
 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 l2vpn atom vc json", "show_l2vpn_vc.ref"
 262         )
 263
 264
 265 def test_ldp_pseudowires_after_link_down():
 266     logger.info("Test: verify LDP pseudowires after r1-r2 link goes down")
 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     # Shut down r1-r2 link */
 274     tgen = get_topogen()
 275     rname = "r1"
 276     tgen.gears[rname].peer_link_enable("r1-eth1", False)
 277     router_compare_json_output(
 278         rname,
 279         "show ip route json",
 280         "show_ip_route_after_link_down.ref",
 281         count=160,
 282         wait=1,
 283     )
 284     # check if the pseudowire is still up (using an alternate path
 285     # for nexthop resolution). Give some extra wait time.
 286     for rname in ["r1", "r2", "r3"]:
 287         router_compare_json_output(
 288             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref", count=160, wait=1
 289         )
 290
 291
 292 # Memory leak test template
 293 def test_memory_leak():
 294     "Run the memory leak test and report results."
 295     tgen = get_topogen()
 296     if not tgen.is_memleak_enabled():
 297         pytest.skip("Memory leak test/report is disabled")
 298
 299     tgen.report_memory_leaks()
 300
 301
 302 if __name__ == "__main__":
 303     args = ["-s"] + sys.argv[1:]
 304     sys.exit(pytest.main(args))