test/resources/linux/node.py

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