Lib/test/test_posix.py

   1 "Test posix functions"
   2
   3 from test import test_support
   4
   5 # Skip these tests if there is no posix module.
   6 posix = test_support.import_module('posix')
   7
   8 import errno
   9 import time
  10 import os
  11 import pwd
  12 import shutil
  13 import unittest
  14 import warnings
  15
  16
  17 warnings.filterwarnings('ignore', '.* potential security risk .*',
  18                         RuntimeWarning)
  19
  20 class PosixTester(unittest.TestCase):
  21
  22     def setUp(self):
  23         # create empty file
  24         fp = open(test_support.TESTFN, 'w+')
  25         fp.close()
  26
  27     def tearDown(self):
  28         os.unlink(test_support.TESTFN)
  29
  30     def testNoArgFunctions(self):
  31         # test posix functions which take no arguments and have
  32         # no side-effects which we need to cleanup (e.g., fork, wait, abort)
  33         NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdu", "uname",
  34                              "times", "getloadavg", "tmpnam",
  35                              "getegid", "geteuid", "getgid", "getgroups",
  36                              "getpid", "getpgrp", "getppid", "getuid",
  37                            ]
  38
  39         for name in NO_ARG_FUNCTIONS:
  40             posix_func = getattr(posix, name, None)
  41             if posix_func is not None:
  42                 posix_func()
  43                 self.assertRaises(TypeError, posix_func, 1)
  44
  45     if hasattr(posix, 'getresuid'):
  46         def test_getresuid(self):
  47             user_ids = posix.getresuid()
  48             self.assertEqual(len(user_ids), 3)
  49             for val in user_ids:
  50                 self.assertGreaterEqual(val, 0)
  51
  52     if hasattr(posix, 'getresgid'):
  53         def test_getresgid(self):
  54             group_ids = posix.getresgid()
  55             self.assertEqual(len(group_ids), 3)
  56             for val in group_ids:
  57                 self.assertGreaterEqual(val, 0)
  58
  59     if hasattr(posix, 'setresuid'):
  60         def test_setresuid(self):
  61             current_user_ids = posix.getresuid()
  62             self.assertIsNone(posix.setresuid(*current_user_ids))
  63             # -1 means don't change that value.
  64             self.assertIsNone(posix.setresuid(-1, -1, -1))
  65
  66         def test_setresuid_exception(self):
  67             # Don't do this test if someone is silly enough to run us as root.
  68             current_user_ids = posix.getresuid()
  69             if 0 not in current_user_ids:
  70                 new_user_ids = (current_user_ids[0]+1, -1, -1)
  71                 self.assertRaises(OSError, posix.setresuid, *new_user_ids)
  72
  73     if hasattr(posix, 'setresgid'):
  74         def test_setresgid(self):
  75             current_group_ids = posix.getresgid()
  76             self.assertIsNone(posix.setresgid(*current_group_ids))
  77             # -1 means don't change that value.
  78             self.assertIsNone(posix.setresgid(-1, -1, -1))
  79
  80         def test_setresgid_exception(self):
  81             # Don't do this test if someone is silly enough to run us as root.
  82             current_group_ids = posix.getresgid()
  83             if 0 not in current_group_ids:
  84                 new_group_ids = (current_group_ids[0]+1, -1, -1)
  85                 self.assertRaises(OSError, posix.setresgid, *new_group_ids)
  86
  87     @unittest.skipUnless(hasattr(posix, 'initgroups'),
  88                          "test needs os.initgroups()")
  89     def test_initgroups(self):
  90         # It takes a string and an integer; check that it raises a TypeError
  91         # for other argument lists.
  92         self.assertRaises(TypeError, posix.initgroups)
  93         self.assertRaises(TypeError, posix.initgroups, None)
  94         self.assertRaises(TypeError, posix.initgroups, 3, "foo")
  95         self.assertRaises(TypeError, posix.initgroups, "foo", 3, object())
  96
  97         # If a non-privileged user invokes it, it should fail with OSError
  98         # EPERM.
  99         if os.getuid() != 0:
 100             name = pwd.getpwuid(posix.getuid()).pw_name
 101             try:
 102                 posix.initgroups(name, 13)
 103             except OSError as e:
 104                 self.assertEquals(e.errno, errno.EPERM)
 105             else:
 106                 self.fail("Expected OSError to be raised by initgroups")
 107
 108     def test_statvfs(self):
 109         if hasattr(posix, 'statvfs'):
 110             self.assertTrue(posix.statvfs(os.curdir))
 111
 112     def test_fstatvfs(self):
 113         if hasattr(posix, 'fstatvfs'):
 114             fp = open(test_support.TESTFN)
 115             try:
 116                 self.assertTrue(posix.fstatvfs(fp.fileno()))
 117             finally:
 118                 fp.close()
 119
 120     def test_ftruncate(self):
 121         if hasattr(posix, 'ftruncate'):
 122             fp = open(test_support.TESTFN, 'w+')
 123             try:
 124                 # we need to have some data to truncate
 125                 fp.write('test')
 126                 fp.flush()
 127                 posix.ftruncate(fp.fileno(), 0)
 128             finally:
 129                 fp.close()
 130
 131     def test_dup(self):
 132         if hasattr(posix, 'dup'):
 133             fp = open(test_support.TESTFN)
 134             try:
 135                 fd = posix.dup(fp.fileno())
 136                 self.assertIsInstance(fd, int)
 137                 os.close(fd)
 138             finally:
 139                 fp.close()
 140
 141     def test_confstr(self):
 142         if hasattr(posix, 'confstr'):
 143             self.assertRaises(ValueError, posix.confstr, "CS_garbage")
 144             self.assertEqual(len(posix.confstr("CS_PATH")) > 0, True)
 145
 146     def test_dup2(self):
 147         if hasattr(posix, 'dup2'):
 148             fp1 = open(test_support.TESTFN)
 149             fp2 = open(test_support.TESTFN)
 150             try:
 151                 posix.dup2(fp1.fileno(), fp2.fileno())
 152             finally:
 153                 fp1.close()
 154                 fp2.close()
 155
 156     def fdopen_helper(self, *args):
 157         fd = os.open(test_support.TESTFN, os.O_RDONLY)
 158         fp2 = posix.fdopen(fd, *args)
 159         fp2.close()
 160
 161     def test_fdopen(self):
 162         if hasattr(posix, 'fdopen'):
 163             self.fdopen_helper()
 164             self.fdopen_helper('r')
 165             self.fdopen_helper('r', 100)
 166
 167     def test_osexlock(self):
 168         if hasattr(posix, "O_EXLOCK"):
 169             fd = os.open(test_support.TESTFN,
 170                          os.O_WRONLY|os.O_EXLOCK|os.O_CREAT)
 171             self.assertRaises(OSError, os.open, test_support.TESTFN,
 172                               os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
 173             os.close(fd)
 174
 175             if hasattr(posix, "O_SHLOCK"):
 176                 fd = os.open(test_support.TESTFN,
 177                              os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 178                 self.assertRaises(OSError, os.open, test_support.TESTFN,
 179                                   os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
 180                 os.close(fd)
 181
 182     def test_osshlock(self):
 183         if hasattr(posix, "O_SHLOCK"):
 184             fd1 = os.open(test_support.TESTFN,
 185                          os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 186             fd2 = os.open(test_support.TESTFN,
 187                           os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 188             os.close(fd2)
 189             os.close(fd1)
 190
 191             if hasattr(posix, "O_EXLOCK"):
 192                 fd = os.open(test_support.TESTFN,
 193                              os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 194                 self.assertRaises(OSError, os.open, test_support.TESTFN,
 195                                   os.O_RDONLY|os.O_EXLOCK|os.O_NONBLOCK)
 196                 os.close(fd)
 197
 198     def test_fstat(self):
 199         if hasattr(posix, 'fstat'):
 200             fp = open(test_support.TESTFN)
 201             try:
 202                 self.assertTrue(posix.fstat(fp.fileno()))
 203             finally:
 204                 fp.close()
 205
 206     def test_stat(self):
 207         if hasattr(posix, 'stat'):
 208             self.assertTrue(posix.stat(test_support.TESTFN))
 209
 210     def _test_all_chown_common(self, chown_func, first_param):
 211         """Common code for chown, fchown and lchown tests."""
 212         if os.getuid() == 0:
 213             try:
 214                 # Many linux distros have a nfsnobody user as MAX_UID-2
 215                 # that makes a good test case for signedness issues.
 216                 #   http://bugs.python.org/issue1747858
 217                 # This part of the test only runs when run as root.
 218                 # Only scary people run their tests as root.
 219                 ent = pwd.getpwnam('nfsnobody')
 220                 chown_func(first_param, ent.pw_uid, ent.pw_gid)
 221             except KeyError:
 222                 pass
 223         else:
 224             # non-root cannot chown to root, raises OSError
 225             self.assertRaises(OSError, chown_func,
 226                               first_param, 0, 0)
 227
 228         # test a successful chown call
 229         chown_func(first_param, os.getuid(), os.getgid())
 230
 231     @unittest.skipUnless(hasattr(posix, 'chown'), "test needs os.chown()")
 232     def test_chown(self):
 233         # raise an OSError if the file does not exist
 234         os.unlink(test_support.TESTFN)
 235         self.assertRaises(OSError, posix.chown, test_support.TESTFN, -1, -1)
 236
 237         # re-create the file
 238         open(test_support.TESTFN, 'w').close()
 239         self._test_all_chown_common(posix.chown, test_support.TESTFN)
 240
 241     @unittest.skipUnless(hasattr(posix, 'fchown'), "test needs os.fchown()")
 242     def test_fchown(self):
 243         os.unlink(test_support.TESTFN)
 244
 245         # re-create the file
 246         test_file = open(test_support.TESTFN, 'w')
 247         try:
 248             fd = test_file.fileno()
 249             self._test_all_chown_common(posix.fchown, fd)
 250         finally:
 251             test_file.close()
 252
 253     @unittest.skipUnless(hasattr(posix, 'lchown'), "test needs os.lchown()")
 254     def test_lchown(self):
 255         os.unlink(test_support.TESTFN)
 256         # create a symlink
 257         os.symlink('/tmp/dummy-symlink-target', test_support.TESTFN)
 258         self._test_all_chown_common(posix.lchown, test_support.TESTFN)
 259
 260     def test_chdir(self):
 261         if hasattr(posix, 'chdir'):
 262             posix.chdir(os.curdir)
 263             self.assertRaises(OSError, posix.chdir, test_support.TESTFN)
 264
 265     def test_lsdir(self):
 266         if hasattr(posix, 'lsdir'):
 267             self.assertIn(test_support.TESTFN, posix.lsdir(os.curdir))
 268
 269     def test_access(self):
 270         if hasattr(posix, 'access'):
 271             self.assertTrue(posix.access(test_support.TESTFN, os.R_OK))
 272
 273     def test_umask(self):
 274         if hasattr(posix, 'umask'):
 275             old_mask = posix.umask(0)
 276             self.assertIsInstance(old_mask, int)
 277             posix.umask(old_mask)
 278
 279     def test_strerror(self):
 280         if hasattr(posix, 'strerror'):
 281             self.assertTrue(posix.strerror(0))
 282
 283     def test_pipe(self):
 284         if hasattr(posix, 'pipe'):
 285             reader, writer = posix.pipe()
 286             os.close(reader)
 287             os.close(writer)
 288
 289     def test_tempnam(self):
 290         if hasattr(posix, 'tempnam'):
 291             self.assertTrue(posix.tempnam())
 292             self.assertTrue(posix.tempnam(os.curdir))
 293             self.assertTrue(posix.tempnam(os.curdir, 'blah'))
 294
 295     def test_tmpfile(self):
 296         if hasattr(posix, 'tmpfile'):
 297             fp = posix.tmpfile()
 298             fp.close()
 299
 300     def test_utime(self):
 301         if hasattr(posix, 'utime'):
 302             now = time.time()
 303             posix.utime(test_support.TESTFN, None)
 304             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, None))
 305             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (now, None))
 306             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, now))
 307             posix.utime(test_support.TESTFN, (int(now), int(now)))
 308             posix.utime(test_support.TESTFN, (now, now))
 309
 310     def test_chflags(self):
 311         if hasattr(posix, 'chflags'):
 312             st = os.stat(test_support.TESTFN)
 313             if hasattr(st, 'st_flags'):
 314                 posix.chflags(test_support.TESTFN, st.st_flags)
 315
 316     def test_lchflags(self):
 317         if hasattr(posix, 'lchflags'):
 318             st = os.stat(test_support.TESTFN)
 319             if hasattr(st, 'st_flags'):
 320                 posix.lchflags(test_support.TESTFN, st.st_flags)
 321
 322     def test_getcwd_long_pathnames(self):
 323         if hasattr(posix, 'getcwd'):
 324             dirname = 'getcwd-test-directory-0123456789abcdef-01234567890abcdef'
 325             curdir = os.getcwd()
 326             base_path = os.path.abspath(test_support.TESTFN) + '.getcwd'
 327
 328             try:
 329                 os.mkdir(base_path)
 330                 os.chdir(base_path)
 331             except:
 332 #               Just returning nothing instead of the SkipTest exception,
 333 #               because the test results in Error in that case.
 334 #               Is that ok?
 335 #                raise unittest.SkipTest, "cannot create directory for testing"
 336                 return
 337
 338             try:
 339                 def _create_and_do_getcwd(dirname, current_path_length = 0):
 340                     try:
 341                         os.mkdir(dirname)
 342                     except:
 343                         raise unittest.SkipTest, "mkdir cannot create directory sufficiently deep for getcwd test"
 344
 345                     os.chdir(dirname)
 346                     try:
 347                         os.getcwd()
 348                         if current_path_length < 1027:
 349                             _create_and_do_getcwd(dirname, current_path_length + len(dirname) + 1)
 350                     finally:
 351                         os.chdir('..')
 352                         os.rmdir(dirname)
 353
 354                 _create_and_do_getcwd(dirname)
 355
 356             finally:
 357                 os.chdir(curdir)
 358                 shutil.rmtree(base_path)
 359
 360
 361 def test_main():
 362     test_support.run_unittest(PosixTester)
 363
 364 if __name__ == '__main__':
 365     test_main()