test/resources/linux/node.py

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