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