adding first draft of tests for all examples, they need comments and clean up,...

adding first draft of tests for all examples, they need comments and clean up, some could be made more rebust

examples/test/runner.py

+import glob
+import unittest
+
+test_file_strings = glob.glob('test_*.py')
+module_strings = [str[0:len(str)-3] for str in test_file_strings]
+print module_strings
+suites = [unittest.defaultTestLoader.loadTestsFromName(str) for str
+          in module_strings]
+testSuite = unittest.TestSuite(suites)
+text_runner = unittest.TextTestRunner().run(testSuite)
\ No newline at end of file

examples/test/test_baresshd.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.log import setLogLevel
+from mininet.clean import cleanup, sh
+
+class testBareSSHD( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    opts = [ '\(yes/no\)\?', 'Welcome to h1', 'refused', pexpect.EOF, pexpect.TIMEOUT ]
+
+    def connected( self ):
+        "check connected"
+        p = pexpect.spawn( 'ssh 10.0.0.1 -i /tmp/ssh/test_rsa ' )
+        while True:
+            index = p.expect( self.opts )
+            if index == 0:
+                p.sendline( 'yes' )
+            elif index == 1:
+                return True
+            else:
+                return False
+
+    def setUp( self ):
+        self.assertFalse( self.connected() )
+        # create public key pair for testing
+        sh( 'mkdir /tmp/ssh' )
+        sh( "ssh-keygen -t rsa -P '' -f /tmp/ssh/test_rsa" )
+        sh( 'cat /tmp/ssh/test_rsa.pub >> /tmp/ssh/authorized_keys' )
+        cmd = ( 'python -m mininet.examples.baresshd '
+                '-o AuthorizedKeysFile=/tmp/ssh/authorized_keys '
+                '-o StrictModes=no' )
+        sh( cmd )
+
+    def testSSH( self ):
+        result = False
+        # try to connect up to 3 times
+        for _ in range( 3 ):
+            result = self.connected()
+            if result:
+                break
+            else:
+                sleep( 1 )
+        self.assertTrue( result )
+
+    def tearDown( self ):
+        # kill the ssh process
+        sh( "ps aux | grep 'ssh.*Banner' | awk '{ print $2 }' | xargs kill" )
+        cleanup()
+        # remove public key pair
+        sh( 'rm -rf /tmp/ssh' )
+
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_bind.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.log import setLogLevel
+from mininet.clean import cleanup, sh
+
+class testBind( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def connected( self, ip ):
+        "check connected"
+        p = pexpect.spawn( 'ssh -i /tmp/ssh/test_rsa %s' % ip )
+        while True:
+            index = p.expect( self.opts )
+            if index == 0:
+                print p.match.group(0)
+                p.sendline( 'yes' )
+            elif index == 1:
+                return False
+            elif index == 2:
+                p.sendline( 'exit' )
+                p.wait()    
+                return True
+            else:
+                return False
+
+    def setUp( self ):
+        self.net = pexpect.spawn( 'python -m mininet.examples.bind' )
+        self.net.expect( "Private Directories: \[([\w\s,'/]+)\]" )
+        self.directories = []
+        # parse directories from mn output
+        for d in self.net.match.group(1).split(', '):
+            self.directories.append( d.strip("'") )
+        self.net.expect( self.prompt )
+        self.assertTrue( len( self.directories ) > 0 )
+
+    def testCreateFile( self ):
+        fileName = 'a.txt'
+        directory = self.directories[ 0 ]
+        self.net.sendline( 'h1 touch %s/%s; ls %s' % ( directory, fileName, directory ) )
+        index = self.net.expect( [ fileName, self.prompt ] )
+        self.assertTrue( index == 0 )
+
+    # TODO: need more tests
+
+    def tearDown( self ):
+        self.net.sendline( 'exit' )
+        self.net.wait()
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()
+

examples/test/test_controllers.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+#from time import sleep
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testControllers( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def connectedTest( self, name, cmap ):
+        p = pexpect.spawn( 'python -m %s' % name )
+        p.expect( self.prompt )
+        p.sendline( 'pingall' )
+        p.expect ( '(\d+)% dropped' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        self.assertEqual( percent, 0 ) # or this
+        p.expect( self.prompt )
+        for switch in cmap:
+            p.sendline( 'sh ovs-vsctl get-controller %s' % switch )
+            p.expect( 'tcp:([\d.:]+)')
+            actual = p.match.group(1)
+            expected = cmap[ switch ]
+            self.assertEqual( actual, expected)
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+        #TODO remove this
+        self.assertEqual( percent, 0 )
+
+    def testControllers( self ):
+        c0 = '127.0.0.1:6633'
+        c1 = '127.0.0.1:6634'
+        cmap = { 's1': c0, 's2': c1, 's3': c0 }
+        self.connectedTest( 'mininet.examples.controllers', cmap )
+
+    def testControllers2( self ):
+        c0 = '127.0.0.1:6633'
+        c1 = '127.0.0.1:6634'
+        cmap = { 's1': c0, 's2': c1 }
+        self.connectedTest( 'mininet.examples.controllers2', cmap )
+
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_controlnet.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testControlNet( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testPingall( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.controlnet' )
+        p.expect( self.prompt )
+        p.sendline( 'pingall' )
+        p.expect ( '(\d+)% dropped' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        self.assertEqual( percent, 0 )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+
+    def testFailover( self ):
+        count = 1
+        p = pexpect.spawn( 'python -m mininet.examples.controlnet' )
+        p.expect( self.prompt )
+        lp = pexpect.spawn( 'tail -f /tmp/s1-ofp.log' )
+        lp.expect( 'tcp:\d+\.\d+\.\d+\.(\d+):\d+: connected' )
+        ip = int( lp.match.group( 1 ) )
+        self.assertEqual( count, ip )
+        count += 1
+        for c in [ 'c0', 'c1' ]:
+            p.sendline( '%s ifconfig %s-eth0 down' % ( c, c) )
+            p.expect( self.prompt )
+            lp.expect( 'tcp:\d+\.\d+\.\d+\.(\d+):\d+: connected' )
+            ip = int( lp.match.group( 1 ) )
+            self.assertEqual( count, ip )
+            count += 1
+        p.sendline( 'exit' )
+        p.wait()
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_cpu.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+
+class testCPU( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testCPU( self ):
+        opts = [ '([a-z]+)\t([\d\.]+)%\t([\d\.]+)', pexpect.EOF ]
+        p = pexpect.spawn( 'python -m mininet.examples.cpu' )
+        scheds = []
+        while True:
+            index = p.expect( opts, timeout=600 )
+            if index == 0:
+                sched = p.match.group( 1 ) 
+                cpu = float( p.match.group( 2 ) )
+                bw = float( p.match.group( 3 ) )
+                if sched not in scheds:
+                    scheds.append( sched )
+                    previous_bw = 10 ** 4 # 10 GB/s
+                self.assertTrue( bw < previous_bw )
+                previous_bw = bw
+            else:
+                break
+
+        self.assertTrue( len( scheds ) > 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_emptynet.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+#from time import sleep
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testEmptyNet( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testEmptyNet( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.emptynet' )
+        p.expect( self.prompt )
+        p.sendline( 'pingall' )
+        p.expect ( '(\d+)% dropped' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        self.assertEqual( percent, 0 ) # or this
+        p.expect( self.prompt )
+        p.sendline( 'iperf' )
+        p.expect( "Results: \['[\d.]+ .bits/sec', '[\d.]+ .bits/sec'\]" )
+        #TODO check the results? maybe we dont care
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+        #TODO remove this
+        self.assertEqual( percent, 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_hwintf.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+import re
+from time import sleep
+from mininet.log import setLogLevel
+from mininet.net import Mininet
+from mininet.node import Node
+from mininet.link import Link, Intf
+
+class testHwintf( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def _testE2E( self ):
+        results = [ "Results:", pexpect.EOF, pexpect.TIMEOUT ]
+        p = pexpect.spawn( 'python -m mininet.examples.simpleperf' )
+        index = p.expect( results, timeout=600 )
+        self.assertEqual( index, 0 )
+        p.wait()
+
+    def setUp( self ):
+        self.h3 = Node( 't0', ip='10.0.0.3/8' )
+        self.n0 = Node( 't1', inNamespace=False)
+        Link( self.h3, self.n0 )
+        self.h3.configDefault()
+
+    def testLocalPing( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.hwintf %s' % self.n0.intf() )
+        p.expect( self.prompt )
+        p.sendline( 'pingall' )
+        p.expect ( '(\d+)% dropped' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        self.assertEqual( percent, 0 )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+
+    def testExternalPing( self ):
+        expectStr = '(\d+) packets transmitted, (\d+) received'
+        p = pexpect.spawn( 'python -m mininet.examples.hwintf %s' % self.n0.intf() )
+        p.expect( self.prompt )
+
+        m = re.search( expectStr, self.h3.cmd( 'ping -v -c 1 10.0.0.1' ) )
+        tx = m.group( 1 )
+        rx = m.group( 2 )
+        self.assertEqual( tx, rx )
+
+        p.sendline( 'h1 ping -c 1 10.0.0.3')
+        p.expect( expectStr )
+        tx = p.match.group( 1 )
+        rx = p.match.group( 2 )
+        self.assertEqual( tx, rx )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+
+    def tearDown( self ):
+        self.h3.terminate()
+        self.n0.terminate()
+
+
+            
+    ''' TAP garbage 
+     def testHwintf( self ):
+        ifname = 'br3'
+        #sudo ip tuntap add mode tap br0
+        #sudo ip tuntap del mode tap br0
+        #sudo ip link add name test0 type veth peer name test1
+        #sudo ip link del test0
+        t0 = Node( 't0', inNamespace=False, ip='10.0.0.3/8' )
+
+        t1 = Node( 't1', inNamespace=False)
+        #t0.cmd( 'ip tuntap add mode tap %s' % ifname )
+        #Intf( ifname, t0 )
+        print Link( t0, t1 )
+        t0.configDefault()
+
+        print t0.cmd( 'ifconfig' )
+        ifname =  t1.intf()
+
+
+        try:
+            foo = pexpect.spawn( 'wireshark' )
+            p = pexpect.spawn( 'python -m mininet.examples.hwintf %s' % ifname )
+            p.expect( self.prompt )
+            #t0.cmd( 'ip link set dev %s up' % ifname )
+            #t0.cmd( "bash -c 'echo 1 > /proc/sys/net/ipv4/ip_forward'" )
+            #t0.cmd( "bash -c 'echo 1 > /proc/sys/net/ipv4/conf/%s/proxy_arp'" % ifname)
+            #t0.cmd( 'arp -Ds 10.0.0.3 s1 pub' )
+
+            #p.sendline( 'x s1 wireshark' )
+            print t0.cmd( 'ifconfig %s' % ifname )
+            print t0.cmd( 'ip route' )
+            print t0.cmd( 'ping -v -c 1 10.0.0.3' )
+            print t0.cmd( 'ping -v -c 1 10.0.0.1' )
+
+            p.interact()
+            #p.wait()
+        finally:
+            #t0.cmd( 'ip tuntap del mode tap %s' % ifname )
+            t0.terminate()
+            t1.terminate()
+        #t0.configDefault()
+    '''
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_limit.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+
+class testLimit( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    def testLimit( self ):
+        opts = [ '\*\*\* Testing network ([\d\.]+) Mbps', 
+                 '\*\*\* Results: \[([\d\., ]+)\]', 
+                 pexpect.EOF ]
+        p = pexpect.spawn( 'python -m mininet.examples.limit' )
+        count = 0
+        bw = 0
+        tolerance = 1
+        while True:
+            index = p.expect( opts )
+            if index == 0:
+                bw = float( p.match.group( 1 ) )
+                count += 1
+            elif index == 1:
+                results = p.match.group( 1 )
+                for x in results.split(','):
+                    result = float( x )
+                    self.assertTrue( result < bw + tolerance )
+                    self.assertTrue( result > bw - tolerance)
+            else:
+                break
+
+        self.assertTrue( count > 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_linearbandwidth.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testLinearBandwidth( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testLinearBandwidth( self ):
+        count = 0
+        tolerance = 0.5
+        opts = [ '\*\*\* Linear network results', '(\d+)\s+([\d\.]+) (.bits)', pexpect.EOF ]
+        p = pexpect.spawn( 'python -m mininet.examples.linearbandwidth' )
+        while True:
+            index = p.expect( opts, timeout=600 )
+            if index == 0:
+                previous_bw = 10 ** 10 # 10 Gbits
+                count += 1
+            elif index == 1:
+                n = int( p.match.group( 1 ) )
+                bw = float( p.match.group( 2 ) )
+                unit = p.match.group( 3 )
+                if unit[ 0 ] == 'K':
+                    bw *= 10 ** 3
+                elif unit[ 0 ] == 'M':
+                    bw *= 10 ** 6
+                elif unit[ 0 ] == 'G':
+                    bw *= 10 ** 9
+                self.assertTrue( bw < previous_bw )
+                previous_bw = bw
+            else:
+                break
+
+        self.assertTrue( count > 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_multiping.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from collections import defaultdict
+from mininet.log import setLogLevel
+
+class testMultiPing( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    def testMultiPing( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.multiping' )
+        opts = []
+        opts.append( "Host (h\d+) \(([\d.]+)\) will be pinging ips: ([\d. ]+)" )
+        opts.append( "(h\d+): ([\d.]+) -> ([\d.]+) \d packets transmitted, (\d) received" )
+        opts.append( pexpect.EOF )
+        pings = defaultdict( list )
+        while True:
+            index = p.expect( opts )
+            if index == 0:
+                name = p.match.group(1)
+                ip = p.match.group(2)
+                targets = p.match.group(3).split()
+                pings[ name ] += targets
+            elif index == 1:
+                name = p.match.group(1)
+                ip = p.match.group(2)
+                target = p.match.group(3)
+                received = int( p.match.group(4) )
+                if target == '10.0.0.200':
+                    self.assertEqual( received, 0 )
+                else:
+                    self.assertEqual( received, 1 )
+                try:
+                    pings[ name ].remove( target )
+                except:
+                    pass
+            else:
+                break
+        self.assertTrue( len(pings) > 0 )
+        for t in pings.values():
+            self.assertEqual( len( t ), 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_multipoll.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from collections import defaultdict
+from mininet.log import setLogLevel
+
+class testMultiPoll( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    def testMultiPoll( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.multipoll' )
+        opts = []
+        opts.append( "\*\*\* (h\d) :" )
+        opts.append( "(h\d+): \d+ bytes from" )
+        opts.append( "Monitoring output for (\d+) seconds" )
+        opts.append( pexpect.EOF )
+        pings = {}
+        while True:
+            index = p.expect( opts )
+            if index == 0:
+                name = p.match.group( 1 )
+                pings[ name ] = 0
+            elif index == 1:
+                name = p.match.group( 1 )
+                pings[ name ] += 1
+            elif index == 2:
+                seconds = int( p.match.group( 1 ) )
+            else:
+                break
+        self.assertTrue( len(pings) > 0 )
+        # make sure we have received at least one ping per second
+        for count in pings.values():
+            self.assertTrue( count >= seconds )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_multitest.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+
+class testMultiTest( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testMultiTest( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.multitest' )
+        p.expect( '(\d+)% dropped' )
+        dropped = int( p.match.group(1) )
+        self.assertEqual( dropped, 0 )
+        ifCount = 0
+        while True:
+            index = p.expect( [ 'h\d-eth0', self.prompt ] )
+            if index == 0:
+                ifCount += 1
+            elif index == 1:
+                p.sendline( 'exit' )
+                break
+        p.wait()
+        self.assertEqual( ifCount, 4 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_nat.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+#from time import sleep
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testNAT( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    # skip if 8.8.8.8 unreachable
+    def testNAT( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.nat' )
+        p.expect( self.prompt )
+        p.sendline( 'h1 ping -c 1 8.8.8.8' )
+        p.expect ( '(\d+)% packet loss' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+        self.assertEqual( percent, 0 )
+    '''
+    def testTopo( self ):
+        topo = SingleSwitchTopo(n=4)
+        net = Mininet(topo=topo,
+                  host=CPULimitedHost, link=TCLink)
+        net.start()
+        h1, h4 = net.get('h1', 'h4')
+        h1.cmd( 'ping -c 1 %s' % h4.IP() )
+        net.stop()
+    '''
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_popen.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from collections import defaultdict
+from mininet.log import setLogLevel
+
+class testPopen( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    def pingTest( self, name ):
+        p = pexpect.spawn( 'python -m %s' % name )
+        opts = []
+        opts.append( "<(h\d+)>: PING " )
+        opts.append( "<(h\d+)>: (\d+) packets transmitted, (\d+) received" )
+        opts.append( pexpect.EOF )
+        pings = {}
+        while True:
+            index = p.expect( opts )
+            if index == 0:
+                name = p.match.group(1)
+                pings[ name ] = 0
+            elif index == 1:
+                name = p.match.group(1)
+                transmitted = p.match.group(2)
+                received = p.match.group(3)
+                self.assertEqual( received, transmitted )
+                pings[ name ] += 1
+            else:
+                break
+        self.assertTrue( len(pings) > 0 )
+        for count in pings.values():
+            self.assertEqual( count, 1 )
+
+    def testPopen( self ):
+        self.pingTest( 'mininet.examples.popen' )
+
+    def testPopenPoll( self ):
+        self.pingTest( 'mininet.examples.popenpoll')
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_scratchnet.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from mininet.log import setLogLevel
+
+class testScratchNet( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    results = [ "1 packets transmitted, 1 received, 0% packet loss", pexpect.EOF ]
+
+    def pingTest( self, name ):
+        p = pexpect.spawn( 'python -m %s' % name )
+        index = p.expect( self.results )
+        self.assertEqual( index, 0 )
+
+
+    def testPingKernel( self ):
+        self.pingTest( 'mininet.examples.scratchnet' )
+
+
+    def testPingUser( self ):
+        self.pingTest( 'mininet.examples.scratchnetuser' )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_simpleperf.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.log import setLogLevel
+from mininet.net import Mininet
+from mininet.node import CPULimitedHost
+from mininet.link import TCLink
+
+from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testSimplePerf( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+
+    def testE2E( self ):
+        results = [ "Results:", pexpect.EOF, pexpect.TIMEOUT ]
+        p = pexpect.spawn( 'python -m mininet.examples.simpleperf' )
+        index = p.expect( results, timeout=600 )
+        self.assertEqual( index, 0 )
+        p.wait()
+
+    def testTopo( self ):
+        topo = SingleSwitchTopo(n=4)
+        net = Mininet(topo=topo,
+                  host=CPULimitedHost, link=TCLink)
+        net.start()
+        h1, h4 = net.get('h1', 'h4')
+        h1.cmd( 'ping -c 1 %s' % h4.IP() )
+        net.stop()
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_sshd.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.log import setLogLevel
+from mininet.clean import cleanup, sh
+
+class testBareSSHD( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    opts = [ '\(yes/no\)\?', 'refused', 'Welcome', pexpect.EOF, pexpect.TIMEOUT ]
+
+    def connected( self, ip ):
+        "check connected"
+        p = pexpect.spawn( 'ssh -i /tmp/ssh/test_rsa %s' % ip )
+        while True:
+            index = p.expect( self.opts )
+            if index == 0:
+                print p.match.group(0)
+                p.sendline( 'yes' )
+            elif index == 1:
+                return False
+            elif index == 2:
+                p.sendline( 'exit' )
+                p.wait()    
+                return True
+            else:
+                return False
+
+    def setUp( self ):
+        # create public key pair for testing
+        sh( 'mkdir /tmp/ssh' )
+        sh( "ssh-keygen -t rsa -P '' -f /tmp/ssh/test_rsa" )
+        sh( 'cat /tmp/ssh/test_rsa.pub >> /tmp/ssh/authorized_keys' )
+        cmd = ( 'python -m mininet.examples.sshd -D '
+                '-o AuthorizedKeysFile=/tmp/ssh/authorized_keys '
+                '-o StrictModes=no' )
+        self.net = pexpect.spawn( cmd )
+        self.net.expect( 'mininet>' )
+
+    def testSSH( self ):
+        for h in range( 1, 5 ):
+            self.assertTrue( self.connected( '10.0.0.%d' % h ) )
+
+    def tearDown( self ):
+        self.net.sendline( 'exit' )
+        self.net.wait()
+        # remove public key pair
+        sh( 'rm -rf /tmp/ssh' )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()
+

examples/test/test_tree1024.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+#from time import sleep
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testTree1024( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testTree1024( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.tree1024' )
+        p.expect( self.prompt, timeout=6000 ) # it takes awhile to set up
+        p.sendline( 'h1 ping -c 1 h1024' )
+        p.expect ( '(\d+)% packet loss' )
+        percent = int( p.match.group( 1 ) ) if p.match else -1
+        #self.assertEqual( percent, 0 )
+        #p.expect( self.prompt )
+        #p.sendline( 'iperf' )
+        #p.expect( "Results: \['\d+ .bits/sec', '\d+ .bits/sec'\]" )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+        self.assertEqual( percent, 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_treeping64.py

+#!/usr/bin/env python
+
+"""TEST"""
+
+import unittest
+import pexpect
+#from time import sleep
+from mininet.log import setLogLevel
+#from mininet.net import Mininet
+#from mininet.node import CPULimitedHost
+#from mininet.link import TCLink
+
+#from mininet.examples.simpleperf import SingleSwitchTopo
+
+class testTreePing64( unittest.TestCase ):
+    "Test ping with single switch topology (common code)."
+
+    prompt = 'mininet>'
+
+    def testTreePing64( self ):
+        p = pexpect.spawn( 'python -m mininet.examples.treeping64' )
+        p.expect( 'Tree network ping results:', timeout=6000 )
+        count = 0
+        while True:
+            index = p.expect( [ '(\d+)% packet loss', pexpect.EOF ] )
+            if index == 0:
+                percent = int( p.match.group( 1 ) ) if p.match else -1
+                self.assertEqual( percent, 0 )
+                count += 1
+            else:
+                break
+        self.assertTrue( count > 0 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()