tests/topotests/ldp_snmp/test_ldp_snmp_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ldp_isis_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by Volta Networks
   8 #
   9 # Permission to use, copy, modify, and/or distribute this software
  10 # for any purpose with or without fee is hereby granted, provided
  11 # that the above copyright notice and this permission notice appear
  12 # in all copies.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 """
  25 test_ldp_vpls_topo1.py:
  26
  27                    +---------+                +---------+
  28                    |         |                |         |
  29                    |   CE1   |                |   CE2   |
  30                    |         |                |         |
  31                    +---------+                +---------+
  32 ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
  33                         |                          |
  34                         |                          |
  35                 rt1-eth0|                          |rt2-eth0
  36                    +---------+  10.0.1.0/24   +---------+
  37                    |         |rt1-eth1        |         |
  38                    |   RT1   +----------------+   RT2   |
  39                    | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
  40                    |         |                |         |
  41                    +---------+                +---------+
  42                 rt1-eth2|                          |rt2-eth2
  43                         |                          |
  44                         |                          |
  45              10.0.2.0/24|        +---------+       |10.0.3.0/24
  46                         |        |         |       |
  47                         |        |   RT3   |       |
  48                         +--------+ 3.3.3.3 +-------+
  49                          rt3-eth2|         |rt3-eth1
  50                                  +---------+
  51                                       |rt3-eth0
  52                                       |
  53                                       |
  54               ce3-eth0 (172.16.1.3/24)|
  55                                  +---------+
  56                                  |         |
  57                                  |   CE3   |
  58                                  |         |
  59                                  +---------+
  60 """
  61
  62 import os
  63 import re
  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 from lib.snmptest import SnmpTester
  80
  81 # Required to instantiate the topology builder class.
  82 from mininet.topo import Topo
  83
  84
  85 class TemplateTopo(Topo):
  86     "Test topology builder"
  87
  88     def build(self, *_args, **_opts):
  89         "Build function"
  90         tgen = get_topogen(self)
  91
  92         #
  93         # Define FRR Routers
  94         #
  95         for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  96             tgen.add_router(router)
  97
  98         #
  99         # Define connections
 100         #
 101         switch = tgen.add_switch("s1")
 102         switch.add_link(tgen.gears["ce1"])
 103         switch.add_link(tgen.gears["r1"])
 104
 105         switch = tgen.add_switch("s2")
 106         switch.add_link(tgen.gears["ce2"])
 107         switch.add_link(tgen.gears["r2"])
 108
 109         switch = tgen.add_switch("s3")
 110         switch.add_link(tgen.gears["ce3"])
 111         switch.add_link(tgen.gears["r3"])
 112
 113         switch = tgen.add_switch("s4")
 114         switch.add_link(tgen.gears["r1"])
 115         switch.add_link(tgen.gears["r2"])
 116
 117         switch = tgen.add_switch("s5")
 118         switch.add_link(tgen.gears["r1"])
 119         switch.add_link(tgen.gears["r3"])
 120
 121         switch = tgen.add_switch("s6")
 122         switch.add_link(tgen.gears["r2"])
 123         switch.add_link(tgen.gears["r3"])
 124
 125
 126 def setup_module(mod):
 127     "Sets up the pytest environment"
 128     tgen = Topogen(TemplateTopo, mod.__name__)
 129     tgen.start_topology()
 130
 131     router_list = tgen.routers()
 132
 133     # For all registered routers, load the zebra configuration file
 134     for rname, router in router_list.items():
 135         router.load_config(
 136             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 137         )
 138         # Don't start isisd and ldpd in the CE nodes
 139         if router.name[0] == "r":
 140             router.load_config(
 141                 TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 142             )
 143             router.load_config(
 144                 TopoRouter.RD_LDP,
 145                 os.path.join(CWD, "{}/ldpd.conf".format(rname)),
 146                 "-M snmp",
 147             )
 148             router.load_config(
 149                 TopoRouter.RD_SNMP,
 150                 os.path.join(CWD, "{}/snmpd.conf".format(rname)),
 151                 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
 152             )
 153
 154     tgen.start_router()
 155
 156
 157 def teardown_module(mod):
 158     "Teardown the pytest environment"
 159     tgen = get_topogen()
 160
 161     # This function tears down the whole topology.
 162     tgen.stop_topology()
 163
 164
 165 def router_compare_json_output(rname, command, reference):
 166     "Compare router JSON output"
 167
 168     logger.info('Comparing router "%s" "%s" output', rname, command)
 169
 170     tgen = get_topogen()
 171     filename = "{}/{}/{}".format(CWD, rname, reference)
 172     expected = json.loads(open(filename).read())
 173
 174     # Run test function until we get an result.
 175     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 176     _, diff = topotest.run_and_expect(test_func, None, count=320, wait=0.5)
 177     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 178     assert diff is None, assertmsg
 179
 180
 181 def test_isis_convergence():
 182     logger.info("Test: check ISIS 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,
 192             "show yang operational-data /frr-interface:lib isisd",
 193             "show_yang_interface_isis_adjacencies.ref",
 194         )
 195
 196
 197 def test_rib():
 198     logger.info("Test: verify RIB")
 199     tgen = get_topogen()
 200
 201     # Skip if previous fatal error condition is raised
 202     # TODO: disabling this check to avoid 'snmpd not running' errors
 203     # if tgen.routers_have_failure():
 204     #    pytest.skip(tgen.errors)
 205
 206     for rname in ["r1", "r2", "r3"]:
 207         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 208
 209
 210 def test_ldp_adjacencies():
 211     logger.info("Test: verify LDP adjacencies")
 212     tgen = get_topogen()
 213
 214     # Skip if previous fatal error condition is raised
 215     # TODO: disabling this check to avoid 'snmpd not running' errors
 216     # if tgen.routers_have_failure():
 217     #    pytest.skip(tgen.errors)
 218
 219     for rname in ["r1", "r2", "r3"]:
 220         router_compare_json_output(
 221             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 222         )
 223
 224
 225 def test_ldp_neighbors():
 226     logger.info("Test: verify LDP neighbors")
 227     tgen = get_topogen()
 228
 229     # Skip if previous fatal error condition is raised
 230     # if tgen.routers_have_failure():
 231     #    pytest.skip(tgen.errors)
 232
 233     for rname in ["r1", "r2", "r3"]:
 234         router_compare_json_output(
 235             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 236         )
 237
 238
 239 def test_r1_ldp_lsr_objects():
 240     "Test mplsLdpLsrObjects objects"
 241     tgen = get_topogen()
 242
 243     r1 = tgen.net.get("r1")
 244     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 245
 246     assert r1_snmp.test_oid("mplsLdpLsrId", "01 01 01 01")
 247     assert r1_snmp.test_oid("mplsLdpLsrLoopDetectionCapable", "none(1)")
 248
 249
 250 def test_r1_ldp_entity_table():
 251     "Test mplsLdpEntityTable"
 252     tgen = get_topogen()
 253
 254     r1 = tgen.net.get("r1")
 255     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 256
 257     assert r1_snmp.test_oid_walk("mplsLdpEntityLdpId", ["1.1.1.1:0"])
 258     assert r1_snmp.test_oid_walk("mplsLdpEntityIndex", ["1"])
 259     assert r1_snmp.test_oid_walk("mplsLdpEntityProtocolVersion", ["1"])
 260     assert r1_snmp.test_oid_walk("mplsLdpEntityAdminStatus", ["enable(1)"])
 261     assert r1_snmp.test_oid_walk("mplsLdpEntityOperStatus", ["enabled(2)"])
 262     assert r1_snmp.test_oid_walk("mplsLdpEntityTcpPort", ["646"])
 263     assert r1_snmp.test_oid_walk("mplsLdpEntityUdpDscPort", ["646"])
 264     assert r1_snmp.test_oid_walk("mplsLdpEntityMaxPduLength", ["4096 octets"])
 265     assert r1_snmp.test_oid_walk("mplsLdpEntityKeepAliveHoldTimer", ["180 seconds"])
 266     assert r1_snmp.test_oid_walk("mplsLdpEntityHelloHoldTimer", ["0 seconds"])
 267     assert r1_snmp.test_oid_walk("mplsLdpEntityInitSessionThreshold", ["0"])
 268     assert r1_snmp.test_oid_walk(
 269         "mplsLdpEntityLabelDistMethod", ["downstreamUnsolicited(2)"]
 270     )
 271     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelRetentionMode", ["liberal(2)"])
 272     assert r1_snmp.test_oid_walk("mplsLdpEntityPathVectorLimit", ["0"])
 273     assert r1_snmp.test_oid_walk("mplsLdpEntityHopCountLimit", ["0"])
 274     assert r1_snmp.test_oid_walk("mplsLdpEntityTransportAddrKind", ["loopback(2)"])
 275     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeer", ["true(1)"])
 276     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddrType", ["ipv4(1)"])
 277     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddr", ["01 01 01 01"])
 278     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelType", ["generic(1)"])
 279     assert r1_snmp.test_oid_walk("mplsLdpEntityDiscontinuityTime", ["(0) 0:00:00.00"])
 280     assert r1_snmp.test_oid_walk("mplsLdpEntityStorageType", ["nonVolatile(3)"])
 281     assert r1_snmp.test_oid_walk("mplsLdpEntityRowStatus", ["createAndGo(4)"])
 282
 283
 284 def test_r1_ldp_entity_stats_table():
 285     "Test mplsLdpEntityStatsTable"
 286     tgen = get_topogen()
 287
 288     r1 = tgen.net.get("r1")
 289     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 290
 291     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionAttempts", ["0"])
 292     assert r1_snmp.test_oid_walk(
 293         "mplsLdpEntityStatsSessionRejectedNoHelloErrors", ["0"]
 294     )
 295     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedAdErrors", ["0"])
 296     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedMaxPduErrors", ["0"])
 297     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedLRErrors", ["0"])
 298     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadLdpIdentifierErrors", ["0"])
 299     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadPduLengthErrors", ["0"])
 300     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadMessageLengthErrors", ["0"])
 301     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadTlvLengthErrors", ["0"])
 302     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsMalformedTlvValueErrors", ["0"])
 303     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsKeepAliveTimerExpErrors", ["0"])
 304     assert r1_snmp.test_oid_walk(
 305         "mplsLdpEntityStatsShutdownReceivedNotifications", ["0"]
 306     )
 307     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsShutdownSentNotifications", ["0"])
 308
 309
 310 def test_r1_ldp_peer_table():
 311     "Test mplsLdpPeerTable"
 312     tgen = get_topogen()
 313
 314     r1 = tgen.net.get("r1")
 315     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 316
 317     assert r1_snmp.test_oid_walk("mplsLdpPeerLdpId", ["2.2.2.2:0", "3.3.3.3:0"])
 318     assert r1_snmp.test_oid_walk(
 319         "mplsLdpPeerLabelDistMethod",
 320         ["downstreamUnsolicited(2)", "downstreamUnsolicited(2)"],
 321     )
 322     assert r1_snmp.test_oid_walk("mplsLdpPeerPathVectorLimit", ["0", "0"])
 323     assert r1_snmp.test_oid_walk("mplsLdpPeerTransportAddrType", ["ipv4(1)", "ipv4(1)"])
 324     assert r1_snmp.test_oid_walk(
 325         "mplsLdpPeerTransportAddr", ["02 02 02 02", "03 03 03 03"]
 326     )
 327
 328
 329 def test_r1_ldp_session_table():
 330     "Test mplsLdpSessionTable"
 331     tgen = get_topogen()
 332
 333     r1 = tgen.net.get("r1")
 334     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 335
 336     assert r1_snmp.test_oid_walk(
 337         "mplsLdpSessionState", ["operational(5)", "operational(5)"]
 338     )
 339     assert r1_snmp.test_oid_walk("mplsLdpSessionRole", ["passive(3)", "passive(3)"])
 340     assert r1_snmp.test_oid_walk("mplsLdpSessionProtocolVersion", ["1", "1"])
 341     assert r1_snmp.test_oid_walk(
 342         "mplsLdpSessionKeepAliveTime", ["180 seconds", "180 seconds"]
 343     )
 344     assert r1_snmp.test_oid_walk(
 345         "mplsLdpSessionMaxPduLength", ["4096 octets", "4096 octets"]
 346     )
 347     assert r1_snmp.test_oid_walk(
 348         "mplsLdpSessionDiscontinuityTime", ["(0) 0:00:00.00", "(0) 0:00:00.00"]
 349     )
 350
 351
 352 def test_r1_ldp_session_stats_table():
 353     "Test mplsLdpSessionStatsTable"
 354     tgen = get_topogen()
 355
 356     r1 = tgen.net.get("r1")
 357     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 358
 359     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownMesTypeErrors", ["0", "0"])
 360     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownTlvErrors", ["0", "0"])
 361
 362
 363 def test_r1_ldp_hello_adjacency_table():
 364     "Test mplsLdpHelloAdjacencyTable"
 365     tgen = get_topogen()
 366
 367     r1 = tgen.net.get("r1")
 368     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 369
 370     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyIndex", ["1", "2", "1"])
 371     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyHoldTime", ["15", "45", "15"])
 372     assert r1_snmp.test_oid_walk(
 373         "mplsLdpHelloAdjacencyType", ["link(1)", "targeted(2)", "link(1)"]
 374     )
 375
 376
 377 # Memory leak test template
 378 # disabling memory leak
 379 def test_memory_leak():
 380     "Run the memory leak test and report results."
 381     tgen = get_topogen()
 382     if not tgen.is_memleak_enabled():
 383         pytest.skip("Memory leak test/report is disabled")
 384
 385     tgen.report_memory_leaks()
 386
 387
 388 if __name__ == "__main__":
 389     args = ["-s"] + sys.argv[1:]
 390     sys.exit(pytest.main(args))