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