Implement #1578269. Patch by Jason R. Coombs.

Added Windows support for os.symlink when run on Windows 6.0 or greater,
aka Vista. Previous Windows versions will raise NotImplementedError
when trying to symlink.

Includes numerous test updates and additions to test_os, including
a symlink_support module because of the fact that privilege escalation
is required in order to run the tests to ensure that the user is able
to create symlinks. By default, accounts do not have the required
privilege, so the escalation code will have to be exposed later (or
documented on how to do so). I'll be following up with that work next.

Note that the tests use ctypes, which was agreed on during the PyCon
language summit.

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)

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)

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)

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)

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.

Lib/test/symlink_support.py

+"""
+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()

Lib/test/test_glob.py

 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():

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:

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,
     )
 

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):

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

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)

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:

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")
 

Misc/ACKS

@@ -154,6 +154,7 @@ Benjamin Collar
 Jeffery Collins
 Robert Collins
 Paul Colomiets
+Jason R. Coombs
 Geremy Condra
 Juan José Conti
 Matt Conway

Misc/NEWS

@@ -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.