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 from test_utils import skipIfNotAlive, skipInteractive, create_node
   6
   7 import os
   8 import time
   9 import tempfile
  10 import unittest
  11
  12 class LinuxNodeTestCase(unittest.TestCase):
  13     def setUp(self):
  14         self.fedora_host = 'nepi2.pl.sophia.inria.fr'
  15         self.fedora_user = 'inria_nepi'
  16
  17         self.ubuntu_host = 'roseval.pl.sophia.inria.fr'
  18         self.ubuntu_user = 'alina'
  19
  20         self.target = 'nepi5.pl.sophia.inria.fr'
  21
  22     @skipIfNotAlive
  23     def t_xterm(self, host, user):
  24         node, ec = create_node(host, user)
  25
  26         node.install_packages('xterm')
  27
  28         (out, err), proc = node.execute('xterm', forward_x11 = True)
  29
  30         self.assertEquals(out, "")
  31
  32         (out, err), proc = node.remove_packages('xterm')
  33
  34         self.assertEquals(out, "")
  35
  36     @skipIfNotAlive
  37     def t_execute(self, host, user):
  38         node, ec = create_node(host, user)
  39
  40         command = "ping -qc3 %s" % self.target
  41
  42         (out, err), proc = node.execute(command)
  43
  44         expected = """3 packets transmitted, 3 received, 0% packet loss"""
  45
  46         self.assertTrue(out.find(expected) > 0)
  47
  48     @skipIfNotAlive
  49     def t_run(self, host, user):
  50         node, ec = create_node(host, user)
  51
  52         app_home = os.path.join(node.exp_dir, "my-app")
  53         node.mkdir(app_home, clean = True)
  54
  55         command = "ping %s" % self.target
  56         node.run(command, app_home)
  57         pid, ppid = node.checkpid(app_home)
  58
  59         status = node.status(pid, ppid)
  60         self.assertTrue(status, RUNNING)
  61
  62         node.kill(pid, ppid)
  63         status = node.status(pid, ppid)
  64         self.assertTrue(status, FINISHED)
  65
  66         (out, err), proc = node.check_output(app_home, "stdout")
  67
  68         expected = """64 bytes from"""
  69
  70         self.assertTrue(out.find(expected) > 0)
  71
  72         node.rmdir(app_home)
  73
  74     @skipIfNotAlive
  75     def t_install(self, host, user):
  76         node, ec = create_node(host, user)
  77
  78         app_home = os.path.join(node.exp_dir, "my-app")
  79         node.mkdir(app_home, clean = True)
  80
  81         prog = """#include <stdio.h>
  82
  83 int
  84 main (void)
  85 {
  86     printf ("Hello, world!\\n");
  87     return 0;
  88 }
  89 """
  90         # upload the test program
  91         dst = os.path.join(app_home, "hello.c")
  92         node.upload(prog, dst, text = True)
  93
  94         # install gcc
  95         node.install_packages('gcc')
  96
  97         # compile the program using gcc
  98         command = "cd %s; gcc -Wall hello.c -o hello" % app_home
  99         (out, err), proc = node.execute(command)
 100
 101         # execute the program and get the output from stdout
 102         command = "%s/hello" % app_home
 103         (out, err), proc = node.execute(command)
 104
 105         self.assertEquals(out, "Hello, world!\n")
 106
 107         # execute the program and get the output from a file
 108         command = "%(home)s/hello > %(home)s/hello.out" % {
 109                 'home': app_home}
 110         (out, err), proc = node.execute(command)
 111
 112         # retrieve the output file
 113         src = os.path.join(app_home, "hello.out")
 114         f = tempfile.NamedTemporaryFile(delete=False)
 115         dst = f.name
 116         node.download(src, dst)
 117         f.close()
 118
 119         node.remove_packages('gcc')
 120         node.rmdir(app_home)
 121
 122         f = open(dst, "r")
 123         out = f.read()
 124         f.close()
 125
 126         self.assertEquals(out, "Hello, world!\n")
 127
 128     def test_execute_fedora(self):
 129         self.t_execute(self.fedora_host, self.fedora_user)
 130
 131     def test_execute_ubuntu(self):
 132         self.t_execute(self.ubuntu_host, self.ubuntu_user)
 133
 134     def test_run_fedora(self):
 135         self.t_run(self.fedora_host, self.fedora_user)
 136
 137     def test_run_ubuntu(self):
 138         self.t_run(self.ubuntu_host, self.ubuntu_user)
 139
 140     def test_intall_fedora(self):
 141         self.t_install(self.fedora_host, self.fedora_user)
 142
 143     def test_install_ubuntu(self):
 144         self.t_install(self.ubuntu_host, self.ubuntu_user)
 145
 146     @skipInteractive
 147     def test_xterm_ubuntu(self):
 148         """ Interactive test. Should not run automatically """
 149         self.t_xterm(self.ubuntu_host, self.ubuntu_user)
 150
 151
 152 if __name__ == '__main__':
 153     unittest.main()
 154