test/resources/linux/node.py

   1 #!/usr/bin/env python
   2 #
   3 #    NEPI, a framework to manage network experiments
   4 #    Copyright (C) 2013 INRIA
   5 #
   6 #    This program is free software: you can redistribute it and/or modify
   7 #    it under the terms of the GNU General Public License as published by
   8 #    the Free Software Foundation, either version 3 of the License, or
   9 #    (at your option) any later version.
  10 #
  11 #    This program is distributed in the hope that it will be useful,
  12 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 #    GNU General Public License for more details.
  15 #
  16 #    You should have received a copy of the GNU General Public License
  17 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  20
  21
  22 from nepi.resources.linux.node import LinuxNode, ExitCode
  23 from nepi.util.sshfuncs import ProcStatus
  24
  25 from test_utils import skipIfNotAlive, skipInteractive, create_node
  26
  27 import os
  28 import time
  29 import tempfile
  30 import unittest
  31
  32 class LinuxNodeTestCase(unittest.TestCase):
  33     def setUp(self):
  34         self.fedora_host = "nepi2.pl.sophia.inria.fr"
  35         self.fedora_user = "inria_nepi"
  36
  37         self.ubuntu_host = "roseval.pl.sophia.inria.fr"
  38         self.ubuntu_user = "alina"
  39
  40         self.target = "nepi5.pl.sophia.inria.fr"
  41
  42     @skipIfNotAlive
  43     def t_execute(self, host, user):
  44         node, ec = create_node(host, user)
  45
  46         command = "ping -qc3 %s" % self.target
  47
  48         (out, err), proc = node.execute(command)
  49
  50         expected = """3 packets transmitted, 3 received, 0% packet loss"""
  51
  52         self.assertTrue(out.find(expected) > 0)
  53
  54     @skipIfNotAlive
  55     def t_run(self, host, user):
  56         node, ec = create_node(host, user)
  57
  58         app_home = os.path.join(node.exp_home, "my-app")
  59         node.mkdir(app_home, clean = True)
  60
  61         command = "ping %s" % self.target
  62         node.run(command, app_home)
  63         pid, ppid = node.getpid(app_home)
  64
  65         status = node.status(pid, ppid)
  66         self.assertTrue(status, ProcStatus.RUNNING)
  67
  68         node.kill(pid, ppid)
  69         status = node.status(pid, ppid)
  70         self.assertTrue(status, ProcStatus.FINISHED)
  71
  72         (out, err), proc = node.check_output(app_home, "stdout")
  73
  74         expected = """64 bytes from"""
  75
  76         self.assertTrue(out.find(expected) > 0)
  77
  78         node.rmdir(app_home)
  79
  80     @skipIfNotAlive
  81     def t_exitcode_ok(self, host, user):
  82         command = "echo 'OK!'"
  83
  84         node, ec = create_node(host, user)
  85
  86         app_home = os.path.join(node.exp_home, "my-app")
  87         node.mkdir(app_home, clean = True)
  88
  89         (out, err), proc = node.run_and_wait(command, app_home,
  90             shfile = "cmd.sh",
  91             pidfile = "pid",
  92             ecodefile = "exitcode",
  93             stdout = "stdout",
  94             stderr = "stderr",
  95             raise_on_error = True)
  96
  97         # get the pid of the process
  98         ecode = node.exitcode(app_home)
  99         self.assertEquals(ecode, ExitCode.OK)
 100
 101     @skipIfNotAlive
 102     def t_exitcode_kill(self, host, user):
 103         node, ec = create_node(host, user)
 104
 105         app_home = os.path.join(node.exp_home, "my-app")
 106         node.mkdir(app_home, clean = True)
 107
 108         # Upload command that will not finish
 109         command = "ping localhost"
 110         (out, err), proc = node.upload_command(command, app_home,
 111             shfile = "cmd.sh",
 112             ecodefile = "exitcode")
 113
 114         (out, err), proc = node.run(command, app_home,
 115             pidfile = "pidfile",
 116             stdout = "stdout",
 117             stderr = "stderr")
 118
 119         # Just wait to make sure the ping started
 120         time.sleep(5)
 121
 122         # The process is still running, so no retfile has been created yet
 123         ecode = node.exitcode(app_home)
 124         self.assertEquals(ecode, ExitCode.FILENOTFOUND)
 125
 126         (out, err), proc = node.check_errors(app_home)
 127         self.assertEquals(err, "")
 128
 129         # Now kill the app
 130         pid, ppid = node.getpid(app_home)
 131         node.kill(pid, ppid)
 132
 133         (out, err), proc = node.check_errors(app_home)
 134         self.assertEquals(err, "")
 135
 136     @skipIfNotAlive
 137     def t_exitcode_error(self, host, user):
 138         # Try to execute a command that doesn't exist
 139         command = "unexistent-command"
 140
 141         node, ec = create_node(host, user)
 142
 143         app_home = os.path.join(node.exp_home, "my-app")
 144         node.mkdir(app_home, clean = True)
 145
 146         (out, err), proc = node.run_and_wait(command, app_home,
 147             shfile = "cmd.sh",
 148             pidfile = "pid",
 149             ecodefile = "exitcode",
 150             stdout = "stdout",
 151             stderr = "stderr",
 152             raise_on_error = False)
 153
 154         # get the pid of the process
 155         ecode = node.exitcode(app_home)
 156
 157         # bash erro 127 - command not found
 158         self.assertEquals(ecode, 127)
 159
 160         (out, err), proc = node.check_errors(app_home)
 161
 162         self.assertEquals(err.strip(), "./cmd.sh: line 1: unexistent-command: command not found")
 163
 164     @skipIfNotAlive
 165     def t_install(self, host, user):
 166         node, ec = create_node(host, user)
 167
 168         (out, err), proc = node.mkdir(node.node_home, clean = True)
 169         self.assertEquals(out, "")
 170
 171         (out, err), proc = node.install_packages("gcc", node.node_home)
 172         self.assertEquals(out, "")
 173
 174         (out, err), proc = node.remove_packages("gcc", node.node_home)
 175         self.assertEquals(out, "")
 176
 177         (out, err), proc = node.rmdir(node.exp_home)
 178         self.assertEquals(out, "")
 179
 180     @skipIfNotAlive
 181     def t_xterm(self, host, user):
 182         node, ec = create_node(host, user)
 183
 184         (out, err), proc = node.mkdir(node.node_home, clean = True)
 185         self.assertEquals(out, "")
 186
 187         node.install_packages("xterm", node.node_home)
 188         self.assertEquals(out, "")
 189
 190         (out, err), proc = node.execute("xterm", forward_x11 = True)
 191         self.assertEquals(out, "")
 192
 193         (out, err), proc = node.remove_packages("xterm", node.node_home)
 194         self.assertEquals(out, "")
 195
 196     @skipIfNotAlive
 197     def t_compile(self, host, user):
 198         node, ec = create_node(host, user)
 199
 200         app_home = os.path.join(node.exp_home, "my-app")
 201         node.mkdir(app_home, clean = True)
 202
 203         prog = """#include <stdio.h>
 204
 205 int
 206 main (void)
 207 {
 208     printf ("Hello, world!\\n");
 209     return 0;
 210 }
 211 """
 212         # upload the test program
 213         dst = os.path.join(app_home, "hello.c")
 214         node.upload(prog, dst, text = True)
 215
 216         # install gcc
 217         node.install_packages('gcc', app_home)
 218
 219         # compile the program using gcc
 220         command = "cd %s; gcc -Wall hello.c -o hello" % app_home
 221         (out, err), proc = node.execute(command)
 222
 223         # execute the program and get the output from stdout
 224         command = "%s/hello" % app_home
 225         (out, err), proc = node.execute(command)
 226
 227         self.assertEquals(out, "Hello, world!\n")
 228
 229         # execute the program and get the output from a file
 230         command = "%(home)s/hello > %(home)s/hello.out" % {
 231                 'home': app_home}
 232         (out, err), proc = node.execute(command)
 233
 234         # retrieve the output file
 235         src = os.path.join(app_home, "hello.out")
 236         f = tempfile.NamedTemporaryFile(delete = False)
 237         dst = f.name
 238         node.download(src, dst)
 239         f.close()
 240
 241         node.remove_packages("gcc", app_home)
 242         node.rmdir(app_home)
 243
 244         f = open(dst, "r")
 245         out = f.read()
 246         f.close()
 247
 248         self.assertEquals(out, "Hello, world!\n")
 249
 250     def test_execute_fedora(self):
 251         self.t_execute(self.fedora_host, self.fedora_user)
 252
 253     def test_execute_ubuntu(self):
 254         self.t_execute(self.ubuntu_host, self.ubuntu_user)
 255
 256     def test_run_fedora(self):
 257         self.t_run(self.fedora_host, self.fedora_user)
 258
 259     def test_run_ubuntu(self):
 260         self.t_run(self.ubuntu_host, self.ubuntu_user)
 261
 262     def test_intall_fedora(self):
 263         self.t_install(self.fedora_host, self.fedora_user)
 264
 265     def test_install_ubuntu(self):
 266         self.t_install(self.ubuntu_host, self.ubuntu_user)
 267
 268     def test_compile_fedora(self):
 269         self.t_compile(self.fedora_host, self.fedora_user)
 270
 271     def test_compile_ubuntu(self):
 272         self.t_compile(self.ubuntu_host, self.ubuntu_user)
 273
 274     def test_exitcode_ok_fedora(self):
 275         self.t_exitcode_ok(self.fedora_host, self.fedora_user)
 276
 277     def test_exitcode_ok_ubuntu(self):
 278         self.t_exitcode_ok(self.ubuntu_host, self.ubuntu_user)
 279
 280     def test_exitcode_kill_fedora(self):
 281         self.t_exitcode_kill(self.fedora_host, self.fedora_user)
 282
 283     def test_exitcode_kill_ubuntu(self):
 284         self.t_exitcode_kill(self.ubuntu_host, self.ubuntu_user)
 285
 286     def test_exitcode_error_fedora(self):
 287         self.t_exitcode_error(self.fedora_host, self.fedora_user)
 288
 289     def test_exitcode_error_ubuntu(self):
 290         self.t_exitcode_error(self.ubuntu_host, self.ubuntu_user)
 291
 292     @skipInteractive
 293     def test_xterm_ubuntu(self):
 294         """ Interactive test. Should not run automatically """
 295         self.t_xterm(self.ubuntu_host, self.ubuntu_user)
 296
 297
 298 if __name__ == '__main__':
 299     unittest.main()
 300