summaryrefslogtreecommitdiffstats
path: root/tests/topotests/lib/ltemplate.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/topotests/lib/ltemplate.py')
-rw-r--r--tests/topotests/lib/ltemplate.py320
1 files changed, 320 insertions, 0 deletions
diff --git a/tests/topotests/lib/ltemplate.py b/tests/topotests/lib/ltemplate.py
new file mode 100644
index 0000000..2544023
--- /dev/null
+++ b/tests/topotests/lib/ltemplate.py
@@ -0,0 +1,320 @@
+#!/usr/bin/env python
+
+#
+# Part of NetDEF Topology Tests
+#
+# Copyright (c) 2017 by
+# Network Device Education Foundation, Inc. ("NetDEF")
+#
+# 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.
+#
+
+"""
+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))