2 from neco.resources.linux.ssh_api import SSHApiFactory
3 from neco.util.sshfuncs import RUNNING, FINISHED
10 def skipIfNotAlive(func):
12 def wrapped(*args, **kwargs):
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)
21 return func(*args, **kwargs)
25 def skipInteractive(func):
27 def wrapped(*args, **kwargs):
28 mode = os.environ.get("NEPI_INTERACTIVE", False) in ['True', 'true', 'yes', 'YES']
30 print "*** WARNING: Skipping test %s: Interactive mode off \n" % name
33 return func(*args, **kwargs)
37 class SSHApiTestCase(unittest.TestCase):
39 self.host_fedora = 'nepi2.pl.sophia.inria.fr'
40 self.user_fedora = 'inria_nepi'
42 self.host_ubuntu = 'roseval.pl.sophia.inria.fr'
43 self.user_ubuntu = 'alina'
45 self.target = 'nepi5.pl.sophia.inria.fr'
46 self.home = '${HOME}/test-app'
49 def t_xterm(self, host, user):
50 api = SSHApiFactory.get_api(host, user)
56 out = api.execute('xterm')
58 api.uninstall('xterm')
60 self.assertEquals(out, "")
63 def t_execute(self, host, user):
64 api = SSHApiFactory.get_api(host, user)
66 command = "ping -qc3 %s" % self.target
67 out, err = api.execute(command)
69 expected = """3 packets transmitted, 3 received, 0% packet loss"""
71 self.assertTrue(out.find(expected) > 0)
74 def t_run(self, host, user):
75 api = SSHApiFactory.get_api(host, user)
77 api.mkdir(self.home, clean = True)
79 command = "ping %s" % self.target
80 dst = os.path.join(self.home, "app.sh")
81 api.upload(command, dst)
84 api.run(cmd, self.home)
85 pid, ppid = api.checkpid(self.home)
87 status = api.status(pid, ppid)
88 self.assertTrue(status, RUNNING)
91 status = api.status(pid, ppid)
92 self.assertTrue(status, FINISHED)
97 def t_install(self, host, user):
98 api = SSHApiFactory.get_api(host, user)
100 api.mkdir(self.home, clean = True)
102 prog = """#include <stdio.h>
107 printf ("Hello, world!\\n");
111 # upload the test program
112 dst = os.path.join(self.home, "hello.c")
113 api.upload(prog, dst)
118 # compile the program using gcc
119 command = "cd %s; gcc -Wall hello.c -o hello" % self.home
120 out = api.execute(command)
122 # execute the program and get the output from stout
123 command = "%s/hello" % self.home
124 out, err = api.execute(command)
126 # execute the program and get the output from a file
127 command = "%s/hello > %s/hello.out" % (self.home, self.home)
130 # retrieve the output file
131 src = os.path.join(self.home, "hello.out")
132 f = tempfile.NamedTemporaryFile(delete=False)
134 api.download(src, dst)
140 self.assertEquals(out, "Hello, world!\n")
146 self.assertEquals(out, "Hello, world!\n")
148 def test_execute_fedora(self):
149 self.t_execute(self.host_fedora, self.user_fedora)
151 def test_execute_ubuntu(self):
152 self.t_execute(self.host_ubuntu, self.user_ubuntu)
154 def test_run_fedora(self):
155 self.t_run(self.host_fedora, self.user_fedora)
157 def test_run_ubuntu(self):
158 self.t_run(self.host_ubuntu, self.user_ubuntu)
160 def test_intall_fedora(self):
161 self.t_install(self.host_fedora, self.user_fedora)
163 def test_install_ubuntu(self):
164 self.t_install(self.host_ubuntu, self.user_ubuntu)
167 def test_xterm_ubuntu(self):
168 """ Interactive test. Should not run automatically """
169 self.t_xterm(self.host_ubuntu, self.user_ubuntu)
172 if __name__ == '__main__':