Merge pull request #207 from mininet/devel/exampletest

Added tests for all non-X11 examples

examples/baresshd.py

@@ -2,6 +2,7 @@
 
 "This example doesn't use OpenFlow, but attempts to run sshd in a namespace."
 
+import sys
 from mininet.node import Host
 from mininet.util import ensureRoot
 
@@ -27,6 +28,10 @@ f.write( 'Welcome to %s at %s\n' % ( h1.name, h1.IP() ) )
 f.close()
 
 print "*** Running sshd"
-h1.cmd( '/usr/sbin/sshd -o "Banner /tmp/%s.banner"' % h1.name )
+cmd = '/usr/sbin/sshd -o "Banner /tmp/%s.banner"' % h1.name
+# add arguments from the command line
+if len( sys.argv ) > 1:
+    cmd += ' ' + ' '.join( sys.argv[ 1: ] )
+h1.cmd( cmd )
 
 print "*** You may now ssh into", h1.name, "at", h1.IP()

examples/controlnet.py

@@ -14,6 +14,8 @@ We also use a Mininet Facade to talk to both the
 control and data networks from a single CLI.
 """
 
+from functools import partial
+
 from mininet.net import Mininet
 from mininet.node import Controller, UserSwitch
 from mininet.cli import CLI
@@ -124,7 +126,8 @@ def run():
     info( '* Creating Data Network\n' )
     topo = TreeTopo( depth=2, fanout=2 )
     # UserSwitch so we can easily test failover
-    net = Mininet( topo=topo, switch=UserSwitch, controller=None )
+    sw = partial( UserSwitch, opts='--inactivity-probe=1 --max-backoff=1' )
+    net = Mininet( topo=topo, switch=sw, controller=None )
     info( '* Adding Controllers to Data Network\n' )
     for host in cnet.hosts:
         if isinstance(host, Controller):

examples/hwintf.py

@@ -5,7 +5,7 @@ This example shows how to add an interface (for example a real
 hardware interface) to a network after the network is created.
 """
 
-import re
+import re, sys
 
 from mininet.cli import CLI
 from mininet.log import setLogLevel, info, error
@@ -28,7 +28,10 @@ def checkIntf( intf ):
 if __name__ == '__main__':
     setLogLevel( 'info' )
 
-    intfName = 'eth1'
+    # try to get hw intf from the command line; by default, use eth1
+    intfName = sys.argv[ 1 ] if len( sys.argv ) > 1 else 'eth1'
+    info( '*** Connecting to hw intf: %s' % intfName )
+
     info( '*** Checking', intfName, '\n' )
     checkIntf( intfName )
 

examples/popenpoll.py

@@ -23,7 +23,7 @@ def pmonitorTest( N=3, seconds=10 ):
     endTime = time() + seconds
     for h, line in pmonitor( popens, timeoutms=500 ):
         if h:
-            print '%s: %s' % ( h.name, line ),
+            print '<%s>: %s' % ( h.name, line ),
         if time() >= endTime:
             for p in popens.values():
                 p.send_signal( SIGINT )

examples/sshd.py

@@ -16,6 +16,8 @@ demonstrates:
 - running server processes (sshd in this case) on hosts
 """
 
+import sys
+
 from mininet.net import Mininet
 from mininet.cli import CLI
 from mininet.log import lg
@@ -68,4 +70,6 @@ def sshd( network, cmd='/usr/sbin/sshd', opts='-D' ):
 if __name__ == '__main__':
     lg.setLogLevel( 'info')
     net = TreeNet( depth=1, fanout=4, switch=OVSKernelSwitch )
-    sshd( net )
+    # get sshd args from the command line; default: -D
+    opts = ' '.join( sys.argv[ 1: ] ) if len( sys.argv ) > 1 else '-D'
+    sshd( net, opts=opts )

examples/test/runner.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+"""
+Run all mininet.examples tests
+ -v : verbose output
+ -quick : skip tests that take more than ~30 seconds
+"""
+
+import unittest
+import os
+import sys
+from mininet.util import ensureRoot
+from mininet.clean import cleanup
+
+def runTests( testDir, verbosity=1 ):
+    "discover and run all tests in testDir"
+    # ensure root and cleanup before starting tests
+    ensureRoot()
+    cleanup()
+    # discover all tests in testDir
+    testSuite = unittest.defaultTestLoader.discover( testDir )
+    # run tests
+    unittest.TextTestRunner( verbosity=verbosity ).run( testSuite )
+
+if __name__ == '__main__':
+    # get the directory containing example tests
+    testDir = os.path.dirname( os.path.realpath( __file__ ) )
+    verbosity = 2 if '-v' in sys.argv else 1
+    runTests( testDir, verbosity )

examples/test/test_baresshd.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python
+
+"""
+Tests for baresshd.py
+"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.clean import cleanup, sh
+
+class testBareSSHD( unittest.TestCase ):
+
+    opts = [ '\(yes/no\)\?', 'Welcome to h1', 'refused', pexpect.EOF, pexpect.TIMEOUT ]
+
+    def connected( self ):
+        "Log into ssh server, check banner, then exit"
+        p = pexpect.spawn( 'ssh 10.0.0.1 -i /tmp/ssh/test_rsa exit' )
+        while True:
+            index = p.expect( self.opts )
+            if index == 0:
+                p.sendline( 'yes' )
+            elif index == 1:
+                return True
+            else:
+                return False
+
+    def setUp( self ):
+        # verify that sshd is not running
+        self.assertFalse( self.connected() )
+        # create public key pair for testing
+        sh( 'rm -rf /tmp/ssh' )
+        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' )
+        # run example with custom sshd args
+        cmd = ( 'python -m mininet.examples.baresshd '
+                '-o AuthorizedKeysFile=/tmp/ssh/authorized_keys '
+                '-o StrictModes=no' )
+        sh( cmd )
+
+    def testSSH( self ):
+        "Simple test to verify that we can ssh into h1"
+        result = False
+        # try to connect up to 3 times; sshd can take a while to start
+        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__':
+    unittest.main()

examples/test/test_bind.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python
+
+"""
+Tests for bind.py
+"""
+
+import unittest
+import pexpect
+
+class testBind( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    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 ):
+        "Create a file, a.txt, in the first private directory and verify"
+        fileName = 'a.txt'
+        directory = self.directories[ 0 ]
+        path = directory + '/' + fileName
+        self.net.sendline( 'h1 touch %s; ls %s' % ( path, directory ) )
+        index = self.net.expect( [ fileName, self.prompt ] )
+        self.assertTrue( index == 0 )
+        self.net.expect( self.prompt )
+        self.net.sendline( 'h1 rm %s' % path )
+        self.net.expect( self.prompt )
+
+    def testIsolation( self ):
+        "Create a file in two hosts and verify that contents are different"
+        fileName = 'b.txt'
+        directory = self.directories[ 0 ]
+        path = directory + '/' + fileName
+        contents = { 'h1' : '1', 'h2' : '2' }
+        # Verify file doesn't exist, then write private copy of file
+        for host in contents:
+            value = contents[ host ]
+            self.net.sendline( '%s cat %s' % ( host, path ) )
+            self.net.expect( 'No such file' )
+            self.net.expect( self.prompt )
+            self.net.sendline( '%s echo %s > %s' % ( host, value, path ) )
+            self.net.expect( self.prompt )
+        # Verify file contents
+        for host in contents:
+            value = contents[ host ]
+            self.net.sendline( '%s cat %s' % ( host, path ) )
+            self.net.expect( value )
+            self.net.expect( self.prompt )
+            self.net.sendline( '%s rm %s' % ( host, path ) )
+            self.net.expect( self.prompt )
+
+    # TODO: need more tests
+
+    def tearDown( self ):
+        self.net.sendline( 'exit' )
+        self.net.wait()
+
+if __name__ == '__main__':
+    unittest.main()

examples/test/test_controllers.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+"""
+Tests for controllers.py and controllers2.py
+"""
+
+import unittest
+import pexpect
+
+class testControllers( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    def connectedTest( self, name, cmap ):
+        "Verify that switches are connected to the controller specified by cmap"
+        p = pexpect.spawn( 'python -m %s' % name )
+        p.expect( self.prompt )
+        # but first a simple ping test
+        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 )
+        # verify connected controller
+        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()
+
+    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__':
+    unittest.main()

examples/test/test_controlnet.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+
+"""
+Test for controlnet.py
+"""
+
+import unittest
+import pexpect
+
+class testControlNet( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    def testPingall( self ):
+        "Simple pingall test that verifies 0% packet drop in data network"
+        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 ):
+        "Kill controllers and verity that switch, s1, fails over properly"
+        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__':
+    unittest.main()

examples/test/test_cpu.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+"""
+Test for cpu.py
+"""
+
+import unittest
+import pexpect
+import sys
+
+class testCPU( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testCPU( self ):
+        "Verify that CPU utilization is monotonically decreasing for each scheduler"
+        p = pexpect.spawn( 'python -m mininet.examples.cpu' )
+        opts = [ '([a-z]+)\t([\d\.]+)%\t([\d\.]+)', pexpect.EOF ]
+        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__':
+    unittest.main()

examples/test/test_emptynet.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+"""
+Test for emptynet.py
+"""
+
+import unittest
+import pexpect
+
+class testEmptyNet( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    def testEmptyNet( self ):
+        "Run simple CLI tests: pingall (verify 0% drop) and iperf (sanity)"
+        p = pexpect.spawn( 'python -m mininet.examples.emptynet' )
+        p.expect( self.prompt )
+        # pingall test
+        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 )
+        # iperf test
+        p.sendline( 'iperf' )
+        p.expect( "Results: \['[\d.]+ .bits/sec', '[\d.]+ .bits/sec'\]" )
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+
+if __name__ == '__main__':
+    unittest.main()

examples/test/test_hwintf.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+
+"""
+Test for hwintf.py
+"""
+
+import unittest
+import pexpect
+import re
+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 ):
+
+    prompt = 'mininet>'
+
+    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 ):
+        "Verify connectivity between virtual hosts using pingall"
+        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 ):
+        "Verify connnectivity between virtual host and virtual-physical 'external' host "
+        p = pexpect.spawn( 'python -m mininet.examples.hwintf %s' % self.n0.intf() )
+        p.expect( self.prompt )
+        # test ping external to internal
+        expectStr = '(\d+) packets transmitted, (\d+) received'
+        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 )
+        # test ping internal to external
+        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()
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_limit.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+"""
+Test for limit.py
+"""
+
+import unittest
+import pexpect
+import sys
+
+class testLimit( unittest.TestCase ):
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testLimit( self ):
+        "Verify that CPU limits are within a 2% tolerance of limit for each scheduler"
+        p = pexpect.spawn( 'python -m mininet.examples.limit' )
+        opts = [ '\*\*\* Testing network ([\d\.]+) Mbps', 
+                 '\*\*\* Results: \[([\d\., ]+)\]', 
+                 pexpect.EOF ]
+        count = 0
+        bw = 0
+        tolerance = 2
+        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__':
+    unittest.main()

examples/test/test_linearbandwidth.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+
+"""
+Test for linearbandwidth.py
+"""
+
+import unittest
+import pexpect
+import sys
+
+class testLinearBandwidth( unittest.TestCase ):
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testLinearBandwidth( self ):
+        "Verify that bandwidth is monotonically decreasing as # of hops increases"
+        p = pexpect.spawn( 'python -m mininet.examples.linearbandwidth' )
+        count = 0
+        opts = [ '\*\*\* Linear network results', 
+                 '(\d+)\s+([\d\.]+) (.bits)', 
+                 pexpect.EOF ]
+        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__':
+    unittest.main()

examples/test/test_multiping.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+"""
+Test for multiping.py
+"""
+
+import unittest
+import pexpect
+from collections import defaultdict
+
+class testMultiPing( unittest.TestCase ):
+
+    def testMultiPing( self ):
+        """Verify that each target is pinged at least once, and 
+           that pings to 'real' targets are successful and unknown targets fail"""
+        p = pexpect.spawn( 'python -m mininet.examples.multiping' )
+        opts = [ "Host (h\d+) \(([\d.]+)\) will be pinging ips: ([\d\. ]+)",
+                 "(h\d+): ([\d.]+) -> ([\d.]+) \d packets transmitted, (\d) received",
+                 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__':
+    unittest.main()

examples/test/test_multipoll.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+"""
+Test for multipoll.py
+"""
+
+import unittest
+import pexpect
+
+class testMultiPoll( unittest.TestCase ):
+
+    def testMultiPoll( self ):
+        "Verify that we receive one ping per second per host"
+        p = pexpect.spawn( 'python -m mininet.examples.multipoll' )
+        opts = [ "\*\*\* (h\d) :" ,
+                 "(h\d+): \d+ bytes from",
+                 "Monitoring output for (\d+) seconds",
+                 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__':
+    unittest.main()

examples/test/test_multitest.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+"""
+Test for multitest.py
+"""
+
+import unittest
+import pexpect
+
+class testMultiTest( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    def testMultiTest( self ):
+        "Verify pingall (0% dropped) and hX-eth0 interface for each host (ifconfig)"
+        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__':
+    unittest.main()

examples/test/test_nat.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+"""
+Test for nat.py
+"""
+
+import unittest
+import pexpect
+from mininet.util import quietRun
+
+destIP = '8.8.8.8' # Google DNS
+
+class testNAT( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    @unittest.skipIf( '0 received' in quietRun( 'ping -c 1 %s' % destIP ), 
+                      'Destination IP is not reachable' )
+    def testNAT( self ):
+        "Attempt to ping an IP on the Internet and verify 0% packet loss"
+        p = pexpect.spawn( 'python -m mininet.examples.nat' )
+        p.expect( self.prompt )
+        p.sendline( 'h1 ping -c 1 %s' % destIP )
+        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 )
+
+if __name__ == '__main__':
+    unittest.main()

examples/test/test_popen.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python
+
+"""
+Test for popen.py and popenpoll.py
+"""
+
+import unittest
+import pexpect
+
+class testPopen( unittest.TestCase ):
+
+    def pingTest( self, name ):
+        "Verify that there are no dropped packets for each host"
+        p = pexpect.spawn( 'python -m %s' % name )
+        opts = [ "<(h\d+)>: PING ",
+                 "<(h\d+)>: (\d+) packets transmitted, (\d+) received",
+                 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)
+                # verify no dropped packets
+                self.assertEqual( received, transmitted )
+                pings[ name ] += 1
+            else:
+                break
+        self.assertTrue( len(pings) > 0 )
+        # verify that each host has gotten results
+        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__':
+    unittest.main()

examples/test/test_scratchnet.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python
+
+"""
+Test for scratchnet.py
+"""
+
+import unittest
+import pexpect
+
+class testScratchNet( unittest.TestCase ):
+
+    opts = [ "1 packets transmitted, 1 received, 0% packet loss", pexpect.EOF ]
+
+    def pingTest( self, name ):
+        "Verify that no ping packets were dropped"
+        p = pexpect.spawn( 'python -m %s' % name )
+        index = p.expect( self.opts )
+        self.assertEqual( index, 0 )
+
+    def testPingKernel( self ):
+        self.pingTest( 'mininet.examples.scratchnet' )
+
+    def testPingUser( self ):
+        self.pingTest( 'mininet.examples.scratchnetuser' )
+
+if __name__ == '__main__':
+    unittest.main()

examples/test/test_simpleperf.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+"""
+Test for simpleperf.py
+"""
+
+import unittest
+import pexpect
+import re
+import sys
+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 ):
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testE2E( self ):
+        "Run the example and verify ping and iperf results"
+        p = pexpect.spawn( 'python -m mininet.examples.simpleperf' )
+        # check ping results
+        p.expect( "Results: (\d+)% dropped", timeout=120 )
+        loss = int( p.match.group( 1 ) )
+        self.assertTrue( loss > 0 and loss < 100 )
+        # check iperf results
+        p.expect( "Results: \['([\d\.]+) .bits/sec", timeout=480 )
+        bw = float( p.match.group( 1 ) )
+        self.assertTrue( bw > 0 )
+        p.wait()
+
+    def testTopo( self ):
+        """Import SingleSwitchTopo from example and test connectivity between two hosts
+           Note: this test may fail very rarely because it is non-deterministic
+           i.e. links are configured with 10% packet loss, but if we get unlucky and 
+           none or all of the packets are dropped, the test will fail"""
+        topo = SingleSwitchTopo( n=4 )
+        net = Mininet( topo=topo, host=CPULimitedHost, link=TCLink )
+        net.start()
+        h1, h4 = net.get( 'h1', 'h4' )
+        # have h1 ping h4 ten times
+        expectStr = '(\d+) packets transmitted, (\d+) received, (\d+)% packet loss'
+        output = h1.cmd( 'ping -c 10 %s' % h4.IP() )
+        m = re.search( expectStr, output )
+        loss = int( m.group( 3 ) )
+        net.stop()
+        self.assertTrue( loss > 0 and loss < 100 )
+
+if __name__ == '__main__':
+    setLogLevel( 'warning' )
+    unittest.main()

examples/test/test_sshd.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+
+"""
+Test for sshd.py
+"""
+
+import unittest
+import pexpect
+from time import sleep
+from mininet.clean import sh
+
+class testSSHD( unittest.TestCase ):
+
+    opts = [ '\(yes/no\)\?', 'refused', 'Welcome|\$|#', pexpect.EOF, pexpect.TIMEOUT ]
+
+    def connected( self, ip ):
+        "Log into ssh server, check banner, then exit"
+        # Note: this test will fail if "Welcome" is not in the sshd banner 
+        # and '#'' or '$'' are not in the prompt
+        p = pexpect.spawn( 'ssh -i /tmp/ssh/test_rsa %s' % ip, timeout=10 )
+        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( 'rm -rf /tmp/ssh' )
+        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' )
+        # run example with custom sshd args
+        self.net = pexpect.spawn( cmd )
+        self.net.expect( 'mininet>' )
+
+    def testSSH( self ):
+        "Verify that we can ssh into all hosts (h1 to h4)"
+        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__':
+    unittest.main()
+

examples/test/test_tree1024.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+"""
+Test for tree1024.py
+"""
+
+import unittest
+import pexpect
+import sys
+
+class testTree1024( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testTree1024( self ):
+        "Run the example and do a simple ping test from h1 to h1024"
+        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
+        p.expect( self.prompt )
+        p.sendline( 'exit' )
+        p.wait()
+        self.assertEqual( percent, 0 )
+
+if __name__ == '__main__':
+    unittest.main()

examples/test/test_treeping64.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+"""
+Test for treeping64.py
+"""
+
+import unittest
+import pexpect
+import sys
+
+class testTreePing64( unittest.TestCase ):
+
+    prompt = 'mininet>'
+
+    @unittest.skipIf( '-quick' in sys.argv, 'long test' )
+    def testTreePing64( self ):
+        "Run the example and verify ping results"
+        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__':
+    unittest.main()