ensure remove_directory is used instead of shutil.rmtree

This change renames `safe_rmtree` into a more appropriate name with
relevant arguments. We ensure that the force behaviour is opt-in
and not the default to avoid ambiguity.

src/poetry/installation/executor.py

@@ -25,7 +25,7 @@
 from poetry.utils.authenticator import Authenticator
 from poetry.utils.env import EnvCommandError
 from poetry.utils.helpers import pluralize
-from poetry.utils.helpers import safe_rmtree
+from poetry.utils.helpers import remove_directory
 from poetry.utils.pip import pip_install
 
 
@@ -488,7 +488,7 @@ def _remove(self, operation: Uninstall) -> int:
         if package.source_type == "git":
             src_dir = self._env.path / "src" / package.name
             if src_dir.exists():
-                safe_rmtree(str(src_dir))
+                remove_directory(src_dir, force=True)
 
         try:
             return self.run_pip("uninstall", package.name, "-y")
@@ -588,7 +588,7 @@ def _install_git(self, operation: Install | Update) -> int:
 
         src_dir = self._env.path / "src" / package.name
         if src_dir.exists():
-            safe_rmtree(str(src_dir))
+            remove_directory(str(src_dir))
 
         src_dir.parent.mkdir(exist_ok=True)
 

src/poetry/installation/pip_installer.py

@@ -13,7 +13,7 @@
 
 from poetry.installation.base_installer import BaseInstaller
 from poetry.utils._compat import encode
-from poetry.utils.helpers import safe_rmtree
+from poetry.utils.helpers import remove_directory
 from poetry.utils.pip import pip_install
 
 
@@ -128,7 +128,7 @@ def remove(self, package: Package) -> None:
         if package.source_type == "git":
             src_dir = self._env.path / "src" / package.name
             if src_dir.exists():
-                safe_rmtree(str(src_dir))
+                remove_directory(src_dir, force=True)
 
     def run(self, *args: Any, **kwargs: Any) -> str:
         return self._env.run_pip(*args, **kwargs)
@@ -252,7 +252,7 @@ def install_git(self, package: Package) -> None:
 
         src_dir = self._env.path / "src" / package.name
         if src_dir.exists():
-            safe_rmtree(str(src_dir))
+            remove_directory(src_dir, force=True)
 
         src_dir.parent.mkdir(exist_ok=True)
 

src/poetry/puzzle/provider.py

@@ -34,7 +34,7 @@
 from poetry.packages.package_collection import PackageCollection
 from poetry.puzzle.exceptions import OverrideNeeded
 from poetry.utils.helpers import download_file
-from poetry.utils.helpers import safe_rmtree
+from poetry.utils.helpers import remove_directory
 
 
 if TYPE_CHECKING:
@@ -211,7 +211,7 @@ def get_package_from_vcs(
         except Exception:
             raise
         finally:
-            safe_rmtree(str(tmp_dir))
+            remove_directory(tmp_dir, force=True)
 
         return package
 

src/poetry/utils/env.py

@@ -7,7 +7,6 @@
 import os
 import platform
 import re
-import shutil
 import subprocess
 import sys
 import sysconfig
@@ -44,6 +43,7 @@
 from poetry.utils._compat import metadata
 from poetry.utils.helpers import is_dir_writable
 from poetry.utils.helpers import paths_csv
+from poetry.utils.helpers import remove_directory
 from poetry.utils.helpers import temporary_directory
 
 
@@ -341,7 +341,7 @@ def remove_distribution_files(self, distribution_name: str) -> list[Path]:
                     file.unlink()
 
             if distribution._path.exists():
-                shutil.rmtree(str(distribution._path))
+                remove_directory(str(distribution._path), force=True)
 
             paths.append(distribution._path)
 
@@ -1046,7 +1046,7 @@ def remove_venv(cls, path: Path | str) -> None:
             path = Path(path)
         assert path.is_dir()
         try:
-            shutil.rmtree(str(path))
+            remove_directory(path)
             return
         except OSError as e:
             # Continue only if e.errno == 16
@@ -1061,7 +1061,7 @@ def remove_venv(cls, path: Path | str) -> None:
             if file_path.is_file() or file_path.is_symlink():
                 file_path.unlink()
             elif file_path.is_dir():
-                shutil.rmtree(str(file_path))
+                remove_directory(file_path, force=True)
 
     @classmethod
     def get_system_env(cls, naive: bool = False) -> SystemEnv | GenericEnv:

src/poetry/utils/helpers.py

@@ -33,18 +33,13 @@ def module_name(name: str) -> str:
     return canonicalize_name(name).replace(".", "_").replace("-", "_")
 
 
-def _del_ro(action: Callable, name: str, exc: Exception) -> None:
-    os.chmod(name, stat.S_IWRITE)
-    os.remove(name)
-
-
 @contextmanager
 def temporary_directory(*args: Any, **kwargs: Any) -> Iterator[str]:
     name = tempfile.mkdtemp(*args, **kwargs)
 
     yield name
 
-    shutil.rmtree(name, onerror=_del_ro)
+    remove_directory(name, force=True)
 
 
 def get_cert(config: Config, repository_name: str) -> Path | None:
@@ -71,11 +66,21 @@ def _on_rm_error(func: Callable, path: str, exc_info: Exception) -> None:
     func(path)
 
 
-def safe_rmtree(path: str) -> None:
+def remove_directory(
+    path: Path | str, *args: Any, force: bool = False, **kwargs: Any
+) -> None:
+    """
+    Helper function handle safe removal, and optionally forces stubborn file removal.
+    This is particularly useful when dist files are read-only or git writes read-only
+    files on Windows.
+
+    Internally, all arguments are passed to `shutil.rmtree`.
+    """
     if Path(path).is_symlink():
         return os.unlink(str(path))
 
-    shutil.rmtree(path, onerror=_on_rm_error)
+    kwargs["onerror"] = kwargs.pop("onerror", _on_rm_error if force else None)
+    shutil.rmtree(path, *args, **kwargs)
 
 
 def merge_dicts(d1: dict, d2: dict) -> None:

tests/utils/test_env.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import os
-import shutil
 import subprocess
 import sys
 
@@ -28,6 +27,7 @@
 from poetry.utils.env import NoCompatiblePythonVersionFound
 from poetry.utils.env import SystemEnv
 from poetry.utils.env import VirtualEnv
+from poetry.utils.helpers import remove_directory
 
 
 if TYPE_CHECKING:
@@ -692,19 +692,19 @@ def test_remove_keeps_dir_if_not_deleteable(
         side_effect=check_output_wrapper(Version.parse("3.6.6")),
     )
 
-    original_rmtree = shutil.rmtree
-
-    def err_on_rm_venv_only(path: str, *args: Any, **kwargs: Any) -> None:
-        if path == str(venv_path):
+    def err_on_rm_venv_only(path: Path | str, *args: Any, **kwargs: Any) -> None:
+        if str(path) == str(venv_path):
             raise OSError(16, "Test error")  # ERRNO 16: Device or resource busy
         else:
-            original_rmtree(path)
+            remove_directory(path)
 
-    m = mocker.patch("shutil.rmtree", side_effect=err_on_rm_venv_only)
+    m = mocker.patch(
+        "poetry.utils.env.remove_directory", side_effect=err_on_rm_venv_only
+    )
 
     venv = manager.remove(f"{venv_name}-py3.6")
 
-    m.assert_any_call(str(venv_path))
+    m.assert_any_call(venv_path)
 
     assert venv_path == venv.path
     assert venv_path.exists()
@@ -713,7 +713,7 @@ def err_on_rm_venv_only(path: str, *args: Any, **kwargs: Any) -> None:
     assert not file1_path.exists()
     assert not file2_path.exists()
 
-    m.side_effect = original_rmtree  # Avoid teardown using `err_on_rm_venv_only`
+    m.side_effect = remove_directory  # Avoid teardown using `err_on_rm_venv_only`
 
 
 @pytest.mark.skipif(os.name == "nt", reason="Symlinks are not support for Windows")