#!/usr/bin/env python # # test_isis_snmp.py # Part of NetDEF Topology Tests # # Copyright (c) 2020 by Volta Networks # # Permission to use, copy, modify, and/or distribute this software # for any purpose with or without fee is hereby granted, provided # that the above copyright notice and this permission notice appear # in all copies. # # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE # OF THIS SOFTWARE. # """ test_isis_snmp.py: +---------+ 45.0.0.0/24 +---------+ | | rt4-eth1 | | | RT4 +----------------+ RT5 | | | rt5-eth1| | +---------+ +---------+ rt4-eth0| |rt5-eth0 | | 14.0.0.0/24| |25.0.0.0/24 | | rt1-eth0| |rt2-eth0 +---------+ +---------+ | | | | | RT1 | | RT2 | | 1.1.1.1 | | 2.2.2.2 | | | | | +---------+ +---------+ rt1-eth1| |rt2-eth1 | | | | 13.0.0.0/24| +---------+ |23.0.0.0/24 | | | | | | RT3 | | +--------+ 3.3.3.3 +-------+ rt3-eth1| |rt3-eth2 +---------+ |rt3-eth0 | | ce3-eth0 (172.16.1.3/24)| +---------+ | | | CE3 | | | +---------+ """ import os import sys import pytest import json from functools import partial # Save the Current Working Directory to find configuration files. CWD = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.join(CWD, "../")) # pylint: disable=C0413 # Import topogen and topotest helpers from lib import topotest from lib.topogen import Topogen, TopoRouter, get_topogen from lib.topolog import logger from lib.snmptest import SnmpTester # Required to instantiate the topology builder class. pytestmark = [pytest.mark.isisd, pytest.mark.ldpd, pytest.mark.snmp] def build_topo(tgen): "Build function" # # Define FRR Routers # for router in ["ce3", "r1", "r2", "r3", "r4", "r5"]: tgen.add_router(router) # # Define connections # switch = tgen.add_switch("s1") switch.add_link(tgen.gears["r1"]) switch.add_link(tgen.gears["r4"]) switch = tgen.add_switch("s2") switch.add_link(tgen.gears["r5"]) switch.add_link(tgen.gears["r2"]) switch = tgen.add_switch("s3") switch.add_link(tgen.gears["ce3"]) switch.add_link(tgen.gears["r3"]) switch = tgen.add_switch("s4") switch.add_link(tgen.gears["r4"]) switch.add_link(tgen.gears["r5"]) switch = tgen.add_switch("s5") switch.add_link(tgen.gears["r1"]) switch.add_link(tgen.gears["r3"]) switch = tgen.add_switch("s6") switch.add_link(tgen.gears["r2"]) switch.add_link(tgen.gears["r3"]) def setup_module(mod): "Sets up the pytest environment" # skip tests is SNMP not installed if not os.path.isfile("/usr/sbin/snmpd"): error_msg = "SNMP not installed - skipping" pytest.skip(error_msg) # This function initiates the topology build with Topogen... tgen = Topogen(build_topo, mod.__name__) # ... and here it calls Mininet initialization functions. tgen.start_topology() router_list = tgen.routers() # For all registered routers, load the zebra configuration file for rname, router in router_list.items(): router.load_config( TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) ) # Don't start the following in the CE nodes if router.name[0] == "r": router.load_config( TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname)), "-M snmp", ) router.load_config( TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname)), ) router.load_config( TopoRouter.RD_SNMP, os.path.join(CWD, "{}/snmpd.conf".format(rname)), "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap", ) # After loading the configurations, this function loads configured daemons. tgen.start_router() def teardown_module(mod): "Teardown the pytest environment" tgen = get_topogen() # This function tears down the whole topology. tgen.stop_topology() def router_compare_json_output(rname, command, reference): "Compare router JSON output" logger.info('Comparing router "%s" "%s" output', rname, command) tgen = get_topogen() filename = "{}/{}/{}".format(CWD, rname, reference) expected = json.loads(open(filename).read()) # Run test function until we get an result. Wait at most 80 seconds. test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected) _, diff = topotest.run_and_expect(test_func, None, count=160, wait=0.5) assertmsg = '"{}" JSON output mismatches the expected result'.format(rname) assert diff is None, assertmsg def generate_oid(numoids, index1, index2): if numoids == 1: oid = "{}".format(index1) else: oid = "{}.{}".format(index1, index2) return oid def test_isis_convergence(): logger.info("Test: check ISIS adjacencies") tgen = get_topogen() for rname in ["r1", "r2", "r3", "r4", "r5"]: router_compare_json_output( rname, "show yang operational-data /frr-interface:lib isisd", "show_yang_interface_isis_adjacencies.ref", ) def test_r1_scalar_snmp(): "Wait for protocol convergence" tgen = get_topogen() # Skip if previous fatal error condition is raised if tgen.routers_have_failure(): pytest.skip(tgen.errors) r1 = tgen.gears["r1"] r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c") assert r1_snmp.test_oid("isisSysVersion", "one(1)") assert r1_snmp.test_oid("isisSysLevelType", "level1and2(3)") assert r1_snmp.test_oid("isisSysID", "00 00 00 00 00 01") assert r1_snmp.test_oid("isisSysMaxPathSplits", "32") assert r1_snmp.test_oid("isisSysMaxLSPGenInt", "900 seconds") assert r1_snmp.test_oid("isisSysAdminState", "on(1)") assert r1_snmp.test_oid("isisSysMaxAge", "1200 seconds") assert r1_snmp.test_oid("isisSysProtSupported", "07 5 6 7") r2 = tgen.gears["r2"] r2_snmp = SnmpTester(r2, "2.2.2.2", "public", "2c") assert r2_snmp.test_oid("isisSysVersion", "one(1)") assert r2_snmp.test_oid("isisSysLevelType", "level1and2(3)") assert r2_snmp.test_oid("isisSysID", "00 00 00 00 00 02") assert r2_snmp.test_oid("isisSysMaxPathSplits", "32") assert r2_snmp.test_oid("isisSysMaxLSPGenInt", "900 seconds") assert r2_snmp.test_oid("isisSysAdminState", "on(1)") assert r2_snmp.test_oid("isisSysMaxAge", "1200 seconds") assert r2_snmp.test_oid("isisSysProtSupported", "07 5 6 7") circtable_test = { "isisCircAdminState": ["on(1)", "on(1)"], "isisCircExistState": ["active(1)", "active(1)"], "isisCircType": ["broadcast(1)", "ptToPt(2)"], "isisCircExtDomain": ["false(2)", "false(2)"], "isisCircLevelType": ["level1(1)", "level1(1)"], "isisCircPassiveCircuit": ["false(2)", "false(2)"], "isisCircMeshGroupEnabled": ["inactive(1)", "inactive(1)"], "isisCircSmallHellos": ["false(2)", "false(2)"], "isisCirc3WayEnabled": ["false(2)", "false(2)"], } def test_r1_isisCircTable(): tgen = get_topogen() r1 = tgen.gears["r1"] r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c") oids = [] oids.append(generate_oid(1, 1, 0)) oids.append(generate_oid(1, 2, 0)) # check items for item in circtable_test.keys(): assertmsg = "{} should be {} oids {} full dict {}:".format( item, circtable_test[item], oids, r1_snmp.walk(item) ) assert r1_snmp.test_oid_walk(item, circtable_test[item], oids), assertmsg circleveltable_test = { "isisCircLevelMetric": ["10", "10"], "isisCircLevelWideMetric": ["10", "10"], "isisCircLevelISPriority": ["64", "64"], "isisCircLevelHelloMultiplier": ["10", "10"], "isisCircLevelHelloTimer": [ "3000 milliseconds", "3000 milliseconds", ], "isisCircLevelMinLSPRetransInt": [ "1 seconds", "1 seconds", ], } def test_r1_isislevelCircTable(): tgen = get_topogen() r1 = tgen.gears["r1"] r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c") oids = [] oids.append(generate_oid(2, 1, "area")) oids.append(generate_oid(2, 2, "area")) # check items for item in circleveltable_test.keys(): assertmsg = "{} should be {} oids {} full dict {}:".format( item, circleveltable_test[item], oids, r1_snmp.walk(item) ) assert r1_snmp.test_oid_walk(item, circleveltable_test[item], oids), assertmsg adjtable_test = { "isisISAdjState": ["up(3)", "up(3)"], "isisISAdj3WayState": ["down(2)", "up(0)"], "isisISAdjNeighSysType": ["l1IntermediateSystem(1)", "l1IntermediateSystem(1)"], "isisISAdjNeighSysID": ["00 00 00 00 00 04", "00 00 00 00 00 03"], "isisISAdjUsage": ["0", "level1(1)"], "isisISAdjNeighPriority": ["64", "0"], } adjtable_down_test = { "isisISAdjState": ["up(3)"], "isisISAdj3WayState": ["down(2)"], "isisISAdjNeighSysType": ["l1IntermediateSystem(1)"], "isisISAdjNeighSysID": ["00 00 00 00 00 04"], "isisISAdjUsage": ["0"], "isisISAdjNeighPriority": ["64"], } def test_r1_isisAdjTable(): "check ISIS Adjacency Table" tgen = get_topogen() r1 = tgen.gears["r1"] r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c") oids = [] oids.append(generate_oid(2, 1, 1)) oids.append(generate_oid(2, 2, 1)) oids_down = [] oids_down.append(generate_oid(2, 1, 1)) # check items for item in adjtable_test.keys(): assertmsg = "{} should be {} oids {} full dict {}:".format( item, adjtable_test[item], oids, r1_snmp.walk(item) ) assert r1_snmp.test_oid_walk(item, adjtable_test[item], oids), assertmsg # shutdown interface and one adjacency should be removed "check ISIS adjacency is removed when interface is shutdown" r1.vtysh_cmd("conf t\ninterface r1-eth1\nshutdown") r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c") for item in adjtable_down_test.keys(): assertmsg = "{} should be {} oids {} full dict {}:".format( item, adjtable_down_test[item], oids_down, r1_snmp.walk(item) ) assert r1_snmp.test_oid_walk( item, adjtable_down_test[item], oids_down ), assertmsg # no shutdown interface and adjacency should be restored r1.vtysh_cmd("conf t\ninterface r1-eth1\nno shutdown") # Memory leak test template # disabling memory leak def test_memory_leak(): "Run the memory leak test and report results." tgen = get_topogen() if not tgen.is_memleak_enabled(): pytest.skip("Memory leak test/report is disabled") tgen.report_memory_leaks() if __name__ == "__main__": args = ["-s"] + sys.argv[1:] sys.exit(pytest.main(args))