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