diff options
| -rw-r--r-- | Doc/library/os.path.rst | 12 | ||||
| -rw-r--r-- | Doc/library/os.rst | 24 | ||||
| -rw-r--r-- | Lib/ntpath.py | 43 | ||||
| -rw-r--r-- | Lib/tarfile.py | 12 | ||||
| -rw-r--r-- | Lib/test/support.py | 25 | ||||
| -rw-r--r-- | Lib/test/symlink_support.py | 203 | ||||
| -rw-r--r-- | Lib/test/test_glob.py | 14 | ||||
| -rw-r--r-- | Lib/test/test_httpservers.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_os.py | 82 | ||||
| -rw-r--r-- | Lib/test/test_platform.py | 34 | ||||
| -rw-r--r-- | Lib/test/test_posixpath.py | 345 | ||||
| -rw-r--r-- | Lib/test/test_shutil.py | 57 | ||||
| -rw-r--r-- | Lib/test/test_sys.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_sysconfig.py | 14 | ||||
| -rw-r--r-- | Lib/test/test_tarfile.py | 29 | ||||
| -rw-r--r-- | Misc/ACKS | 1 | ||||
| -rw-r--r-- | Misc/NEWS | 3 | ||||
| -rw-r--r-- | Modules/posixmodule.c | 546 |
18 files changed, 1167 insertions, 281 deletions
diff --git a/Doc/library/os.path.rst b/Doc/library/os.path.rst index c838983..7fcb2a9 100644 --- a/Doc/library/os.path.rst +++ b/Doc/library/os.path.rst @@ -231,11 +231,15 @@ applications should use string objects to access all files. .. function:: samefile(path1, path2) - Return ``True`` if both pathname arguments refer to the same file or directory - (as indicated by device number and i-node number). Raise an exception if a - :func:`os.stat` call on either pathname fails. + Return ``True`` if both pathname arguments refer to the same file or directory. + On Unix, this is determined by the device number and i-node number and raises an + exception if a :func:`os.stat` call on either pathname fails. - Availability: Unix. + On Windows, two files are the same if they resolve to the same final path + name using the Windows API call GetFinalPathNameByHandle and this function + raises an exception if handles cannot be obtained to either file. + + Availability: Windows, Unix. .. function:: sameopenfile(fp1, fp2) diff --git a/Doc/library/os.rst b/Doc/library/os.rst index b74350d..d8835f6 100644 --- a/Doc/library/os.rst +++ b/Doc/library/os.rst @@ -1065,7 +1065,7 @@ Files and Directories Like :func:`stat`, but do not follow symbolic links. This is an alias for :func:`stat` on platforms that do not support symbolic links, such as - Windows. + Windows prior to 6.0 (Vista). .. function:: mkfifo(path[, mode]) @@ -1181,7 +1181,7 @@ Files and Directories and the call may raise an UnicodeDecodeError. If the *path* is a bytes object, the result will be a bytes object. - Availability: Unix. + Availability: Unix, Windows. .. function:: remove(path) @@ -1341,9 +1341,25 @@ Files and Directories .. function:: symlink(source, link_name) - Create a symbolic link pointing to *source* named *link_name*. + Create a symbolic link pointing to *source* named *link_name*. On Windows, + symlink version takes an additional, optional parameter, + *target_is_directory*, which defaults to False. - Availability: Unix. + symlink(source, link_name, target_is_directory=False) + + On Windows, a symlink represents a file or a directory, and does not + morph to the target dynamically. For this reason, when creating a + symlink on Windows, if the target is not already present, the symlink + will default to being a file symlink. If *target_is_directory* is set to + True, the symlink will be created as a directory symlink. This + parameter is ignored if the target exists (and the symlink is created + with the same type as the target). + + Symbolic link support was introduced in Windows 6.0 (Vista). *symlink* + will raise a NotImplementedError on Windows versions earlier than 6.0. The + SeCreateSymbolicLinkPrivilege is required in order to create symlinks. + + Availability: Unix, Windows 6.0. .. function:: unlink(path) diff --git a/Lib/ntpath.py b/Lib/ntpath.py index 7972aa9..ab0b318 100644 --- a/Lib/ntpath.py +++ b/Lib/ntpath.py @@ -16,7 +16,8 @@ __all__ = ["normcase","isabs","join","splitdrive","split","splitext", "getatime","getctime", "islink","exists","lexists","isdir","isfile", "ismount", "expanduser","expandvars","normpath","abspath", "splitunc","curdir","pardir","sep","pathsep","defpath","altsep", - "extsep","devnull","realpath","supports_unicode_filenames","relpath"] + "extsep","devnull","realpath","supports_unicode_filenames","relpath", + "samefile",] # strings representing various path-related bits and pieces # These are primarily for export; internally, they are hardcoded. @@ -309,16 +310,28 @@ def dirname(p): return split(p)[0] # Is a path a symbolic link? -# This will always return false on systems where posix.lstat doesn't exist. +# This will always return false on systems where os.lstat doesn't exist. def islink(path): - """Test for symbolic link. - On WindowsNT/95 and OS/2 always returns false + """Test whether a path is a symbolic link. + This will always return false for Windows prior to 6.0 + and for OS/2. """ - return False - -# alias exists to lexists -lexists = exists + try: + st = os.lstat(path) + except (os.error, AttributeError): + return False + return stat.S_ISLNK(st.st_mode) + +# Being true for dangling symbolic links is also useful. + +def lexists(path): + """Test whether a path exists. Returns True for broken symbolic links""" + try: + st = os.lstat(path) + except (os.error, WindowsError): + return False + return True # Is a path a mount point? Either a root (with or without drive letter) # or an UNC path with at most a / or \ after the mount point. @@ -612,3 +625,17 @@ def relpath(path, start=curdir): if not rel_list: return _get_dot(path) return join(*rel_list) + + +# determine if two files are in fact the same file +def samefile(f1, f2): + "Test whether two pathnames reference the same actual file" + try: + from nt import _getfinalpathname + return _getfinalpathname(f1) == _getfinalpathname(f2) + except (NotImplementedError, ImportError): + # On Windows XP and earlier, two files are the same if their + # absolute pathnames are the same. + # Also, on other operating systems, fake this method with a + # Windows-XP approximation. + return abspath(f1) == abspath(f2) diff --git a/Lib/tarfile.py b/Lib/tarfile.py index 4839eb1..aca934a 100644 --- a/Lib/tarfile.py +++ b/Lib/tarfile.py @@ -2273,7 +2273,7 @@ class TarFile(object): (platform limitation), we try to make a copy of the referenced file instead of a link. """ - if hasattr(os, "symlink") and hasattr(os, "link"): + try: # For systems that support symbolic and hard links. if tarinfo.issym(): os.symlink(tarinfo.linkname, targetpath) @@ -2282,7 +2282,15 @@ class TarFile(object): if os.path.exists(tarinfo._link_target): os.link(tarinfo._link_target, targetpath) else: - self._extract_member(self._find_link_target(tarinfo), targetpath) + self._extract_mem + except (AttributeError, NotImplementedError, WindowsError): + # AttributeError if no os.symlink + # NotImplementedError if on Windows XP + # WindowsError (1314) if the required privilege is not held by the client + if tarinfo.issym(): + linkpath = os.path.join(os.path.dirname(tarinfo.name),tarinfo.linkname) + else: + linkpath = tarinfo.linkname else: try: self._extract_member(self._find_link_target(tarinfo), targetpath) diff --git a/Lib/test/support.py b/Lib/test/support.py index 08105df..83d7ba8 100644 --- a/Lib/test/support.py +++ b/Lib/test/support.py @@ -17,6 +17,7 @@ import unittest import importlib import collections import re +import subprocess import imp import time try: @@ -38,8 +39,7 @@ __all__ = [ "set_memlimit", "bigmemtest", "bigaddrspacetest", "BasicTestRunner", "run_unittest", "run_doctest", "threading_setup", "threading_cleanup", "reap_children", "cpython_only", "check_impl_detail", "get_attribute", - "swap_item", "swap_attr", - ] + "swap_item", "swap_attr", "can_symlink", "skip_unless_symlink"] class Error(Exception): @@ -1169,6 +1169,27 @@ def reap_children(): except: break +try: + from .symlink_support import enable_symlink_privilege +except: + enable_symlink_privilege = lambda: True + +def can_symlink(): + """It's no longer sufficient to test for the presence of symlink in the + os module - on Windows XP and earlier, os.symlink exists but a + NotImplementedError is thrown. + """ + has_symlink = hasattr(os, 'symlink') + is_old_windows = sys.platform == "win32" and sys.getwindowsversion().major < 6 + has_privilege = False if is_old_windows else enable_symlink_privilege() + return has_symlink and (not is_old_windows) and has_privilege + +def skip_unless_symlink(test): + """Skip decorator for tests that require functional symlink""" + selector = can_symlink() + msg = "Requires functional symlink implementation" + return [unittest.skip(msg)(test), test][selector] + @contextlib.contextmanager def swap_attr(obj, attr, new_val): """Temporary swap out an attribute with a new object. diff --git a/Lib/test/symlink_support.py b/Lib/test/symlink_support.py new file mode 100644 index 0000000..eab7667 --- /dev/null +++ b/Lib/test/symlink_support.py @@ -0,0 +1,203 @@ +""" +A module built to test if the current process has the privilege to +create symlinks on Windows. +""" + +# allow script to run natively under python 2.6+ +from __future__ import print_function + +import ctypes +from ctypes import wintypes + +GetCurrentProcess = ctypes.windll.kernel32.GetCurrentProcess +GetCurrentProcess.restype = wintypes.HANDLE +OpenProcessToken = ctypes.windll.advapi32.OpenProcessToken +OpenProcessToken.argtypes = (wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)) +OpenProcessToken.restype = wintypes.BOOL + +class LUID(ctypes.Structure): + _fields_ = [ + ('low_part', wintypes.DWORD), + ('high_part', wintypes.LONG), + ] + + def __eq__(self, other): + return ( + self.high_part == other.high_part and + self.low_part == other.low_part + ) + + def __ne__(self, other): + return not (self==other) + +LookupPrivilegeValue = ctypes.windll.advapi32.LookupPrivilegeValueW +LookupPrivilegeValue.argtypes = ( + wintypes.LPWSTR, # system name + wintypes.LPWSTR, # name + ctypes.POINTER(LUID), + ) +LookupPrivilegeValue.restype = wintypes.BOOL + +class TOKEN_INFORMATION_CLASS: + TokenUser = 1 + TokenGroups = 2 + TokenPrivileges = 3 + # ... see http://msdn.microsoft.com/en-us/library/aa379626%28VS.85%29.aspx + +SE_PRIVILEGE_ENABLED_BY_DEFAULT = (0x00000001) +SE_PRIVILEGE_ENABLED = (0x00000002) +SE_PRIVILEGE_REMOVED = (0x00000004) +SE_PRIVILEGE_USED_FOR_ACCESS = (0x80000000) + +class LUID_AND_ATTRIBUTES(ctypes.Structure): + _fields_ = [ + ('LUID', LUID), + ('attributes', wintypes.DWORD), + ] + + def is_enabled(self): + return bool(self.attributes & SE_PRIVILEGE_ENABLED) + + def enable(self): + self.attributes |= SE_PRIVILEGE_ENABLED + + def get_name(self): + size = wintypes.DWORD(10240) + buf = ctypes.create_unicode_buffer(size.value) + res = LookupPrivilegeName(None, self.LUID, buf, size) + if res == 0: + raise RuntimeError + return buf[:size.value] + + def __str__(self): + name = self.name + fmt = ['{name}', '{name} (enabled)'][self.is_enabled()] + return fmt.format(**vars()) + +LookupPrivilegeName = ctypes.windll.advapi32.LookupPrivilegeNameW +LookupPrivilegeName.argtypes = ( + wintypes.LPWSTR, # lpSystemName + ctypes.POINTER(LUID), # lpLuid + wintypes.LPWSTR, # lpName + ctypes.POINTER(wintypes.DWORD), #cchName + ) +LookupPrivilegeName.restype = wintypes.BOOL + +class TOKEN_PRIVILEGES(ctypes.Structure): + _fields_ = [ + ('count', wintypes.DWORD), + ('privileges', LUID_AND_ATTRIBUTES*0), + ] + + def get_array(self): + array_type = LUID_AND_ATTRIBUTES*self.count + privileges = ctypes.cast(self.privileges, ctypes.POINTER(array_type)).contents + return privileges + + def __iter__(self): + return iter(self.get_array()) + +PTOKEN_PRIVILEGES = ctypes.POINTER(TOKEN_PRIVILEGES) + +GetTokenInformation = ctypes.windll.advapi32.GetTokenInformation +GetTokenInformation.argtypes = [ + wintypes.HANDLE, # TokenHandle + ctypes.c_uint, # TOKEN_INFORMATION_CLASS value + ctypes.c_void_p, # TokenInformation + wintypes.DWORD, # TokenInformationLength + ctypes.POINTER(wintypes.DWORD), # ReturnLength + ] +GetTokenInformation.restype = wintypes.BOOL + +# http://msdn.microsoft.com/en-us/library/aa375202%28VS.85%29.aspx +AdjustTokenPrivileges = ctypes.windll.advapi32.AdjustTokenPrivileges +AdjustTokenPrivileges.restype = wintypes.BOOL +AdjustTokenPrivileges.argtypes = [ + wintypes.HANDLE, # TokenHandle + wintypes.BOOL, # DisableAllPrivileges + PTOKEN_PRIVILEGES, # NewState (optional) + wintypes.DWORD, # BufferLength of PreviousState + PTOKEN_PRIVILEGES, # PreviousState (out, optional) + ctypes.POINTER(wintypes.DWORD), # ReturnLength + ] + +def get_process_token(): + "Get the current process token" + token = wintypes.HANDLE() + TOKEN_ALL_ACCESS = 0xf01ff + res = OpenProcessToken(GetCurrentProcess(), TOKEN_ALL_ACCESS, token) + if not res > 0: + raise RuntimeError("Couldn't get process token") + return token + +def get_symlink_luid(): + "Get the LUID for the SeCreateSymbolicLinkPrivilege" + symlink_luid = LUID() + res = LookupPrivilegeValue(None, "SeCreateSymbolicLinkPrivilege", symlink_luid) + if not res > 0: + raise RuntimeError("Couldn't lookup privilege value") + return symlink_luid + +def get_privilege_information(): + "Get all privileges associated with the current process." + # first call with zero length to determine what size buffer we need + + return_length = wintypes.DWORD() + params = [ + get_process_token(), + TOKEN_INFORMATION_CLASS.TokenPrivileges, + None, + 0, + return_length, + ] + + res = GetTokenInformation(*params) + + # assume we now have the necessary length in return_length + + buffer = ctypes.create_string_buffer(return_length.value) + params[2] = buffer + params[3] = return_length.value + + res = GetTokenInformation(*params) + assert res > 0, "Error in second GetTokenInformation (%d)" % res + + privileges = ctypes.cast(buffer, ctypes.POINTER(TOKEN_PRIVILEGES)).contents + return privileges + +def report_privilege_information(): + "Report all privilege information assigned to the current process." + privileges = get_privilege_information() + print("found {0} privileges".format(privileges.count)) + tuple(map(print, privileges)) + +def enable_symlink_privilege(): + """ + Try to assign the symlink privilege to the current process token. + Return True if the assignment is successful. + """ + # create a space in memory for a TOKEN_PRIVILEGES structure + # with one element + size = ctypes.sizeof(TOKEN_PRIVILEGES) + size += ctypes.sizeof(LUID_AND_ATTRIBUTES) + buffer = ctypes.create_string_buffer(size) + tp = ctypes.cast(buffer, ctypes.POINTER(TOKEN_PRIVILEGES)).contents + tp.count = 1 + tp.get_array()[0].enable() + tp.get_array()[0].LUID = get_symlink_luid() + token = get_process_token() + res = AdjustTokenPrivileges(token, False, tp, 0, None, None) + if res == 0: + raise RuntimeError("Error in AdjustTokenPrivileges") + + ERROR_NOT_ALL_ASSIGNED = 1300 + return ctypes.windll.kernel32.GetLastError() != ERROR_NOT_ALL_ASSIGNED + +def main(): + assigned = enable_symlink_privilege() + msg = ['failure', 'success'][assigned] + + print("Symlink privilege assignment completed with {0}".format(msg)) + +if __name__ == '__main__': + main() diff --git a/Lib/test/test_glob.py b/Lib/test/test_glob.py index 5a3be94..05e29b5 100644 --- a/Lib/test/test_glob.py +++ b/Lib/test/test_glob.py @@ -1,5 +1,5 @@ import unittest -from test.support import run_unittest, TESTFN +from test.support import run_unittest, TESTFN, skip_unless_symlink, can_symlink import glob import os import shutil @@ -25,7 +25,7 @@ class GlobTests(unittest.TestCase): self.mktemp('ZZZ') self.mktemp('a', 'bcd', 'EF') self.mktemp('a', 'bcd', 'efg', 'ha') - if hasattr(os, 'symlink'): + if can_symlink(): os.symlink(self.norm('broken'), self.norm('sym1')) os.symlink(self.norm('broken'), self.norm('sym2')) @@ -98,12 +98,12 @@ class GlobTests(unittest.TestCase): # either of these results are reasonable self.assertIn(res[0], [self.tempdir, self.tempdir + os.sep]) + @skip_unless_symlink def test_glob_broken_symlinks(self): - if hasattr(os, 'symlink'): - eq = self.assertSequencesEqual_noorder - eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2')]) - eq(self.glob('sym1'), [self.norm('sym1')]) - eq(self.glob('sym2'), [self.norm('sym2')]) + eq = self.assertSequencesEqual_noorder + eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2')]) + eq(self.glob('sym1'), [self.norm('sym1')]) + eq(self.glob('sym2'), [self.norm('sym2')]) def test_main(): diff --git a/Lib/test/test_httpservers.py b/Lib/test/test_httpservers.py index 6a93800..4895b65 100644 --- a/Lib/test/test_httpservers.py +++ b/Lib/test/test_httpservers.py @@ -299,7 +299,7 @@ class CGIHTTPServerTestCase(BaseTestCase): # The shebang line should be pure ASCII: use symlink if possible. # See issue #7668. - if hasattr(os, 'symlink'): + if support.can_symlink(): self.pythonexe = os.path.join(self.parent_dir, 'python') os.symlink(sys.executable, self.pythonexe) else: diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index 14a9a7a..4693164 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -520,7 +520,7 @@ class WalkTests(unittest.TestCase): f = open(path, "w") f.write("I'm " + path + " and proud of it. Blame test_os.\n") f.close() - if hasattr(os, "symlink"): + if support.can_symlink(): os.symlink(os.path.abspath(t2_path), link_path) sub2_tree = (sub2_path, ["link"], ["tmp3"]) else: @@ -564,7 +564,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: @@ -1033,6 +1033,83 @@ class Win32KillTests(unittest.TestCase): self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT") +def skipUnlessWindows6(test): + if hasattr(sys, 'getwindowsversion') and sys.getwindowsversion().major >= 6: + return test + return unittest.skip("Requires Windows Vista or later")(test) + +@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) + 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)) + + class MiscTests(unittest.TestCase): @unittest.skipIf(os.name == "nt", "POSIX specific test") @@ -1056,6 +1133,7 @@ def test_main(): PosixUidGidTests, Pep383Tests, Win32KillTests, + Win32SymlinkTests, MiscTests, ) diff --git a/Lib/test/test_platform.py b/Lib/test/test_platform.py index e4051f9..51f47db 100644 --- a/Lib/test/test_platform.py +++ b/Lib/test/test_platform.py @@ -10,20 +10,26 @@ class PlatformTest(unittest.TestCase): def test_architecture(self): res = platform.architecture() - if hasattr(os, "symlink"): - def test_architecture_via_symlink(self): # issue3762 - def get(python): - cmd = [python, '-c', - 'import platform; print(platform.architecture())'] - p = subprocess.Popen(cmd, stdout=subprocess.PIPE) - return p.communicate() - real = os.path.realpath(sys.executable) - link = os.path.abspath(support.TESTFN) - os.symlink(real, link) - try: - self.assertEqual(get(real), get(link)) - finally: - os.remove(link) + @support.skip_unless_symlink + def test_architecture_via_symlink(self): # issue3762 + # On Windows, the EXE needs to know where pythonXY.dll is at so we have + # to add the directory to the path. + if sys.platform == "win32": + os.environ["Path"] = "{};{}".format(os.path.dirname(sys.executable), + os.environ["Path"]) + + def get(python): + cmd = [python, '-c', + 'import platform; print(platform.architecture())'] + p = subprocess.Popen(cmd, stdout=subprocess.PIPE) + return p.communicate() + real = os.path.realpath(sys.executable) + link = os.path.abspath(support.TESTFN) + os.symlink(real, link) + try: + self.assertEqual(get(real), get(link)) + finally: + os.remove(link) def test_platform(self): for aliased in (False, True): diff --git a/Lib/test/test_posixpath.py b/Lib/test/test_posixpath.py index 7ba95f7..34b9689 100644 --- a/Lib/test/test_posixpath.py +++ b/Lib/test/test_posixpath.py @@ -1,7 +1,9 @@ import unittest from test import support, test_genericpath -import posixpath, os +import posixpath +import os +import sys from posixpath import realpath, abspath, dirname, basename # An absolute path to a temporary filename for testing. We can't rely on TESTFN @@ -9,6 +11,16 @@ from posixpath import realpath, abspath, dirname, basename ABSTFN = abspath(support.TESTFN) +def skip_if_ABSTFN_contains_backslash(test): + """ + On Windows, posixpath.abspath still returns paths with backslashes + instead of posix forward slashes. If this is the case, several tests + fail, so skip them. + """ + found_backslash = '\\' in ABSTFN + msg = "ABSTFN is not a posix path - tests fail" + return [test, unittest.skip(msg)(test)][found_backslash] + def safe_rmdir(dirname): try: os.rmdir(dirname) @@ -143,7 +155,7 @@ class PosixPathTest(unittest.TestCase): f.write(b"foo") f.close() self.assertIs(posixpath.islink(support.TESTFN + "1"), False) - if hasattr(os, "symlink"): + if support.can_symlink(): os.symlink(support.TESTFN + "1", support.TESTFN + "2") self.assertIs(posixpath.islink(support.TESTFN + "2"), True) os.remove(support.TESTFN + "1") @@ -154,85 +166,59 @@ class PosixPathTest(unittest.TestCase): if not f.close(): f.close() + @staticmethod + def _create_file(filename): + with open(filename, 'wb') as f: + f.write(b'foo') + def test_samefile(self): - f = open(support.TESTFN + "1", "wb") - try: - f.write(b"foo") - f.close() - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "1" - ), - True - ) - # If we don't have links, assume that os.stat doesn't return resonable - # inode information and thus, that samefile() doesn't work - if hasattr(os, "symlink"): - os.symlink( - support.TESTFN + "1", - support.TESTFN + "2" - ) - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "2" - ), - True - ) - os.remove(support.TESTFN + "2") - f = open(support.TESTFN + "2", "wb") - f.write(b"bar") - f.close() - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "2" - ), - False - ) - finally: - if not f.close(): - f.close() + test_fn = support.TESTFN + "1" + self._create_file(test_fn) + self.assertTrue(posixpath.samefile(test_fn, test_fn)) + self.assertRaises(TypeError, posixpath.samefile) + + @unittest.skipIf( + sys.platform.startswith('win'), + "posixpath.samefile does not work on links in Windows") + @support.skip_unless_symlink + def test_samefile_on_links(self): + test_fn1 = support.TESTFN + "1" + test_fn2 = support.TESTFN + "2" + self._create_file(test_fn1) + + os.symlink(test_fn1, test_fn2) + self.assertTrue(posixpath.samefile(test_fn1, test_fn2)) + os.remove(test_fn2) + + self._create_file(test_fn2) + self.assertFalse(posixpath.samefile(test_fn1, test_fn2)) + def test_samestat(self): - f = open(support.TESTFN + "1", "wb") - try: - f.write(b"foo") - f.close() - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "1") - ), - True - ) - # If we don't have links, assume that os.stat() doesn't return resonable - # inode information and thus, that samefile() doesn't work - if hasattr(os, "symlink"): - if hasattr(os, "symlink"): - os.symlink(support.TESTFN + "1", support.TESTFN + "2") - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "2") - ), - True - ) - os.remove(support.TESTFN + "2") - f = open(support.TESTFN + "2", "wb") - f.write(b"bar") - f.close() - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "2") - ), - False - ) - finally: - if not f.close(): - f.close() + test_fn = support.TESTFN + "1" + self._create_file(test_fn) + test_fns = [test_fn]*2 + stats = map(os.stat, test_fns) + self.assertTrue(posixpath.samestat(*stats)) + + @unittest.skipIf( + sys.platform.startswith('win'), + "posixpath.samestat does not work on links in Windows") + @support.skip_unless_symlink + def test_samestat_on_links(self): + test_fn1 = support.TESTFN + "1" + test_fn2 = support.TESTFN + "2" + test_fns = (test_fn1, test_fn2) + os.symlink(*test_fns) + stats = map(os.stat, test_fns) + self.assertTrue(posixpath.samestat(*stats)) + os.remove(test_fn2) + + self._create_file(test_fn2) + stats = map(os.stat, test_fns) + self.assertFalse(posixpath.samestat(*stats)) + + self.assertRaises(TypeError, posixpath.samestat) def test_ismount(self): self.assertIs(posixpath.ismount("/"), True) @@ -286,103 +272,112 @@ class PosixPathTest(unittest.TestCase): self.assertEqual(posixpath.normpath(b"///..//./foo/.//bar"), b"/foo/bar") - if hasattr(os, "symlink"): - def test_realpath_basic(self): - # Basic operation. - try: - os.symlink(ABSTFN+"1", ABSTFN) - self.assertEqual(realpath(ABSTFN), ABSTFN+"1") - finally: - support.unlink(ABSTFN) - - def test_realpath_symlink_loops(self): - # Bug #930024, return the path unchanged if we get into an infinite - # symlink loop. - try: - old_path = abspath('.') - os.symlink(ABSTFN, ABSTFN) - self.assertEqual(realpath(ABSTFN), ABSTFN) - - os.symlink(ABSTFN+"1", ABSTFN+"2") - os.symlink(ABSTFN+"2", ABSTFN+"1") - self.assertEqual(realpath(ABSTFN+"1"), ABSTFN+"1") - self.assertEqual(realpath(ABSTFN+"2"), ABSTFN+"2") - - # Test using relative path as well. - os.chdir(dirname(ABSTFN)) - self.assertEqual(realpath(basename(ABSTFN)), ABSTFN) - finally: - os.chdir(old_path) - support.unlink(ABSTFN) - support.unlink(ABSTFN+"1") - support.unlink(ABSTFN+"2") - - def test_realpath_resolve_parents(self): - # We also need to resolve any symlinks in the parents of a relative - # path passed to realpath. E.g.: current working directory is - # /usr/doc with 'doc' being a symlink to /usr/share/doc. We call - # realpath("a"). This should return /usr/share/doc/a/. - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/y") - os.symlink(ABSTFN + "/y", ABSTFN + "/k") - - os.chdir(ABSTFN + "/k") - self.assertEqual(realpath("a"), ABSTFN + "/y/a") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "/k") - safe_rmdir(ABSTFN + "/y") - safe_rmdir(ABSTFN) - - def test_realpath_resolve_before_normalizing(self): - # Bug #990669: Symbolic links should be resolved before we - # normalize the path. E.g.: if we have directories 'a', 'k' and 'y' - # in the following hierarchy: - # a/k/y - # - # and a symbolic link 'link-y' pointing to 'y' in directory 'a', - # then realpath("link-y/..") should return 'k', not 'a'. - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/k") - os.mkdir(ABSTFN + "/k/y") - os.symlink(ABSTFN + "/k/y", ABSTFN + "/link-y") - - # Absolute path. - self.assertEqual(realpath(ABSTFN + "/link-y/.."), ABSTFN + "/k") - # Relative path. - os.chdir(dirname(ABSTFN)) - self.assertEqual(realpath(basename(ABSTFN) + "/link-y/.."), - ABSTFN + "/k") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "/link-y") - safe_rmdir(ABSTFN + "/k/y") - safe_rmdir(ABSTFN + "/k") - safe_rmdir(ABSTFN) - - def test_realpath_resolve_first(self): - # Bug #1213894: The first component of the path, if not absolute, - # must be resolved too. - - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/k") - os.symlink(ABSTFN, ABSTFN + "link") - os.chdir(dirname(ABSTFN)) - - base = basename(ABSTFN) - self.assertEqual(realpath(base + "link"), ABSTFN) - self.assertEqual(realpath(base + "link/k"), ABSTFN + "/k") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "link") - safe_rmdir(ABSTFN + "/k") - safe_rmdir(ABSTFN) + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_basic(self): + # Basic operation. + try: + os.symlink(ABSTFN+"1", ABSTFN) + self.assertEqual(realpath(ABSTFN), ABSTFN+"1") + finally: + support.unlink(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_symlink_loops(self): + # Bug #930024, return the path unchanged if we get into an infinite + # symlink loop. + try: + old_path = abspath('.') + os.symlink(ABSTFN, ABSTFN) + self.assertEqual(realpath(ABSTFN), ABSTFN) + + os.symlink(ABSTFN+"1", ABSTFN+"2") + os.symlink(ABSTFN+"2", ABSTFN+"1") + self.assertEqual(realpath(ABSTFN+"1"), ABSTFN+"1") + self.assertEqual(realpath(ABSTFN+"2"), ABSTFN+"2") + + # Test using relative path as well. + os.chdir(dirname(ABSTFN)) + self.assertEqual(realpath(basename(ABSTFN)), ABSTFN) + finally: + os.chdir(old_path) + support.unlink(ABSTFN) + support.unlink(ABSTFN+"1") + support.unlink(ABSTFN+"2") + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_parents(self): + # We also need to resolve any symlinks in the parents of a relative + # path passed to realpath. E.g.: current working directory is + # /usr/doc with 'doc' being a symlink to /usr/share/doc. We call + # realpath("a"). This should return /usr/share/doc/a/. + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/y") + os.symlink(ABSTFN + "/y", ABSTFN + "/k") + + os.chdir(ABSTFN + "/k") + self.assertEqual(realpath("a"), ABSTFN + "/y/a") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "/k") + safe_rmdir(ABSTFN + "/y") + safe_rmdir(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_before_normalizing(self): + # Bug #990669: Symbolic links should be resolved before we + # normalize the path. E.g.: if we have directories 'a', 'k' and 'y' + # in the following hierarchy: + # a/k/y + # + # and a symbolic link 'link-y' pointing to 'y' in directory 'a', + # then realpath("link-y/..") should return 'k', not 'a'. + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/k") + os.mkdir(ABSTFN + "/k/y") + os.symlink(ABSTFN + "/k/y", ABSTFN + "/link-y") + + # Absolute path. + self.assertEqual(realpath(ABSTFN + "/link-y/.."), ABSTFN + "/k") + # Relative path. + os.chdir(dirname(ABSTFN)) + self.assertEqual(realpath(basename(ABSTFN) + "/link-y/.."), + ABSTFN + "/k") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "/link-y") + safe_rmdir(ABSTFN + "/k/y") + safe_rmdir(ABSTFN + "/k") + safe_rmdir(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_first(self): + # Bug #1213894: The first component of the path, if not absolute, + # must be resolved too. + + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/k") + os.symlink(ABSTFN, ABSTFN + "link") + os.chdir(dirname(ABSTFN)) + + base = basename(ABSTFN) + self.assertEqual(realpath(base + "link"), ABSTFN) + self.assertEqual(realpath(base + "link/k"), ABSTFN + "/k") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "link") + safe_rmdir(ABSTFN + "/k") + safe_rmdir(ABSTFN) def test_relpath(self): (real_getcwd, os.getcwd) = (os.getcwd, lambda: r"/home/user/bar") diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py index 9f4871b..baa1943 100644 --- a/Lib/test/test_shutil.py +++ b/Lib/test/test_shutil.py @@ -271,7 +271,7 @@ class TestShutil(unittest.TestCase): shutil.rmtree(src_dir) shutil.rmtree(os.path.dirname(dst_dir)) - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') + @support.skip_unless_symlink def test_dont_copy_file_onto_link_to_itself(self): # bug 851123. os.mkdir(TESTFN) @@ -282,10 +282,11 @@ class TestShutil(unittest.TestCase): f.write('cheddar') f.close() - os.link(src, dst) - self.assertRaises(shutil.Error, shutil.copyfile, src, dst) - self.assertEqual(open(src,'r').read(), 'cheddar') - os.remove(dst) + if hasattr(os, "link"): + os.link(src, dst) + self.assertRaises(shutil.Error, shutil.copyfile, src, dst) + self.assertEqual(open(src,'r').read(), 'cheddar') + os.remove(dst) # Using `src` here would mean we end up with a symlink pointing # to TESTFN/TESTFN/cheese, while it should point at @@ -300,30 +301,30 @@ class TestShutil(unittest.TestCase): except OSError: pass - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') - def test_rmtree_on_symlink(self): - # bug 1669. - os.mkdir(TESTFN) - try: - src = os.path.join(TESTFN, 'cheese') - dst = os.path.join(TESTFN, 'shop') - os.mkdir(src) - os.symlink(src, dst) - self.assertRaises(OSError, shutil.rmtree, dst) - finally: - shutil.rmtree(TESTFN, ignore_errors=True) - - @unittest.skipUnless(hasattr(os, 'mkfifo'), 'requires os.mkfifo') - # Issue #3002: copyfile and copytree block indefinitely on named pipes - def test_copyfile_named_pipe(self): - os.mkfifo(TESTFN) + @support.skip_unless_symlink + def test_rmtree_on_symlink(self): + # bug 1669. + os.mkdir(TESTFN) try: - self.assertRaises(shutil.SpecialFileError, - shutil.copyfile, TESTFN, TESTFN2) - self.assertRaises(shutil.SpecialFileError, - shutil.copyfile, __file__, TESTFN) + src = os.path.join(TESTFN, 'cheese') + dst = os.path.join(TESTFN, 'shop') + os.mkdir(src) + os.symlink(src, dst) + self.assertRaises(OSError, shutil.rmtree, dst) finally: - os.remove(TESTFN) + shutil.rmtree(TESTFN, ignore_errors=True) + + if hasattr(os, "mkfifo"): + # Issue #3002: copyfile and copytree block indefinitely on named pipes + def test_copyfile_named_pipe(self): + os.mkfifo(TESTFN) + try: + self.assertRaises(shutil.SpecialFileError, + shutil.copyfile, TESTFN, TESTFN2) + self.assertRaises(shutil.SpecialFileError, + shutil.copyfile, __file__, TESTFN) + finally: + os.remove(TESTFN) @unittest.skipUnless(hasattr(os, 'mkfifo'), 'requires os.mkfifo') def test_copytree_named_pipe(self): @@ -361,7 +362,7 @@ class TestShutil(unittest.TestCase): shutil.copytree(src_dir, dst_dir, copy_function=_copy) self.assertEquals(len(copied), 2) - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') + @support.skip_unless_symlink def test_copytree_dangling_symlinks(self): # a dangling symlink raises an error at the end diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py index d55da62..2f1fffd 100644 --- a/Lib/test/test_sys.py +++ b/Lib/test/test_sys.py @@ -263,7 +263,7 @@ class SysModuleTest(unittest.TestCase): # Raise SkipTest if sys doesn't have getwindowsversion attribute test.support.get_attribute(sys, "getwindowsversion") v = sys.getwindowsversion() - self.assertEqual(len(v), 5) + self.assertEqual(len(v), 9) self.assertIsInstance(v[0], int) self.assertIsInstance(v[1], int) self.assertIsInstance(v[2], int) diff --git a/Lib/test/test_sysconfig.py b/Lib/test/test_sysconfig.py index f4a3c8e..8314e66 100644 --- a/Lib/test/test_sysconfig.py +++ b/Lib/test/test_sysconfig.py @@ -12,7 +12,7 @@ import shutil from copy import copy, deepcopy from test.support import (run_unittest, TESTFN, unlink, get_attribute, - captured_stdout) + captured_stdout, skip_unless_symlink) import sysconfig from sysconfig import (get_paths, get_platform, get_config_vars, @@ -239,17 +239,23 @@ class TestSysConfig(unittest.TestCase): 'posix_home', 'posix_prefix', 'posix_user') self.assertEquals(get_scheme_names(), wanted) + @skip_unless_symlink def test_symlink(self): + # On Windows, the EXE needs to know where pythonXY.dll is at so we have + # to add the directory to the path. + if sys.platform == "win32": + os.environ["Path"] = "{};{}".format(os.path.dirname(sys.executable), + os.environ["Path"]) + # Issue 7880 - symlink = get_attribute(os, "symlink") def get(python): cmd = [python, '-c', 'import sysconfig; print(sysconfig.get_platform())'] - p = subprocess.Popen(cmd, stdout=subprocess.PIPE) + p = subprocess.Popen(cmd, stdout=subprocess.PIPE, env=os.environ) return p.communicate() real = os.path.realpath(sys.executable) link = os.path.abspath(TESTFN) - symlink(real, link) + os.symlink(real, link) try: self.assertEqual(get(real), get(link)) finally: diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index 858f45b..5c064fc 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -291,6 +291,8 @@ class MiscReadTest(CommonReadTest): self.assertTrue(self.tar.getmembers()[-1].name == "misc/eof", "could not find all members") + @unittest.skipUnless(hasattr(os, "link"), "Missing hardlink implementation") + @support.skip_unless_symlink def test_extract_hardlink(self): # Test hardlink extraction (e.g. bug #857297). tar = tarfile.open(tarname, errorlevel=1, encoding="iso8859-1") @@ -695,16 +697,16 @@ class WriteTest(WriteTestBase): os.remove(target) os.remove(link) + @support.skip_unless_symlink def test_symlink_size(self): - if hasattr(os, "symlink"): - path = os.path.join(TEMPDIR, "symlink") - os.symlink("link_target", path) - try: - tar = tarfile.open(tmpname, self.mode) - tarinfo = tar.gettarinfo(path) - self.assertEqual(tarinfo.size, 0) - finally: - os.remove(path) + path = os.path.join(TEMPDIR, "symlink") + os.symlink("link_target", path) + try: + tar = tarfile.open(tmpname, self.mode) + tarinfo = tar.gettarinfo(path) + self.assertEqual(tarinfo.size, 0) + finally: + os.remove(path) def test_add_self(self): # Test for #1257255. @@ -1408,15 +1410,24 @@ class LinkEmulationTest(ReadTest): data = open(os.path.join(TEMPDIR, name), "rb").read() self.assertEqual(md5sum(data), md5_regtype) + # When 8879 gets fixed, this will need to change. Currently on Windows + # we have os.path.islink but no os.link, so these tests fail without the + # following skip until link is completed. + @unittest.skipIf(hasattr(os.path, "islink"), + "Skip emulation - has os.path.islink but not os.link") def test_hardlink_extraction1(self): self._test_link_extraction("ustar/lnktype") + @unittest.skipIf(hasattr(os.path, "islink"), + "Skip emulation - has os.path.islink but not os.link") def test_hardlink_extraction2(self): self._test_link_extraction("./ustar/linktest2/lnktype") + @unittest.skipIf(hasattr(os, "symlink"), "Skip emulation if symlink exists") def test_symlink_extraction1(self): self._test_link_extraction("ustar/symtype") + @unittest.skipIf(hasattr(os, "symlink"), "Skip emulation if symlink exists") def test_symlink_extraction2(self): self._test_link_extraction("./ustar/linktest2/symtype") @@ -154,6 +154,7 @@ Benjamin Collar Jeffery Collins Robert Collins Paul Colomiets +Jason R. Coombs Geremy Condra Juan José Conti Matt Conway @@ -1406,6 +1406,9 @@ Library Extension Modules ----------------- +- Issue #1578269: Implement os.symlink for Windows 6.0+. Patch by + Jason R. Coombs + - In struct.pack, correctly propogate exceptions from computing the truth of an object in the '?' format. diff --git a/Modules/posixmodule.c b/Modules/posixmodule.c index e9ea6cd..6fb990a 100644 --- a/Modules/posixmodule.c +++ b/Modules/posixmodule.c @@ -571,7 +571,7 @@ posix_error_with_allocated_filename(PyObject* name) #ifdef MS_WINDOWS static PyObject * -win32_error(char* function, char* filename) +win32_error(char* function, const char* filename) { /* XXX We should pass the function name along in the future. (winreg.c also wants to pass the function name.) @@ -978,12 +978,28 @@ attributes_from_dir_w(LPCWSTR pszFile, LPWIN32_FILE_ATTRIBUTE_DATA pfad) return TRUE; } -static int -win32_stat(const char* path, struct win32_stat *result) +/* About the following functions: win32_lstat, win32_lstat_w, win32_stat, + win32_stat_w + + In Posix, stat automatically traverses symlinks and returns the stat + structure for the target. In Windows, the equivalent GetFileAttributes by + default does not traverse symlinks and instead returns attributes for + the symlink. + + Therefore, win32_lstat will get the attributes traditionally, and + win32_stat will first explicitly resolve the symlink target and then will + call win32_lstat on that result. + + The _w represent Unicode equivalents of the aformentioned ANSI functions. */ + +static int +win32_lstat(const char* path, struct win32_stat *result) { WIN32_FILE_ATTRIBUTE_DATA info; int code; char *dot; + WIN32_FIND_DATAA find_data; + HANDLE find_data_handle; if (!GetFileAttributesExA(path, GetFileExInfoStandard, &info)) { if (GetLastError() != ERROR_SHARING_VIOLATION) { /* Protocol violation: we explicitly clear errno, instead of @@ -1000,27 +1016,44 @@ win32_stat(const char* path, struct win32_stat *result) } } } + code = attribute_data_to_stat(&info, result); if (code != 0) return code; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileA(path, &find_data); + if(find_data_handle != INVALID_HANDLE_VALUE) { + if(find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK) { + /* first clear the S_IFMT bits */ + result->st_mode ^= (result->st_mode & 0170000); + /* now set the bits that make this a symlink */ + result->st_mode |= 0120000; + } + FindClose(find_data_handle); + } + } + /* Set S_IFEXEC if it is an .exe, .bat, ... */ dot = strrchr(path, '.'); if (dot) { - if (stricmp(dot, ".bat") == 0 || - stricmp(dot, ".cmd") == 0 || - stricmp(dot, ".exe") == 0 || - stricmp(dot, ".com") == 0) + if (stricmp(dot, ".bat") == 0 || stricmp(dot, ".cmd") == 0 || + stricmp(dot, ".exe") == 0 || stricmp(dot, ".com") == 0) result->st_mode |= 0111; } return code; } static int -win32_wstat(const wchar_t* path, struct win32_stat *result) +win32_lstat_w(const wchar_t* path, struct win32_stat *result) { int code; const wchar_t *dot; WIN32_FILE_ATTRIBUTE_DATA info; + WIN32_FIND_DATAW find_data; + HANDLE find_data_handle; if (!GetFileAttributesExW(path, GetFileExInfoStandard, &info)) { if (GetLastError() != ERROR_SHARING_VIOLATION) { /* Protocol violation: we explicitly clear errno, instead of @@ -1028,8 +1061,8 @@ win32_wstat(const wchar_t* path, struct win32_stat *result) errno = 0; return -1; } else { - /* Could not get attributes on open file. Fall back to - reading the directory. */ + /* Could not get attributes on open file. Fall back to reading + the directory. */ if (!attributes_from_dir_w(path, &info)) { /* Very strange. This should not fail now */ errno = 0; @@ -1040,18 +1073,198 @@ win32_wstat(const wchar_t* path, struct win32_stat *result) code = attribute_data_to_stat(&info, result); if (code < 0) return code; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileW(path, &find_data); + if(find_data_handle != INVALID_HANDLE_VALUE) { + if(find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK) { + /* first clear the S_IFMT bits */ + result->st_mode ^= (result->st_mode & 0170000); + /* now set the bits that make this a symlink */ + result->st_mode |= 0120000; + } + FindClose(find_data_handle); + } + } + /* Set IFEXEC if it is an .exe, .bat, ... */ dot = wcsrchr(path, '.'); if (dot) { - if (_wcsicmp(dot, L".bat") == 0 || - _wcsicmp(dot, L".cmd") == 0 || - _wcsicmp(dot, L".exe") == 0 || - _wcsicmp(dot, L".com") == 0) + if (_wcsicmp(dot, L".bat") == 0 || _wcsicmp(dot, L".cmd") == 0 || + _wcsicmp(dot, L".exe") == 0 || _wcsicmp(dot, L".com") == 0) result->st_mode |= 0111; } return code; } +/* Grab GetFinalPathNameByHandle dynamically from kernel32 */ +static int has_GetFinalPathNameByHandle = 0; +static DWORD (CALLBACK *Py_GetFinalPathNameByHandleA)(HANDLE, LPSTR, DWORD, + DWORD); +static DWORD (CALLBACK *Py_GetFinalPathNameByHandleW)(HANDLE, LPWSTR, DWORD, + DWORD); +static int +check_GetFinalPathNameByHandle() +{ + HINSTANCE hKernel32; + /* only recheck */ + if (!has_GetFinalPathNameByHandle) + { + hKernel32 = GetModuleHandle("KERNEL32"); + *(FARPROC*)&Py_GetFinalPathNameByHandleA = GetProcAddress(hKernel32, + "GetFinalPathNameByHandleA"); + *(FARPROC*)&Py_GetFinalPathNameByHandleW = GetProcAddress(hKernel32, + "GetFinalPathNameByHandleW"); + has_GetFinalPathNameByHandle = Py_GetFinalPathNameByHandleA && Py_GetFinalPathNameByHandleW; + } + return has_GetFinalPathNameByHandle; +} + +static int +win32_stat(const char* path, struct win32_stat *result) +{ + /* Traverse the symlink to the target using + GetFinalPathNameByHandle() + */ + int code; + HANDLE hFile; + int buf_size; + char *target_path; + int result_length; + WIN32_FILE_ATTRIBUTE_DATA info; + + if(!check_GetFinalPathNameByHandle()) { + /* if the OS doesn't have GetFinalPathNameByHandle, it doesn't + have symlinks, so just fall back to the traditional behavior + found in lstat. */ + return win32_lstat(path, result); + } + + hFile = CreateFileA( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + /* Either the target doesn't exist, or we don't have access to + get a handle to it. If the former, we need to return an error. + If the latter, we can use attributes_from_dir. */ + if (GetLastError() != ERROR_SHARING_VIOLATION) { + /* Protocol violation: we explicitly clear errno, instead of + setting it to a POSIX error. Callers should use GetLastError. */ + errno = 0; + return -1; + } else { + /* Could not get attributes on open file. Fall back to + reading the directory. */ + if (!attributes_from_dir(path, &info)) { + /* Very strange. This should not fail now */ + errno = 0; + return -1; + } + } + code = attribute_data_to_stat(&info, result); + } + + buf_size = Py_GetFinalPathNameByHandleA(hFile, 0, 0, VOLUME_NAME_DOS); + if(!buf_size) return -1; + target_path = (char *)malloc((buf_size+1)*sizeof(char)); + result_length = Py_GetFinalPathNameByHandleA(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + + if(!result_length) + return -1; + + if(!CloseHandle(hFile)) + return -1; + + target_path[result_length] = 0; + code = win32_lstat(target_path, result); + free(target_path); + + return code; +} + +static int +win32_stat_w(const wchar_t* path, struct win32_stat *result) +{ + /* Traverse the symlink to the target using GetFinalPathNameByHandle() */ + int code; + HANDLE hFile; + int buf_size; + wchar_t *target_path; + int result_length; + WIN32_FILE_ATTRIBUTE_DATA info; + + if(!check_GetFinalPathNameByHandle()) { + /* If the OS doesn't have GetFinalPathNameByHandle, it doesn't have + symlinks, so just fall back to the traditional behavior found + in lstat. */ + return win32_lstat_w(path, result); + } + + hFile = CreateFileW( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_ATTRIBUTE_NORMAL|FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + /* Either the target doesn't exist, or we don't have access to + get a handle to it. If the former, we need to return an error. + If the latter, we can use attributes_from_dir. */ + if (GetLastError() != ERROR_SHARING_VIOLATION) { + /* Protocol violation: we explicitly clear errno, instead of + setting it to a POSIX error. Callers should use GetLastError. */ + errno = 0; + return -1; + } else { + /* Could not get attributes on open file. Fall back to + reading the directory. */ + if (!attributes_from_dir_w(path, &info)) { + /* Very strange. This should not fail now */ + errno = 0; + return -1; + } + } + code = attribute_data_to_stat(&info, result); + } + else { + /* We have a good handle to the target, use it to determine the target + path name (then we'll call lstat on it). */ + buf_size = Py_GetFinalPathNameByHandleW(hFile, 0, 0, VOLUME_NAME_DOS); + if(!buf_size) + return -1; + + target_path = (wchar_t *)malloc((buf_size+1)*sizeof(wchar_t)); + result_length = Py_GetFinalPathNameByHandleW(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + + if(!result_length) + return -1; + + if(!CloseHandle(hFile)) + return -1; + + target_path[result_length] = 0; + code = win32_lstat_w(target_path, result); + free(target_path); + } + + return code; +} + static int win32_fstat(int file_number, struct win32_stat *result) { @@ -2443,6 +2656,69 @@ posix__getfullpathname(PyObject *self, PyObject *args) } return PyBytes_FromString(outbuf); } /* end of posix__getfullpathname */ + +/* A helper function for samepath on windows */ +static PyObject * +posix__getfinalpathname(PyObject *self, PyObject *args) +{ + HANDLE hFile; + int buf_size; + wchar_t *target_path; + int result_length; + PyObject *result; + wchar_t *path; + + if (!PyArg_ParseTuple(args, "u|:_getfullpathname", &path)) { + return NULL; + } + + if(!check_GetFinalPathNameByHandle()) { + /* If the OS doesn't have GetFinalPathNameByHandle, return a + NotImplementedError. */ + return PyErr_Format(PyExc_NotImplementedError, + "GetFinalPathNameByHandle not available on this platform"); + } + + hFile = CreateFileW( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + return win32_error_unicode("GetFinalPathNamyByHandle", path); + return PyErr_Format(PyExc_RuntimeError, "Could not get a handle to file."); + } + + /* We have a good handle to the target, use it to determine the + target path name. */ + buf_size = Py_GetFinalPathNameByHandleW(hFile, 0, 0, VOLUME_NAME_NT); + + if(!buf_size) + return win32_error_unicode("GetFinalPathNameByHandle", path); + + target_path = (wchar_t *)malloc((buf_size+1)*sizeof(wchar_t)); + if(!target_path) + return PyErr_NoMemory(); + + result_length = Py_GetFinalPathNameByHandleW(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + if(!result_length) + return win32_error_unicode("GetFinalPathNamyByHandle", path); + + if(!CloseHandle(hFile)) + return win32_error_unicode("GetFinalPathNameByHandle", path); + + target_path[result_length] = 0; + result = PyUnicode_FromUnicode(target_path, result_length); + free(target_path); + return result; + +} /* end of posix__getfinalpathname */ #endif /* MS_WINDOWS */ PyDoc_STRVAR(posix_mkdir__doc__, @@ -2623,7 +2899,7 @@ static PyObject * posix_stat(PyObject *self, PyObject *args) { #ifdef MS_WINDOWS - return posix_do_stat(self, args, "O&:stat", STAT, "U:stat", win32_wstat); + return posix_do_stat(self, args, "O&:stat", STAT, "U:stat", win32_stat_w); #else return posix_do_stat(self, args, "O&:stat", STAT, NULL, NULL); #endif @@ -2681,6 +2957,41 @@ posix_umask(PyObject *self, PyObject *args) return PyLong_FromLong((long)i); } +#ifdef MS_WINDOWS + +/* override the default DeleteFileW behavior so that directory +symlinks can be removed with this function, the same as with +Unix symlinks */ +BOOL WINAPI Py_DeleteFileW(LPCWSTR lpFileName) +{ + WIN32_FILE_ATTRIBUTE_DATA info; + WIN32_FIND_DATAW find_data; + HANDLE find_data_handle; + int is_directory = 0; + int is_link = 0; + + if (GetFileAttributesExW(lpFileName, GetFileExInfoStandard, &info)) { + is_directory = info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(is_directory && + info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileW(lpFileName, &find_data); + + if(find_data_handle != INVALID_HANDLE_VALUE) { + is_link = find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK; + FindClose(find_data_handle); + } + } + } + + if (is_directory && is_link) + return RemoveDirectoryW(lpFileName); + + return DeleteFileW(lpFileName); +} +#endif /* MS_WINDOWS */ PyDoc_STRVAR(posix_unlink__doc__, "unlink(path)\n\n\ @@ -2694,7 +3005,7 @@ static PyObject * posix_unlink(PyObject *self, PyObject *args) { #ifdef MS_WINDOWS - return win32_1str(args, "remove", "y:remove", DeleteFileA, "U:remove", DeleteFileW); + return win32_1str(args, "remove", "y:remove", DeleteFileA, "U:remove", Py_DeleteFileW); #else return posix_1str(args, "O&:remove", unlink); #endif @@ -4544,7 +4855,8 @@ posix_lstat(PyObject *self, PyObject *args) return posix_do_stat(self, args, "O&:lstat", lstat, NULL, NULL); #else /* !HAVE_LSTAT */ #ifdef MS_WINDOWS - return posix_do_stat(self, args, "O&:lstat", STAT, "U:lstat", win32_wstat); + return posix_do_stat(self, args, "O&:lstat", STAT, "U:lstat", + win32_lstat_w); #else return posix_do_stat(self, args, "O&:lstat", STAT, NULL, NULL); #endif @@ -4609,6 +4921,193 @@ posix_symlink(PyObject *self, PyObject *args) } #endif /* HAVE_SYMLINK */ +#if !defined(HAVE_READLINK) && defined(MS_WINDOWS) + +PyDoc_STRVAR(win_readlink__doc__, +"readlink(path) -> path\n\n\ +Return a string representing the path to which the symbolic link points."); + +/* The following structure was copied from + http://msdn.microsoft.com/en-us/library/ms791514.aspx as the required + include doesn't seem to be present in the Windows SDK (at least as included + with Visual Studio Express). */ +typedef struct _REPARSE_DATA_BUFFER { + ULONG ReparseTag; + USHORT ReparseDataLength; + USHORT Reserved; + union { + struct { + USHORT SubstituteNameOffset; + USHORT SubstituteNameLength; + USHORT PrintNameOffset; + USHORT PrintNameLength; + ULONG Flags; + WCHAR PathBuffer[1]; + } SymbolicLinkReparseBuffer; + + struct { + USHORT SubstituteNameOffset; + USHORT SubstituteNameLength; + USHORT PrintNameOffset; + USHORT PrintNameLength; + WCHAR PathBuffer[1]; + } MountPointReparseBuffer; + + struct { + UCHAR DataBuffer[1]; + } GenericReparseBuffer; + }; +} REPARSE_DATA_BUFFER, *PREPARSE_DATA_BUFFER; + +#define REPARSE_DATA_BUFFER_HEADER_SIZE FIELD_OFFSET(REPARSE_DATA_BUFFER, GenericReparseBuffer) + +#define MAXIMUM_REPARSE_DATA_BUFFER_SIZE ( 16 * 1024 ) + +/* Windows readlink implementation */ +static PyObject * +win_readlink(PyObject *self, PyObject *args) +{ + wchar_t *path; + DWORD n_bytes_returned; + DWORD io_result; + PyObject *result; + HANDLE reparse_point_handle; + + char target_buffer[MAXIMUM_REPARSE_DATA_BUFFER_SIZE]; + REPARSE_DATA_BUFFER *rdb = (REPARSE_DATA_BUFFER *)target_buffer; + wchar_t *print_name; + + if (!PyArg_ParseTuple(args, + "u:readlink", + &path)) + return NULL; + + /* First get a handle to the reparse point */ + Py_BEGIN_ALLOW_THREADS + reparse_point_handle = CreateFileW( + path, + 0, + 0, + 0, + OPEN_EXISTING, + FILE_FLAG_OPEN_REPARSE_POINT|FILE_FLAG_BACKUP_SEMANTICS, + 0); + Py_END_ALLOW_THREADS + + if (reparse_point_handle==INVALID_HANDLE_VALUE) + { + return win32_error_unicode("readlink", path); + } + + Py_BEGIN_ALLOW_THREADS + /* New call DeviceIoControl to read the reparse point */ + io_result = DeviceIoControl( + reparse_point_handle, + FSCTL_GET_REPARSE_POINT, + 0, 0, /* in buffer */ + target_buffer, sizeof(target_buffer), + &n_bytes_returned, + 0 /* we're not using OVERLAPPED_IO */ + ); + CloseHandle(reparse_point_handle); + Py_END_ALLOW_THREADS + + if (io_result==0) + { + return win32_error_unicode("readlink", path); + } + + if (rdb->ReparseTag != IO_REPARSE_TAG_SYMLINK) + { + PyErr_SetString(PyExc_ValueError, + "not a symbolic link"); + return NULL; + } + print_name = rdb->SymbolicLinkReparseBuffer.PathBuffer + rdb->SymbolicLinkReparseBuffer.PrintNameOffset; + result = PyUnicode_FromWideChar(print_name, rdb->SymbolicLinkReparseBuffer.PrintNameLength/2); + return result; +} + +#endif /* !defined(HAVE_READLINK) && defined(MS_WINDOWS) */ + +#if !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) + +/* Grab CreateSymbolicLinkW dynamically from kernel32 */ +static int has_CreateSymbolicLinkW = 0; +static DWORD (CALLBACK *Py_CreateSymbolicLinkW)(LPWSTR, LPWSTR, DWORD); +static int +check_CreateSymbolicLinkW() +{ + HINSTANCE hKernel32; + /* only recheck */ + if (has_CreateSymbolicLinkW) + return has_CreateSymbolicLinkW; + hKernel32 = GetModuleHandle("KERNEL32"); + *(FARPROC*)&Py_CreateSymbolicLinkW = GetProcAddress(hKernel32, "CreateSymbolicLinkW"); + if (Py_CreateSymbolicLinkW) + has_CreateSymbolicLinkW = 1; + return has_CreateSymbolicLinkW; +} + +PyDoc_STRVAR(win_symlink__doc__, +"symlink(src, dst, target_is_directory=False)\n\n\ +Create a symbolic link pointing to src named dst.\n\ +target_is_directory is required if the target is to be interpreted as\n\ +a directory.\n\ +This function requires Windows 6.0 or greater, and raises a\n\ +NotImplementedError otherwise."); + +static PyObject * +win_symlink(PyObject *self, PyObject *args, PyObject *kwargs) +{ + static char *kwlist[] = {"src", "dest", "target_is_directory", NULL}; + PyObject *src, *dest; + int target_is_directory = 0; + DWORD res; + WIN32_FILE_ATTRIBUTE_DATA src_info; + + if (!check_CreateSymbolicLinkW()) + { + /* raise NotImplementedError */ + return PyErr_Format(PyExc_NotImplementedError, + "CreateSymbolicLinkW not found"); + } + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OO|i:symlink", + kwlist, &src, &dest, &target_is_directory)) + return NULL; + if (!convert_to_unicode(&src)) { return NULL; } + if (!convert_to_unicode(&dest)) { + Py_DECREF(src); + return NULL; + } + + /* if src is a directory, ensure target_is_directory==1 */ + if( + GetFileAttributesExW( + PyUnicode_AsUnicode(src), GetFileExInfoStandard, &src_info + )) + { + target_is_directory = target_is_directory || + (src_info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY); + } + + Py_BEGIN_ALLOW_THREADS + res = Py_CreateSymbolicLinkW( + PyUnicode_AsUnicode(dest), + PyUnicode_AsUnicode(src), + target_is_directory); + Py_END_ALLOW_THREADS + Py_DECREF(src); + Py_DECREF(dest); + if (!res) + { + return win32_error_unicode("symlink", PyUnicode_AsUnicode(src)); + } + + Py_INCREF(Py_None); + return Py_None; +} +#endif /* !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) */ #ifdef HAVE_TIMES #if defined(PYCC_VACPP) && defined(PYOS_OS2) @@ -7076,13 +7575,19 @@ static PyMethodDef posix_methods[] = { #ifdef HAVE_READLINK {"readlink", posix_readlink, METH_VARARGS, posix_readlink__doc__}, #endif /* HAVE_READLINK */ - {"rename", posix_rename, METH_VARARGS, posix_rename__doc__}, - {"rmdir", posix_rmdir, METH_VARARGS, posix_rmdir__doc__}, - {"stat", posix_stat, METH_VARARGS, posix_stat__doc__}, +#if !defined(HAVE_READLINK) && defined(MS_WINDOWS) + {"readlink", win_readlink, METH_VARARGS, win_readlink__doc__}, +#endif /* !defined(HAVE_READLINK) && defined(MS_WINDOWS) */ + {"rename", posix_rename, METH_VARARGS, posix_rename__doc__}, + {"rmdir", posix_rmdir, METH_VARARGS, posix_rmdir__doc__}, + {"stat", posix_stat, METH_VARARGS, posix_stat__doc__}, {"stat_float_times", stat_float_times, METH_VARARGS, stat_float_times__doc__}, #ifdef HAVE_SYMLINK {"symlink", posix_symlink, METH_VARARGS, posix_symlink__doc__}, #endif /* HAVE_SYMLINK */ +#if !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) + {"symlink", (PyCFunction)win_symlink, METH_VARARGS | METH_KEYWORDS, win_symlink__doc__}, +#endif /* !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) */ #ifdef HAVE_SYSTEM {"system", posix_system, METH_VARARGS, posix_system__doc__}, #endif @@ -7307,6 +7812,7 @@ static PyMethodDef posix_methods[] = { {"abort", posix_abort, METH_NOARGS, posix_abort__doc__}, #ifdef MS_WINDOWS {"_getfullpathname", posix__getfullpathname, METH_VARARGS, NULL}, + {"_getfinalpathname", posix__getfinalpathname, METH_VARARGS, NULL}, #endif #ifdef HAVE_GETLOADAVG {"getloadavg", posix_getloadavg, METH_NOARGS, posix_getloadavg__doc__}, |