2 from neco.resources.linux.node import LinuxNode
3 from neco.util.sshfuncs import RUNNING, FINISHED
5 from test_utils import skipIfNotAlive, skipInteractive, create_node
12 class LinuxNodeTestCase(unittest.TestCase):
14 self.fedora_host = 'nepi2.pl.sophia.inria.fr'
15 self.fedora_user = 'inria_nepi'
17 self.ubuntu_host = 'roseval.pl.sophia.inria.fr'
18 self.ubuntu_user = 'alina'
20 self.target = 'nepi5.pl.sophia.inria.fr'
23 def t_xterm(self, host, user):
24 node, ec = create_node(host, user)
26 node.install_packages('xterm')
28 (out, err), proc = node.execute('xterm', forward_x11 = True)
30 self.assertEquals(out, "")
32 (out, err), proc = node.remove_packages('xterm')
34 self.assertEquals(out, "")
37 def t_execute(self, host, user):
38 node, ec = create_node(host, user)
40 command = "ping -qc3 %s" % self.target
42 (out, err), proc = node.execute(command)
44 expected = """3 packets transmitted, 3 received, 0% packet loss"""
46 self.assertTrue(out.find(expected) > 0)
49 def t_run(self, host, user):
50 node, ec = create_node(host, user)
52 app_home = os.path.join(node.exp_dir, "my-app")
53 node.mkdir(app_home, clean = True)
55 command = "ping %s" % self.target
56 node.run(command, app_home)
57 pid, ppid = node.checkpid(app_home)
59 status = node.status(pid, ppid)
60 self.assertTrue(status, RUNNING)
63 status = node.status(pid, ppid)
64 self.assertTrue(status, FINISHED)
66 (out, err), proc = node.check_output(app_home, "stdout")
68 expected = """64 bytes from"""
70 self.assertTrue(out.find(expected) > 0)
75 def t_install(self, host, user):
76 node, ec = create_node(host, user)
78 app_home = os.path.join(node.exp_dir, "my-app")
79 node.mkdir(app_home, clean = True)
81 prog = """#include <stdio.h>
86 printf ("Hello, world!\\n");
90 # upload the test program
91 dst = os.path.join(app_home, "hello.c")
92 node.upload(prog, dst, text = True)
95 node.install_packages('gcc')
97 # compile the program using gcc
98 command = "cd %s; gcc -Wall hello.c -o hello" % app_home
99 (out, err), proc = node.execute(command)
101 # execute the program and get the output from stdout
102 command = "%s/hello" % app_home
103 (out, err), proc = node.execute(command)
105 self.assertEquals(out, "Hello, world!\n")
107 # execute the program and get the output from a file
108 command = "%(home)s/hello > %(home)s/hello.out" % {
110 (out, err), proc = node.execute(command)
112 # retrieve the output file
113 src = os.path.join(app_home, "hello.out")
114 f = tempfile.NamedTemporaryFile(delete=False)
116 node.download(src, dst)
119 node.remove_packages('gcc')
126 self.assertEquals(out, "Hello, world!\n")
128 def test_execute_fedora(self):
129 self.t_execute(self.fedora_host, self.fedora_user)
131 def test_execute_ubuntu(self):
132 self.t_execute(self.ubuntu_host, self.ubuntu_user)
134 def test_run_fedora(self):
135 self.t_run(self.fedora_host, self.fedora_user)
137 def test_run_ubuntu(self):
138 self.t_run(self.ubuntu_host, self.ubuntu_user)
140 def test_intall_fedora(self):
141 self.t_install(self.fedora_host, self.fedora_user)
143 def test_install_ubuntu(self):
144 self.t_install(self.ubuntu_host, self.ubuntu_user)
147 def test_xterm_ubuntu(self):
148 """ Interactive test. Should not run automatically """
149 self.t_xterm(self.ubuntu_host, self.ubuntu_user)
152 if __name__ == '__main__':