tests/topotests/lib/ltemplate.py

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