diff options
Diffstat (limited to 'tests/topotests/lib/ltemplate.py')
| -rw-r--r-- | tests/topotests/lib/ltemplate.py | 119 |
1 files changed, 71 insertions, 48 deletions
diff --git a/tests/topotests/lib/ltemplate.py b/tests/topotests/lib/ltemplate.py index 3c93e1ac5c..d211be8836 100644 --- a/tests/topotests/lib/ltemplate.py +++ b/tests/topotests/lib/ltemplate.py @@ -43,7 +43,8 @@ from mininet.topo import Topo customize = None -class LTemplate(): + +class LTemplate: test = None testdir = None scriptdir = None @@ -54,12 +55,12 @@ class LTemplate(): def __init__(self, test, testdir): global customize - customize = imp.load_source('customize', os.path.join(testdir, 'customize.py')) + customize = imp.load_source("customize", os.path.join(testdir, "customize.py")) self.test = test self.testdir = testdir self.scriptdir = testdir - self.logdir = '/tmp/topotests/{0}.test_{0}'.format(test) - logger.info('LTemplate: '+test) + self.logdir = "/tmp/topotests/{0}.test_{0}".format(test) + logger.info("LTemplate: " + test) def setup_module(self, mod): "Sets up the pytest environment" @@ -68,14 +69,14 @@ class LTemplate(): # ... and here it calls Mininet initialization functions. tgen.start_topology() - logger.info('Topology started') + logger.info("Topology started") try: self.prestarthooksuccess = customize.ltemplatePreRouterStartHook() except AttributeError: - #not defined + # not defined logger.debug("ltemplatePreRouterStartHook() not defined") if self.prestarthooksuccess != True: - logger.info('ltemplatePreRouterStartHook() failed, skipping test') + logger.info("ltemplatePreRouterStartHook() failed, skipping test") return # This is a sample of configuration loading. @@ -85,48 +86,57 @@ class LTemplate(): 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])) + 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])) + 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') + logger.info("Starting routers") tgen.start_router() try: self.poststarthooksuccess = customize.ltemplatePostRouterStartHook() except AttributeError: - #not defined + # not defined logger.debug("ltemplatePostRouterStartHook() not defined") luStart(baseScriptDir=self.scriptdir, baseLogDir=self.logdir, net=tgen.net) -#initialized by ltemplate_start + +# 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(".")] + 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: + # 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: + # except ValueError: # logger.debug(testdir+" not found in original sys.path") - #add testdir - #sys.path.append(testdir) + # add testdir + # sys.path.append(testdir) - #init class + # init class _lt = LTemplate(test, testdir) _lt.setup_module(mod) - #drop testdir - #sys.path.remove(testdir) + # drop testdir + # sys.path.remove(testdir) + def teardown_module(mod): global _lt @@ -141,7 +151,10 @@ def teardown_module(mod): tgen.stop_topology() _lt = None -def ltemplateTest(script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, KeepGoing=False): + +def ltemplateTest( + script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, KeepGoing=False +): global _lt if _lt == None or _lt.prestarthooksuccess != True: return @@ -149,8 +162,8 @@ def ltemplateTest(script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, 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.error("Could not find script file: " + script) + assert "Could not find script file: " + script logger.info("Starting template test: " + script) numEntry = luNumFail() @@ -163,7 +176,7 @@ def ltemplateTest(script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, if CheckFuncStr != None: check = eval(CheckFuncStr) if check != True: - pytest.skip("Check function '"+CheckFuncStr+"' returned: " + check) + pytest.skip("Check function '" + CheckFuncStr + "' returned: " + check) if CallOnFail != None: CallOnFail = eval(CallOnFail) @@ -173,22 +186,26 @@ def ltemplateTest(script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, luShowFail() fatal_error = "%d tests failed" % numFail if not KeepGoing: - assert "scripts/cleanup_all.py failed" == "See summary output above", fatal_error + 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') + pytest.skip("Memory leak test/report is disabled") tgen.report_memory_leaks() -class ltemplateRtrCmd(): + +class ltemplateRtrCmd: def __init__(self): self.resetCounts() - def doCmd(self, tgen, rtr, cmd, checkstr = None): + def doCmd(self, tgen, rtr, cmd, checkstr=None): output = tgen.net[rtr].cmd(cmd).strip() if len(output): self.output += 1 @@ -199,8 +216,8 @@ class ltemplateRtrCmd(): else: self.match += 1 return ret - logger.info('command: {} {}'.format(rtr, cmd)) - logger.info('output: ' + output) + logger.info("command: {} {}".format(rtr, cmd)) + logger.info("output: " + output) self.none += 1 return None @@ -222,63 +239,69 @@ class ltemplateRtrCmd(): def getNone(self): return self.none -def ltemplateVersionCheck(vstr, rname='r1', compstr='<',cli=False, kernel='4.9', iproute2=None, mpls=True): + +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') + logger.info("calling mininet CLI") tgen.mininet_cli() - logger.info('exited mininet CLI') + logger.info("exited mininet CLI") if _lt == None: - ret = 'Template not initialized' + ret = "Template not initialized" return ret if _lt.prestarthooksuccess != True: - ret = 'ltemplatePreRouterStartHook failed' + ret = "ltemplatePreRouterStartHook failed" return ret if _lt.poststarthooksuccess != True: - ret = 'ltemplatePostRouterStartHook failed' + ret = "ltemplatePostRouterStartHook failed" return ret if mpls == True and tgen.hasmpls != True: - ret = 'MPLS not initialized' + 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) + ret = "Skipping tests, old kernel ({} < {})".format(krel, kernel) return ret if iproute2 != None: if _lt.iproute2Ver == None: - #collect/log info on iproute2 + # collect/log info on iproute2 cc = ltemplateRtrCmd() - found = cc.doCmd(tgen, rname, 'apt-cache policy iproute2', 'Installed: ([\d\.]*)') + found = cc.doCmd( + tgen, rname, "apt-cache policy iproute2", "Installed: ([\d\.]*)" + ) if found != None: iproute2Ver = found.group(1) else: - iproute2Ver = '0-unknown' - logger.info('Have iproute2 version=' + iproute2Ver) + iproute2Ver = "0-unknown" + logger.info("Have iproute2 version=" + iproute2Ver) if topotest.version_cmp(iproute2Ver, iproute2) < 0: - ret = 'Skipping tests, old iproute2 ({} < {})'.format(iproute2Ver, iproute2) + 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) + ret = "Skipping tests, old FRR version {} {}".format(compstr, vstr) return ret except: ret = True return ret -#for testing -if __name__ == '__main__': + +# for testing +if __name__ == "__main__": args = ["-s"] + sys.argv[1:] sys.exit(pytest.main(args)) |