mininet/examples/test/test_baresshd.py

   1 #!/usr/bin/env python
   2
   3 """
   4 Tests for baresshd.py
   5 """
   6
   7 import unittest
   8 from mininet.util import pexpect
   9 from mininet.clean import cleanup, sh
  10 from sys import stdout
  11
  12 class testBareSSHD( unittest.TestCase ):
  13
  14     opts = [ 'Welcome to h1', pexpect.EOF, pexpect.TIMEOUT ]
  15
  16     def connected( self ):
  17         "Log into ssh server, check banner, then exit"
  18         p = pexpect.spawn( 'ssh 10.0.0.1 -o ConnectTimeout=1 '
  19                            '-o StrictHostKeyChecking=no '
  20                            '-i /tmp/ssh/test_rsa exit' )
  21         while True:
  22             index = p.expect( self.opts )
  23             if index == 0:
  24                 return True
  25             else:
  26                 return False
  27
  28
  29     def setUp( self ):
  30         # verify that sshd is not running
  31         self.assertFalse( self.connected() )
  32         # create public key pair for testing
  33         sh( 'rm -rf /tmp/ssh' )
  34         sh( 'mkdir /tmp/ssh' )
  35         sh( "ssh-keygen -t rsa -P '' -f /tmp/ssh/test_rsa" )
  36         sh( 'cat /tmp/ssh/test_rsa.pub >> /tmp/ssh/authorized_keys' )
  37         # run example with custom sshd args
  38         cmd = ( 'python -m mininet.examples.baresshd '
  39                 '-o AuthorizedKeysFile=/tmp/ssh/authorized_keys '
  40                 '-o StrictModes=no' )
  41         p = pexpect.spawn( cmd )
  42         runOpts = [ 'You may now ssh into h1 at 10.0.0.1',
  43                     'after 5 seconds, h1 is not listening on port 22',
  44                     pexpect.EOF, pexpect.TIMEOUT ]
  45         while True:
  46             index = p.expect( runOpts )
  47             if index == 0:
  48                 break
  49             else:
  50                 self.tearDown()
  51                 self.fail( 'sshd failed to start in host h1' )
  52
  53     def testSSH( self ):
  54         "Simple test to verify that we can ssh into h1"
  55         result = False
  56         # try to connect up to 3 times; sshd can take a while to start
  57         result = self.connected()
  58         self.assertTrue( result )
  59
  60     def tearDown( self ):
  61         # kill the ssh process
  62         sh( "ps aux | grep ssh |grep Banner| awk '{ print $2 }' | xargs kill" )
  63         cleanup()
  64         # remove public key pair
  65         sh( 'rm -rf /tmp/ssh' )
  66
  67 if __name__ == '__main__':
  68     unittest.main()