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 unittest
   9
  10
  11 class DummyEC(object):
  12     pass
  13
  14 class LinuxBoxTestCase(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         box = Box()
  29         ec = DummyEC()
  30
  31         node = LinuxNode(box, ec)
  32         node.host = host
  33         node.user = user
  34
  35         return node
  36
  37     def t_execute(self, node, target):
  38         if not node.is_alive():
  39             print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
  40             return
  41
  42         command = "ping -qc3 %s" % target
  43         out = node.execute(command)
  44
  45         expected = """3 packets transmitted, 3 received, 0% packet loss"""
  46
  47         self.assertTrue(out.find(expected) > 0)
  48
  49     def t_run(self, node, target):
  50         if not node.is_alive():
  51             print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
  52             return
  53
  54         node.mkdir(self.home, clean = True)
  55
  56         command = "ping %s" % target
  57         dst = os.path.join(self.home, "app.sh")
  58         node.upload(command, dst)
  59
  60         cmd = "bash ./app.sh"
  61         node.run(cmd, self.home)
  62         pid, ppid = node.checkpid(self.home)
  63
  64         status = node.status(pid, ppid)
  65         self.assertTrue(status, RUNNING)
  66
  67         node.kill(pid, ppid)
  68         status = node.status(pid, ppid)
  69         self.assertTrue(status, FINISHED)
  70
  71         node.rmdir(self.home)
  72
  73     def t_install(self, node, target):
  74         if not node.is_alive():
  75             print "*** WARNING: Skipping test: Node %s is not alive\n" % (node.host)
  76             return
  77
  78         node.mkdir(self.home, clean = True)
  79
  80         prog = """#include <stdio.h>
  81
  82 int
  83 main (void)
  84 {
  85     printf ("Hello, world!\\n");
  86     return 0;
  87 }
  88 """
  89         dst = os.path.join(self.home, "hello.c")
  90         node.upload(prog, dst)
  91
  92         node.install('gcc')
  93
  94         command = "cd %s; gcc -Wall hello.c -o hello" % self.home
  95         out = node.execute(command)
  96
  97         command = "%s/hello" % self.home
  98         out = node.execute(command)
  99
 100         self.assertEquals(out, "Hello, world!\n")
 101
 102         node.uninstall('gcc')
 103         node.rmdir(self.home)
 104
 105     def test_execute_fedora(self):
 106         self.t_execute(self.node_fedora, self.target)
 107
 108     def test_execute_ubuntu(self):
 109         self.t_execute(self.node_ubuntu, self.target)
 110
 111     def test_run_fedora(self):
 112         self.t_run(self.node_fedora, self.target)
 113
 114     def test_run_ubuntu(self):
 115         self.t_run(self.node_ubuntu, self.target)
 116
 117     def test_intall_fedora(self):
 118         self.t_install(self.node_fedora, self.target)
 119
 120     def test_install_ubuntu(self):
 121         self.t_install(self.node_ubuntu, self.target)
 122
 123 if __name__ == '__main__':
 124     unittest.main()
 125