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