test/resources/linux/ssh_api.py

   1 #!/usr/bin/env python
   2 from neco.resources.linux.ssh_api import SSHApiFactory
   3 from neco.util.sshfuncs import RUNNING, FINISHED
   4
   5 import os
   6 import time
   7 import tempfile
   8 import unittest
   9
  10 def skipIfNotAlive(func):
  11     name = func.__name__
  12     def wrapped(*args, **kwargs):
  13         host = args[1]
  14         user = args[2]
  15
  16         api = SSHApiFactory.get_api(host, user)
  17         if not api.is_alive():
  18             print "*** WARNING: Skipping test %s: Node %s is not alive\n" % (name, host)
  19             return
  20
  21         return func(*args, **kwargs)
  22
  23     return wrapped
  24
  25 def skipInteractive(func):
  26     name = func.__name__
  27     def wrapped(*args, **kwargs):
  28         mode = os.environ.get("NEPI_INTERACTIVE", False) in ['True', 'true', 'yes', 'YES']
  29         if not mode:
  30             print "*** WARNING: Skipping test %s: Interactive mode off \n" % name
  31             return
  32
  33         return func(*args, **kwargs)
  34
  35     return wrapped
  36
  37 class SSHApiTestCase(unittest.TestCase):
  38     def setUp(self):
  39         self.host_fedora = 'nepi2.pl.sophia.inria.fr'
  40         self.user_fedora = 'inria_nepi'
  41
  42         self.host_ubuntu = 'roseval.pl.sophia.inria.fr'
  43         self.user_ubuntu = 'alina'
  44
  45         self.target = 'nepi5.pl.sophia.inria.fr'
  46         self.home = '${HOME}/test-app'
  47
  48     @skipIfNotAlive
  49     def t_xterm(self, host, user):
  50         api = SSHApiFactory.get_api(host, user)
  51
  52         api.enable_x11 = True
  53
  54         api.install('xterm')
  55
  56         out = api.execute('xterm')
  57
  58         api.uninstall('xterm')
  59
  60         self.assertEquals(out, "")
  61
  62     @skipIfNotAlive
  63     def t_execute(self, host, user):
  64         api = SSHApiFactory.get_api(host, user)
  65
  66         command = "ping -qc3 %s" % self.target
  67         out, err = api.execute(command)
  68
  69         expected = """3 packets transmitted, 3 received, 0% packet loss"""
  70
  71         self.assertTrue(out.find(expected) > 0)
  72
  73     @skipIfNotAlive
  74     def t_run(self, host, user):
  75         api = SSHApiFactory.get_api(host, user)
  76
  77         api.mkdir(self.home, clean = True)
  78
  79         command = "ping %s" % self.target
  80         dst = os.path.join(self.home, "app.sh")
  81         api.upload(command, dst)
  82
  83         cmd = "bash ./app.sh"
  84         api.run(cmd, self.home)
  85         pid, ppid = api.checkpid(self.home)
  86
  87         status = api.status(pid, ppid)
  88         self.assertTrue(status, RUNNING)
  89
  90         api.kill(pid, ppid)
  91         status = api.status(pid, ppid)
  92         self.assertTrue(status, FINISHED)
  93
  94         api.rmdir(self.home)
  95
  96     @skipIfNotAlive
  97     def t_install(self, host, user):
  98         api = SSHApiFactory.get_api(host, user)
  99
 100         api.mkdir(self.home, clean = True)
 101
 102         prog = """#include <stdio.h>
 103
 104 int
 105 main (void)
 106 {
 107     printf ("Hello, world!\\n");
 108     return 0;
 109 }
 110 """
 111         # upload the test program
 112         dst = os.path.join(self.home, "hello.c")
 113         api.upload(prog, dst)
 114
 115         # install gcc
 116         api.install('gcc')
 117
 118         # compile the program using gcc
 119         command = "cd %s; gcc -Wall hello.c -o hello" % self.home
 120         out = api.execute(command)
 121
 122         # execute the program and get the output from stout
 123         command = "%s/hello" % self.home
 124         out, err = api.execute(command)
 125
 126         # execute the program and get the output from a file
 127         command = "%s/hello > %s/hello.out" % (self.home, self.home)
 128         api.execute(command)
 129
 130         # retrieve the output file
 131         src = os.path.join(self.home, "hello.out")
 132         f = tempfile.NamedTemporaryFile(delete=False)
 133         dst = f.name
 134         api.download(src, dst)
 135         f.close()
 136
 137         api.uninstall('gcc')
 138         api.rmdir(self.home)
 139
 140         self.assertEquals(out, "Hello, world!\n")
 141
 142         f = open(dst, "r")
 143         out = f.read()
 144         f.close()
 145
 146         self.assertEquals(out, "Hello, world!\n")
 147
 148     def test_execute_fedora(self):
 149         self.t_execute(self.host_fedora, self.user_fedora)
 150
 151     def test_execute_ubuntu(self):
 152         self.t_execute(self.host_ubuntu, self.user_ubuntu)
 153
 154     def test_run_fedora(self):
 155         self.t_run(self.host_fedora, self.user_fedora)
 156
 157     def test_run_ubuntu(self):
 158         self.t_run(self.host_ubuntu, self.user_ubuntu)
 159
 160     def test_intall_fedora(self):
 161         self.t_install(self.host_fedora, self.user_fedora)
 162
 163     def test_install_ubuntu(self):
 164         self.t_install(self.host_ubuntu, self.user_ubuntu)
 165
 166     @skipInteractive
 167     def test_xterm_ubuntu(self):
 168         """ Interactive test. Should not run automatically """
 169         self.t_xterm(self.host_ubuntu, self.user_ubuntu)
 170
 171
 172 if __name__ == '__main__':
 173     unittest.main()
 174