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 sys
  10 import time
  11 import os
  12 import pwd
  13 import shutil
  14 import stat
  15 import sys
  16 import tempfile
  17 import unittest
  18 import warnings
  19
  20 _DUMMY_SYMLINK = os.path.join(tempfile.gettempdir(),
  21                               test_support.TESTFN + '-dummy-symlink')
  22
  23 warnings.filterwarnings('ignore', '.* potential security risk .*',
  24                         RuntimeWarning)
  25
  26 class PosixTester(unittest.TestCase):
  27
  28     def setUp(self):
  29         # create empty file
  30         fp = open(test_support.TESTFN, 'w+')
  31         fp.close()
  32         self.teardown_files = [ test_support.TESTFN ]
  33
  34     def tearDown(self):
  35         for teardown_file in self.teardown_files:
  36             os.unlink(teardown_file)
  37
  38     def testNoArgFunctions(self):
  39         # test posix functions which take no arguments and have
  40         # no side-effects which we need to cleanup (e.g., fork, wait, abort)
  41         NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdu", "uname",
  42                              "times", "getloadavg", "tmpnam",
  43                              "getegid", "geteuid", "getgid", "getgroups",
  44                              "getpid", "getpgrp", "getppid", "getuid",
  45                            ]
  46
  47         with warnings.catch_warnings():
  48             warnings.filterwarnings("ignore", "", DeprecationWarning)
  49             for name in NO_ARG_FUNCTIONS:
  50                 posix_func = getattr(posix, name, None)
  51                 if posix_func is not None:
  52                     posix_func()
  53                     self.assertRaises(TypeError, posix_func, 1)
  54
  55     if hasattr(posix, 'getresuid'):
  56         def test_getresuid(self):
  57             user_ids = posix.getresuid()
  58             self.assertEqual(len(user_ids), 3)
  59             for val in user_ids:
  60                 self.assertGreaterEqual(val, 0)
  61
  62     if hasattr(posix, 'getresgid'):
  63         def test_getresgid(self):
  64             group_ids = posix.getresgid()
  65             self.assertEqual(len(group_ids), 3)
  66             for val in group_ids:
  67                 self.assertGreaterEqual(val, 0)
  68
  69     if hasattr(posix, 'setresuid'):
  70         def test_setresuid(self):
  71             current_user_ids = posix.getresuid()
  72             self.assertIsNone(posix.setresuid(*current_user_ids))
  73             # -1 means don't change that value.
  74             self.assertIsNone(posix.setresuid(-1, -1, -1))
  75
  76         def test_setresuid_exception(self):
  77             # Don't do this test if someone is silly enough to run us as root.
  78             current_user_ids = posix.getresuid()
  79             if 0 not in current_user_ids:
  80                 new_user_ids = (current_user_ids[0]+1, -1, -1)
  81                 self.assertRaises(OSError, posix.setresuid, *new_user_ids)
  82
  83     if hasattr(posix, 'setresgid'):
  84         def test_setresgid(self):
  85             current_group_ids = posix.getresgid()
  86             self.assertIsNone(posix.setresgid(*current_group_ids))
  87             # -1 means don't change that value.
  88             self.assertIsNone(posix.setresgid(-1, -1, -1))
  89
  90         def test_setresgid_exception(self):
  91             # Don't do this test if someone is silly enough to run us as root.
  92             current_group_ids = posix.getresgid()
  93             if 0 not in current_group_ids:
  94                 new_group_ids = (current_group_ids[0]+1, -1, -1)
  95                 self.assertRaises(OSError, posix.setresgid, *new_group_ids)
  96
  97     @unittest.skipUnless(hasattr(posix, 'initgroups'),
  98                          "test needs os.initgroups()")
  99     def test_initgroups(self):
 100         # It takes a string and an integer; check that it raises a TypeError
 101         # for other argument lists.
 102         self.assertRaises(TypeError, posix.initgroups)
 103         self.assertRaises(TypeError, posix.initgroups, None)
 104         self.assertRaises(TypeError, posix.initgroups, 3, "foo")
 105         self.assertRaises(TypeError, posix.initgroups, "foo", 3, object())
 106
 107         # If a non-privileged user invokes it, it should fail with OSError
 108         # EPERM.
 109         if os.getuid() != 0:
 110             name = pwd.getpwuid(posix.getuid()).pw_name
 111             try:
 112                 posix.initgroups(name, 13)
 113             except OSError as e:
 114                 self.assertEqual(e.errno, errno.EPERM)
 115             else:
 116                 self.fail("Expected OSError to be raised by initgroups")
 117
 118     def test_statvfs(self):
 119         if hasattr(posix, 'statvfs'):
 120             self.assertTrue(posix.statvfs(os.curdir))
 121
 122     def test_fstatvfs(self):
 123         if hasattr(posix, 'fstatvfs'):
 124             fp = open(test_support.TESTFN)
 125             try:
 126                 self.assertTrue(posix.fstatvfs(fp.fileno()))
 127             finally:
 128                 fp.close()
 129
 130     def test_ftruncate(self):
 131         if hasattr(posix, 'ftruncate'):
 132             fp = open(test_support.TESTFN, 'w+')
 133             try:
 134                 # we need to have some data to truncate
 135                 fp.write('test')
 136                 fp.flush()
 137                 posix.ftruncate(fp.fileno(), 0)
 138             finally:
 139                 fp.close()
 140
 141     def test_dup(self):
 142         if hasattr(posix, 'dup'):
 143             fp = open(test_support.TESTFN)
 144             try:
 145                 fd = posix.dup(fp.fileno())
 146                 self.assertIsInstance(fd, int)
 147                 os.close(fd)
 148             finally:
 149                 fp.close()
 150
 151     def test_confstr(self):
 152         if hasattr(posix, 'confstr'):
 153             self.assertRaises(ValueError, posix.confstr, "CS_garbage")
 154             self.assertEqual(len(posix.confstr("CS_PATH")) > 0, True)
 155
 156     def test_dup2(self):
 157         if hasattr(posix, 'dup2'):
 158             fp1 = open(test_support.TESTFN)
 159             fp2 = open(test_support.TESTFN)
 160             try:
 161                 posix.dup2(fp1.fileno(), fp2.fileno())
 162             finally:
 163                 fp1.close()
 164                 fp2.close()
 165
 166     def fdopen_helper(self, *args):
 167         fd = os.open(test_support.TESTFN, os.O_RDONLY)
 168         fp2 = posix.fdopen(fd, *args)
 169         fp2.close()
 170
 171     def test_fdopen(self):
 172         if hasattr(posix, 'fdopen'):
 173             self.fdopen_helper()
 174             self.fdopen_helper('r')
 175             self.fdopen_helper('r', 100)
 176
 177     def test_osexlock(self):
 178         if hasattr(posix, "O_EXLOCK"):
 179             fd = os.open(test_support.TESTFN,
 180                          os.O_WRONLY|os.O_EXLOCK|os.O_CREAT)
 181             self.assertRaises(OSError, os.open, test_support.TESTFN,
 182                               os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
 183             os.close(fd)
 184
 185             if hasattr(posix, "O_SHLOCK"):
 186                 fd = os.open(test_support.TESTFN,
 187                              os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 188                 self.assertRaises(OSError, os.open, test_support.TESTFN,
 189                                   os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
 190                 os.close(fd)
 191
 192     def test_osshlock(self):
 193         if hasattr(posix, "O_SHLOCK"):
 194             fd1 = os.open(test_support.TESTFN,
 195                          os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 196             fd2 = os.open(test_support.TESTFN,
 197                           os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 198             os.close(fd2)
 199             os.close(fd1)
 200
 201             if hasattr(posix, "O_EXLOCK"):
 202                 fd = os.open(test_support.TESTFN,
 203                              os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
 204                 self.assertRaises(OSError, os.open, test_support.TESTFN,
 205                                   os.O_RDONLY|os.O_EXLOCK|os.O_NONBLOCK)
 206                 os.close(fd)
 207
 208     def test_fstat(self):
 209         if hasattr(posix, 'fstat'):
 210             fp = open(test_support.TESTFN)
 211             try:
 212                 self.assertTrue(posix.fstat(fp.fileno()))
 213             finally:
 214                 fp.close()
 215
 216     def test_stat(self):
 217         if hasattr(posix, 'stat'):
 218             self.assertTrue(posix.stat(test_support.TESTFN))
 219
 220     def _test_all_chown_common(self, chown_func, first_param):
 221         """Common code for chown, fchown and lchown tests."""
 222         if os.getuid() == 0:
 223             try:
 224                 # Many linux distros have a nfsnobody user as MAX_UID-2
 225                 # that makes a good test case for signedness issues.
 226                 #   http://bugs.python.org/issue1747858
 227                 # This part of the test only runs when run as root.
 228                 # Only scary people run their tests as root.
 229                 ent = pwd.getpwnam('nfsnobody')
 230                 chown_func(first_param, ent.pw_uid, ent.pw_gid)
 231             except KeyError:
 232                 pass
 233         else:
 234             # non-root cannot chown to root, raises OSError
 235             self.assertRaises(OSError, chown_func,
 236                               first_param, 0, 0)
 237
 238         # test a successful chown call
 239         chown_func(first_param, os.getuid(), os.getgid())
 240
 241     @unittest.skipUnless(hasattr(posix, 'chown'), "test needs os.chown()")
 242     def test_chown(self):
 243         # raise an OSError if the file does not exist
 244         os.unlink(test_support.TESTFN)
 245         self.assertRaises(OSError, posix.chown, test_support.TESTFN, -1, -1)
 246
 247         # re-create the file
 248         open(test_support.TESTFN, 'w').close()
 249         self._test_all_chown_common(posix.chown, test_support.TESTFN)
 250
 251     @unittest.skipUnless(hasattr(posix, 'fchown'), "test needs os.fchown()")
 252     def test_fchown(self):
 253         os.unlink(test_support.TESTFN)
 254
 255         # re-create the file
 256         test_file = open(test_support.TESTFN, 'w')
 257         try:
 258             fd = test_file.fileno()
 259             self._test_all_chown_common(posix.fchown, fd)
 260         finally:
 261             test_file.close()
 262
 263     @unittest.skipUnless(hasattr(posix, 'lchown'), "test needs os.lchown()")
 264     def test_lchown(self):
 265         os.unlink(test_support.TESTFN)
 266         # create a symlink
 267         os.symlink(_DUMMY_SYMLINK, test_support.TESTFN)
 268         self._test_all_chown_common(posix.lchown, test_support.TESTFN)
 269
 270     def test_chdir(self):
 271         if hasattr(posix, 'chdir'):
 272             posix.chdir(os.curdir)
 273             self.assertRaises(OSError, posix.chdir, test_support.TESTFN)
 274
 275     def test_lsdir(self):
 276         if hasattr(posix, 'lsdir'):
 277             self.assertIn(test_support.TESTFN, posix.lsdir(os.curdir))
 278
 279     def test_access(self):
 280         if hasattr(posix, 'access'):
 281             self.assertTrue(posix.access(test_support.TESTFN, os.R_OK))
 282
 283     def test_umask(self):
 284         if hasattr(posix, 'umask'):
 285             old_mask = posix.umask(0)
 286             self.assertIsInstance(old_mask, int)
 287             posix.umask(old_mask)
 288
 289     def test_strerror(self):
 290         if hasattr(posix, 'strerror'):
 291             self.assertTrue(posix.strerror(0))
 292
 293     def test_pipe(self):
 294         if hasattr(posix, 'pipe'):
 295             reader, writer = posix.pipe()
 296             os.close(reader)
 297             os.close(writer)
 298
 299     def test_tempnam(self):
 300         if hasattr(posix, 'tempnam'):
 301             with warnings.catch_warnings():
 302                 warnings.filterwarnings("ignore", "tempnam", DeprecationWarning)
 303                 self.assertTrue(posix.tempnam())
 304                 self.assertTrue(posix.tempnam(os.curdir))
 305                 self.assertTrue(posix.tempnam(os.curdir, 'blah'))
 306
 307     def test_tmpfile(self):
 308         if hasattr(posix, 'tmpfile'):
 309             with warnings.catch_warnings():
 310                 warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
 311                 fp = posix.tmpfile()
 312                 fp.close()
 313
 314     def test_utime(self):
 315         if hasattr(posix, 'utime'):
 316             now = time.time()
 317             posix.utime(test_support.TESTFN, None)
 318             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, None))
 319             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (now, None))
 320             self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, now))
 321             posix.utime(test_support.TESTFN, (int(now), int(now)))
 322             posix.utime(test_support.TESTFN, (now, now))
 323
 324     def _test_chflags_regular_file(self, chflags_func, target_file):
 325         st = os.stat(target_file)
 326         self.assertTrue(hasattr(st, 'st_flags'))
 327         chflags_func(target_file, st.st_flags | stat.UF_IMMUTABLE)
 328         try:
 329             new_st = os.stat(target_file)
 330             self.assertEqual(st.st_flags | stat.UF_IMMUTABLE, new_st.st_flags)
 331             try:
 332                 fd = open(target_file, 'w+')
 333             except IOError as e:
 334                 self.assertEqual(e.errno, errno.EPERM)
 335         finally:
 336             posix.chflags(target_file, st.st_flags)
 337
 338     @unittest.skipUnless(hasattr(posix, 'chflags'), 'test needs os.chflags()')
 339     def test_chflags(self):
 340         self._test_chflags_regular_file(posix.chflags, test_support.TESTFN)
 341
 342     @unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
 343     def test_lchflags_regular_file(self):
 344         self._test_chflags_regular_file(posix.lchflags, test_support.TESTFN)
 345
 346     @unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
 347     def test_lchflags_symlink(self):
 348         testfn_st = os.stat(test_support.TESTFN)
 349
 350         self.assertTrue(hasattr(testfn_st, 'st_flags'))
 351
 352         os.symlink(test_support.TESTFN, _DUMMY_SYMLINK)
 353         self.teardown_files.append(_DUMMY_SYMLINK)
 354         dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
 355
 356         posix.lchflags(_DUMMY_SYMLINK,
 357                        dummy_symlink_st.st_flags | stat.UF_IMMUTABLE)
 358         try:
 359             new_testfn_st = os.stat(test_support.TESTFN)
 360             new_dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
 361
 362             self.assertEqual(testfn_st.st_flags, new_testfn_st.st_flags)
 363             self.assertEqual(dummy_symlink_st.st_flags | stat.UF_IMMUTABLE,
 364                              new_dummy_symlink_st.st_flags)
 365         finally:
 366             posix.lchflags(_DUMMY_SYMLINK, dummy_symlink_st.st_flags)
 367
 368     def test_getcwd_long_pathnames(self):
 369         if hasattr(posix, 'getcwd'):
 370             dirname = 'getcwd-test-directory-0123456789abcdef-01234567890abcdef'
 371             curdir = os.getcwd()
 372             base_path = os.path.abspath(test_support.TESTFN) + '.getcwd'
 373
 374             try:
 375                 os.mkdir(base_path)
 376                 os.chdir(base_path)
 377             except:
 378 #               Just returning nothing instead of the SkipTest exception,
 379 #               because the test results in Error in that case.
 380 #               Is that ok?
 381 #                raise unittest.SkipTest, "cannot create directory for testing"
 382                 return
 383
 384             try:
 385                 def _create_and_do_getcwd(dirname, current_path_length = 0):
 386                     try:
 387                         os.mkdir(dirname)
 388                     except:
 389                         raise unittest.SkipTest, "mkdir cannot create directory sufficiently deep for getcwd test"
 390
 391                     os.chdir(dirname)
 392                     try:
 393                         os.getcwd()
 394                         if current_path_length < 4099:
 395                             _create_and_do_getcwd(dirname, current_path_length + len(dirname) + 1)
 396                     except OSError as e:
 397                         expected_errno = errno.ENAMETOOLONG
 398                         if 'sunos' in sys.platform or 'openbsd' in sys.platform:
 399                             expected_errno = errno.ERANGE # Issue 9185
 400                         self.assertEqual(e.errno, expected_errno)
 401                     finally:
 402                         os.chdir('..')
 403                         os.rmdir(dirname)
 404
 405                 _create_and_do_getcwd(dirname)
 406
 407             finally:
 408                 os.chdir(curdir)
 409                 shutil.rmtree(base_path)
 410
 411     @unittest.skipUnless(hasattr(os, 'getegid'), "test needs os.getegid()")
 412     def test_getgroups(self):
 413         with os.popen('id -G') as idg:
 414             groups = idg.read().strip()
 415
 416         if not groups:
 417             raise unittest.SkipTest("need working 'id -G'")
 418
 419         # 'id -G' and 'os.getgroups()' should return the same
 420         # groups, ignoring order and duplicates.
 421         # #10822 - it is implementation defined whether posix.getgroups()
 422         # includes the effective gid so we include it anyway, since id -G does
 423         self.assertEqual(
 424                 set([int(x) for x in groups.split()]),
 425                 set(posix.getgroups() + [posix.getegid()]))
 426
 427 class PosixGroupsTester(unittest.TestCase):
 428
 429     def setUp(self):
 430         if posix.getuid() != 0:
 431             raise unittest.SkipTest("not enough privileges")
 432         if not hasattr(posix, 'getgroups'):
 433             raise unittest.SkipTest("need posix.getgroups")
 434         if sys.platform == 'darwin':
 435             raise unittest.SkipTest("getgroups(2) is broken on OSX")
 436         self.saved_groups = posix.getgroups()
 437
 438     def tearDown(self):
 439         if hasattr(posix, 'setgroups'):
 440             posix.setgroups(self.saved_groups)
 441         elif hasattr(posix, 'initgroups'):
 442             name = pwd.getpwuid(posix.getuid()).pw_name
 443             posix.initgroups(name, self.saved_groups[0])
 444
 445     @unittest.skipUnless(hasattr(posix, 'initgroups'),
 446                          "test needs posix.initgroups()")
 447     def test_initgroups(self):
 448         # find missing group
 449
 450         g = max(self.saved_groups) + 1
 451         name = pwd.getpwuid(posix.getuid()).pw_name
 452         posix.initgroups(name, g)
 453         self.assertIn(g, posix.getgroups())
 454
 455     @unittest.skipUnless(hasattr(posix, 'setgroups'),
 456                          "test needs posix.setgroups()")
 457     def test_setgroups(self):
 458         for groups in [[0], range(16)]:
 459             posix.setgroups(groups)
 460             self.assertListEqual(groups, posix.getgroups())
 461
 462
 463 def test_main():
 464     test_support.run_unittest(PosixTester, PosixGroupsTester)
 465
 466 if __name__ == '__main__':
 467     test_main()