Merge branch 'mr/add_support_for_wsl_symlinks' into 'master'

Add support for WSL links on Windows when calling sync_tree

See merge request it/e3-core!83

src/e3/fs.py

@@ -715,14 +715,28 @@ def isdir(fi: FileInfo) -> bool:
         """
         return fi.stat is not None and stat.S_ISDIR(fi.stat.st_mode)
 
+    def is_native_link(fi: FileInfo) -> bool:
+        """Check if a file is a native link.
+
+        :param fi: a FileInfo namedtuple
+        :return: return True if fi is a native symbolic link. The notion
+            of native link is only meaningful on Windows platform for which
+            some links are not well understood by the Win32 API (WSL links)
+        """
+        return fi.stat is not None and stat.S_ISLNK(fi.stat.st_mode)
+
     def islink(fi: FileInfo) -> bool:
         """Check if a file is a link.
 
         :param fi: a FileInfo namedtuple
 
         :return: True if fi is a symbolic link
         """
-        return fi.stat is not None and stat.S_ISLNK(fi.stat.st_mode)
+        return fi.stat is not None and (
+            stat.S_ISLNK(fi.stat.st_mode)
+            # Check for WSL links on Windows
+            or (sys.platform == "win32" and fi.stat.st_reparse_tag == 0xA000001D)
+        )
 
     def isfile(fi: FileInfo) -> bool:
         """Check if a file is a regular file.
@@ -822,8 +836,10 @@ def safe_copy(src: FileInfo, dst: FileInfo, is_directory: bool = False) -> None:
         :param dst: the target FileInfo object
         """
         if islink(src):  # windows: no cover
-            linkto = os.readlink(src.path)
-            if not islink(dst) or os.readlink(dst.path) != linkto:
+            linkto = e3.os.fs.readlink(src.path)
+            if not is_native_link(dst) or e3.os.fs.readlink(dst.path) != linkto:
+                # Checking here if the file is a native link allows us on Windows
+                # to transform Cygwin links into Win32 symlinks
                 if dst.stat is not None:
                     rm(dst.path, recursive=True, glob=False)
                 os.symlink(linkto, dst.path, target_is_directory=is_directory)

src/e3/os/fs.py

@@ -274,6 +274,27 @@ def mv(source: str | Path, target: str | Path) -> None:
         shutil.move(source, target)
 
 
+def readlink(filename: str | Path) -> str:
+    """Get target path of a symlink.
+
+    Equivalent of os.readlink with support for WSL Windows links.
+
+    :param filename: path containing a symlink
+    :return: target of the symlink
+    """
+    try:
+        return os.readlink(filename)
+    except Exception:
+        if sys.platform == "win32":
+            # This might be a WSL link
+            from e3.os.windows.fs import NTFile
+
+            f = NTFile(filename)
+            return f.wsl_reparse_link_target()
+        else:
+            raise
+
+
 def touch(filename: str | Path) -> None:
     """Update file access and modification times. Create the file if needed.
 

src/e3/os/windows/fs.py

@@ -28,6 +28,7 @@
     Share,
     Status,
     UnicodeString,
+    ReparseGUIDDataBuffer,
 )
 
 if TYPE_CHECKING:
@@ -243,6 +244,42 @@ def reparse_tag(self) -> int:
 
         return result.reserved0
 
+    def wsl_reparse_link_target(self) -> str | None:
+        """Get target of a WSL link (also used by Cygwin).
+
+        :return: the link target
+        """
+        FSCTL_GET_REPARSE_POINT = 0x900A8
+        self.read_attributes_internal()
+        if self.reparse_tag != IOReparseTag.WSL_SYMLINK:
+            return None
+
+        self.open(open_options=OpenOptions.OPEN_REPARSE_POINT)
+        result = ReparseGUIDDataBuffer()
+        fs_control_file: Callable = NT.FsControlFile  # type: ignore
+        status = fs_control_file(
+            self.handle,
+            None,
+            None,
+            None,
+            pointer(self.io_status),
+            FSCTL_GET_REPARSE_POINT,
+            None,
+            0,
+            pointer(result),
+            sizeof(result),
+        )
+        self.close()
+        if status < 0:
+            raise NTException(
+                status=status,
+                message=f"cannot find target of WSL link for {self.path}",
+                origin="NTFile.wsl_reparse_link_target",
+            )
+        return os.path.join(
+            os.path.dirname(self.path), result.data[: result.length - 4].decode("utf-8")
+        )
+
     def read_attributes_internal(self) -> None:
         """Retrieve file basic attributes (internal function).
 

src/e3/os/windows/native_api.py

@@ -156,6 +156,16 @@ class IOStatusBlock(Structure):
     _fields_ = [("status", NTSTATUS), ("information", POINTER(ULONG))]
 
 
+class ReparseGUIDDataBuffer(Structure):
+    _fields_ = [
+        ("tag", DWORD),
+        ("length", WORD),
+        ("reserved", WORD),
+        ("guid", DWORD),
+        ("data", ctypes.c_char * (16 * 1024)),
+    ]
+
+
 class UnicodeString(Structure):
     """Map UNICODE_STRING structure."""
 
@@ -403,6 +413,21 @@ def init_api(cls) -> None:
             kernel32 = ctypes.windll.kernel32
             ntdll = ctypes.windll.ntdll
 
+            cls.FsControlFile = ntdll.NtFsControlFile
+            cls.FsControlFile.restype = NTSTATUS
+            cls.FsControlFile.argtypes = [
+                HANDLE,
+                HANDLE,
+                LPVOID,
+                LPVOID,
+                POINTER(IOStatusBlock),
+                ULONG,
+                LPVOID,
+                ULONG,
+                POINTER(ReparseGUIDDataBuffer),
+                ULONG,
+            ]
+
             cls.FindFirstFile = kernel32.FindFirstFileW
             cls.FindFirstFile.restype = HANDLE
             cls.FindFirstFile.argtypes = [c_wchar_p, POINTER(FindData)]