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
  97         # After loading the configurations, this function loads configured daemons.
  98         logger.info('Starting routers')
  99         tgen.start_router()
 100         try:
 101             self.poststarthooksuccess = customize.ltemplatePostRouterStartHook()
 102         except AttributeError:
 103             #not defined
 104             logger.debug("ltemplatePostRouterStartHook() not defined")
 105         luStart(baseScriptDir=self.scriptdir, baseLogDir=self.logdir, net=tgen.net)
 106
 107 #initialized by ltemplate_start
 108 _lt = None
 109
 110 def setup_module(mod):
 111     global _lt
 112     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 113     test = mod.__name__[:mod.__name__.rfind(".")]
 114     testdir = os.path.join(root, test)
 115
 116     #don't do this for now as reload didn't work as expected
 117     #fixup sys.path, want test dir there only once
 118     #try:
 119     #    sys.path.remove(testdir)
 120     #except ValueError:
 121     #    logger.debug(testdir+" not found in original sys.path")
 122     #add testdir
 123     #sys.path.append(testdir)
 124
 125     #init class
 126     _lt = LTemplate(test, testdir)
 127     _lt.setup_module(mod)
 128
 129     #drop testdir
 130     #sys.path.remove(testdir)
 131
 132 def teardown_module(mod):
 133     global _lt
 134     "Teardown the pytest environment"
 135     tgen = get_topogen()
 136
 137     if _lt != None and _lt.scriptdir != None:
 138         print(luFinish())
 139
 140     # This function tears down the whole topology.
 141     tgen.stop_topology()
 142     _lt = None
 143
 144 def ltemplateTest(script, SkipIfFailed=True, CallOnFail=None, CheckFuncStr=None):
 145     tgen = get_topogen()
 146     if not os.path.isfile(script):
 147         if not os.path.isfile(os.path.join(_lt.scriptdir, script)):
 148             logger.error('Could not find script file: ' + script)
 149             assert 'Could not find script file: ' + script
 150     logger.info("Starting template test: " + script)
 151     numEntry = luNumFail()
 152
 153     if SkipIfFailed and tgen.routers_have_failure():
 154         pytest.skip(tgen.errors)
 155     if numEntry > 0:
 156         pytest.skip("Have %d errors" % numEntry)
 157
 158     if CheckFuncStr != None:
 159         check = eval(CheckFuncStr)
 160         if check != True:
 161             pytest.skip("Check function '"+CheckFuncStr+"' returned: " + check)
 162
 163     if CallOnFail != None:
 164         CallOnFail = eval(CallOnFail)
 165     luInclude(script, CallOnFail)
 166     numFail = luNumFail() - numEntry
 167     if numFail > 0:
 168         luShowFail()
 169         fatal_error = "%d tests failed" % numFail
 170         assert "scripts/cleanup_all.py failed" == "See summary output above", fatal_error
 171
 172 # Memory leak test template
 173 def test_memory_leak():
 174     "Run the memory leak test and report results."
 175     tgen = get_topogen()
 176     if not tgen.is_memleak_enabled():
 177         pytest.skip('Memory leak test/report is disabled')
 178
 179     tgen.report_memory_leaks()
 180
 181 class ltemplateRtrCmd():
 182     def __init__(self):
 183         self.resetCounts()
 184
 185     def doCmd(self, tgen, rtr, cmd, checkstr = None):
 186         output = tgen.net[rtr].cmd(cmd).strip()
 187         if len(output):
 188             self.output += 1
 189             if checkstr != None:
 190                 ret = re.search(checkstr, output)
 191                 if ret == None:
 192                     self.nomatch += 1
 193                 else:
 194                     self.match += 1
 195                 return ret
 196             logger.info('command: {} {}'.format(rtr, cmd))
 197             logger.info('output: ' + output)
 198         self.none += 1
 199         return None
 200
 201     def resetCounts(self):
 202         self.match = 0
 203         self.nomatch = 0
 204         self.output = 0
 205         self.none = 0
 206
 207     def getMatch(self):
 208         return self.match
 209
 210     def getNoMatch(self):
 211         return self.nomatch
 212
 213     def getOutput(self):
 214         return self.output
 215
 216     def getNone(self):
 217         return self.none
 218
 219 def ltemplateVersionCheck(vstr, rname='r1', compstr='<',cli=False, kernel='4.9', iproute2=None, mpls=True):
 220     tgen = get_topogen()
 221     router = tgen.gears[rname]
 222
 223     if cli:
 224         logger.info('calling mininet CLI')
 225         tgen.mininet_cli()
 226         logger.info('exited mininet CLI')
 227
 228     if _lt == None:
 229         ret = 'Template not initialized'
 230         return ret
 231
 232     if _lt.prestarthooksuccess != True:
 233         ret = 'ltemplatePreRouterStartHook failed'
 234         return ret
 235
 236     if _lt.poststarthooksuccess != True:
 237         ret = 'ltemplatePostRouterStartHook failed'
 238         return ret
 239
 240     if mpls == True and tgen.hasmpls != True:
 241         ret = 'MPLS not initialized'
 242         return ret
 243
 244     if kernel != None:
 245         krel = platform.release()
 246         if topotest.version_cmp(krel, kernel) < 0:
 247             ret = 'Skipping tests, old kernel ({} < {})'.format(krel, kernel)
 248             return ret
 249
 250     if iproute2 != None:
 251         if _lt.iproute2Ver == None:
 252             #collect/log info on iproute2
 253             cc = ltemplateRtrCmd()
 254             found = cc.doCmd(tgen, rname, 'apt-cache policy iproute2', 'Installed: ([\d\.]*)')
 255             if found != None:
 256                 iproute2Ver = found.group(1)
 257             else:
 258                 iproute2Ver = '0-unknown'
 259             logger.info('Have iproute2 version=' + iproute2Ver)
 260
 261         if topotest.version_cmp(iproute2Ver, iproute2) < 0:
 262             ret = 'Skipping tests, old iproute2 ({} < {})'.format(iproute2Ver, iproute2)
 263             return ret
 264
 265     ret = True
 266     try:
 267         if router.has_version(compstr, vstr):
 268             ret = 'Skipping tests, old FRR version {} {}'.format(compstr, vstr)
 269             return ret
 270     except:
 271         ret = True
 272
 273     return ret
 274
 275 #for testing
 276 if __name__ == '__main__':
 277     args = ["-s"] + sys.argv[1:]
 278     sys.exit(pytest.main(args))