code.grnet.gr Git - ganeti-local/blob - test/ganeti.utils.wrapper_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2010, 2011 Google Inc.
   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 2 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, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # 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, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Script for testing ganeti.utils.wrapper"""
  23
  24 import errno
  25 import fcntl
  26 import os
  27 import socket
  28 import tempfile
  29 import unittest
  30
  31 from ganeti import constants
  32 from ganeti import utils
  33
  34 import testutils
  35
  36
  37 class TestSetCloseOnExecFlag(unittest.TestCase):
  38   """Tests for SetCloseOnExecFlag"""
  39
  40   def setUp(self):
  41     self.tmpfile = tempfile.TemporaryFile()
  42
  43   def testEnable(self):
  44     utils.SetCloseOnExecFlag(self.tmpfile.fileno(), True)
  45     self.failUnless(fcntl.fcntl(self.tmpfile.fileno(), fcntl.F_GETFD) &
  46                     fcntl.FD_CLOEXEC)
  47
  48   def testDisable(self):
  49     utils.SetCloseOnExecFlag(self.tmpfile.fileno(), False)
  50     self.failIf(fcntl.fcntl(self.tmpfile.fileno(), fcntl.F_GETFD) &
  51                 fcntl.FD_CLOEXEC)
  52
  53
  54 class TestSetNonblockFlag(unittest.TestCase):
  55   def setUp(self):
  56     self.tmpfile = tempfile.TemporaryFile()
  57
  58   def testEnable(self):
  59     utils.SetNonblockFlag(self.tmpfile.fileno(), True)
  60     self.failUnless(fcntl.fcntl(self.tmpfile.fileno(), fcntl.F_GETFL) &
  61                     os.O_NONBLOCK)
  62
  63   def testDisable(self):
  64     utils.SetNonblockFlag(self.tmpfile.fileno(), False)
  65     self.failIf(fcntl.fcntl(self.tmpfile.fileno(), fcntl.F_GETFL) &
  66                 os.O_NONBLOCK)
  67
  68
  69 class TestIgnoreProcessNotFound(unittest.TestCase):
  70   @staticmethod
  71   def _WritePid(fd):
  72     os.write(fd, str(os.getpid()))
  73     os.close(fd)
  74     return True
  75
  76   def test(self):
  77     (pid_read_fd, pid_write_fd) = os.pipe()
  78
  79     # Start short-lived process which writes its PID to pipe
  80     self.assert_(utils.RunInSeparateProcess(self._WritePid, pid_write_fd))
  81     os.close(pid_write_fd)
  82
  83     # Read PID from pipe
  84     pid = int(os.read(pid_read_fd, 1024))
  85     os.close(pid_read_fd)
  86
  87     # Try to send signal to process which exited recently
  88     self.assertFalse(utils.IgnoreProcessNotFound(os.kill, pid, 0))
  89
  90
  91 class TestIgnoreSignals(unittest.TestCase):
  92   """Test the IgnoreSignals decorator"""
  93
  94   @staticmethod
  95   def _Raise(exception):
  96     raise exception
  97
  98   @staticmethod
  99   def _Return(rval):
 100     return rval
 101
 102   def testIgnoreSignals(self):
 103     sock_err_intr = socket.error(errno.EINTR, "Message")
 104     sock_err_inval = socket.error(errno.EINVAL, "Message")
 105
 106     env_err_intr = EnvironmentError(errno.EINTR, "Message")
 107     env_err_inval = EnvironmentError(errno.EINVAL, "Message")
 108
 109     self.assertRaises(socket.error, self._Raise, sock_err_intr)
 110     self.assertRaises(socket.error, self._Raise, sock_err_inval)
 111     self.assertRaises(EnvironmentError, self._Raise, env_err_intr)
 112     self.assertRaises(EnvironmentError, self._Raise, env_err_inval)
 113
 114     self.assertEquals(utils.IgnoreSignals(self._Raise, sock_err_intr), None)
 115     self.assertEquals(utils.IgnoreSignals(self._Raise, env_err_intr), None)
 116     self.assertRaises(socket.error, utils.IgnoreSignals, self._Raise,
 117                       sock_err_inval)
 118     self.assertRaises(EnvironmentError, utils.IgnoreSignals, self._Raise,
 119                       env_err_inval)
 120
 121     self.assertEquals(utils.IgnoreSignals(self._Return, True), True)
 122     self.assertEquals(utils.IgnoreSignals(self._Return, 33), 33)
 123
 124
 125 if __name__ == "__main__":
 126   testutils.GanetiTestProgram()