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 time import sleep
  68 from functools import partial
  69
  70 # Save the Current Working Directory to find configuration files.
  71 CWD = os.path.dirname(os.path.realpath(__file__))
  72 sys.path.append(os.path.join(CWD, '../'))
  73
  74 # pylint: disable=C0413
  75 # Import topogen and topotest helpers
  76 from lib import topotest
  77 from lib.topogen import Topogen, TopoRouter, get_topogen
  78 from lib.topolog import logger
  79
  80 # Required to instantiate the topology builder class.
  81 from mininet.topo import Topo
  82
  83 class TemplateTopo(Topo):
  84     "Test topology builder"
  85     def build(self, *_args, **_opts):
  86         "Build function"
  87         tgen = get_topogen(self)
  88
  89         #
  90         # Define FRR Routers
  91         #
  92         for router in ['ce1', 'ce2', 'ce3', 'r1', 'r2', 'r3']:
  93             tgen.add_router(router)
  94
  95         #
  96         # Define connections
  97         #
  98         switch = tgen.add_switch('s1')
  99         switch.add_link(tgen.gears['ce1'])
 100         switch.add_link(tgen.gears['r1'])
 101
 102         switch = tgen.add_switch('s2')
 103         switch.add_link(tgen.gears['ce2'])
 104         switch.add_link(tgen.gears['r2'])
 105
 106         switch = tgen.add_switch('s3')
 107         switch.add_link(tgen.gears['ce3'])
 108         switch.add_link(tgen.gears['r3'])
 109
 110         switch = tgen.add_switch('s4')
 111         switch.add_link(tgen.gears['r1'])
 112         switch.add_link(tgen.gears['r2'])
 113
 114         switch = tgen.add_switch('s5')
 115         switch.add_link(tgen.gears['r1'])
 116         switch.add_link(tgen.gears['r3'])
 117
 118         switch = tgen.add_switch('s6')
 119         switch.add_link(tgen.gears['r2'])
 120         switch.add_link(tgen.gears['r3'])
 121
 122 def setup_module(mod):
 123     "Sets up the pytest environment"
 124     tgen = Topogen(TemplateTopo, mod.__name__)
 125     tgen.start_topology()
 126
 127     router_list = tgen.routers()
 128
 129     # For all registered routers, load the zebra configuration file
 130     for rname, router in router_list.iteritems():
 131         router.load_config(
 132             TopoRouter.RD_ZEBRA,
 133             os.path.join(CWD, '{}/zebra.conf'.format(rname))
 134         )
 135         # Don't start ospfd and ldpd in the CE nodes
 136         if router.name[0] == 'r':
 137             router.load_config(
 138                 TopoRouter.RD_OSPF,
 139                 os.path.join(CWD, '{}/ospfd.conf'.format(rname))
 140             )
 141             router.load_config(
 142                 TopoRouter.RD_LDP,
 143                 os.path.join(CWD, '{}/ldpd.conf'.format(rname))
 144             )
 145
 146     tgen.start_router()
 147
 148 def teardown_module(mod):
 149     "Teardown the pytest environment"
 150     tgen = get_topogen()
 151
 152     # This function tears down the whole topology.
 153     tgen.stop_topology()
 154
 155
 156 def router_compare_json_output(rname, command, reference):
 157     "Compare router JSON output"
 158
 159     logger.info('Comparing router "%s" "%s" output', rname, command)
 160
 161     tgen = get_topogen()
 162     filename = '{}/{}/{}'.format(CWD, rname, reference)
 163     expected = json.loads(open(filename).read())
 164
 165     # Run test function until we get an result. Wait at most 80 seconds.
 166     test_func = partial(topotest.router_json_cmp,
 167         tgen.gears[rname], command, expected)
 168     _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
 169     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 170     assert diff is None, assertmsg
 171
 172 def test_ospf_convergence():
 173     logger.info("Test: check OSPF adjacencies")
 174     tgen = get_topogen()
 175
 176     # Skip if previous fatal error condition is raised
 177     if tgen.routers_have_failure():
 178         pytest.skip(tgen.errors)
 179
 180     for rname in ['r1', 'r2', 'r3']:
 181         router_compare_json_output(rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json")
 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 def test_ldp_adjacencies():
 195     logger.info("Test: verify LDP adjacencies")
 196     tgen = get_topogen()
 197
 198     # Skip if previous fatal error condition is raised
 199     if tgen.routers_have_failure():
 200         pytest.skip(tgen.errors)
 201
 202     for rname in ['r1', 'r2', 'r3']:
 203         router_compare_json_output(rname, "show mpls ldp discovery json", "show_ldp_discovery.ref")
 204
 205 def test_ldp_neighbors():
 206     logger.info("Test: verify LDP neighbors")
 207     tgen = get_topogen()
 208
 209     # Skip if previous fatal error condition is raised
 210     if tgen.routers_have_failure():
 211         pytest.skip(tgen.errors)
 212
 213     for rname in ['r1', 'r2', 'r3']:
 214         router_compare_json_output(rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref")
 215
 216 def test_ldp_bindings():
 217     logger.info("Test: verify LDP bindings")
 218     tgen = get_topogen()
 219
 220     # Skip if previous fatal error condition is raised
 221     if tgen.routers_have_failure():
 222         pytest.skip(tgen.errors)
 223
 224     for rname in ['r1', 'r2', 'r3']:
 225         router_compare_json_output(rname, "show mpls ldp binding json", "show_ldp_binding.ref")
 226
 227 def test_ldp_pwid_bindings():
 228     logger.info("Test: verify LDP PW-ID bindings")
 229     tgen = get_topogen()
 230
 231     # Skip if previous fatal error condition is raised
 232     if tgen.routers_have_failure():
 233         pytest.skip(tgen.errors)
 234
 235     for rname in ['r1', 'r2', 'r3']:
 236         router_compare_json_output(rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref")
 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(rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref")
 248
 249 def test_ldp_pseudowires_after_link_down():
 250     logger.info("Test: verify LDP pseudowires after r1-r2 link goes down")
 251     tgen = get_topogen()
 252
 253     # Skip if previous fatal error condition is raised
 254     if tgen.routers_have_failure():
 255         pytest.skip(tgen.errors)
 256
 257     # Shut down r1-r2 link */
 258     tgen = get_topogen()
 259     tgen.gears['r1'].peer_link_enable('r1-eth1', False)
 260
 261     # check if the pseudowire is still up (using an alternate path for nexthop resolution)
 262     for rname in ['r1', 'r2', 'r3']:
 263         router_compare_json_output(rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref")
 264
 265 # Memory leak test template
 266 def test_memory_leak():
 267     "Run the memory leak test and report results."
 268     tgen = get_topogen()
 269     if not tgen.is_memleak_enabled():
 270         pytest.skip('Memory leak test/report is disabled')
 271
 272     tgen.report_memory_leaks()
 273
 274 if __name__ == '__main__':
 275     args = ["-s"] + sys.argv[1:]
 276     sys.exit(pytest.main(args))