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_home, "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         (out, err), proc = node.install_packages('gcc')
  79         self.assertEquals(out, "")
  80
  81         (out, err), proc = node.remove_packages('gcc')
  82
  83         self.assertEquals(out, "")
  84
  85
  86     @skipIfNotAlive
  87     def t_compile(self, host, user):
  88         node, ec = create_node(host, user)
  89
  90         app_home = os.path.join(node.exp_home, "my-app")
  91         node.mkdir(app_home, clean = True)
  92
  93         prog = """#include <stdio.h>
  94
  95 int
  96 main (void)
  97 {
  98     printf ("Hello, world!\\n");
  99     return 0;
 100 }
 101 """
 102         # upload the test program
 103         dst = os.path.join(app_home, "hello.c")
 104         node.upload(prog, dst, text = True)
 105
 106         # install gcc
 107         node.install_packages('gcc')
 108
 109         # compile the program using gcc
 110         command = "cd %s; gcc -Wall hello.c -o hello" % app_home
 111         (out, err), proc = node.execute(command)
 112
 113         # execute the program and get the output from stdout
 114         command = "%s/hello" % app_home
 115         (out, err), proc = node.execute(command)
 116
 117         self.assertEquals(out, "Hello, world!\n")
 118
 119         # execute the program and get the output from a file
 120         command = "%(home)s/hello > %(home)s/hello.out" % {
 121                 'home': app_home}
 122         (out, err), proc = node.execute(command)
 123
 124         # retrieve the output file
 125         src = os.path.join(app_home, "hello.out")
 126         f = tempfile.NamedTemporaryFile(delete=False)
 127         dst = f.name
 128         node.download(src, dst)
 129         f.close()
 130
 131         node.remove_packages('gcc')
 132         node.rmdir(app_home)
 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.fedora_host, self.fedora_user)
 142
 143     def test_execute_ubuntu(self):
 144         self.t_execute(self.ubuntu_host, self.ubuntu_user)
 145
 146     def test_run_fedora(self):
 147         self.t_run(self.fedora_host, self.fedora_user)
 148
 149     def test_run_ubuntu(self):
 150         self.t_run(self.ubuntu_host, self.ubuntu_user)
 151
 152     def test_intall_fedora(self):
 153         self.t_install(self.fedora_host, self.fedora_user)
 154
 155     def test_install_ubuntu(self):
 156         self.t_install(self.ubuntu_host, self.ubuntu_user)
 157
 158     def test_compile_fedora(self):
 159         self.t_compile(self.fedora_host, self.fedora_user)
 160
 161     def test_compile_ubuntu(self):
 162         self.t_compile(self.ubuntu_host, self.ubuntu_user)
 163
 164     @skipInteractive
 165     def test_xterm_ubuntu(self):
 166         """ Interactive test. Should not run automatically """
 167         self.t_xterm(self.ubuntu_host, self.ubuntu_user)
 168
 169
 170 if __name__ == '__main__':
 171     unittest.main()
 172