#!/usr/bin/env python # SPDX-License-Identifier: ISC # # Part of NetDEF Topology Tests # # Copyright (c) 2017 by # Network Device Education Foundation, Inc. ("NetDEF") # """ ltemplate.py: LabN template for FRR tests. """ import os import sys import platform import pytest # 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.lutil import * # Required to instantiate the topology builder class. customize = None class LTemplate: test = None testdir = None scriptdir = None logdir = None prestarthooksuccess = True poststarthooksuccess = True iproute2Ver = None def __init__(self, test, testdir): pathname = os.path.join(testdir, "customize.py") global customize if sys.version_info >= (3, 5): import importlib.util spec = importlib.util.spec_from_file_location("customize", pathname) customize = importlib.util.module_from_spec(spec) spec.loader.exec_module(customize) else: import imp customize = imp.load_source("customize", pathname) self.test = test self.testdir = testdir self.scriptdir = testdir self.logdir = "" logger.info("LTemplate: " + test) def setup_module(self, mod): "Sets up the pytest environment" # This function initiates the topology build with Topogen... tgen = Topogen(customize.build_topo, mod.__name__) # ... and here it calls Mininet initialization functions. tgen.start_topology() self.logdir = tgen.logdir logger.info("Topology started") try: self.prestarthooksuccess = customize.ltemplatePreRouterStartHook() except AttributeError: # not defined logger.debug("ltemplatePreRouterStartHook() not defined") if self.prestarthooksuccess != True: logger.info("ltemplatePreRouterStartHook() failed, skipping test") return # This is a sample of configuration loading. router_list = tgen.routers() # For all registered routers, load the zebra configuration file for rname, router in router_list.items(): logger.info("Setting up %s" % rname) for rd_val in TopoRouter.RD: config = os.path.join( self.testdir, "{}/{}.conf".format(rname, TopoRouter.RD[rd_val]) ) prog = os.path.join(tgen.net[rname].daemondir, TopoRouter.RD[rd_val]) if os.path.exists(config): if os.path.exists(prog): router.load_config(rd_val, config) else: logger.warning( "{} not found, but have {}.conf file".format( prog, TopoRouter.RD[rd_val] ) ) # After loading the configurations, this function loads configured daemons. logger.info("Starting routers") tgen.start_router() try: self.poststarthooksuccess = customize.ltemplatePostRouterStartHook() except AttributeError: # not defined logger.debug("ltemplatePostRouterStartHook() not defined") luStart(baseScriptDir=self.scriptdir, baseLogDir=self.logdir, net=tgen.net) # initialized by ltemplate_start _lt = None def setup_module(mod): global _lt root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) test = mod.__name__[: mod.__name__.rfind(".")] testdir = os.path.join(root, test) # don't do this for now as reload didn't work as expected # fixup sys.path, want test dir there only once # try: # sys.path.remove(testdir) # except ValueError: # logger.debug(testdir+" not found in original sys.path") # add testdir # sys.path.append(testdir) # init class _lt = LTemplate(test, testdir) _lt.setup_module(mod) # drop testdir # sys.path.remove(testdir) def teardown_module(mod): global _lt "Teardown the pytest environment" tgen = get_topogen() if _lt != None and _lt.scriptdir != None and _lt.prestarthooksuccess == True: luShowResults(logger.info) print(luFinish()) # This function tears down the whole topology. tgen.stop_topology() _lt = None def ltemplateTest( script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, KeepGoing=False ): global _lt if _lt == None or _lt.prestarthooksuccess != True: return tgen = get_topogen() if not os.path.isfile(script): if not os.path.isfile(os.path.join(_lt.scriptdir, script)): logger.error("Could not find script file: " + script) assert "Could not find script file: " + script logger.info("Starting template test: " + script) numEntry = luNumFail() if SkipIfFailed and tgen.routers_have_failure(): pytest.skip(tgen.errors) if numEntry > 0: if not KeepGoing: pytest.skip("Have %d errors" % numEntry) if CheckFuncStr != None: check = eval(CheckFuncStr) if check != True: pytest.skip("Check function '" + CheckFuncStr + "' returned: " + check) if CallOnFail != None: CallOnFail = eval(CallOnFail) luInclude(script, CallOnFail) numFail = luNumFail() - numEntry if numFail > 0: luShowFail() fatal_error = "%d tests failed" % numFail if not KeepGoing: assert ( "scripts/cleanup_all.py failed" == "See summary output above" ), fatal_error # Memory leak test template 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() class ltemplateRtrCmd: def __init__(self): self.resetCounts() def doCmd(self, tgen, rtr, cmd, checkstr=None): logger.info("doCmd: {} {}".format(rtr, cmd)) output = tgen.net[rtr].cmd(cmd).strip() if len(output): self.output += 1 if checkstr != None: ret = re.search(checkstr, output) if ret == None: self.nomatch += 1 else: self.match += 1 return ret logger.info("output: " + output) else: logger.info("No output") self.none += 1 return None def resetCounts(self): self.match = 0 self.nomatch = 0 self.output = 0 self.none = 0 def getMatch(self): return self.match def getNoMatch(self): return self.nomatch def getOutput(self): return self.output def getNone(self): return self.none def ltemplateVersionCheck( vstr, rname="r1", compstr="<", cli=False, kernel="4.9", iproute2=None, mpls=True ): tgen = get_topogen() router = tgen.gears[rname] if cli: logger.info("calling mininet CLI") tgen.mininet_cli() logger.info("exited mininet CLI") if _lt == None: ret = "Template not initialized" return ret if _lt.prestarthooksuccess != True: ret = "ltemplatePreRouterStartHook failed" return ret if _lt.poststarthooksuccess != True: ret = "ltemplatePostRouterStartHook failed" return ret if mpls == True and tgen.hasmpls != True: ret = "MPLS not initialized" return ret if kernel != None: krel = platform.release() if topotest.version_cmp(krel, kernel) < 0: ret = "Skipping tests, old kernel ({} < {})".format(krel, kernel) return ret if iproute2 != None: if _lt.iproute2Ver == None: # collect/log info on iproute2 cc = ltemplateRtrCmd() found = cc.doCmd( tgen, rname, "apt-cache policy iproute2", r"Installed: ([\d\.]*)" ) if found != None: iproute2Ver = found.group(1) else: iproute2Ver = "0-unknown" logger.info("Have iproute2 version=" + iproute2Ver) if topotest.version_cmp(iproute2Ver, iproute2) < 0: ret = "Skipping tests, old iproute2 ({} < {})".format(iproute2Ver, iproute2) return ret ret = True try: if router.has_version(compstr, vstr): ret = "Skipping tests, old FRR version {} {}".format(compstr, vstr) return ret except: ret = True return ret # for testing if __name__ == "__main__": args = ["-s"] + sys.argv[1:] sys.exit(pytest.main(args))