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