Lib/test/test_os.py

   1 # As a test suite for the os module, this is woefully inadequate, but this
   2 # does add tests for a few functions which have been determined to be more
   3 # portable than they had been thought to be.
   4
   5 import os
   6 import errno
   7 import unittest
   8 import warnings
   9 import sys
  10 import signal
  11 import subprocess
  12 import time
  13 from test import test_support
  14
  15 warnings.filterwarnings("ignore", "tempnam", RuntimeWarning, __name__)
  16 warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, __name__)
  17
  18 # Tests creating TESTFN
  19 class FileTests(unittest.TestCase):
  20     def setUp(self):
  21         if os.path.exists(test_support.TESTFN):
  22             os.unlink(test_support.TESTFN)
  23     tearDown = setUp
  24
  25     def test_access(self):
  26         f = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
  27         os.close(f)
  28         self.assertTrue(os.access(test_support.TESTFN, os.W_OK))
  29
  30     def test_closerange(self):
  31         first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
  32         # We must allocate two consecutive file descriptors, otherwise
  33         # it will mess up other file descriptors (perhaps even the three
  34         # standard ones).
  35         second = os.dup(first)
  36         try:
  37             retries = 0
  38             while second != first + 1:
  39                 os.close(first)
  40                 retries += 1
  41                 if retries > 10:
  42                     # XXX test skipped
  43                     self.skipTest("couldn't allocate two consecutive fds")
  44                 first, second = second, os.dup(second)
  45         finally:
  46             os.close(second)
  47         # close a fd that is open, and one that isn't
  48         os.closerange(first, first + 2)
  49         self.assertRaises(OSError, os.write, first, "a")
  50
  51     @test_support.cpython_only
  52     def test_rename(self):
  53         path = unicode(test_support.TESTFN)
  54         old = sys.getrefcount(path)
  55         self.assertRaises(TypeError, os.rename, path, 0)
  56         new = sys.getrefcount(path)
  57         self.assertEqual(old, new)
  58
  59
  60 class TemporaryFileTests(unittest.TestCase):
  61     def setUp(self):
  62         self.files = []
  63         os.mkdir(test_support.TESTFN)
  64
  65     def tearDown(self):
  66         for name in self.files:
  67             os.unlink(name)
  68         os.rmdir(test_support.TESTFN)
  69
  70     def check_tempfile(self, name):
  71         # make sure it doesn't already exist:
  72         self.assertFalse(os.path.exists(name),
  73                     "file already exists for temporary file")
  74         # make sure we can create the file
  75         open(name, "w")
  76         self.files.append(name)
  77
  78     def test_tempnam(self):
  79         if not hasattr(os, "tempnam"):
  80             return
  81         warnings.filterwarnings("ignore", "tempnam", RuntimeWarning,
  82                                 r"test_os$")
  83         self.check_tempfile(os.tempnam())
  84
  85         name = os.tempnam(test_support.TESTFN)
  86         self.check_tempfile(name)
  87
  88         name = os.tempnam(test_support.TESTFN, "pfx")
  89         self.assertTrue(os.path.basename(name)[:3] == "pfx")
  90         self.check_tempfile(name)
  91
  92     def test_tmpfile(self):
  93         if not hasattr(os, "tmpfile"):
  94             return
  95         # As with test_tmpnam() below, the Windows implementation of tmpfile()
  96         # attempts to create a file in the root directory of the current drive.
  97         # On Vista and Server 2008, this test will always fail for normal users
  98         # as writing to the root directory requires elevated privileges.  With
  99         # XP and below, the semantics of tmpfile() are the same, but the user
 100         # running the test is more likely to have administrative privileges on
 101         # their account already.  If that's the case, then os.tmpfile() should
 102         # work.  In order to make this test as useful as possible, rather than
 103         # trying to detect Windows versions or whether or not the user has the
 104         # right permissions, just try and create a file in the root directory
 105         # and see if it raises a 'Permission denied' OSError.  If it does, then
 106         # test that a subsequent call to os.tmpfile() raises the same error. If
 107         # it doesn't, assume we're on XP or below and the user running the test
 108         # has administrative privileges, and proceed with the test as normal.
 109         if sys.platform == 'win32':
 110             name = '\\python_test_os_test_tmpfile.txt'
 111             if os.path.exists(name):
 112                 os.remove(name)
 113             try:
 114                 fp = open(name, 'w')
 115             except IOError, first:
 116                 # open() failed, assert tmpfile() fails in the same way.
 117                 # Although open() raises an IOError and os.tmpfile() raises an
 118                 # OSError(), 'args' will be (13, 'Permission denied') in both
 119                 # cases.
 120                 try:
 121                     fp = os.tmpfile()
 122                 except OSError, second:
 123                     self.assertEqual(first.args, second.args)
 124                 else:
 125                     self.fail("expected os.tmpfile() to raise OSError")
 126                 return
 127             else:
 128                 # open() worked, therefore, tmpfile() should work.  Close our
 129                 # dummy file and proceed with the test as normal.
 130                 fp.close()
 131                 os.remove(name)
 132
 133         fp = os.tmpfile()
 134         fp.write("foobar")
 135         fp.seek(0,0)
 136         s = fp.read()
 137         fp.close()
 138         self.assertTrue(s == "foobar")
 139
 140     def test_tmpnam(self):
 141         if not hasattr(os, "tmpnam"):
 142             return
 143         warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning,
 144                                 r"test_os$")
 145         name = os.tmpnam()
 146         if sys.platform in ("win32",):
 147             # The Windows tmpnam() seems useless.  From the MS docs:
 148             #
 149             #     The character string that tmpnam creates consists of
 150             #     the path prefix, defined by the entry P_tmpdir in the
 151             #     file STDIO.H, followed by a sequence consisting of the
 152             #     digit characters '0' through '9'; the numerical value
 153             #     of this string is in the range 1 - 65,535.  Changing the
 154             #     definitions of L_tmpnam or P_tmpdir in STDIO.H does not
 155             #     change the operation of tmpnam.
 156             #
 157             # The really bizarre part is that, at least under MSVC6,
 158             # P_tmpdir is "\\".  That is, the path returned refers to
 159             # the root of the current drive.  That's a terrible place to
 160             # put temp files, and, depending on privileges, the user
 161             # may not even be able to open a file in the root directory.
 162             self.assertFalse(os.path.exists(name),
 163                         "file already exists for temporary file")
 164         else:
 165             self.check_tempfile(name)
 166
 167 # Test attributes on return values from os.*stat* family.
 168 class StatAttributeTests(unittest.TestCase):
 169     def setUp(self):
 170         os.mkdir(test_support.TESTFN)
 171         self.fname = os.path.join(test_support.TESTFN, "f1")
 172         f = open(self.fname, 'wb')
 173         f.write("ABC")
 174         f.close()
 175
 176     def tearDown(self):
 177         os.unlink(self.fname)
 178         os.rmdir(test_support.TESTFN)
 179
 180     def test_stat_attributes(self):
 181         if not hasattr(os, "stat"):
 182             return
 183
 184         import stat
 185         result = os.stat(self.fname)
 186
 187         # Make sure direct access works
 188         self.assertEquals(result[stat.ST_SIZE], 3)
 189         self.assertEquals(result.st_size, 3)
 190
 191         # Make sure all the attributes are there
 192         members = dir(result)
 193         for name in dir(stat):
 194             if name[:3] == 'ST_':
 195                 attr = name.lower()
 196                 if name.endswith("TIME"):
 197                     def trunc(x): return int(x)
 198                 else:
 199                     def trunc(x): return x
 200                 self.assertEquals(trunc(getattr(result, attr)),
 201                                   result[getattr(stat, name)])
 202                 self.assertIn(attr, members)
 203
 204         try:
 205             result[200]
 206             self.fail("No exception thrown")
 207         except IndexError:
 208             pass
 209
 210         # Make sure that assignment fails
 211         try:
 212             result.st_mode = 1
 213             self.fail("No exception thrown")
 214         except (AttributeError, TypeError):
 215             pass
 216
 217         try:
 218             result.st_rdev = 1
 219             self.fail("No exception thrown")
 220         except (AttributeError, TypeError):
 221             pass
 222
 223         try:
 224             result.parrot = 1
 225             self.fail("No exception thrown")
 226         except AttributeError:
 227             pass
 228
 229         # Use the stat_result constructor with a too-short tuple.
 230         try:
 231             result2 = os.stat_result((10,))
 232             self.fail("No exception thrown")
 233         except TypeError:
 234             pass
 235
 236         # Use the constructr with a too-long tuple.
 237         try:
 238             result2 = os.stat_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
 239         except TypeError:
 240             pass
 241
 242
 243     def test_statvfs_attributes(self):
 244         if not hasattr(os, "statvfs"):
 245             return
 246
 247         try:
 248             result = os.statvfs(self.fname)
 249         except OSError, e:
 250             # On AtheOS, glibc always returns ENOSYS
 251             if e.errno == errno.ENOSYS:
 252                 return
 253
 254         # Make sure direct access works
 255         self.assertEquals(result.f_bfree, result[3])
 256
 257         # Make sure all the attributes are there.
 258         members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
 259                     'ffree', 'favail', 'flag', 'namemax')
 260         for value, member in enumerate(members):
 261             self.assertEquals(getattr(result, 'f_' + member), result[value])
 262
 263         # Make sure that assignment really fails
 264         try:
 265             result.f_bfree = 1
 266             self.fail("No exception thrown")
 267         except TypeError:
 268             pass
 269
 270         try:
 271             result.parrot = 1
 272             self.fail("No exception thrown")
 273         except AttributeError:
 274             pass
 275
 276         # Use the constructor with a too-short tuple.
 277         try:
 278             result2 = os.statvfs_result((10,))
 279             self.fail("No exception thrown")
 280         except TypeError:
 281             pass
 282
 283         # Use the constructr with a too-long tuple.
 284         try:
 285             result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
 286         except TypeError:
 287             pass
 288
 289     def test_utime_dir(self):
 290         delta = 1000000
 291         st = os.stat(test_support.TESTFN)
 292         # round to int, because some systems may support sub-second
 293         # time stamps in stat, but not in utime.
 294         os.utime(test_support.TESTFN, (st.st_atime, int(st.st_mtime-delta)))
 295         st2 = os.stat(test_support.TESTFN)
 296         self.assertEquals(st2.st_mtime, int(st.st_mtime-delta))
 297
 298     # Restrict test to Win32, since there is no guarantee other
 299     # systems support centiseconds
 300     if sys.platform == 'win32':
 301         def get_file_system(path):
 302             root = os.path.splitdrive(os.path.abspath(path))[0] + '\\'
 303             import ctypes
 304             kernel32 = ctypes.windll.kernel32
 305             buf = ctypes.create_string_buffer("", 100)
 306             if kernel32.GetVolumeInformationA(root, None, 0, None, None, None, buf, len(buf)):
 307                 return buf.value
 308
 309         if get_file_system(test_support.TESTFN) == "NTFS":
 310             def test_1565150(self):
 311                 t1 = 1159195039.25
 312                 os.utime(self.fname, (t1, t1))
 313                 self.assertEquals(os.stat(self.fname).st_mtime, t1)
 314
 315         def test_1686475(self):
 316             # Verify that an open file can be stat'ed
 317             try:
 318                 os.stat(r"c:\pagefile.sys")
 319             except WindowsError, e:
 320                 if e.errno == 2: # file does not exist; cannot run test
 321                     return
 322                 self.fail("Could not stat pagefile.sys")
 323
 324 from test import mapping_tests
 325
 326 class EnvironTests(mapping_tests.BasicTestMappingProtocol):
 327     """check that os.environ object conform to mapping protocol"""
 328     type2test = None
 329     def _reference(self):
 330         return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"}
 331     def _empty_mapping(self):
 332         os.environ.clear()
 333         return os.environ
 334     def setUp(self):
 335         self.__save = dict(os.environ)
 336         os.environ.clear()
 337     def tearDown(self):
 338         os.environ.clear()
 339         os.environ.update(self.__save)
 340
 341     # Bug 1110478
 342     def test_update2(self):
 343         if os.path.exists("/bin/sh"):
 344             os.environ.update(HELLO="World")
 345             value = os.popen("/bin/sh -c 'echo $HELLO'").read().strip()
 346             self.assertEquals(value, "World")
 347
 348 class WalkTests(unittest.TestCase):
 349     """Tests for os.walk()."""
 350
 351     def test_traversal(self):
 352         import os
 353         from os.path import join
 354
 355         # Build:
 356         #     TESTFN/
 357         #       TEST1/              a file kid and two directory kids
 358         #         tmp1
 359         #         SUB1/             a file kid and a directory kid
 360         #           tmp2
 361         #           SUB11/          no kids
 362         #         SUB2/             a file kid and a dirsymlink kid
 363         #           tmp3
 364         #           link/           a symlink to TESTFN.2
 365         #       TEST2/
 366         #         tmp4              a lone file
 367         walk_path = join(test_support.TESTFN, "TEST1")
 368         sub1_path = join(walk_path, "SUB1")
 369         sub11_path = join(sub1_path, "SUB11")
 370         sub2_path = join(walk_path, "SUB2")
 371         tmp1_path = join(walk_path, "tmp1")
 372         tmp2_path = join(sub1_path, "tmp2")
 373         tmp3_path = join(sub2_path, "tmp3")
 374         link_path = join(sub2_path, "link")
 375         t2_path = join(test_support.TESTFN, "TEST2")
 376         tmp4_path = join(test_support.TESTFN, "TEST2", "tmp4")
 377
 378         # Create stuff.
 379         os.makedirs(sub11_path)
 380         os.makedirs(sub2_path)
 381         os.makedirs(t2_path)
 382         for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path:
 383             f = file(path, "w")
 384             f.write("I'm " + path + " and proud of it.  Blame test_os.\n")
 385             f.close()
 386         if hasattr(os, "symlink"):
 387             os.symlink(os.path.abspath(t2_path), link_path)
 388             sub2_tree = (sub2_path, ["link"], ["tmp3"])
 389         else:
 390             sub2_tree = (sub2_path, [], ["tmp3"])
 391
 392         # Walk top-down.
 393         all = list(os.walk(walk_path))
 394         self.assertEqual(len(all), 4)
 395         # We can't know which order SUB1 and SUB2 will appear in.
 396         # Not flipped:  TESTFN, SUB1, SUB11, SUB2
 397         #     flipped:  TESTFN, SUB2, SUB1, SUB11
 398         flipped = all[0][1][0] != "SUB1"
 399         all[0][1].sort()
 400         self.assertEqual(all[0], (walk_path, ["SUB1", "SUB2"], ["tmp1"]))
 401         self.assertEqual(all[1 + flipped], (sub1_path, ["SUB11"], ["tmp2"]))
 402         self.assertEqual(all[2 + flipped], (sub11_path, [], []))
 403         self.assertEqual(all[3 - 2 * flipped], sub2_tree)
 404
 405         # Prune the search.
 406         all = []
 407         for root, dirs, files in os.walk(walk_path):
 408             all.append((root, dirs, files))
 409             # Don't descend into SUB1.
 410             if 'SUB1' in dirs:
 411                 # Note that this also mutates the dirs we appended to all!
 412                 dirs.remove('SUB1')
 413         self.assertEqual(len(all), 2)
 414         self.assertEqual(all[0], (walk_path, ["SUB2"], ["tmp1"]))
 415         self.assertEqual(all[1], sub2_tree)
 416
 417         # Walk bottom-up.
 418         all = list(os.walk(walk_path, topdown=False))
 419         self.assertEqual(len(all), 4)
 420         # We can't know which order SUB1 and SUB2 will appear in.
 421         # Not flipped:  SUB11, SUB1, SUB2, TESTFN
 422         #     flipped:  SUB2, SUB11, SUB1, TESTFN
 423         flipped = all[3][1][0] != "SUB1"
 424         all[3][1].sort()
 425         self.assertEqual(all[3], (walk_path, ["SUB1", "SUB2"], ["tmp1"]))
 426         self.assertEqual(all[flipped], (sub11_path, [], []))
 427         self.assertEqual(all[flipped + 1], (sub1_path, ["SUB11"], ["tmp2"]))
 428         self.assertEqual(all[2 - 2 * flipped], sub2_tree)
 429
 430         if hasattr(os, "symlink"):
 431             # Walk, following symlinks.
 432             for root, dirs, files in os.walk(walk_path, followlinks=True):
 433                 if root == link_path:
 434                     self.assertEqual(dirs, [])
 435                     self.assertEqual(files, ["tmp4"])
 436                     break
 437             else:
 438                 self.fail("Didn't follow symlink with followlinks=True")
 439
 440     def tearDown(self):
 441         # Tear everything down.  This is a decent use for bottom-up on
 442         # Windows, which doesn't have a recursive delete command.  The
 443         # (not so) subtlety is that rmdir will fail unless the dir's
 444         # kids are removed first, so bottom up is essential.
 445         for root, dirs, files in os.walk(test_support.TESTFN, topdown=False):
 446             for name in files:
 447                 os.remove(os.path.join(root, name))
 448             for name in dirs:
 449                 dirname = os.path.join(root, name)
 450                 if not os.path.islink(dirname):
 451                     os.rmdir(dirname)
 452                 else:
 453                     os.remove(dirname)
 454         os.rmdir(test_support.TESTFN)
 455
 456 class MakedirTests (unittest.TestCase):
 457     def setUp(self):
 458         os.mkdir(test_support.TESTFN)
 459
 460     def test_makedir(self):
 461         base = test_support.TESTFN
 462         path = os.path.join(base, 'dir1', 'dir2', 'dir3')
 463         os.makedirs(path)             # Should work
 464         path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4')
 465         os.makedirs(path)
 466
 467         # Try paths with a '.' in them
 468         self.assertRaises(OSError, os.makedirs, os.curdir)
 469         path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', os.curdir)
 470         os.makedirs(path)
 471         path = os.path.join(base, 'dir1', os.curdir, 'dir2', 'dir3', 'dir4',
 472                             'dir5', 'dir6')
 473         os.makedirs(path)
 474
 475
 476
 477
 478     def tearDown(self):
 479         path = os.path.join(test_support.TESTFN, 'dir1', 'dir2', 'dir3',
 480                             'dir4', 'dir5', 'dir6')
 481         # If the tests failed, the bottom-most directory ('../dir6')
 482         # may not have been created, so we look for the outermost directory
 483         # that exists.
 484         while not os.path.exists(path) and path != test_support.TESTFN:
 485             path = os.path.dirname(path)
 486
 487         os.removedirs(path)
 488
 489 class DevNullTests (unittest.TestCase):
 490     def test_devnull(self):
 491         f = file(os.devnull, 'w')
 492         f.write('hello')
 493         f.close()
 494         f = file(os.devnull, 'r')
 495         self.assertEqual(f.read(), '')
 496         f.close()
 497
 498 class URandomTests (unittest.TestCase):
 499     def test_urandom(self):
 500         try:
 501             self.assertEqual(len(os.urandom(1)), 1)
 502             self.assertEqual(len(os.urandom(10)), 10)
 503             self.assertEqual(len(os.urandom(100)), 100)
 504             self.assertEqual(len(os.urandom(1000)), 1000)
 505             # see http://bugs.python.org/issue3708
 506             self.assertRaises(TypeError, os.urandom, 0.9)
 507             self.assertRaises(TypeError, os.urandom, 1.1)
 508             self.assertRaises(TypeError, os.urandom, 2.0)
 509         except NotImplementedError:
 510             pass
 511
 512     def test_execvpe_with_bad_arglist(self):
 513         self.assertRaises(ValueError, os.execvpe, 'notepad', [], None)
 514
 515 class Win32ErrorTests(unittest.TestCase):
 516     def test_rename(self):
 517         self.assertRaises(WindowsError, os.rename, test_support.TESTFN, test_support.TESTFN+".bak")
 518
 519     def test_remove(self):
 520         self.assertRaises(WindowsError, os.remove, test_support.TESTFN)
 521
 522     def test_chdir(self):
 523         self.assertRaises(WindowsError, os.chdir, test_support.TESTFN)
 524
 525     def test_mkdir(self):
 526         f = open(test_support.TESTFN, "w")
 527         try:
 528             self.assertRaises(WindowsError, os.mkdir, test_support.TESTFN)
 529         finally:
 530             f.close()
 531             os.unlink(test_support.TESTFN)
 532
 533     def test_utime(self):
 534         self.assertRaises(WindowsError, os.utime, test_support.TESTFN, None)
 535
 536     def test_chmod(self):
 537         self.assertRaises(WindowsError, os.chmod, test_support.TESTFN, 0)
 538
 539 class TestInvalidFD(unittest.TestCase):
 540     singles = ["fchdir", "fdopen", "dup", "fdatasync", "fstat",
 541                "fstatvfs", "fsync", "tcgetpgrp", "ttyname"]
 542     #singles.append("close")
 543     #We omit close because it doesn'r raise an exception on some platforms
 544     def get_single(f):
 545         def helper(self):
 546             if  hasattr(os, f):
 547                 self.check(getattr(os, f))
 548         return helper
 549     for f in singles:
 550         locals()["test_"+f] = get_single(f)
 551
 552     def check(self, f, *args):
 553         try:
 554             f(test_support.make_bad_fd(), *args)
 555         except OSError as e:
 556             self.assertEqual(e.errno, errno.EBADF)
 557         else:
 558             self.fail("%r didn't raise a OSError with a bad file descriptor"
 559                       % f)
 560
 561     def test_isatty(self):
 562         if hasattr(os, "isatty"):
 563             self.assertEqual(os.isatty(test_support.make_bad_fd()), False)
 564
 565     def test_closerange(self):
 566         if hasattr(os, "closerange"):
 567             fd = test_support.make_bad_fd()
 568             # Make sure none of the descriptors we are about to close are
 569             # currently valid (issue 6542).
 570             for i in range(10):
 571                 try: os.fstat(fd+i)
 572                 except OSError:
 573                     pass
 574                 else:
 575                     break
 576             if i < 2:
 577                 raise unittest.SkipTest(
 578                     "Unable to acquire a range of invalid file descriptors")
 579             self.assertEqual(os.closerange(fd, fd + i-1), None)
 580
 581     def test_dup2(self):
 582         if hasattr(os, "dup2"):
 583             self.check(os.dup2, 20)
 584
 585     def test_fchmod(self):
 586         if hasattr(os, "fchmod"):
 587             self.check(os.fchmod, 0)
 588
 589     def test_fchown(self):
 590         if hasattr(os, "fchown"):
 591             self.check(os.fchown, -1, -1)
 592
 593     def test_fpathconf(self):
 594         if hasattr(os, "fpathconf"):
 595             self.check(os.fpathconf, "PC_NAME_MAX")
 596
 597     def test_ftruncate(self):
 598         if hasattr(os, "ftruncate"):
 599             self.check(os.ftruncate, 0)
 600
 601     def test_lseek(self):
 602         if hasattr(os, "lseek"):
 603             self.check(os.lseek, 0, 0)
 604
 605     def test_read(self):
 606         if hasattr(os, "read"):
 607             self.check(os.read, 1)
 608
 609     def test_tcsetpgrpt(self):
 610         if hasattr(os, "tcsetpgrp"):
 611             self.check(os.tcsetpgrp, 0)
 612
 613     def test_write(self):
 614         if hasattr(os, "write"):
 615             self.check(os.write, " ")
 616
 617 if sys.platform != 'win32':
 618     class Win32ErrorTests(unittest.TestCase):
 619         pass
 620
 621     class PosixUidGidTests(unittest.TestCase):
 622         if hasattr(os, 'setuid'):
 623             def test_setuid(self):
 624                 if os.getuid() != 0:
 625                     self.assertRaises(os.error, os.setuid, 0)
 626                 self.assertRaises(OverflowError, os.setuid, 1<<32)
 627
 628         if hasattr(os, 'setgid'):
 629             def test_setgid(self):
 630                 if os.getuid() != 0:
 631                     self.assertRaises(os.error, os.setgid, 0)
 632                 self.assertRaises(OverflowError, os.setgid, 1<<32)
 633
 634         if hasattr(os, 'seteuid'):
 635             def test_seteuid(self):
 636                 if os.getuid() != 0:
 637                     self.assertRaises(os.error, os.seteuid, 0)
 638                 self.assertRaises(OverflowError, os.seteuid, 1<<32)
 639
 640         if hasattr(os, 'setegid'):
 641             def test_setegid(self):
 642                 if os.getuid() != 0:
 643                     self.assertRaises(os.error, os.setegid, 0)
 644                 self.assertRaises(OverflowError, os.setegid, 1<<32)
 645
 646         if hasattr(os, 'setreuid'):
 647             def test_setreuid(self):
 648                 if os.getuid() != 0:
 649                     self.assertRaises(os.error, os.setreuid, 0, 0)
 650                 self.assertRaises(OverflowError, os.setreuid, 1<<32, 0)
 651                 self.assertRaises(OverflowError, os.setreuid, 0, 1<<32)
 652
 653             def test_setreuid_neg1(self):
 654                 # Needs to accept -1.  We run this in a subprocess to avoid
 655                 # altering the test runner's process state (issue8045).
 656                 subprocess.check_call([
 657                         sys.executable, '-c',
 658                         'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
 659
 660         if hasattr(os, 'setregid'):
 661             def test_setregid(self):
 662                 if os.getuid() != 0:
 663                     self.assertRaises(os.error, os.setregid, 0, 0)
 664                 self.assertRaises(OverflowError, os.setregid, 1<<32, 0)
 665                 self.assertRaises(OverflowError, os.setregid, 0, 1<<32)
 666
 667             def test_setregid_neg1(self):
 668                 # Needs to accept -1.  We run this in a subprocess to avoid
 669                 # altering the test runner's process state (issue8045).
 670                 subprocess.check_call([
 671                         sys.executable, '-c',
 672                         'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
 673 else:
 674     class PosixUidGidTests(unittest.TestCase):
 675         pass
 676
 677 @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
 678 class Win32KillTests(unittest.TestCase):
 679     def _kill(self, sig):
 680         # Start sys.executable as a subprocess and communicate from the
 681         # subprocess to the parent that the interpreter is ready. When it
 682         # becomes ready, send *sig* via os.kill to the subprocess and check
 683         # that the return code is equal to *sig*.
 684         import ctypes
 685         from ctypes import wintypes
 686         import msvcrt
 687
 688         # Since we can't access the contents of the process' stdout until the
 689         # process has exited, use PeekNamedPipe to see what's inside stdout
 690         # without waiting. This is done so we can tell that the interpreter
 691         # is started and running at a point where it could handle a signal.
 692         PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
 693         PeekNamedPipe.restype = wintypes.BOOL
 694         PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
 695                                   ctypes.POINTER(ctypes.c_char), # stdout buf
 696                                   wintypes.DWORD, # Buffer size
 697                                   ctypes.POINTER(wintypes.DWORD), # bytes read
 698                                   ctypes.POINTER(wintypes.DWORD), # bytes avail
 699                                   ctypes.POINTER(wintypes.DWORD)) # bytes left
 700         msg = "running"
 701         proc = subprocess.Popen([sys.executable, "-c",
 702                                  "import sys;"
 703                                  "sys.stdout.write('{}');"
 704                                  "sys.stdout.flush();"
 705                                  "input()".format(msg)],
 706                                 stdout=subprocess.PIPE,
 707                                 stderr=subprocess.PIPE,
 708                                 stdin=subprocess.PIPE)
 709
 710         count, max = 0, 100
 711         while count < max and proc.poll() is None:
 712             # Create a string buffer to store the result of stdout from the pipe
 713             buf = ctypes.create_string_buffer(len(msg))
 714             # Obtain the text currently in proc.stdout
 715             # Bytes read/avail/left are left as NULL and unused
 716             rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
 717                                  buf, ctypes.sizeof(buf), None, None, None)
 718             self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
 719             if buf.value:
 720                 self.assertEqual(msg, buf.value)
 721                 break
 722             time.sleep(0.1)
 723             count += 1
 724         else:
 725             self.fail("Did not receive communication from the subprocess")
 726
 727         os.kill(proc.pid, sig)
 728         self.assertEqual(proc.wait(), sig)
 729
 730     def test_kill_sigterm(self):
 731         # SIGTERM doesn't mean anything special, but make sure it works
 732         self._kill(signal.SIGTERM)
 733
 734     def test_kill_int(self):
 735         # os.kill on Windows can take an int which gets set as the exit code
 736         self._kill(100)
 737
 738     def _kill_with_event(self, event, name):
 739         # Run a script which has console control handling enabled.
 740         proc = subprocess.Popen([sys.executable,
 741                    os.path.join(os.path.dirname(__file__),
 742                                 "win_console_handler.py")],
 743                    creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
 744         # Let the interpreter startup before we send signals. See #3137.
 745         time.sleep(0.5)
 746         os.kill(proc.pid, event)
 747         # proc.send_signal(event) could also be done here.
 748         # Allow time for the signal to be passed and the process to exit.
 749         time.sleep(0.5)
 750         if not proc.poll():
 751             # Forcefully kill the process if we weren't able to signal it.
 752             os.kill(proc.pid, signal.SIGINT)
 753             self.fail("subprocess did not stop on {}".format(name))
 754
 755     @unittest.skip("subprocesses aren't inheriting CTRL+C property")
 756     def test_CTRL_C_EVENT(self):
 757         from ctypes import wintypes
 758         import ctypes
 759
 760         # Make a NULL value by creating a pointer with no argument.
 761         NULL = ctypes.POINTER(ctypes.c_int)()
 762         SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
 763         SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
 764                                           wintypes.BOOL)
 765         SetConsoleCtrlHandler.restype = wintypes.BOOL
 766
 767         # Calling this with NULL and FALSE causes the calling process to
 768         # handle CTRL+C, rather than ignore it. This property is inherited
 769         # by subprocesses.
 770         SetConsoleCtrlHandler(NULL, 0)
 771
 772         self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
 773
 774     def test_CTRL_BREAK_EVENT(self):
 775         self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
 776
 777
 778 def test_main():
 779     test_support.run_unittest(
 780         FileTests,
 781         TemporaryFileTests,
 782         StatAttributeTests,
 783         EnvironTests,
 784         WalkTests,
 785         MakedirTests,
 786         DevNullTests,
 787         URandomTests,
 788         Win32ErrorTests,
 789         TestInvalidFD,
 790         PosixUidGidTests,
 791         Win32KillTests
 792     )
 793
 794 if __name__ == "__main__":
 795     test_main()