tests/topotests/lib/ltemplate.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2017 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10
  11 """
  12 ltemplate.py: LabN template for FRR tests.
  13 """
  14
  15 import os
  16 import sys
  17 import platform
  18
  19 import pytest
  20
  21 # pylint: disable=C0413
  22 # Import topogen and topotest helpers
  23 from lib import topotest
  24 from lib.topogen import Topogen, TopoRouter, get_topogen
  25 from lib.topolog import logger
  26 from lib.lutil import *
  27
  28 # Required to instantiate the topology builder class.
  29
  30 customize = None
  31
  32
  33 class LTemplate:
  34     test = None
  35     testdir = None
  36     scriptdir = None
  37     logdir = None
  38     prestarthooksuccess = True
  39     poststarthooksuccess = True
  40     iproute2Ver = None
  41
  42     def __init__(self, test, testdir):
  43         pathname = os.path.join(testdir, "customize.py")
  44         global customize
  45         if sys.version_info >= (3, 5):
  46             import importlib.util
  47
  48             spec = importlib.util.spec_from_file_location("customize", pathname)
  49             customize = importlib.util.module_from_spec(spec)
  50             spec.loader.exec_module(customize)
  51         else:
  52             import imp
  53
  54             customize = imp.load_source("customize", pathname)
  55         self.test = test
  56         self.testdir = testdir
  57         self.scriptdir = testdir
  58         self.logdir = ""
  59         logger.info("LTemplate: " + test)
  60
  61     def setup_module(self, mod):
  62         "Sets up the pytest environment"
  63         # This function initiates the topology build with Topogen...
  64         tgen = Topogen(customize.build_topo, mod.__name__)
  65         # ... and here it calls Mininet initialization functions.
  66         tgen.start_topology()
  67
  68         self.logdir = tgen.logdir
  69
  70         logger.info("Topology started")
  71         try:
  72             self.prestarthooksuccess = customize.ltemplatePreRouterStartHook()
  73         except AttributeError:
  74             # not defined
  75             logger.debug("ltemplatePreRouterStartHook() not defined")
  76         if self.prestarthooksuccess != True:
  77             logger.info("ltemplatePreRouterStartHook() failed, skipping test")
  78             return
  79
  80         # This is a sample of configuration loading.
  81         router_list = tgen.routers()
  82
  83         # For all registered routers, load the zebra configuration file
  84         for rname, router in router_list.items():
  85             logger.info("Setting up %s" % rname)
  86             for rd_val in TopoRouter.RD:
  87                 config = os.path.join(
  88                     self.testdir, "{}/{}.conf".format(rname, TopoRouter.RD[rd_val])
  89                 )
  90                 prog = os.path.join(tgen.net[rname].daemondir, TopoRouter.RD[rd_val])
  91                 if os.path.exists(config):
  92                     if os.path.exists(prog):
  93                         router.load_config(rd_val, config)
  94                     else:
  95                         logger.warning(
  96                             "{} not found, but have {}.conf file".format(
  97                                 prog, TopoRouter.RD[rd_val]
  98                             )
  99                         )
 100
 101         # After loading the configurations, this function loads configured daemons.
 102         logger.info("Starting routers")
 103         tgen.start_router()
 104         try:
 105             self.poststarthooksuccess = customize.ltemplatePostRouterStartHook()
 106         except AttributeError:
 107             # not defined
 108             logger.debug("ltemplatePostRouterStartHook() not defined")
 109         luStart(baseScriptDir=self.scriptdir, baseLogDir=self.logdir, net=tgen.net)
 110
 111
 112 # initialized by ltemplate_start
 113 _lt = None
 114
 115
 116 def setup_module(mod):
 117     global _lt
 118     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 119     test = mod.__name__[: mod.__name__.rfind(".")]
 120     testdir = os.path.join(root, test)
 121
 122     # don't do this for now as reload didn't work as expected
 123     # fixup sys.path, want test dir there only once
 124     # try:
 125     #    sys.path.remove(testdir)
 126     # except ValueError:
 127     #    logger.debug(testdir+" not found in original sys.path")
 128     # add testdir
 129     # sys.path.append(testdir)
 130
 131     # init class
 132     _lt = LTemplate(test, testdir)
 133     _lt.setup_module(mod)
 134
 135     # drop testdir
 136     # sys.path.remove(testdir)
 137
 138
 139 def teardown_module(mod):
 140     global _lt
 141     "Teardown the pytest environment"
 142     tgen = get_topogen()
 143
 144     if _lt != None and _lt.scriptdir != None and _lt.prestarthooksuccess == True:
 145         luShowResults(logger.info)
 146         print(luFinish())
 147
 148     # This function tears down the whole topology.
 149     tgen.stop_topology()
 150     _lt = None
 151
 152
 153 def ltemplateTest(
 154     script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None, KeepGoing=False
 155 ):
 156     global _lt
 157     if _lt == None or _lt.prestarthooksuccess != True:
 158         return
 159
 160     tgen = get_topogen()
 161     if not os.path.isfile(script):
 162         if not os.path.isfile(os.path.join(_lt.scriptdir, script)):
 163             logger.error("Could not find script file: " + script)
 164             assert "Could not find script file: " + script
 165     logger.info("Starting template test: " + script)
 166     numEntry = luNumFail()
 167
 168     if SkipIfFailed and tgen.routers_have_failure():
 169         pytest.skip(tgen.errors)
 170     if numEntry > 0:
 171         if not KeepGoing:
 172             pytest.skip("Have %d errors" % numEntry)
 173
 174     if CheckFuncStr != None:
 175         check = eval(CheckFuncStr)
 176         if check != True:
 177             pytest.skip("Check function '" + CheckFuncStr + "' returned: " + check)
 178
 179     if CallOnFail != None:
 180         CallOnFail = eval(CallOnFail)
 181     luInclude(script, CallOnFail)
 182     numFail = luNumFail() - numEntry
 183     if numFail > 0:
 184         luShowFail()
 185         fatal_error = "%d tests failed" % numFail
 186         if not KeepGoing:
 187             assert (
 188                 "scripts/cleanup_all.py failed" == "See summary output above"
 189             ), fatal_error
 190
 191
 192 # Memory leak test template
 193 def test_memory_leak():
 194     "Run the memory leak test and report results."
 195     tgen = get_topogen()
 196     if not tgen.is_memleak_enabled():
 197         pytest.skip("Memory leak test/report is disabled")
 198
 199     tgen.report_memory_leaks()
 200
 201
 202 class ltemplateRtrCmd:
 203     def __init__(self):
 204         self.resetCounts()
 205
 206     def doCmd(self, tgen, rtr, cmd, checkstr=None):
 207         logger.info("doCmd: {} {}".format(rtr, cmd))
 208         output = tgen.net[rtr].cmd(cmd).strip()
 209         if len(output):
 210             self.output += 1
 211             if checkstr != None:
 212                 ret = re.search(checkstr, output)
 213                 if ret == None:
 214                     self.nomatch += 1
 215                 else:
 216                     self.match += 1
 217                 return ret
 218             logger.info("output: " + output)
 219         else:
 220             logger.info("No output")
 221             self.none += 1
 222         return None
 223
 224     def resetCounts(self):
 225         self.match = 0
 226         self.nomatch = 0
 227         self.output = 0
 228         self.none = 0
 229
 230     def getMatch(self):
 231         return self.match
 232
 233     def getNoMatch(self):
 234         return self.nomatch
 235
 236     def getOutput(self):
 237         return self.output
 238
 239     def getNone(self):
 240         return self.none
 241
 242
 243 def ltemplateVersionCheck(
 244     vstr, rname="r1", compstr="<", cli=False, kernel="4.9", iproute2=None, mpls=True
 245 ):
 246     tgen = get_topogen()
 247     router = tgen.gears[rname]
 248
 249     if cli:
 250         logger.info("calling mininet CLI")
 251         tgen.mininet_cli()
 252         logger.info("exited mininet CLI")
 253
 254     if _lt == None:
 255         ret = "Template not initialized"
 256         return ret
 257
 258     if _lt.prestarthooksuccess != True:
 259         ret = "ltemplatePreRouterStartHook failed"
 260         return ret
 261
 262     if _lt.poststarthooksuccess != True:
 263         ret = "ltemplatePostRouterStartHook failed"
 264         return ret
 265
 266     if mpls == True and tgen.hasmpls != True:
 267         ret = "MPLS not initialized"
 268         return ret
 269
 270     if kernel != None:
 271         krel = platform.release()
 272         if topotest.version_cmp(krel, kernel) < 0:
 273             ret = "Skipping tests, old kernel ({} < {})".format(krel, kernel)
 274             return ret
 275
 276     if iproute2 != None:
 277         if _lt.iproute2Ver == None:
 278             # collect/log info on iproute2
 279             cc = ltemplateRtrCmd()
 280             found = cc.doCmd(
 281                 tgen, rname, "apt-cache policy iproute2", r"Installed: ([\d\.]*)"
 282             )
 283             if found != None:
 284                 iproute2Ver = found.group(1)
 285             else:
 286                 iproute2Ver = "0-unknown"
 287             logger.info("Have iproute2 version=" + iproute2Ver)
 288
 289         if topotest.version_cmp(iproute2Ver, iproute2) < 0:
 290             ret = "Skipping tests, old iproute2 ({} < {})".format(iproute2Ver, iproute2)
 291             return ret
 292
 293     ret = True
 294     try:
 295         if router.has_version(compstr, vstr):
 296             ret = "Skipping tests, old FRR version {} {}".format(compstr, vstr)
 297             return ret
 298     except:
 299         ret = True
 300
 301     return ret
 302
 303
 304 # for testing
 305 if __name__ == "__main__":
 306     args = ["-s"] + sys.argv[1:]
 307     sys.exit(pytest.main(args))