Add utility functions to rename a file or folder src to dst with retr…

…ying.

Signed-off-by: loonghao <[email protected]>

src/rez/package_cache.py

@@ -28,6 +28,7 @@
 from rez.utils.logging_ import print_warning
 from rez.packages import get_variant
 from rez.system import system
+from rez.utils.filesystem import rename
 
 
 class PackageCache(object):
@@ -337,7 +338,7 @@ def remove_variant(self, variant):
                     os.chmod(rootpath, st.st_mode | stat.S_IWUSR)
 
                 # actually a mv
-                os.rename(rootpath, dest_rootpath)
+                rename(rootpath, dest_rootpath)
 
             except OSError as e:
                 if e.errno == errno.ENOENT:

src/rez/tests/test_utils_filesystem.py

@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright Contributors to the Rez Project
+
+
+"""
+unit tests for 'rez.utils.filesystem' module
+"""
+import os.path
+import shutil
+import tempfile
+
+from rez.tests.util import TestBase
+from rez.tests.util import TempdirMixin
+from rez.utils import filesystem
+from rez.utils.platform_ import platform_
+import unittest
+
+
+class TestFileSystem(TestBase, TempdirMixin):
+
+    def __init__(self, *nargs, **kwargs):
+        super(TestFileSystem, self).__init__(*nargs, **kwargs)
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestFileSystem, cls).setUpClass()
+        TempdirMixin.setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestFileSystem, cls).tearDownClass()
+        TempdirMixin.tearDownClass()
+
+    def test_windows_rename_fallback_to_robocopy(self):
+        if platform_.name != 'windows':
+            self.skipTest('Robocopy is only available on windows.')
+        src = tempfile.mkdtemp(dir=self.root)
+        dst = tempfile.mkdtemp(dir=self.root)
+        with unittest.mock.patch("os.rename") as mock_rename:
+            mock_rename.side_effect = PermissionError("Permission denied")
+            filesystem.rename(src, dst)
+            self.assertTrue(os.path.exists(dst))
+            self.assertFalse(os.path.exists(src))
+
+    def test_windows_robocopy_failed(self):
+        if platform_.name != 'windows':
+            self.skipTest('Robocopy is only available on windows.')
+        src = tempfile.mkdtemp(dir=self.root)
+        dst = tempfile.mkdtemp(dir=self.root)
+        with unittest.mock.patch("os.rename") as mock_rename:
+            mock_rename.side_effect = PermissionError("Permission denied")
+            with unittest.mock.patch("rez.utils.filesystem.Popen") as mock_subprocess:
+                mock_subprocess.return_value = unittest.mock.Mock(returncode=9)
+                with self.assertRaises(OSError) as err:
+                    filesystem.rename(src, dst)
+                self.assertEqual(str(err.exception), "Rename {} to {} failed.".format(src, dst))
+
+    def test_rename_folder_with_permission_error_and_no_robocopy(self):
+        src = tempfile.mkdtemp(dir=self.root)
+        dst = tempfile.mkdtemp(dir=self.root)
+        with unittest.mock.patch("os.rename") as mock_rename:
+            mock_rename.side_effect = PermissionError("Permission denied")
+            with unittest.mock.patch("rez.utils.filesystem.which") as mock_which:
+                mock_which.return_value = False
+                with self.assertRaises(PermissionError) as err:
+                    filesystem.rename(src, dst)
+                self.assertEqual(str(err.exception), "Permission denied")
+
+    def test_rename_folder_with_permission_error_and_src_is_file(self):
+        src = tempfile.mktemp(dir=self.root)
+        dst = tempfile.mktemp(dir=self.root)
+        with open(src, "w") as file_:
+            file_.write("content.")
+        with unittest.mock.patch("os.rename") as mock_rename:
+            mock_rename.side_effect = PermissionError("Permission denied")
+            with self.assertRaises(PermissionError) as err:
+                filesystem.rename(src, dst)
+            self.assertEqual(str(err.exception), "Permission denied")
+        self.assertFalse(os.path.exists(dst))
+        self.assertTrue(os.path.exists(src))
+
+    def test_rename_file(self):
+        src = tempfile.mktemp(dir=self.root)
+        dst = tempfile.mktemp(dir=self.root)
+        with open(src, "w") as file_:
+            file_.write("content.")
+        filesystem.rename(src, dst)
+        self.assertTrue(os.path.exists(dst))
+        self.assertFalse(os.path.exists(src))

src/rez/utils/filesystem.py

@@ -23,7 +23,8 @@
 import uuid
 
 from rez.utils.platform_ import platform_
-
+from rez.util import which
+from rez.utils.execution import Popen
 
 is_windows = platform.system() == "Windows"
 
@@ -213,6 +214,7 @@ def forceful_rmtree(path):
         * path length over 259 char (on Windows)
         * unicode path
     """
+
     def _on_error(func, path, exc_info):
         try:
             if is_windows:
@@ -281,7 +283,7 @@ def replace_file_or_dir(dest, source):
 
     if not os.path.exists(dest):
         try:
-            os.rename(source, dest)
+            rename(source, dest)
             return
         except:
             if not os.path.exists(dest):
@@ -294,8 +296,8 @@ def replace_file_or_dir(dest, source):
         pass
 
     with make_tmp_name(dest) as tmp_dest:
-        os.rename(dest, tmp_dest)
-        os.rename(source, dest)
+        rename(dest, tmp_dest)
+        rename(source, dest)
 
 
 def additive_copytree(src, dst, symlinks=False, ignore=None):
@@ -690,3 +692,55 @@ def windows_long_path(dos_path):
         path = "\\\\?\\" + path
 
     return path
+
+
+def rename(src, dst):
+    """Utility function to rename a file or folder src to dst with retrying.
+
+    This function uses the built-in `os.rename()` function and falls back to `robocopy` tool
+    if `os.rename` raises a `PermissionError` exception.
+
+    Args:
+        src (str): The original name (path) of the file or folder.
+        dst (str): The new name (path) for the file or folder.
+
+    Raises:
+        OSError: If renaming fails after all attempts.
+
+    """
+    # Inspired by https://github.com/conan-io/conan/blob/2.1.0/conan/tools/files/files.py#L207
+    try:
+        os.rename(src, dst)
+    except PermissionError as err:
+        if is_windows and which("robocopy") and os.path.isdir(src):
+            # https://learn.microsoft.com/en-us/windows-server/administration/windows-commands/robocopy
+            args = [
+                "robocopy",
+                # /move Moves files and directories, and deletes them from the source after they are copied.
+                "/move",
+                # /e Copies subdirectories. Note that this option includes empty directories.
+                "/e",
+                # /ndl Specifies that directory names are not to be logged.
+                "/ndl",
+                # /nfl Specifies that file names are not to be logged.
+                "/nfl",
+                # /njs Specifies that there's no job summary.
+                "/njs",
+                # /njh Specifies that there's no job header.
+                "/njh",
+                # /np Specifies that the progress of the copying operation
+                # (the number of files or directories copied so far) won't be displayed.
+                "/np",
+                # /ns Specifies that file sizes aren't to be logged.
+                "/ns",
+                # /nc Specifies that file classes aren't to be logged.
+                "/nc",
+                src,
+                dst,
+            ]
+            process = Popen(args)
+            process.communicate()
+            if process.returncode > 7:  # https://ss64.com/nt/robocopy-exit.html
+                raise OSError("Rename {} to {} failed.".format(src, dst))
+        else:
+            raise err