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 sys
  64 import pytest
  65 import json
  66 from functools import partial
  67
  68 # Save the Current Working Directory to find configuration files.
  69 CWD = os.path.dirname(os.path.realpath(__file__))
  70 sys.path.append(os.path.join(CWD, "../"))
  71
  72 # pylint: disable=C0413
  73 # Import topogen and topotest helpers
  74 from lib import topotest
  75 from lib.topogen import Topogen, TopoRouter, get_topogen
  76 from lib.topolog import logger
  77 from lib.snmptest import SnmpTester
  78
  79 # Required to instantiate the topology builder class.
  80
  81 pytestmark = [pytest.mark.ldpd, pytest.mark.isisd, pytest.mark.snmp]
  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 isisd and ldpd in the CE nodes
 134         if router.name[0] == "r":
 135             router.load_config(
 136                 TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 137             )
 138             router.load_config(
 139                 TopoRouter.RD_LDP,
 140                 os.path.join(CWD, "{}/ldpd.conf".format(rname)),
 141                 "-M snmp",
 142             )
 143             router.load_config(
 144                 TopoRouter.RD_SNMP,
 145                 os.path.join(CWD, "{}/snmpd.conf".format(rname)),
 146                 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
 147             )
 148
 149     tgen.start_router()
 150
 151
 152 def teardown_module(mod):
 153     "Teardown the pytest environment"
 154     tgen = get_topogen()
 155
 156     # This function tears down the whole topology.
 157     tgen.stop_topology()
 158
 159
 160 def router_compare_json_output(rname, command, reference):
 161     "Compare router JSON output"
 162
 163     logger.info('Comparing router "%s" "%s" output', rname, command)
 164
 165     tgen = get_topogen()
 166     filename = "{}/{}/{}".format(CWD, rname, reference)
 167     expected = json.loads(open(filename).read())
 168
 169     # Run test function until we get an result.
 170     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 171     _, diff = topotest.run_and_expect(test_func, None, count=320, wait=0.5)
 172     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 173     assert diff is None, assertmsg
 174
 175
 176 def test_isis_convergence():
 177     logger.info("Test: check ISIS adjacencies")
 178     tgen = get_topogen()
 179
 180     # Skip if previous fatal error condition is raised
 181     if tgen.routers_have_failure():
 182         pytest.skip(tgen.errors)
 183
 184     for rname in ["r1", "r2", "r3"]:
 185         router_compare_json_output(
 186             rname,
 187             "show yang operational-data /frr-interface:lib isisd",
 188             "show_yang_interface_isis_adjacencies.ref",
 189         )
 190
 191
 192 def test_rib():
 193     logger.info("Test: verify RIB")
 194     tgen = get_topogen()
 195
 196     # Skip if previous fatal error condition is raised
 197     # TODO: disabling this check to avoid 'snmpd not running' errors
 198     # if tgen.routers_have_failure():
 199     #    pytest.skip(tgen.errors)
 200
 201     for rname in ["r1", "r2", "r3"]:
 202         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 203
 204
 205 def test_ldp_adjacencies():
 206     logger.info("Test: verify LDP adjacencies")
 207     tgen = get_topogen()
 208
 209     # Skip if previous fatal error condition is raised
 210     # TODO: disabling this check to avoid 'snmpd not running' errors
 211     # if tgen.routers_have_failure():
 212     #    pytest.skip(tgen.errors)
 213
 214     for rname in ["r1", "r2", "r3"]:
 215         router_compare_json_output(
 216             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 217         )
 218
 219
 220 def test_ldp_neighbors():
 221     logger.info("Test: verify LDP neighbors")
 222     tgen = get_topogen()
 223
 224     # Skip if previous fatal error condition is raised
 225     # if tgen.routers_have_failure():
 226     #    pytest.skip(tgen.errors)
 227
 228     for rname in ["r1", "r2", "r3"]:
 229         router_compare_json_output(
 230             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 231         )
 232
 233
 234 def test_r1_ldp_lsr_objects():
 235     "Test mplsLdpLsrObjects objects"
 236     tgen = get_topogen()
 237
 238     r1 = tgen.gears["r1"]
 239     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 240
 241     assert r1_snmp.test_oid("mplsLdpLsrId", "01 01 01 01")
 242     assert r1_snmp.test_oid("mplsLdpLsrLoopDetectionCapable", "none(1)")
 243
 244
 245 def test_r1_ldp_entity_table():
 246     "Test mplsLdpEntityTable"
 247     tgen = get_topogen()
 248
 249     r1 = tgen.gears["r1"]
 250     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 251
 252     assert r1_snmp.test_oid_walk("mplsLdpEntityLdpId", ["1.1.1.1:0"])
 253     assert r1_snmp.test_oid_walk("mplsLdpEntityIndex", ["1"])
 254     assert r1_snmp.test_oid_walk("mplsLdpEntityProtocolVersion", ["1"])
 255     assert r1_snmp.test_oid_walk("mplsLdpEntityAdminStatus", ["enable(1)"])
 256     assert r1_snmp.test_oid_walk("mplsLdpEntityOperStatus", ["enabled(2)"])
 257     assert r1_snmp.test_oid_walk("mplsLdpEntityTcpPort", ["646"])
 258     assert r1_snmp.test_oid_walk("mplsLdpEntityUdpDscPort", ["646"])
 259     assert r1_snmp.test_oid_walk("mplsLdpEntityMaxPduLength", ["4096 octets"])
 260     assert r1_snmp.test_oid_walk("mplsLdpEntityKeepAliveHoldTimer", ["180 seconds"])
 261     assert r1_snmp.test_oid_walk("mplsLdpEntityHelloHoldTimer", ["0 seconds"])
 262     assert r1_snmp.test_oid_walk("mplsLdpEntityInitSessionThreshold", ["0"])
 263     assert r1_snmp.test_oid_walk(
 264         "mplsLdpEntityLabelDistMethod", ["downstreamUnsolicited(2)"]
 265     )
 266     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelRetentionMode", ["liberal(2)"])
 267     assert r1_snmp.test_oid_walk("mplsLdpEntityPathVectorLimit", ["0"])
 268     assert r1_snmp.test_oid_walk("mplsLdpEntityHopCountLimit", ["0"])
 269     assert r1_snmp.test_oid_walk("mplsLdpEntityTransportAddrKind", ["loopback(2)"])
 270     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeer", ["true(1)"])
 271     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddrType", ["ipv4(1)"])
 272     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddr", ["01 01 01 01"])
 273     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelType", ["generic(1)"])
 274     assert r1_snmp.test_oid_walk("mplsLdpEntityDiscontinuityTime", ["(0) 0:00:00.00"])
 275     assert r1_snmp.test_oid_walk("mplsLdpEntityStorageType", ["nonVolatile(3)"])
 276     assert r1_snmp.test_oid_walk("mplsLdpEntityRowStatus", ["createAndGo(4)"])
 277
 278
 279 def test_r1_ldp_entity_stats_table():
 280     "Test mplsLdpEntityStatsTable"
 281     tgen = get_topogen()
 282
 283     r1 = tgen.gears["r1"]
 284     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 285
 286     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionAttempts", ["0"])
 287     assert r1_snmp.test_oid_walk(
 288         "mplsLdpEntityStatsSessionRejectedNoHelloErrors", ["0"]
 289     )
 290     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedAdErrors", ["0"])
 291     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedMaxPduErrors", ["0"])
 292     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedLRErrors", ["0"])
 293     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadLdpIdentifierErrors", ["0"])
 294     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadPduLengthErrors", ["0"])
 295     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadMessageLengthErrors", ["0"])
 296     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadTlvLengthErrors", ["0"])
 297     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsMalformedTlvValueErrors", ["0"])
 298     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsKeepAliveTimerExpErrors", ["0"])
 299     assert r1_snmp.test_oid_walk(
 300         "mplsLdpEntityStatsShutdownReceivedNotifications", ["0"]
 301     )
 302     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsShutdownSentNotifications", ["0"])
 303
 304
 305 def test_r1_ldp_peer_table():
 306     "Test mplsLdpPeerTable"
 307     tgen = get_topogen()
 308
 309     r1 = tgen.gears["r1"]
 310     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 311
 312     assert r1_snmp.test_oid_walk("mplsLdpPeerLdpId", ["2.2.2.2:0", "3.3.3.3:0"])
 313     assert r1_snmp.test_oid_walk(
 314         "mplsLdpPeerLabelDistMethod",
 315         ["downstreamUnsolicited(2)", "downstreamUnsolicited(2)"],
 316     )
 317     assert r1_snmp.test_oid_walk("mplsLdpPeerPathVectorLimit", ["0", "0"])
 318     assert r1_snmp.test_oid_walk("mplsLdpPeerTransportAddrType", ["ipv4(1)", "ipv4(1)"])
 319     assert r1_snmp.test_oid_walk(
 320         "mplsLdpPeerTransportAddr", ["02 02 02 02", "03 03 03 03"]
 321     )
 322
 323
 324 def test_r1_ldp_session_table():
 325     "Test mplsLdpSessionTable"
 326     tgen = get_topogen()
 327
 328     r1 = tgen.gears["r1"]
 329     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 330
 331     assert r1_snmp.test_oid_walk(
 332         "mplsLdpSessionState", ["operational(5)", "operational(5)"]
 333     )
 334     assert r1_snmp.test_oid_walk("mplsLdpSessionRole", ["passive(3)", "passive(3)"])
 335     assert r1_snmp.test_oid_walk("mplsLdpSessionProtocolVersion", ["1", "1"])
 336     assert r1_snmp.test_oid_walk(
 337         "mplsLdpSessionKeepAliveTime", ["180 seconds", "180 seconds"]
 338     )
 339     assert r1_snmp.test_oid_walk(
 340         "mplsLdpSessionMaxPduLength", ["4096 octets", "4096 octets"]
 341     )
 342     assert r1_snmp.test_oid_walk(
 343         "mplsLdpSessionDiscontinuityTime", ["(0) 0:00:00.00", "(0) 0:00:00.00"]
 344     )
 345
 346
 347 def test_r1_ldp_session_stats_table():
 348     "Test mplsLdpSessionStatsTable"
 349     tgen = get_topogen()
 350
 351     r1 = tgen.gears["r1"]
 352     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 353
 354     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownMesTypeErrors", ["0", "0"])
 355     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownTlvErrors", ["0", "0"])
 356
 357
 358 def test_r1_ldp_hello_adjacency_table():
 359     "Test mplsLdpHelloAdjacencyTable"
 360     tgen = get_topogen()
 361
 362     r1 = tgen.gears["r1"]
 363     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 364
 365     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyIndex", ["1", "2", "1"])
 366     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyHoldTime", ["15", "45", "15"])
 367     assert r1_snmp.test_oid_walk(
 368         "mplsLdpHelloAdjacencyType", ["link(1)", "targeted(2)", "link(1)"]
 369     )
 370
 371
 372 # Memory leak test template
 373 # disabling memory leak
 374 def test_memory_leak():
 375     "Run the memory leak test and report results."
 376     tgen = get_topogen()
 377     if not tgen.is_memleak_enabled():
 378         pytest.skip("Memory leak test/report is disabled")
 379
 380     tgen.report_memory_leaks()
 381
 382
 383 if __name__ == "__main__":
 384     args = ["-s"] + sys.argv[1:]
 385     sys.exit(pytest.main(args))