#!/usr/bin/env python
#
-# test_bgp_multiview_topo1.py
+# <template>.py
# Part of NetDEF Topology Tests
#
-# Copyright (c) 2016 by
+# Copyright (c) 2017 by
# Network Device Education Foundation, Inc. ("NetDEF")
#
# Permission to use, copy, modify, and/or distribute this software
+-----+-----+ +----+-----+
| |
-"""
+"""
import os
-import re
import sys
import pytest
-from time import sleep
-
-from mininet.topo import Topo
-from mininet.net import Mininet
-from mininet.node import Node, OVSSwitch, Host
-from mininet.log import setLogLevel, info
-from mininet.cli import CLI
-from mininet.link import Intf
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + '/utilities')
+# Save the Current Working Directory to find configuration files.
+CWD = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(CWD, '../'))
+sys.path.append(os.path.join(CWD, '../utilities'))
+# pylint: disable=C0413
+# Import topogen and topotest helpers
from lib import topotest
-
+from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topolog import logger
from lutil import luStart, luInclude, luFinish, luNumFail
-fatal_error = ""
-
-# Expected version of CLI Output - Appendix to filename
-# empty string = current, latest output (default)
-# "-1" ... "-NNN" previous versions (incrementing with each version)
-cli_version = ""
-
-#####################################################
-##
-## Network Topology Definition
-##
-#####################################################
-
-class NetworkTopo(Topo):
- "VPN Test Topology 1"
-
- def build(self, **_opts):
-
- # Setup Routers
- router = {}
- for i in range(1, 5):
- router[i] = topotest.addRouter(self, 'r%s' % i)
- ce = {}
- for i in range(1, 4):
- ce[i] = topotest.addRouter(self, 'ce%s' % i)
+# Required to instantiate the topology builder class.
+from mininet.topo import Topo
- self.addLink(ce[1], router[1], intfName1='ce1-eth0', intfName2='r1-eth4', addr1='00:11:01:00:00:00', addr2='00:11:00:01:00:04')
- self.addLink(ce[2], router[3], intfName1='ce2-eth0', intfName2='r3-eth4', addr1='00:11:02:00:00:00', addr2='00:11:00:03:00:04')
- self.addLink(ce[3], router[4], intfName1='ce3-eth0', intfName2='r4-eth4', addr1='00:11:03:00:00:00', addr2='00:11:00:04:00:04')
- # Setup Switches, add Interfaces and Connections
+class TemplateTopo(Topo):
+ "Test topology builder"
+ def build(self, *_args, **_opts):
+ "Build function"
+ tgen = get_topogen(self)
+
+ # This function only purpose is to define allocation and relationship
+ # between routers, switches and hosts.
+ #
+ # Create P/PE routers
+ for routern in range(1, 5):
+ tgen.add_router('r{}'.format(routern))
+ # Create CE routers
+ for routern in range(1, 4):
+ tgen.add_router('ce{}'.format(routern))
+
+ #CE/PE links
+ tgen.add_link(tgen.gears['ce1'], tgen.gears['r1'], 'ce1-eth0', 'r1-eth4')
+ tgen.add_link(tgen.gears['ce2'], tgen.gears['r3'], 'ce2-eth0', 'r3-eth4')
+ tgen.add_link(tgen.gears['ce3'], tgen.gears['r4'], 'ce3-eth0', 'r4-eth4')
+
+ # Create a switch with just one router connected to it to simulate a
+ # empty network.
switch = {}
- # First switch
- switch[0] = self.addSwitch('sw0', cls=topotest.LegacySwitch)
- self.addLink(switch[0], router[1], intfName2='r1-eth0', addr1='80:AA:00:00:00:00', addr2='00:11:00:01:00:00')
- self.addLink(switch[0], router[2], intfName2='r2-eth0', addr1='80:AA:00:00:00:01', addr2='00:11:00:02:00:00')
- # Second switch
- switch[1] = self.addSwitch('sw1', cls=topotest.LegacySwitch)
- self.addLink(switch[1], router[2], intfName2='r2-eth1', addr1='80:AA:00:01:00:00', addr2='00:11:00:02:00:01')
- self.addLink(switch[1], router[3], intfName2='r3-eth0', addr1='80:AA:00:01:00:01', addr2='00:11:00:03:00:00')
- self.addLink(switch[1], router[4], intfName2='r4-eth0', addr1='80:AA:00:01:00:02', addr2='00:11:00:04:00:00')
- # Third switch
- switch[2] = self.addSwitch('sw2', cls=topotest.LegacySwitch)
- self.addLink(switch[2], router[2], intfName2='r2-eth2', addr1='80:AA:00:02:00:00', addr2='00:11:00:02:00:02')
- self.addLink(switch[2], router[3], intfName2='r3-eth1', addr1='80:AA:00:02:00:01', addr2='00:11:00:03:00:01')
-
-#####################################################
-##
-## Tests starting
-##
-#####################################################
-
-def setup_module(module):
- global topo, net
- global fatal_error
- global thisDir
-
- print("\n\n** %s: Setup Topology" % module.__name__)
- print("******************************************\n")
-
- print("Cleanup old Mininet runs")
- os.system('sudo mn -c > /dev/null 2>&1')
-
- thisDir = os.path.dirname(os.path.realpath(__file__))
- topo = NetworkTopo()
-
- net = Mininet(controller=None, topo=topo)
- net.start()
- luStart(thisDir, net)
-
- # Starting Routers
- for i in range(1, 5):
- net['r%s' % i].loadConf('zebra', '%s/r%s/zebra.conf' % (thisDir, i))
- net['r%s' % i].loadConf('ospfd', '%s/r%s/ospfd.conf' % (thisDir, i))
- net['r%s' % i].loadConf('ldpd', '%s/r%s/ldpd.conf' % (thisDir, i))
- net['r%s' % i].loadConf('bgpd', '%s/r%s/bgpd.conf' % (thisDir, i))
- fatal_error = net['r%s' % i].startRouter()
-
- if fatal_error != "":
- break
-
- # Starting CE Routers
- for i in range(1, 4):
- net['ce%s' % i].loadConf('zebra', '%s/ce%s/zebra.conf' % (thisDir, i))
- net['ce%s' % i].loadConf('bgpd', '%s/ce%s/bgpd.conf' % (thisDir, i))
- fatal_error = net['ce%s' % i].startRouter()
-
- if fatal_error != "":
- break
-
- # For debugging after starting FRR/Quagga daemons, uncomment the next line
- #CLI(net)
-
-def teardown_module(module):
- global net
-
- print("\n\n** %s: Shutdown Topology" % module.__name__)
- print("******************************************\n")
-
- # End - Shutdown network
- net.stop()
-
-def test_add_vnc_routes():
- global fatal_error
- global net
- global cli_version
-
- # Skip if previous fatal error condition is raised
- if (fatal_error != ""):
- pytest.skip(fatal_error)
-
- print("\n\n** Running main test cases")
- print("******************************\n")
- luInclude('teststart.py')
- # For debugging after starting FRR/Quagga daemons, uncomment the next line
- #CLI(net)
-
- luInclude('testfinish.py')
- # For debugging after starting FRR/Quagga daemons, uncomment the next line
- #CLI(net)
-
- # Make sure that all daemons are running
- numFail = luNumFail()
- if numFail > 0:
- fatal_error = '%d tests failed' % numFail
- assert fatal_error == "", fatal_error
-
-def test_shutdown_check_stderr():
- global fatal_error
- global net
-
- # Skip if previous fatal error condition is raised
- if (fatal_error != ""):
- pytest.skip(fatal_error)
-
- if os.environ.get('TOPOTESTS_CHECK_STDERR') is None:
- print("SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n")
- pytest.skip('Skipping test for Stderr output')
-
- thisDir = os.path.dirname(os.path.realpath(__file__))
-
- print("\n\n** Verifying unexpected STDERR output from daemons")
- print("******************************************\n")
-
- for i in range(1, 5):
- net['r%s' % i].stopRouter()
- log = net['r%s' % i].getStdErr('bgpd')
- if log:
- print("\nRouter r%s BGPd StdErr Log:\n%s" % (i, log))
- log = net['r%s' % i].getStdErr('ldpd')
- if log:
- print("\nRouter r%s LDPd StdErr Log:\n%s" % (i, log))
- log = net['r%s' % i].getStdErr('ospfd')
- if log:
- print("\nRouter r%s OSPFd StdErr Log:\n%s" % (i, log))
- log = net['r%s' % i].getStdErr('zebra')
- if log:
- print("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log))
-
- for i in range(1, 4):
- net['ce%s' % i].stopRouter()
- log = net['ce%s' % i].getStdErr('bgpd')
- if log:
- print("\nRouter r%s BGPd StdErr Log:\n%s" % (i, log))
- log = net['ce%s' % i].getStdErr('zebra')
- if log:
- print("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log))
-
-
-def test_shutdown_check_memleak():
- global fatal_error
- global net
-
- # Skip if previous fatal error condition is raised
- if (fatal_error != ""):
- pytest.skip(fatal_error)
-
- if os.environ.get('TOPOTESTS_CHECK_MEMLEAK') is None:
- print("SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)\n")
- pytest.skip('Skipping test for memory leaks')
-
- thisDir = os.path.dirname(os.path.realpath(__file__))
-
- for i in range(1, 5):
- net['r%s' % i].stopRouter()
- net['r%s' % i].report_memory_leaks(os.environ.get('TOPOTESTS_CHECK_MEMLEAK'), os.path.basename(__file__))
-
- for i in range(1, 4):
- net['ce%s' % i].stopRouter()
- net['ce%s' % i].report_memory_leaks(os.environ.get('TOPOTESTS_CHECK_MEMLEAK'), os.path.basename(__file__))
+ switch[0] = tgen.add_switch('sw0')
+ switch[0].add_link(tgen.gears['r1'], nodeif='r1-eth0')
+ switch[0].add_link(tgen.gears['r2'], nodeif='r2-eth0')
+
+ switch[1] = tgen.add_switch('sw1')
+ switch[1].add_link(tgen.gears['r2'], nodeif='r2-eth1')
+ switch[1].add_link(tgen.gears['r3'], nodeif='r3-eth0')
+ switch[1].add_link(tgen.gears['r4'], nodeif='r4-eth0')
+
+ switch[1] = tgen.add_switch('sw2')
+ switch[1].add_link(tgen.gears['r2'], nodeif='r2-eth2')
+ switch[1].add_link(tgen.gears['r3'], nodeif='r3-eth1')
+
+def setup_module(mod):
+ "Sets up the pytest environment"
+ # This function initiates the topology build with Topogen...
+ tgen = Topogen(TemplateTopo, mod.__name__)
+ # ... and here it calls Mininet initialization functions.
+ tgen.start_topology()
+
+ # This is a sample of configuration loading.
+ router_list = tgen.routers()
+
+ # For all registred routers, load the zebra configuration file
+ for rname, router in router_list.iteritems():
+ config = os.path.join(CWD, '{}/zebra.conf'.format(rname))
+ if os.path.exists(config):
+ router.load_config(TopoRouter.RD_ZEBRA, config)
+ config = os.path.join(CWD, '{}/ospfd.conf'.format(rname))
+ if os.path.exists(config):
+ router.load_config(TopoRouter.RD_OSPF, config)
+ config = os.path.join(CWD, '{}/ldpd.conf'.format(rname))
+ if os.path.exists(config):
+ router.load_config(TopoRouter.RD_LDP, config)
+ config = os.path.join(CWD, '{}/bgpd.conf'.format(rname))
+ if os.path.exists(config):
+ router.load_config(TopoRouter.RD_BGP, config)
+
+ # After loading the configurations, this function loads configured daemons.
+ tgen.start_router()
+
+def teardown_module(mod):
+ "Teardown the pytest environment"
+ tgen = get_topogen()
+
+ # This function tears down the whole topology.
+ tgen.stop_topology()
+
+def no_test_call_mininet_cli():
+ "Dummy test that just calls mininet CLI so we can interact with the build."
+ tgen = get_topogen()
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ logger.info('calling mininet CLI')
+ tgen.mininet_cli()
+
+def test_run_lu_tests():
+ tgen = get_topogen()
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ router = tgen.gears['r1']
+ is_pre31 = False
+ try:
+ if router.has_version('<', '3.1'):
+ is_pre31 = True
+ print("\nversion check failed, version < 3.1")
+ except:
+ is_pre31 = False
+
+ if is_pre31 == True:
+ print("\n\n** Skipping main tests on old version (<3.1)")
+ else:
+ print("\n\n** Running main test cases")
+ print("******************************\n")
+
+ luStart(os.path.dirname(os.path.realpath(__file__)), tgen.net)
+
+ luInclude('teststart.py')
+ # For debugging after starting FRR/Quagga daemons, uncomment the next line
+ #CLI(net)
+
+ luInclude('testfinish.py')
+ print(luFinish())
+
+ # For debugging after starting FRR/Quagga daemons, uncomment the next line
+ #CLI(net)
+
+ # Make sure that all daemons are running
+ numFail = luNumFail()
+ if numFail > 0:
+ fatal_error = '%d tests failed' % numFail
+ assert fatal_error == "", fatal_error
+
+# Memory leak test template
+def test_memory_leak():
+ "Run the memory leak test and report results."
+ tgen = get_topogen()
+ if not tgen.is_memleak_enabled():
+ pytest.skip('Memory leak test/report is disabled')
+
+ tgen.report_memory_leaks()
if __name__ == '__main__':
-
- setLogLevel('info')
- # To suppress tracebacks, either use the following pytest call or add "--tb=no" to cli
- # retval = pytest.main(["-s", "--tb=no"])
- retval = pytest.main(["-s"])
- print(luFinish())
- sys.exit(retval)
+ args = ["-s"] + sys.argv[1:]
+ sys.exit(pytest.main(args))
projects / mirror / frr.git / commitdiff