tests/topotests/ldp_snmp/test_ldp_snmp_topo1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_ldp_isis_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 from lib.snmptest import SnmpTester
  65
  66 # Required to instantiate the topology builder class.
  67
  68 pytestmark = [pytest.mark.ldpd, pytest.mark.isisd, pytest.mark.snmp]
  69
  70
  71 def build_topo(tgen):
  72     "Build function"
  73
  74     #
  75     # Define FRR Routers
  76     #
  77     for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  78         tgen.add_router(router)
  79
  80     #
  81     # Define connections
  82     #
  83     switch = tgen.add_switch("s1")
  84     switch.add_link(tgen.gears["ce1"])
  85     switch.add_link(tgen.gears["r1"])
  86
  87     switch = tgen.add_switch("s2")
  88     switch.add_link(tgen.gears["ce2"])
  89     switch.add_link(tgen.gears["r2"])
  90
  91     switch = tgen.add_switch("s3")
  92     switch.add_link(tgen.gears["ce3"])
  93     switch.add_link(tgen.gears["r3"])
  94
  95     switch = tgen.add_switch("s4")
  96     switch.add_link(tgen.gears["r1"])
  97     switch.add_link(tgen.gears["r2"])
  98
  99     switch = tgen.add_switch("s5")
 100     switch.add_link(tgen.gears["r1"])
 101     switch.add_link(tgen.gears["r3"])
 102
 103     switch = tgen.add_switch("s6")
 104     switch.add_link(tgen.gears["r2"])
 105     switch.add_link(tgen.gears["r3"])
 106
 107
 108 def setup_module(mod):
 109     "Sets up the pytest environment"
 110     tgen = Topogen(build_topo, mod.__name__)
 111     tgen.start_topology()
 112
 113     router_list = tgen.routers()
 114
 115     # For all registered routers, load the zebra configuration file
 116     for rname, router in router_list.items():
 117         router.load_config(
 118             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 119         )
 120         # Don't start isisd and ldpd in the CE nodes
 121         if router.name[0] == "r":
 122             router.load_config(
 123                 TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 124             )
 125             router.load_config(
 126                 TopoRouter.RD_LDP,
 127                 os.path.join(CWD, "{}/ldpd.conf".format(rname)),
 128                 "-M snmp",
 129             )
 130             router.load_config(
 131                 TopoRouter.RD_SNMP,
 132                 os.path.join(CWD, "{}/snmpd.conf".format(rname)),
 133                 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
 134             )
 135
 136     tgen.start_router()
 137
 138
 139 def teardown_module(mod):
 140     "Teardown the pytest environment"
 141     tgen = get_topogen()
 142
 143     # This function tears down the whole topology.
 144     tgen.stop_topology()
 145
 146
 147 def router_compare_json_output(rname, command, reference):
 148     "Compare router JSON output"
 149
 150     logger.info('Comparing router "%s" "%s" output', rname, command)
 151
 152     tgen = get_topogen()
 153     filename = "{}/{}/{}".format(CWD, rname, reference)
 154     expected = json.loads(open(filename).read())
 155
 156     # Run test function until we get an result.
 157     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 158     _, diff = topotest.run_and_expect(test_func, None, count=320, wait=0.5)
 159     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 160     assert diff is None, assertmsg
 161
 162
 163 def test_isis_convergence():
 164     logger.info("Test: check ISIS adjacencies")
 165     tgen = get_topogen()
 166
 167     # Skip if previous fatal error condition is raised
 168     if tgen.routers_have_failure():
 169         pytest.skip(tgen.errors)
 170
 171     for rname in ["r1", "r2", "r3"]:
 172         router_compare_json_output(
 173             rname,
 174             "show yang operational-data /frr-interface:lib isisd",
 175             "show_yang_interface_isis_adjacencies.ref",
 176         )
 177
 178
 179 def test_rib():
 180     logger.info("Test: verify RIB")
 181     tgen = get_topogen()
 182
 183     # Skip if previous fatal error condition is raised
 184     # TODO: disabling this check to avoid 'snmpd not running' errors
 185     # if tgen.routers_have_failure():
 186     #    pytest.skip(tgen.errors)
 187
 188     for rname in ["r1", "r2", "r3"]:
 189         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 190
 191
 192 def test_ldp_adjacencies():
 193     logger.info("Test: verify LDP adjacencies")
 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(
 203             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 204         )
 205
 206
 207 def test_ldp_neighbors():
 208     logger.info("Test: verify LDP neighbors")
 209     tgen = get_topogen()
 210
 211     # Skip if previous fatal error condition is raised
 212     # if tgen.routers_have_failure():
 213     #    pytest.skip(tgen.errors)
 214
 215     for rname in ["r1", "r2", "r3"]:
 216         router_compare_json_output(
 217             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 218         )
 219
 220
 221 def test_r1_ldp_lsr_objects():
 222     "Test mplsLdpLsrObjects objects"
 223     tgen = get_topogen()
 224
 225     r1 = tgen.gears["r1"]
 226     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 227
 228     assert r1_snmp.test_oid("mplsLdpLsrId", "01 01 01 01")
 229     assert r1_snmp.test_oid("mplsLdpLsrLoopDetectionCapable", "none(1)")
 230
 231
 232 def test_r1_ldp_entity_table():
 233     "Test mplsLdpEntityTable"
 234     tgen = get_topogen()
 235
 236     r1 = tgen.gears["r1"]
 237     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 238
 239     assert r1_snmp.test_oid_walk("mplsLdpEntityLdpId", ["1.1.1.1:0"])
 240     assert r1_snmp.test_oid_walk("mplsLdpEntityIndex", ["1"])
 241     assert r1_snmp.test_oid_walk("mplsLdpEntityProtocolVersion", ["1"])
 242     assert r1_snmp.test_oid_walk("mplsLdpEntityAdminStatus", ["enable(1)"])
 243     assert r1_snmp.test_oid_walk("mplsLdpEntityOperStatus", ["enabled(2)"])
 244     assert r1_snmp.test_oid_walk("mplsLdpEntityTcpPort", ["646"])
 245     assert r1_snmp.test_oid_walk("mplsLdpEntityUdpDscPort", ["646"])
 246     assert r1_snmp.test_oid_walk("mplsLdpEntityMaxPduLength", ["4096 octets"])
 247     assert r1_snmp.test_oid_walk("mplsLdpEntityKeepAliveHoldTimer", ["180 seconds"])
 248     assert r1_snmp.test_oid_walk("mplsLdpEntityHelloHoldTimer", ["0 seconds"])
 249     assert r1_snmp.test_oid_walk("mplsLdpEntityInitSessionThreshold", ["0"])
 250     assert r1_snmp.test_oid_walk(
 251         "mplsLdpEntityLabelDistMethod", ["downstreamUnsolicited(2)"]
 252     )
 253     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelRetentionMode", ["liberal(2)"])
 254     assert r1_snmp.test_oid_walk("mplsLdpEntityPathVectorLimit", ["0"])
 255     assert r1_snmp.test_oid_walk("mplsLdpEntityHopCountLimit", ["0"])
 256     assert r1_snmp.test_oid_walk("mplsLdpEntityTransportAddrKind", ["loopback(2)"])
 257     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeer", ["true(1)"])
 258     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddrType", ["ipv4(1)"])
 259     assert r1_snmp.test_oid_walk("mplsLdpEntityTargetPeerAddr", ["01 01 01 01"])
 260     assert r1_snmp.test_oid_walk("mplsLdpEntityLabelType", ["generic(1)"])
 261     assert r1_snmp.test_oid_walk("mplsLdpEntityDiscontinuityTime", ["(0) 0:00:00.00"])
 262     assert r1_snmp.test_oid_walk("mplsLdpEntityStorageType", ["nonVolatile(3)"])
 263     assert r1_snmp.test_oid_walk("mplsLdpEntityRowStatus", ["createAndGo(4)"])
 264
 265
 266 def test_r1_ldp_entity_stats_table():
 267     "Test mplsLdpEntityStatsTable"
 268     tgen = get_topogen()
 269
 270     r1 = tgen.gears["r1"]
 271     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 272
 273     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionAttempts", ["0"])
 274     assert r1_snmp.test_oid_walk(
 275         "mplsLdpEntityStatsSessionRejectedNoHelloErrors", ["0"]
 276     )
 277     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedAdErrors", ["0"])
 278     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedMaxPduErrors", ["0"])
 279     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsSessionRejectedLRErrors", ["0"])
 280     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadLdpIdentifierErrors", ["0"])
 281     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadPduLengthErrors", ["0"])
 282     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadMessageLengthErrors", ["0"])
 283     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsBadTlvLengthErrors", ["0"])
 284     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsMalformedTlvValueErrors", ["0"])
 285     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsKeepAliveTimerExpErrors", ["0"])
 286     assert r1_snmp.test_oid_walk(
 287         "mplsLdpEntityStatsShutdownReceivedNotifications", ["0"]
 288     )
 289     assert r1_snmp.test_oid_walk("mplsLdpEntityStatsShutdownSentNotifications", ["0"])
 290
 291
 292 def test_r1_ldp_peer_table():
 293     "Test mplsLdpPeerTable"
 294     tgen = get_topogen()
 295
 296     r1 = tgen.gears["r1"]
 297     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 298
 299     assert r1_snmp.test_oid_walk("mplsLdpPeerLdpId", ["2.2.2.2:0", "3.3.3.3:0"])
 300     assert r1_snmp.test_oid_walk(
 301         "mplsLdpPeerLabelDistMethod",
 302         ["downstreamUnsolicited(2)", "downstreamUnsolicited(2)"],
 303     )
 304     assert r1_snmp.test_oid_walk("mplsLdpPeerPathVectorLimit", ["0", "0"])
 305     assert r1_snmp.test_oid_walk("mplsLdpPeerTransportAddrType", ["ipv4(1)", "ipv4(1)"])
 306     assert r1_snmp.test_oid_walk(
 307         "mplsLdpPeerTransportAddr", ["02 02 02 02", "03 03 03 03"]
 308     )
 309
 310
 311 def test_r1_ldp_session_table():
 312     "Test mplsLdpSessionTable"
 313     tgen = get_topogen()
 314
 315     r1 = tgen.gears["r1"]
 316     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 317
 318     assert r1_snmp.test_oid_walk(
 319         "mplsLdpSessionState", ["operational(5)", "operational(5)"]
 320     )
 321     assert r1_snmp.test_oid_walk("mplsLdpSessionRole", ["passive(3)", "passive(3)"])
 322     assert r1_snmp.test_oid_walk("mplsLdpSessionProtocolVersion", ["1", "1"])
 323     assert r1_snmp.test_oid_walk(
 324         "mplsLdpSessionKeepAliveTime", ["180 seconds", "180 seconds"]
 325     )
 326     assert r1_snmp.test_oid_walk(
 327         "mplsLdpSessionMaxPduLength", ["4096 octets", "4096 octets"]
 328     )
 329     assert r1_snmp.test_oid_walk(
 330         "mplsLdpSessionDiscontinuityTime", ["(0) 0:00:00.00", "(0) 0:00:00.00"]
 331     )
 332
 333
 334 def test_r1_ldp_session_stats_table():
 335     "Test mplsLdpSessionStatsTable"
 336     tgen = get_topogen()
 337
 338     r1 = tgen.gears["r1"]
 339     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 340
 341     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownMesTypeErrors", ["0", "0"])
 342     assert r1_snmp.test_oid_walk("mplsLdpSessionStatsUnknownTlvErrors", ["0", "0"])
 343
 344
 345 def test_r1_ldp_hello_adjacency_table():
 346     "Test mplsLdpHelloAdjacencyTable"
 347     tgen = get_topogen()
 348
 349     r1 = tgen.gears["r1"]
 350     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 351
 352     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyIndex", ["1", "2", "1"])
 353     assert r1_snmp.test_oid_walk("mplsLdpHelloAdjacencyHoldTime", ["15", "45", "15"])
 354     assert r1_snmp.test_oid_walk(
 355         "mplsLdpHelloAdjacencyType", ["link(1)", "targeted(2)", "link(1)"]
 356     )
 357
 358
 359 # Memory leak test template
 360 # disabling memory leak
 361 def test_memory_leak():
 362     "Run the memory leak test and report results."
 363     tgen = get_topogen()
 364     if not tgen.is_memleak_enabled():
 365         pytest.skip("Memory leak test/report is disabled")
 366
 367     tgen.report_memory_leaks()
 368
 369
 370 if __name__ == "__main__":
 371     args = ["-s"] + sys.argv[1:]
 372     sys.exit(pytest.main(args))