tests/topotests/isis_snmp/test_isis_snmp.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_isis_snmp.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_isis_snmp.py:
  26
  27                    +---------+  45.0.0.0/24   +---------+
  28                    |         | rt4-eth1       |         |
  29                    |   RT4   +----------------+   RT5   |
  30                    |         |        rt5-eth1|         |
  31                    +---------+                +---------+
  32                 rt4-eth0|                          |rt5-eth0
  33                         |                          |
  34              14.0.0.0/24|                          |25.0.0.0/24
  35                         |                          |
  36                 rt1-eth0|                          |rt2-eth0
  37                    +---------+                +---------+
  38                    |         |                |         |
  39                    |   RT1   |                |   RT2   |
  40                    | 1.1.1.1 |                | 2.2.2.2 |
  41                    |         |                |         |
  42                    +---------+                +---------+
  43                 rt1-eth1|                          |rt2-eth1
  44                         |                          |
  45                         |                          |
  46              13.0.0.0/24|        +---------+       |23.0.0.0/24
  47                         |        |         |       |
  48                         |        |   RT3   |       |
  49                         +--------+ 3.3.3.3 +-------+
  50                          rt3-eth1|         |rt3-eth2
  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 re
  65 import sys
  66 import pytest
  67 import json
  68 from time import sleep
  69 from functools import partial
  70
  71 # Save the Current Working Directory to find configuration files.
  72 CWD = os.path.dirname(os.path.realpath(__file__))
  73 sys.path.append(os.path.join(CWD, "../"))
  74
  75 # pylint: disable=C0413
  76 # Import topogen and topotest helpers
  77 from lib import topotest
  78 from lib.topogen import Topogen, TopoRouter, get_topogen
  79 from lib.topolog import logger
  80 from lib.snmptest import SnmpTester
  81
  82 # Required to instantiate the topology builder class.
  83 from mininet.topo import Topo
  84
  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 ["ce3", "r1", "r2", "r3", "r4", "r5"]:
  97             tgen.add_router(router)
  98
  99         #
 100         # Define connections
 101         #
 102         switch = tgen.add_switch("s1")
 103         switch.add_link(tgen.gears["r1"])
 104         switch.add_link(tgen.gears["r4"])
 105
 106         switch = tgen.add_switch("s2")
 107         switch.add_link(tgen.gears["r5"])
 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["r4"])
 116         switch.add_link(tgen.gears["r5"])
 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
 130     # skip tests is SNMP not installed
 131     if not os.path.isfile("/usr/sbin/snmpd"):
 132         error_msg = "SNMP not installed - skipping"
 133         pytest.skip(error_msg)
 134
 135     # This function initiates the topology build with Topogen...
 136     tgen = Topogen(TemplateTopo, mod.__name__)
 137     # ... and here it calls Mininet initialization functions.
 138     tgen.start_topology()
 139
 140     router_list = tgen.routers()
 141
 142     # For all registered routers, load the zebra configuration file
 143     for rname, router in router_list.iteritems():
 144         router.load_config(
 145             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 146         )
 147         # Don't start the following in the CE nodes
 148         if router.name[0] == "r":
 149             router.load_config(
 150                 TopoRouter.RD_ISIS,
 151                 os.path.join(CWD, "{}/isisd.conf".format(rname)),
 152                 "-M snmp",
 153             )
 154             router.load_config(
 155                 TopoRouter.RD_LDP,
 156                 os.path.join(CWD, "{}/ldpd.conf".format(rname)),
 157             )
 158             router.load_config(
 159                 TopoRouter.RD_SNMP,
 160                 os.path.join(CWD, "{}/snmpd.conf".format(rname)),
 161                 "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap",
 162             )
 163
 164     # After loading the configurations, this function loads configured daemons.
 165     tgen.start_router()
 166
 167
 168 def teardown_module(mod):
 169     "Teardown the pytest environment"
 170     tgen = get_topogen()
 171
 172     # This function tears down the whole topology.
 173     tgen.stop_topology()
 174
 175
 176 def router_compare_json_output(rname, command, reference):
 177     "Compare router JSON output"
 178
 179     logger.info('Comparing router "%s" "%s" output', rname, command)
 180
 181     tgen = get_topogen()
 182     filename = "{}/{}/{}".format(CWD, rname, reference)
 183     expected = json.loads(open(filename).read())
 184
 185     # Run test function until we get an result. Wait at most 80 seconds.
 186     test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
 187     _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
 188     assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
 189     assert diff is None, assertmsg
 190
 191
 192 def generate_oid(numoids, index1, index2):
 193     if numoids == 1:
 194         oid = "{}".format(index1)
 195     else:
 196         oid = "{}.{}".format(index1, index2)
 197     return oid
 198
 199
 200 def test_isis_convergence():
 201     logger.info("Test: check ISIS adjacencies")
 202     tgen = get_topogen()
 203
 204     for rname in ["r1", "r2", "r3", "r4", "r5"]:
 205         router_compare_json_output(
 206             rname,
 207             "show yang operational-data /frr-interface:lib isisd",
 208             "show_yang_interface_isis_adjacencies.ref",
 209         )
 210
 211
 212 def test_r1_scalar_snmp():
 213     "Wait for protocol convergence"
 214     tgen = get_topogen()
 215
 216     # Skip if previous fatal error condition is raised
 217     if tgen.routers_have_failure():
 218         pytest.skip(tgen.errors)
 219
 220     r1 = tgen.net.get("r1")
 221     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 222
 223     assert r1_snmp.test_oid("isisSysVersion", "one(1)")
 224     assert r1_snmp.test_oid("isisSysLevelType", "level1and2(3)")
 225     assert r1_snmp.test_oid("isisSysID", "00 00 00 00 00 01")
 226     assert r1_snmp.test_oid("isisSysMaxPathSplits", "32")
 227     assert r1_snmp.test_oid("isisSysMaxLSPGenInt", "900 seconds")
 228     assert r1_snmp.test_oid("isisSysAdminState", "on(1)")
 229     assert r1_snmp.test_oid("isisSysMaxAge", "1200 seconds")
 230     assert r1_snmp.test_oid("isisSysProtSupported", "07 5 6 7")
 231
 232     r2 = tgen.net.get("r2")
 233     r2_snmp = SnmpTester(r2, "2.2.2.2", "public", "2c")
 234
 235     assert r2_snmp.test_oid("isisSysVersion", "one(1)")
 236     assert r2_snmp.test_oid("isisSysLevelType", "level1and2(3)")
 237     assert r2_snmp.test_oid("isisSysID", "00 00 00 00 00 02")
 238     assert r2_snmp.test_oid("isisSysMaxPathSplits", "32")
 239     assert r2_snmp.test_oid("isisSysMaxLSPGenInt", "900 seconds")
 240     assert r2_snmp.test_oid("isisSysAdminState", "on(1)")
 241     assert r2_snmp.test_oid("isisSysMaxAge", "1200 seconds")
 242     assert r2_snmp.test_oid("isisSysProtSupported", "07 5 6 7")
 243
 244
 245 circtable_test = {
 246     "isisCircAdminState": ["on(1)", "on(1)"],
 247     "isisCircExistState": ["active(1)", "active(1)"],
 248     "isisCircType": ["broadcast(1)", "ptToPt(2)"],
 249     "isisCircExtDomain": ["false(2)", "false(2)"],
 250     "isisCircLevelType": ["level1(1)", "level1(1)"],
 251     "isisCircPassiveCircuit": ["false(2)", "false(2)"],
 252     "isisCircMeshGroupEnabled": ["inactive(1)", "inactive(1)"],
 253     "isisCircSmallHellos": ["false(2)", "false(2)"],
 254     "isisCirc3WayEnabled": ["false(2)", "false(2)"],
 255 }
 256
 257
 258 def test_r1_isisCircTable():
 259     tgen = get_topogen()
 260
 261     r1 = tgen.net.get("r1")
 262     r1r = tgen.gears["r1"]
 263
 264     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 265
 266     oids = []
 267     oids.append(generate_oid(1, 1, 0))
 268     oids.append(generate_oid(1, 2, 0))
 269
 270     # check items
 271     for item in circtable_test.keys():
 272         assertmsg = "{} should be {} oids {} full dict {}:".format(
 273             item, circtable_test[item], oids, r1_snmp.walk(item)
 274         )
 275         assert r1_snmp.test_oid_walk(item, circtable_test[item], oids), assertmsg
 276
 277
 278 circleveltable_test = {
 279     "isisCircLevelMetric": ["10", "10"],
 280     "isisCircLevelWideMetric": ["10", "10"],
 281     "isisCircLevelISPriority": ["64", "64"],
 282     "isisCircLevelHelloMultiplier": ["10", "10"],
 283     "isisCircLevelHelloTimer": [
 284         "3000 milliseconds",
 285         "3000 milliseconds",
 286     ],
 287     "isisCircLevelMinLSPRetransInt": [
 288         "1 seconds",
 289         "1 seconds",
 290     ],
 291 }
 292
 293
 294 def test_r1_isislevelCircTable():
 295     tgen = get_topogen()
 296
 297     r1 = tgen.net.get("r1")
 298     r1r = tgen.gears["r1"]
 299
 300     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 301
 302     oids = []
 303     oids.append(generate_oid(2, 1, "area"))
 304     oids.append(generate_oid(2, 2, "area"))
 305
 306     # check items
 307     for item in circleveltable_test.keys():
 308         assertmsg = "{} should be {} oids {} full dict {}:".format(
 309             item, circleveltable_test[item], oids, r1_snmp.walk(item)
 310         )
 311         assert r1_snmp.test_oid_walk(item, circleveltable_test[item], oids), assertmsg
 312
 313
 314 adjtable_test = {
 315     "isisISAdjState": ["up(3)", "up(3)"],
 316     "isisISAdj3WayState": ["down(2)", "up(0)"],
 317     "isisISAdjNeighSysType": ["l1IntermediateSystem(1)", "l1IntermediateSystem(1)"],
 318     "isisISAdjNeighSysID": ["00 00 00 00 00 04", "00 00 00 00 00 03"],
 319     "isisISAdjUsage": ["0", "level1(1)"],
 320     "isisISAdjNeighPriority": ["64", "0"],
 321 }
 322
 323 adjtable_down_test = {
 324     "isisISAdjState": ["up(3)"],
 325     "isisISAdj3WayState": ["down(2)"],
 326     "isisISAdjNeighSysType": ["l1IntermediateSystem(1)"],
 327     "isisISAdjNeighSysID": ["00 00 00 00 00 04"],
 328     "isisISAdjUsage": ["0"],
 329     "isisISAdjNeighPriority": ["64"],
 330 }
 331
 332
 333 def test_r1_isisAdjTable():
 334     "check ISIS Adjacency Table"
 335     tgen = get_topogen()
 336     r1 = tgen.net.get("r1")
 337     r1_cmd = tgen.gears["r1"]
 338     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 339
 340     oids = []
 341     oids.append(generate_oid(2, 1, 1))
 342     oids.append(generate_oid(2, 2, 1))
 343
 344     oids_down = []
 345     oids_down.append(generate_oid(2, 1, 1))
 346
 347     # check items
 348     for item in adjtable_test.keys():
 349         assertmsg = "{} should be {} oids {} full dict {}:".format(
 350             item, adjtable_test[item], oids, r1_snmp.walk(item)
 351         )
 352         assert r1_snmp.test_oid_walk(item, adjtable_test[item], oids), assertmsg
 353
 354     # shutdown interface and one adjacency should be removed
 355     "check ISIS adjacency is removed when interface is shutdown"
 356     r1_cmd.vtysh_cmd("conf t\ninterface r1-eth1\nshutdown")
 357     r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")
 358
 359     for item in adjtable_down_test.keys():
 360         assertmsg = "{} should be {} oids {} full dict {}:".format(
 361             item, adjtable_down_test[item], oids_down, r1_snmp.walk(item)
 362         )
 363         assert r1_snmp.test_oid_walk(
 364             item, adjtable_down_test[item], oids_down
 365         ), assertmsg
 366
 367     # no shutdown interface and adjacency should be restored
 368     r1_cmd.vtysh_cmd("conf t\ninterface r1-eth1\nno shutdown")
 369
 370
 371 # Memory leak test template
 372 # disabling memory leak
 373 def test_memory_leak():
 374     "Run the memory leak test and report results."
 375     tgen = get_topogen()
 376     if not tgen.is_memleak_enabled():
 377         pytest.skip("Memory leak test/report is disabled")
 378
 379     tgen.report_memory_leaks()
 380
 381
 382 if __name__ == "__main__":
 383     args = ["-s"] + sys.argv[1:]
 384     sys.exit(pytest.main(args))