diff options
Diffstat (limited to 'Lib/test/test_os.py')
| -rw-r--r-- | Lib/test/test_os.py | 906 |
1 files changed, 703 insertions, 203 deletions
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index bdbeb01..5b67da1 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -10,28 +10,39 @@ import sys import signal import subprocess import time - -from test import test_support +import shutil +from test import support +import contextlib import mmap import uuid - -warnings.filterwarnings("ignore", "tempnam", RuntimeWarning, __name__) -warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, __name__) +import stat +from test.script_helper import assert_python_ok + +# Detect whether we're on a Linux system that uses the (now outdated +# and unmaintained) linuxthreads threading library. There's an issue +# when combining linuxthreads with a failed execv call: see +# http://bugs.python.org/issue4970. +if (hasattr(os, "confstr_names") and + "CS_GNU_LIBPTHREAD_VERSION" in os.confstr_names): + libpthread = os.confstr("CS_GNU_LIBPTHREAD_VERSION") + USING_LINUXTHREADS= libpthread.startswith("linuxthreads") +else: + USING_LINUXTHREADS= False # Tests creating TESTFN class FileTests(unittest.TestCase): def setUp(self): - if os.path.exists(test_support.TESTFN): - os.unlink(test_support.TESTFN) + if os.path.exists(support.TESTFN): + os.unlink(support.TESTFN) tearDown = setUp def test_access(self): - f = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR) + f = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR) os.close(f) - self.assertTrue(os.access(test_support.TESTFN, os.W_OK)) + self.assertTrue(os.access(support.TESTFN, os.W_OK)) def test_closerange(self): - first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR) + first = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR) # We must allocate two consecutive file descriptors, otherwise # it will mess up other file descriptors (perhaps even the three # standard ones). @@ -49,151 +60,90 @@ class FileTests(unittest.TestCase): os.close(second) # close a fd that is open, and one that isn't os.closerange(first, first + 2) - self.assertRaises(OSError, os.write, first, "a") + self.assertRaises(OSError, os.write, first, b"a") - @test_support.cpython_only + @support.cpython_only def test_rename(self): - path = unicode(test_support.TESTFN) + path = support.TESTFN old = sys.getrefcount(path) self.assertRaises(TypeError, os.rename, path, 0) new = sys.getrefcount(path) self.assertEqual(old, new) + def test_read(self): + with open(support.TESTFN, "w+b") as fobj: + fobj.write(b"spam") + fobj.flush() + fd = fobj.fileno() + os.lseek(fd, 0, 0) + s = os.read(fd, 4) + self.assertEqual(type(s), bytes) + self.assertEqual(s, b"spam") -class TemporaryFileTests(unittest.TestCase): - def setUp(self): - self.files = [] - os.mkdir(test_support.TESTFN) + def test_write(self): + # os.write() accepts bytes- and buffer-like objects but not strings + fd = os.open(support.TESTFN, os.O_CREAT | os.O_WRONLY) + self.assertRaises(TypeError, os.write, fd, "beans") + os.write(fd, b"bacon\n") + os.write(fd, bytearray(b"eggs\n")) + os.write(fd, memoryview(b"spam\n")) + os.close(fd) + with open(support.TESTFN, "rb") as fobj: + self.assertEqual(fobj.read().splitlines(), + [b"bacon", b"eggs", b"spam"]) + + def write_windows_console(self, *args): + retcode = subprocess.call(args, + # use a new console to not flood the test output + creationflags=subprocess.CREATE_NEW_CONSOLE, + # use a shell to hide the console window (SW_HIDE) + shell=True) + self.assertEqual(retcode, 0) + + @unittest.skipUnless(sys.platform == 'win32', + 'test specific to the Windows console') + def test_write_windows_console(self): + # Issue #11395: the Windows console returns an error (12: not enough + # space error) on writing into stdout if stdout mode is binary and the + # length is greater than 66,000 bytes (or less, depending on heap + # usage). + code = "print('x' * 100000)" + self.write_windows_console(sys.executable, "-c", code) + self.write_windows_console(sys.executable, "-u", "-c", code) + + def fdopen_helper(self, *args): + fd = os.open(support.TESTFN, os.O_RDONLY) + f = os.fdopen(fd, *args) + f.close() - def tearDown(self): - for name in self.files: - os.unlink(name) - os.rmdir(test_support.TESTFN) - - def check_tempfile(self, name): - # make sure it doesn't already exist: - self.assertFalse(os.path.exists(name), - "file already exists for temporary file") - # make sure we can create the file - open(name, "w") - self.files.append(name) - - def test_tempnam(self): - if not hasattr(os, "tempnam"): - return - with warnings.catch_warnings(): - warnings.filterwarnings("ignore", "tempnam", RuntimeWarning, - r"test_os$") - warnings.filterwarnings("ignore", "tempnam", DeprecationWarning) - self.check_tempfile(os.tempnam()) + def test_fdopen(self): + fd = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR) + os.close(fd) - name = os.tempnam(test_support.TESTFN) - self.check_tempfile(name) + self.fdopen_helper() + self.fdopen_helper('r') + self.fdopen_helper('r', 100) - name = os.tempnam(test_support.TESTFN, "pfx") - self.assertTrue(os.path.basename(name)[:3] == "pfx") - self.check_tempfile(name) - - def test_tmpfile(self): - if not hasattr(os, "tmpfile"): - return - # As with test_tmpnam() below, the Windows implementation of tmpfile() - # attempts to create a file in the root directory of the current drive. - # On Vista and Server 2008, this test will always fail for normal users - # as writing to the root directory requires elevated privileges. With - # XP and below, the semantics of tmpfile() are the same, but the user - # running the test is more likely to have administrative privileges on - # their account already. If that's the case, then os.tmpfile() should - # work. In order to make this test as useful as possible, rather than - # trying to detect Windows versions or whether or not the user has the - # right permissions, just try and create a file in the root directory - # and see if it raises a 'Permission denied' OSError. If it does, then - # test that a subsequent call to os.tmpfile() raises the same error. If - # it doesn't, assume we're on XP or below and the user running the test - # has administrative privileges, and proceed with the test as normal. - with warnings.catch_warnings(): - warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning) - - if sys.platform == 'win32': - name = '\\python_test_os_test_tmpfile.txt' - if os.path.exists(name): - os.remove(name) - try: - fp = open(name, 'w') - except IOError, first: - # open() failed, assert tmpfile() fails in the same way. - # Although open() raises an IOError and os.tmpfile() raises an - # OSError(), 'args' will be (13, 'Permission denied') in both - # cases. - try: - fp = os.tmpfile() - except OSError, second: - self.assertEqual(first.args, second.args) - else: - self.fail("expected os.tmpfile() to raise OSError") - return - else: - # open() worked, therefore, tmpfile() should work. Close our - # dummy file and proceed with the test as normal. - fp.close() - os.remove(name) - - fp = os.tmpfile() - fp.write("foobar") - fp.seek(0,0) - s = fp.read() - fp.close() - self.assertTrue(s == "foobar") - - def test_tmpnam(self): - if not hasattr(os, "tmpnam"): - return - with warnings.catch_warnings(): - warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, - r"test_os$") - warnings.filterwarnings("ignore", "tmpnam", DeprecationWarning) - - name = os.tmpnam() - if sys.platform in ("win32",): - # The Windows tmpnam() seems useless. From the MS docs: - # - # The character string that tmpnam creates consists of - # the path prefix, defined by the entry P_tmpdir in the - # file STDIO.H, followed by a sequence consisting of the - # digit characters '0' through '9'; the numerical value - # of this string is in the range 1 - 65,535. Changing the - # definitions of L_tmpnam or P_tmpdir in STDIO.H does not - # change the operation of tmpnam. - # - # The really bizarre part is that, at least under MSVC6, - # P_tmpdir is "\\". That is, the path returned refers to - # the root of the current drive. That's a terrible place to - # put temp files, and, depending on privileges, the user - # may not even be able to open a file in the root directory. - self.assertFalse(os.path.exists(name), - "file already exists for temporary file") - else: - self.check_tempfile(name) # Test attributes on return values from os.*stat* family. class StatAttributeTests(unittest.TestCase): def setUp(self): - os.mkdir(test_support.TESTFN) - self.fname = os.path.join(test_support.TESTFN, "f1") + os.mkdir(support.TESTFN) + self.fname = os.path.join(support.TESTFN, "f1") f = open(self.fname, 'wb') - f.write("ABC") + f.write(b"ABC") f.close() def tearDown(self): os.unlink(self.fname) - os.rmdir(test_support.TESTFN) + os.rmdir(support.TESTFN) - def test_stat_attributes(self): + def check_stat_attributes(self, fname): if not hasattr(os, "stat"): return import stat - result = os.stat(self.fname) + result = os.stat(fname) # Make sure direct access works self.assertEqual(result[stat.ST_SIZE], 3) @@ -209,7 +159,7 @@ class StatAttributeTests(unittest.TestCase): else: def trunc(x): return x self.assertEqual(trunc(getattr(result, attr)), - result[getattr(stat, name)]) + result[getattr(stat, name)]) self.assertIn(attr, members) try: @@ -222,7 +172,7 @@ class StatAttributeTests(unittest.TestCase): try: result.st_mode = 1 self.fail("No exception raised") - except (AttributeError, TypeError): + except AttributeError: pass try: @@ -250,6 +200,15 @@ class StatAttributeTests(unittest.TestCase): except TypeError: pass + def test_stat_attributes(self): + self.check_stat_attributes(self.fname) + + def test_stat_attributes_bytes(self): + try: + fname = self.fname.encode(sys.getfilesystemencoding()) + except UnicodeEncodeError: + self.skipTest("cannot encode %a for the filesystem" % self.fname) + self.check_stat_attributes(fname) def test_statvfs_attributes(self): if not hasattr(os, "statvfs"): @@ -257,7 +216,7 @@ class StatAttributeTests(unittest.TestCase): try: result = os.statvfs(self.fname) - except OSError, e: + except OSError as e: # On AtheOS, glibc always returns ENOSYS if e.errno == errno.ENOSYS: return @@ -275,7 +234,7 @@ class StatAttributeTests(unittest.TestCase): try: result.f_bfree = 1 self.fail("No exception raised") - except TypeError: + except AttributeError: pass try: @@ -299,11 +258,11 @@ class StatAttributeTests(unittest.TestCase): def test_utime_dir(self): delta = 1000000 - st = os.stat(test_support.TESTFN) + st = os.stat(support.TESTFN) # round to int, because some systems may support sub-second # time stamps in stat, but not in utime. - os.utime(test_support.TESTFN, (st.st_atime, int(st.st_mtime-delta))) - st2 = os.stat(test_support.TESTFN) + os.utime(support.TESTFN, (st.st_atime, int(st.st_mtime-delta))) + st2 = os.stat(support.TESTFN) self.assertEqual(st2.st_mtime, int(st.st_mtime-delta)) # Restrict test to Win32, since there is no guarantee other @@ -313,11 +272,11 @@ class StatAttributeTests(unittest.TestCase): root = os.path.splitdrive(os.path.abspath(path))[0] + '\\' import ctypes kernel32 = ctypes.windll.kernel32 - buf = ctypes.create_string_buffer("", 100) - if kernel32.GetVolumeInformationA(root, None, 0, None, None, None, buf, len(buf)): + buf = ctypes.create_unicode_buffer("", 100) + if kernel32.GetVolumeInformationW(root, None, 0, None, None, None, buf, len(buf)): return buf.value - if get_file_system(test_support.TESTFN) == "NTFS": + if get_file_system(support.TESTFN) == "NTFS": def test_1565150(self): t1 = 1159195039.25 os.utime(self.fname, (t1, t1)) @@ -332,7 +291,7 @@ class StatAttributeTests(unittest.TestCase): # Verify that an open file can be stat'ed try: os.stat(r"c:\pagefile.sys") - except WindowsError, e: + except WindowsError as e: if e.errno == 2: # file does not exist; cannot run test return self.fail("Could not stat pagefile.sys") @@ -342,26 +301,130 @@ from test import mapping_tests class EnvironTests(mapping_tests.BasicTestMappingProtocol): """check that os.environ object conform to mapping protocol""" type2test = None - def _reference(self): - return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"} - def _empty_mapping(self): - os.environ.clear() - return os.environ + def setUp(self): self.__save = dict(os.environ) - os.environ.clear() + if os.supports_bytes_environ: + self.__saveb = dict(os.environb) + for key, value in self._reference().items(): + os.environ[key] = value + def tearDown(self): os.environ.clear() os.environ.update(self.__save) + if os.supports_bytes_environ: + os.environb.clear() + os.environb.update(self.__saveb) + + def _reference(self): + return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"} + + def _empty_mapping(self): + os.environ.clear() + return os.environ # Bug 1110478 def test_update2(self): + os.environ.clear() if os.path.exists("/bin/sh"): os.environ.update(HELLO="World") with os.popen("/bin/sh -c 'echo $HELLO'") as popen: value = popen.read().strip() self.assertEqual(value, "World") + def test_os_popen_iter(self): + if os.path.exists("/bin/sh"): + with os.popen( + "/bin/sh -c 'echo \"line1\nline2\nline3\"'") as popen: + it = iter(popen) + self.assertEqual(next(it), "line1\n") + self.assertEqual(next(it), "line2\n") + self.assertEqual(next(it), "line3\n") + self.assertRaises(StopIteration, next, it) + + # Verify environ keys and values from the OS are of the + # correct str type. + def test_keyvalue_types(self): + for key, val in os.environ.items(): + self.assertEqual(type(key), str) + self.assertEqual(type(val), str) + + def test_items(self): + for key, value in self._reference().items(): + self.assertEqual(os.environ.get(key), value) + + # Issue 7310 + def test___repr__(self): + """Check that the repr() of os.environ looks like environ({...}).""" + env = os.environ + self.assertEqual(repr(env), 'environ({{{}}})'.format(', '.join( + '{!r}: {!r}'.format(key, value) + for key, value in env.items()))) + + def test_get_exec_path(self): + defpath_list = os.defpath.split(os.pathsep) + test_path = ['/monty', '/python', '', '/flying/circus'] + test_env = {'PATH': os.pathsep.join(test_path)} + + saved_environ = os.environ + try: + os.environ = dict(test_env) + # Test that defaulting to os.environ works. + self.assertSequenceEqual(test_path, os.get_exec_path()) + self.assertSequenceEqual(test_path, os.get_exec_path(env=None)) + finally: + os.environ = saved_environ + + # No PATH environment variable + self.assertSequenceEqual(defpath_list, os.get_exec_path({})) + # Empty PATH environment variable + self.assertSequenceEqual(('',), os.get_exec_path({'PATH':''})) + # Supplied PATH environment variable + self.assertSequenceEqual(test_path, os.get_exec_path(test_env)) + + if os.supports_bytes_environ: + # env cannot contain 'PATH' and b'PATH' keys + try: + # ignore BytesWarning warning + with warnings.catch_warnings(record=True): + mixed_env = {'PATH': '1', b'PATH': b'2'} + except BytesWarning: + # mixed_env cannot be created with python -bb + pass + else: + self.assertRaises(ValueError, os.get_exec_path, mixed_env) + + # bytes key and/or value + self.assertSequenceEqual(os.get_exec_path({b'PATH': b'abc'}), + ['abc']) + self.assertSequenceEqual(os.get_exec_path({b'PATH': 'abc'}), + ['abc']) + self.assertSequenceEqual(os.get_exec_path({'PATH': b'abc'}), + ['abc']) + + @unittest.skipUnless(os.supports_bytes_environ, + "os.environb required for this test.") + def test_environb(self): + # os.environ -> os.environb + value = 'euro\u20ac' + try: + value_bytes = value.encode(sys.getfilesystemencoding(), + 'surrogateescape') + except UnicodeEncodeError: + msg = "U+20AC character is not encodable to %s" % ( + sys.getfilesystemencoding(),) + self.skipTest(msg) + os.environ['unicode'] = value + self.assertEqual(os.environ['unicode'], value) + self.assertEqual(os.environb[b'unicode'], value_bytes) + + # os.environb -> os.environ + value = b'\xff' + os.environb[b'bytes'] = value + self.assertEqual(os.environb[b'bytes'], value) + value_str = value.decode(sys.getfilesystemencoding(), 'surrogateescape') + self.assertEqual(os.environ['bytes'], value_str) + # On FreeBSD < 7 and OS X < 10.6, unsetenv() doesn't return a value (issue # #13415). @unittest.skipIf(sys.platform.startswith(('freebsd', 'darwin')), @@ -395,7 +458,7 @@ class WalkTests(unittest.TestCase): # link/ a symlink to TESTFN.2 # TEST2/ # tmp4 a lone file - walk_path = join(test_support.TESTFN, "TEST1") + walk_path = join(support.TESTFN, "TEST1") sub1_path = join(walk_path, "SUB1") sub11_path = join(sub1_path, "SUB11") sub2_path = join(walk_path, "SUB2") @@ -403,19 +466,24 @@ class WalkTests(unittest.TestCase): tmp2_path = join(sub1_path, "tmp2") tmp3_path = join(sub2_path, "tmp3") link_path = join(sub2_path, "link") - t2_path = join(test_support.TESTFN, "TEST2") - tmp4_path = join(test_support.TESTFN, "TEST2", "tmp4") + t2_path = join(support.TESTFN, "TEST2") + tmp4_path = join(support.TESTFN, "TEST2", "tmp4") # Create stuff. os.makedirs(sub11_path) os.makedirs(sub2_path) os.makedirs(t2_path) for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path: - f = file(path, "w") + f = open(path, "w") f.write("I'm " + path + " and proud of it. Blame test_os.\n") f.close() - if hasattr(os, "symlink"): - os.symlink(os.path.abspath(t2_path), link_path) + if support.can_symlink(): + if os.name == 'nt': + def symlink_to_dir(src, dest): + os.symlink(src, dest, True) + else: + symlink_to_dir = os.symlink + symlink_to_dir(os.path.abspath(t2_path), link_path) sub2_tree = (sub2_path, ["link"], ["tmp3"]) else: sub2_tree = (sub2_path, [], ["tmp3"]) @@ -458,7 +526,7 @@ class WalkTests(unittest.TestCase): self.assertEqual(all[flipped + 1], (sub1_path, ["SUB11"], ["tmp2"])) self.assertEqual(all[2 - 2 * flipped], sub2_tree) - if hasattr(os, "symlink"): + if support.can_symlink(): # Walk, following symlinks. for root, dirs, files in os.walk(walk_path, followlinks=True): if root == link_path: @@ -473,7 +541,7 @@ class WalkTests(unittest.TestCase): # Windows, which doesn't have a recursive delete command. The # (not so) subtlety is that rmdir will fail unless the dir's # kids are removed first, so bottom up is essential. - for root, dirs, files in os.walk(test_support.TESTFN, topdown=False): + for root, dirs, files in os.walk(support.TESTFN, topdown=False): for name in files: os.remove(os.path.join(root, name)) for name in dirs: @@ -482,14 +550,14 @@ class WalkTests(unittest.TestCase): os.rmdir(dirname) else: os.remove(dirname) - os.rmdir(test_support.TESTFN) + os.rmdir(support.TESTFN) -class MakedirTests (unittest.TestCase): +class MakedirTests(unittest.TestCase): def setUp(self): - os.mkdir(test_support.TESTFN) + os.mkdir(support.TESTFN) def test_makedir(self): - base = test_support.TESTFN + base = support.TESTFN path = os.path.join(base, 'dir1', 'dir2', 'dir3') os.makedirs(path) # Should work path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4') @@ -503,31 +571,123 @@ class MakedirTests (unittest.TestCase): 'dir5', 'dir6') os.makedirs(path) + def test_exist_ok_existing_directory(self): + path = os.path.join(support.TESTFN, 'dir1') + mode = 0o777 + old_mask = os.umask(0o022) + try: + os.makedirs(path, mode) + self.assertRaises(OSError, os.makedirs, path, mode) + self.assertRaises(OSError, os.makedirs, path, mode, exist_ok=False) + self.assertRaises(OSError, os.makedirs, path, 0o776, exist_ok=True) + os.makedirs(path, mode=mode, exist_ok=True) + finally: + os.umask(old_mask) + def test_exist_ok_s_isgid_directory(self): + path = os.path.join(support.TESTFN, 'dir1') + S_ISGID = stat.S_ISGID + mode = 0o777 + old_mask = os.umask(0o022) + try: + existing_testfn_mode = stat.S_IMODE( + os.lstat(support.TESTFN).st_mode) + try: + os.chmod(support.TESTFN, existing_testfn_mode | S_ISGID) + except OSError: + raise unittest.SkipTest('Cannot set S_ISGID for dir.') + if (os.lstat(support.TESTFN).st_mode & S_ISGID != S_ISGID): + raise unittest.SkipTest('No support for S_ISGID dir mode.') + # The os should apply S_ISGID from the parent dir for us, but + # this test need not depend on that behavior. Be explicit. + os.makedirs(path, mode | S_ISGID) + # http://bugs.python.org/issue14992 + # Should not fail when the bit is already set. + os.makedirs(path, mode, exist_ok=True) + # remove the bit. + os.chmod(path, stat.S_IMODE(os.lstat(path).st_mode) & ~S_ISGID) + with self.assertRaises(OSError): + # Should fail when the bit is not already set when demanded. + os.makedirs(path, mode | S_ISGID, exist_ok=True) + finally: + os.umask(old_mask) + def test_exist_ok_existing_regular_file(self): + base = support.TESTFN + path = os.path.join(support.TESTFN, 'dir1') + f = open(path, 'w') + f.write('abc') + f.close() + self.assertRaises(OSError, os.makedirs, path) + self.assertRaises(OSError, os.makedirs, path, exist_ok=False) + self.assertRaises(OSError, os.makedirs, path, exist_ok=True) + os.remove(path) def tearDown(self): - path = os.path.join(test_support.TESTFN, 'dir1', 'dir2', 'dir3', + path = os.path.join(support.TESTFN, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', 'dir6') # If the tests failed, the bottom-most directory ('../dir6') # may not have been created, so we look for the outermost directory # that exists. - while not os.path.exists(path) and path != test_support.TESTFN: + while not os.path.exists(path) and path != support.TESTFN: path = os.path.dirname(path) os.removedirs(path) -class DevNullTests (unittest.TestCase): + +class RemoveDirsTests(unittest.TestCase): + def setUp(self): + os.makedirs(support.TESTFN) + + def tearDown(self): + support.rmtree(support.TESTFN) + + def test_remove_all(self): + dira = os.path.join(support.TESTFN, 'dira') + os.mkdir(dira) + dirb = os.path.join(dira, 'dirb') + os.mkdir(dirb) + os.removedirs(dirb) + self.assertFalse(os.path.exists(dirb)) + self.assertFalse(os.path.exists(dira)) + self.assertFalse(os.path.exists(support.TESTFN)) + + def test_remove_partial(self): + dira = os.path.join(support.TESTFN, 'dira') + os.mkdir(dira) + dirb = os.path.join(dira, 'dirb') + os.mkdir(dirb) + with open(os.path.join(dira, 'file.txt'), 'w') as f: + f.write('text') + os.removedirs(dirb) + self.assertFalse(os.path.exists(dirb)) + self.assertTrue(os.path.exists(dira)) + self.assertTrue(os.path.exists(support.TESTFN)) + + def test_remove_nothing(self): + dira = os.path.join(support.TESTFN, 'dira') + os.mkdir(dira) + dirb = os.path.join(dira, 'dirb') + os.mkdir(dirb) + with open(os.path.join(dirb, 'file.txt'), 'w') as f: + f.write('text') + with self.assertRaises(OSError): + os.removedirs(dirb) + self.assertTrue(os.path.exists(dirb)) + self.assertTrue(os.path.exists(dira)) + self.assertTrue(os.path.exists(support.TESTFN)) + + +class DevNullTests(unittest.TestCase): def test_devnull(self): - f = file(os.devnull, 'w') - f.write('hello') - f.close() - f = file(os.devnull, 'r') - self.assertEqual(f.read(), '') - f.close() + with open(os.devnull, 'wb') as f: + f.write(b'hello') + f.close() + with open(os.devnull, 'rb') as f: + self.assertEqual(f.read(), b'') -class URandomTests (unittest.TestCase): +class URandomTests(unittest.TestCase): def test_urandom_length(self): self.assertEqual(len(os.urandom(0)), 0) self.assertEqual(len(os.urandom(1)), 1) @@ -541,57 +701,145 @@ class URandomTests (unittest.TestCase): self.assertNotEqual(data1, data2) def get_urandom_subprocess(self, count): - # We need to use repr() and eval() to avoid line ending conversions - # under Windows. code = '\n'.join(( 'import os, sys', 'data = os.urandom(%s)' % count, - 'sys.stdout.write(repr(data))', - 'sys.stdout.flush()', - 'print >> sys.stderr, (len(data), data)')) - cmd_line = [sys.executable, '-c', code] - p = subprocess.Popen(cmd_line, stdin=subprocess.PIPE, - stdout=subprocess.PIPE, stderr=subprocess.PIPE) - out, err = p.communicate() - self.assertEqual(p.wait(), 0, (p.wait(), err)) - out = eval(out) - self.assertEqual(len(out), count, err) - return out + 'sys.stdout.buffer.write(data)', + 'sys.stdout.buffer.flush()')) + out = assert_python_ok('-c', code) + stdout = out[1] + self.assertEqual(len(stdout), 16) + return stdout def test_urandom_subprocess(self): data1 = self.get_urandom_subprocess(16) data2 = self.get_urandom_subprocess(16) self.assertNotEqual(data1, data2) +@contextlib.contextmanager +def _execvpe_mockup(defpath=None): + """ + Stubs out execv and execve functions when used as context manager. + Records exec calls. The mock execv and execve functions always raise an + exception as they would normally never return. + """ + # A list of tuples containing (function name, first arg, args) + # of calls to execv or execve that have been made. + calls = [] + + def mock_execv(name, *args): + calls.append(('execv', name, args)) + raise RuntimeError("execv called") + + def mock_execve(name, *args): + calls.append(('execve', name, args)) + raise OSError(errno.ENOTDIR, "execve called") + + try: + orig_execv = os.execv + orig_execve = os.execve + orig_defpath = os.defpath + os.execv = mock_execv + os.execve = mock_execve + if defpath is not None: + os.defpath = defpath + yield calls + finally: + os.execv = orig_execv + os.execve = orig_execve + os.defpath = orig_defpath + +class ExecTests(unittest.TestCase): + @unittest.skipIf(USING_LINUXTHREADS, + "avoid triggering a linuxthreads bug: see issue #4970") + def test_execvpe_with_bad_program(self): + self.assertRaises(OSError, os.execvpe, 'no such app-', + ['no such app-'], None) + def test_execvpe_with_bad_arglist(self): self.assertRaises(ValueError, os.execvpe, 'notepad', [], None) + @unittest.skipUnless(hasattr(os, '_execvpe'), + "No internal os._execvpe function to test.") + def _test_internal_execvpe(self, test_type): + program_path = os.sep + 'absolutepath' + if test_type is bytes: + program = b'executable' + fullpath = os.path.join(os.fsencode(program_path), program) + native_fullpath = fullpath + arguments = [b'progname', 'arg1', 'arg2'] + else: + program = 'executable' + arguments = ['progname', 'arg1', 'arg2'] + fullpath = os.path.join(program_path, program) + if os.name != "nt": + native_fullpath = os.fsencode(fullpath) + else: + native_fullpath = fullpath + env = {'spam': 'beans'} + + # test os._execvpe() with an absolute path + with _execvpe_mockup() as calls: + self.assertRaises(RuntimeError, + os._execvpe, fullpath, arguments) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0], ('execv', fullpath, (arguments,))) + + # test os._execvpe() with a relative path: + # os.get_exec_path() returns defpath + with _execvpe_mockup(defpath=program_path) as calls: + self.assertRaises(OSError, + os._execvpe, program, arguments, env=env) + self.assertEqual(len(calls), 1) + self.assertSequenceEqual(calls[0], + ('execve', native_fullpath, (arguments, env))) + + # test os._execvpe() with a relative path: + # os.get_exec_path() reads the 'PATH' variable + with _execvpe_mockup() as calls: + env_path = env.copy() + if test_type is bytes: + env_path[b'PATH'] = program_path + else: + env_path['PATH'] = program_path + self.assertRaises(OSError, + os._execvpe, program, arguments, env=env_path) + self.assertEqual(len(calls), 1) + self.assertSequenceEqual(calls[0], + ('execve', native_fullpath, (arguments, env_path))) + + def test_internal_execvpe_str(self): + self._test_internal_execvpe(str) + if os.name != "nt": + self._test_internal_execvpe(bytes) + + class Win32ErrorTests(unittest.TestCase): def test_rename(self): - self.assertRaises(WindowsError, os.rename, test_support.TESTFN, test_support.TESTFN+".bak") + self.assertRaises(WindowsError, os.rename, support.TESTFN, support.TESTFN+".bak") def test_remove(self): - self.assertRaises(WindowsError, os.remove, test_support.TESTFN) + self.assertRaises(WindowsError, os.remove, support.TESTFN) def test_chdir(self): - self.assertRaises(WindowsError, os.chdir, test_support.TESTFN) + self.assertRaises(WindowsError, os.chdir, support.TESTFN) def test_mkdir(self): - f = open(test_support.TESTFN, "w") + f = open(support.TESTFN, "w") try: - self.assertRaises(WindowsError, os.mkdir, test_support.TESTFN) + self.assertRaises(WindowsError, os.mkdir, support.TESTFN) finally: f.close() - os.unlink(test_support.TESTFN) + os.unlink(support.TESTFN) def test_utime(self): - self.assertRaises(WindowsError, os.utime, test_support.TESTFN, None) + self.assertRaises(WindowsError, os.utime, support.TESTFN, None) def test_chmod(self): - self.assertRaises(WindowsError, os.chmod, test_support.TESTFN, 0) + self.assertRaises(WindowsError, os.chmod, support.TESTFN, 0) class TestInvalidFD(unittest.TestCase): - singles = ["fchdir", "fdopen", "dup", "fdatasync", "fstat", + singles = ["fchdir", "dup", "fdopen", "fdatasync", "fstat", "fstatvfs", "fsync", "tcgetpgrp", "ttyname"] #singles.append("close") #We omit close because it doesn'r raise an exception on some platforms @@ -605,7 +853,7 @@ class TestInvalidFD(unittest.TestCase): def check(self, f, *args): try: - f(test_support.make_bad_fd(), *args) + f(support.make_bad_fd(), *args) except OSError as e: self.assertEqual(e.errno, errno.EBADF) else: @@ -614,11 +862,11 @@ class TestInvalidFD(unittest.TestCase): def test_isatty(self): if hasattr(os, "isatty"): - self.assertEqual(os.isatty(test_support.make_bad_fd()), False) + self.assertEqual(os.isatty(support.make_bad_fd()), False) def test_closerange(self): if hasattr(os, "closerange"): - fd = test_support.make_bad_fd() + fd = support.make_bad_fd() # Make sure none of the descriptors we are about to close are # currently valid (issue 6542). for i in range(10): @@ -666,7 +914,43 @@ class TestInvalidFD(unittest.TestCase): def test_write(self): if hasattr(os, "write"): - self.check(os.write, " ") + self.check(os.write, b" ") + + +class LinkTests(unittest.TestCase): + def setUp(self): + self.file1 = support.TESTFN + self.file2 = os.path.join(support.TESTFN + "2") + + def tearDown(self): + for file in (self.file1, self.file2): + if os.path.exists(file): + os.unlink(file) + + def _test_link(self, file1, file2): + with open(file1, "w") as f1: + f1.write("test") + + os.link(file1, file2) + with open(file1, "r") as f1, open(file2, "r") as f2: + self.assertTrue(os.path.sameopenfile(f1.fileno(), f2.fileno())) + + def test_link(self): + self._test_link(self.file1, self.file2) + + def test_link_bytes(self): + self._test_link(bytes(self.file1, sys.getfilesystemencoding()), + bytes(self.file2, sys.getfilesystemencoding())) + + def test_unicode_name(self): + try: + os.fsencode("\xf1") + except UnicodeError: + raise unittest.SkipTest("Unable to encode for this platform.") + + self.file1 += "\xf1" + self.file2 = self.file1 + "2" + self._test_link(self.file1, self.file2) if sys.platform != 'win32': class Win32ErrorTests(unittest.TestCase): @@ -724,9 +1008,63 @@ if sys.platform != 'win32': subprocess.check_call([ sys.executable, '-c', 'import os,sys;os.setregid(-1,-1);sys.exit(0)']) + + class Pep383Tests(unittest.TestCase): + def setUp(self): + if support.TESTFN_UNENCODABLE: + self.dir = support.TESTFN_UNENCODABLE + else: + self.dir = support.TESTFN + self.bdir = os.fsencode(self.dir) + + bytesfn = [] + def add_filename(fn): + try: + fn = os.fsencode(fn) + except UnicodeEncodeError: + return + bytesfn.append(fn) + add_filename(support.TESTFN_UNICODE) + if support.TESTFN_UNENCODABLE: + add_filename(support.TESTFN_UNENCODABLE) + if not bytesfn: + self.skipTest("couldn't create any non-ascii filename") + + self.unicodefn = set() + os.mkdir(self.dir) + try: + for fn in bytesfn: + f = open(os.path.join(self.bdir, fn), "w") + f.close() + fn = os.fsdecode(fn) + if fn in self.unicodefn: + raise ValueError("duplicate filename") + self.unicodefn.add(fn) + except: + shutil.rmtree(self.dir) + raise + + def tearDown(self): + shutil.rmtree(self.dir) + + def test_listdir(self): + expected = self.unicodefn + found = set(os.listdir(self.dir)) + self.assertEqual(found, expected) + + def test_open(self): + for fn in self.unicodefn: + f = open(os.path.join(self.dir, fn), 'rb') + f.close() + + def test_stat(self): + for fn in self.unicodefn: + os.stat(os.path.join(self.dir, fn)) else: class PosixUidGidTests(unittest.TestCase): pass + class Pep383Tests(unittest.TestCase): + pass @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") class Win32KillTests(unittest.TestCase): @@ -774,7 +1112,7 @@ class Win32KillTests(unittest.TestCase): buf, ctypes.sizeof(buf), None, None, None) self.assertNotEqual(rslt, 0, "PeekNamedPipe failed") if buf.value: - self.assertEqual(msg, buf.value) + self.assertEqual(msg, buf.value.decode()) break time.sleep(0.1) count += 1 @@ -795,20 +1133,22 @@ class Win32KillTests(unittest.TestCase): def _kill_with_event(self, event, name): tagname = "test_os_%s" % uuid.uuid1() m = mmap.mmap(-1, 1, tagname) - m[0] = '0' + m[0] = 0 # Run a script which has console control handling enabled. proc = subprocess.Popen([sys.executable, os.path.join(os.path.dirname(__file__), "win_console_handler.py"), tagname], creationflags=subprocess.CREATE_NEW_PROCESS_GROUP) # Let the interpreter startup before we send signals. See #3137. - count, max = 0, 20 + count, max = 0, 100 while count < max and proc.poll() is None: - if m[0] == '1': + if m[0] == 1: break - time.sleep(0.5) + time.sleep(0.1) count += 1 else: + # Forcefully kill the process if we weren't able to signal it. + os.kill(proc.pid, signal.SIGINT) self.fail("Subprocess didn't finish initialization") os.kill(proc.pid, event) # proc.send_signal(event) could also be done here. @@ -842,20 +1182,180 @@ class Win32KillTests(unittest.TestCase): self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT") +@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") +@support.skip_unless_symlink +class Win32SymlinkTests(unittest.TestCase): + filelink = 'filelinktest' + filelink_target = os.path.abspath(__file__) + dirlink = 'dirlinktest' + dirlink_target = os.path.dirname(filelink_target) + missing_link = 'missing link' + + def setUp(self): + assert os.path.exists(self.dirlink_target) + assert os.path.exists(self.filelink_target) + assert not os.path.exists(self.dirlink) + assert not os.path.exists(self.filelink) + assert not os.path.exists(self.missing_link) + + def tearDown(self): + if os.path.exists(self.filelink): + os.remove(self.filelink) + if os.path.exists(self.dirlink): + os.rmdir(self.dirlink) + if os.path.lexists(self.missing_link): + os.remove(self.missing_link) + + def test_directory_link(self): + os.symlink(self.dirlink_target, self.dirlink, True) + self.assertTrue(os.path.exists(self.dirlink)) + self.assertTrue(os.path.isdir(self.dirlink)) + self.assertTrue(os.path.islink(self.dirlink)) + self.check_stat(self.dirlink, self.dirlink_target) + + def test_file_link(self): + os.symlink(self.filelink_target, self.filelink) + self.assertTrue(os.path.exists(self.filelink)) + self.assertTrue(os.path.isfile(self.filelink)) + self.assertTrue(os.path.islink(self.filelink)) + self.check_stat(self.filelink, self.filelink_target) + + def _create_missing_dir_link(self): + 'Create a "directory" link to a non-existent target' + linkname = self.missing_link + if os.path.lexists(linkname): + os.remove(linkname) + target = r'c:\\target does not exist.29r3c740' + assert not os.path.exists(target) + target_is_dir = True + os.symlink(target, linkname, target_is_dir) + + def test_remove_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # For compatibility with Unix, os.remove will check the + # directory status and call RemoveDirectory if the symlink + # was created with target_is_dir==True. + os.remove(self.missing_link) + + @unittest.skip("currently fails; consider for improvement") + def test_isdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider having isdir return true for directory links + self.assertTrue(os.path.isdir(self.missing_link)) + + @unittest.skip("currently fails; consider for improvement") + def test_rmdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider allowing rmdir to remove directory links + os.rmdir(self.missing_link) + + def check_stat(self, link, target): + self.assertEqual(os.stat(link), os.stat(target)) + self.assertNotEqual(os.lstat(link), os.stat(link)) + + bytes_link = os.fsencode(link) + self.assertEqual(os.stat(bytes_link), os.stat(target)) + self.assertNotEqual(os.lstat(bytes_link), os.stat(bytes_link)) + + def test_12084(self): + level1 = os.path.abspath(support.TESTFN) + level2 = os.path.join(level1, "level2") + level3 = os.path.join(level2, "level3") + try: + os.mkdir(level1) + os.mkdir(level2) + os.mkdir(level3) + + file1 = os.path.abspath(os.path.join(level1, "file1")) + + with open(file1, "w") as f: + f.write("file1") + + orig_dir = os.getcwd() + try: + os.chdir(level2) + link = os.path.join(level2, "link") + os.symlink(os.path.relpath(file1), "link") + self.assertIn("link", os.listdir(os.getcwd())) + + # Check os.stat calls from the same dir as the link + self.assertEqual(os.stat(file1), os.stat("link")) + + # Check os.stat calls from a dir below the link + os.chdir(level1) + self.assertEqual(os.stat(file1), + os.stat(os.path.relpath(link))) + + # Check os.stat calls from a dir above the link + os.chdir(level3) + self.assertEqual(os.stat(file1), + os.stat(os.path.relpath(link))) + finally: + os.chdir(orig_dir) + except OSError as err: + self.fail(err) + finally: + os.remove(file1) + shutil.rmtree(level1) + + +class FSEncodingTests(unittest.TestCase): + def test_nop(self): + self.assertEqual(os.fsencode(b'abc\xff'), b'abc\xff') + self.assertEqual(os.fsdecode('abc\u0141'), 'abc\u0141') + + def test_identity(self): + # assert fsdecode(fsencode(x)) == x + for fn in ('unicode\u0141', 'latin\xe9', 'ascii'): + try: + bytesfn = os.fsencode(fn) + except UnicodeEncodeError: + continue + self.assertEqual(os.fsdecode(bytesfn), fn) + + +class PidTests(unittest.TestCase): + @unittest.skipUnless(hasattr(os, 'getppid'), "test needs os.getppid") + def test_getppid(self): + p = subprocess.Popen([sys.executable, '-c', + 'import os; print(os.getppid())'], + stdout=subprocess.PIPE) + stdout, _ = p.communicate() + # We are the parent of our subprocess + self.assertEqual(int(stdout), os.getpid()) + + +# The introduction of this TestCase caused at least two different errors on +# *nix buildbots. Temporarily skip this to let the buildbots move along. +@unittest.skip("Skip due to platform/environment differences on *NIX buildbots") +@unittest.skipUnless(hasattr(os, 'getlogin'), "test needs os.getlogin") +class LoginTests(unittest.TestCase): + def test_getlogin(self): + user_name = os.getlogin() + self.assertNotEqual(len(user_name), 0) + + def test_main(): - test_support.run_unittest( + support.run_unittest( FileTests, - TemporaryFileTests, StatAttributeTests, EnvironTests, WalkTests, MakedirTests, DevNullTests, URandomTests, + ExecTests, Win32ErrorTests, TestInvalidFD, PosixUidGidTests, - Win32KillTests + Pep383Tests, + Win32KillTests, + Win32SymlinkTests, + FSEncodingTests, + PidTests, + LoginTests, + LinkTests, + RemoveDirsTests, ) if __name__ == "__main__": |