2 from neco.resources.linux.node import LinuxNode
3 from neco.util.sshfuncs import RUNNING, FINISHED
11 class DummyEC(object):
14 class LinuxNodeTestCase(unittest.TestCase):
16 host = 'nepi2.pl.sophia.inria.fr'
18 self.node_fedora = self.create_node(host, user)
20 host = 'roseval.pl.sophia.inria.fr'
22 self.node_ubuntu = self.create_node(host, user)
24 self.target = 'nepi5.pl.sophia.inria.fr'
25 self.home = '${HOME}/test-app'
27 def create_node(self, host, user):
30 node = LinuxNode(ec, 1)
36 def t_xterm(self, node):
37 if not node.is_alive():
38 print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
41 node.enable_x11 = True
45 out = node.execute('xterm')
47 node.uninstall('xterm')
49 self.assertEquals(out, "")
51 def t_execute(self, node, target):
52 if not node.is_alive():
53 print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
56 command = "ping -qc3 %s" % target
57 out = node.execute(command)
59 expected = """3 packets transmitted, 3 received, 0% packet loss"""
61 self.assertTrue(out.find(expected) > 0)
63 def t_run(self, node, target):
64 if not node.is_alive():
65 print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
68 node.mkdir(self.home, clean = True)
70 command = "ping %s" % target
71 dst = os.path.join(self.home, "app.sh")
72 node.upload(command, dst)
75 node.run(cmd, self.home)
76 pid, ppid = node.checkpid(self.home)
78 status = node.status(pid, ppid)
79 self.assertTrue(status, RUNNING)
82 status = node.status(pid, ppid)
83 self.assertTrue(status, FINISHED)
87 def t_install(self, node):
88 if not node.is_alive():
89 print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
92 node.mkdir(self.home, clean = True)
94 prog = """#include <stdio.h>
99 printf ("Hello, world!\\n");
103 # upload the test program
104 dst = os.path.join(self.home, "hello.c")
105 node.upload(prog, dst)
110 # compile the program using gcc
111 command = "cd %s; gcc -Wall hello.c -o hello" % self.home
112 out = node.execute(command)
114 # execute the program and get the output from stout
115 command = "%s/hello" % self.home
116 out = node.execute(command)
118 # execute the program and get the output from a file
119 command = "%s/hello > %s/hello.out" % (self.home, self.home)
120 node.execute(command)
122 # retrieve the output file
123 src = os.path.join(self.home, "hello.out")
124 f = tempfile.NamedTemporaryFile(delete=False)
126 node.download(src, dst)
129 node.uninstall('gcc')
130 node.rmdir(self.home)
132 self.assertEquals(out, "Hello, world!\n")
138 self.assertEquals(out, "Hello, world!\n")
140 def test_execute_fedora(self):
141 self.t_execute(self.node_fedora, self.target)
143 def test_execute_ubuntu(self):
144 self.t_execute(self.node_ubuntu, self.target)
146 def test_run_fedora(self):
147 self.t_run(self.node_fedora, self.target)
149 def test_run_ubuntu(self):
150 self.t_run(self.node_ubuntu, self.target)
152 def test_intall_fedora(self):
153 self.t_install(self.node_fedora)
155 def test_install_ubuntu(self):
156 self.t_install(self.node_ubuntu)
158 def xtest_xterm_fedora(self):
159 """ PlanetLab doesn't currently support X11 forwarding.
160 Interactive test. Should not run automatically """
161 self.t_xterm(self.node_fedora)
163 def xtest_xterm_ubuntu(self):
164 """ Interactive test. Should not run automatically """
165 self.t_xterm(self.node_ubuntu)
168 if __name__ == '__main__':