use tempfile.TemporaryDirectory

src/poetry/core/masonry/builders/wheel.py

@@ -14,6 +14,7 @@
 from base64 import urlsafe_b64encode
 from io import StringIO
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING
 from typing import TextIO
 
@@ -26,7 +27,6 @@
 from poetry.core.masonry.utils.helpers import distribution_name
 from poetry.core.masonry.utils.helpers import normalize_file_permissions
 from poetry.core.masonry.utils.package_include import PackageInclude
-from poetry.core.utils.helpers import temporary_directory
 
 
 if TYPE_CHECKING:
@@ -124,7 +124,7 @@ def build(
             self._copy_file_scripts(zip_file)
 
             if self._metadata_directory is None:
-                with temporary_directory() as temp_dir:
+                with TemporaryDirectory() as temp_dir:
                     metadata_directory = self.prepare_metadata(Path(temp_dir))
                     self._copy_dist_info(zip_file, metadata_directory)
             else:

src/poetry/core/utils/helpers.py

@@ -1,28 +1,15 @@
 from __future__ import annotations
 
-import os
-import shutil
-import stat
-import sys
-import tempfile
-import time
 import unicodedata
 import warnings
 
-from contextlib import contextmanager
 from pathlib import Path
-from typing import TYPE_CHECKING
-from typing import Any
 
 from packaging.utils import canonicalize_name
 
 from poetry.core.version.pep440 import PEP440Version
 
 
-if TYPE_CHECKING:
-    from collections.abc import Iterator
-
-
 def combine_unicode(string: str) -> str:
     return unicodedata.normalize("NFC", string)
 
@@ -40,20 +27,6 @@ def normalize_version(version: str) -> str:
     return PEP440Version.parse(version).to_string()
 
 
-@contextmanager
-def temporary_directory(*args: Any, **kwargs: Any) -> Iterator[str]:
-    if sys.version_info >= (3, 10):
-        # mypy reports an error if ignore_cleanup_errors is
-        # specified literally in the call
-        kwargs["ignore_cleanup_errors"] = True
-        with tempfile.TemporaryDirectory(*args, **kwargs) as name:
-            yield name
-    else:
-        name = tempfile.mkdtemp(*args, **kwargs)
-        yield name
-        robust_rmtree(name)
-
-
 def parse_requires(requires: str) -> list[str]:
     lines = requires.split("\n")
 
@@ -91,41 +64,6 @@ def parse_requires(requires: str) -> list[str]:
     return requires_dist
 
 
-def _on_rm_error(func: Any, path: str | Path, exc_info: Any) -> None:
-    if not os.path.exists(path):
-        return
-
-    os.chmod(path, stat.S_IWRITE)
-    func(path)
-
-
-def robust_rmtree(path: str | Path, max_timeout: float = 1) -> None:
-    """
-    Robustly tries to delete paths.
-    Retries several times if an OSError occurs.
-    If the final attempt fails, the Exception is propagated
-    to the caller.
-    """
-    path = Path(path)  # make sure this is a Path object, not str
-    timeout = 0.001
-    while timeout < max_timeout:
-        try:
-            # both os.unlink and shutil.rmtree can throw exceptions on Windows
-            # if the files are in use when called
-            if path.is_symlink():
-                path.unlink()
-            else:
-                shutil.rmtree(path)
-            return  # Only hits this on success
-        except OSError:
-            # Increase the timeout and try again
-            time.sleep(timeout)
-            timeout *= 2
-
-    # Final attempt, pass any Exceptions up to caller.
-    shutil.rmtree(path, onerror=_on_rm_error)
-
-
 def readme_content_type(path: str | Path) -> str:
     suffix = Path(path).suffix
     if suffix == ".rst":

tests/masonry/test_api.py

@@ -5,12 +5,12 @@
 
 from contextlib import contextmanager
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING
 from typing import Iterator
 
 from poetry.core import __version__
 from poetry.core.masonry import api
-from poetry.core.utils.helpers import temporary_directory
 from tests.testutils import validate_sdist_contents
 from tests.testutils import validate_wheel_contents
 
@@ -45,7 +45,7 @@ def test_get_requires_for_build_sdist() -> None:
 
 
 def test_build_wheel() -> None:
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
         filename = api.build_wheel(tmp_dir)
         validate_wheel_contents(
             name="my_package",
@@ -56,7 +56,7 @@ def test_build_wheel() -> None:
 
 
 def test_build_wheel_with_include() -> None:
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "with-include")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "with-include")):
         filename = api.build_wheel(tmp_dir)
         validate_wheel_contents(
             name="with_include",
@@ -67,14 +67,14 @@ def test_build_wheel_with_include() -> None:
 
 
 def test_build_wheel_with_bad_path_dev_dep_succeeds() -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dev_dep")
     ):
         api.build_wheel(tmp_dir)
 
 
 def test_build_wheel_with_bad_path_dep_succeeds(caplog: LogCaptureFixture) -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dep")
     ):
         api.build_wheel(tmp_dir)
@@ -85,15 +85,15 @@ def test_build_wheel_with_bad_path_dep_succeeds(caplog: LogCaptureFixture) -> No
 
 
 def test_build_wheel_extended() -> None:
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "extended")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "extended")):
         filename = api.build_wheel(tmp_dir)
         whl = Path(tmp_dir) / filename
         assert whl.exists()
         validate_wheel_contents(name="extended", version="0.1", path=whl.as_posix())
 
 
 def test_build_sdist() -> None:
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
         filename = api.build_sdist(tmp_dir)
         validate_sdist_contents(
             name="my-package",
@@ -104,7 +104,7 @@ def test_build_sdist() -> None:
 
 
 def test_build_sdist_with_include() -> None:
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "with-include")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "with-include")):
         filename = api.build_sdist(tmp_dir)
         validate_sdist_contents(
             name="with-include",
@@ -115,14 +115,14 @@ def test_build_sdist_with_include() -> None:
 
 
 def test_build_sdist_with_bad_path_dev_dep_succeeds() -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dev_dep")
     ):
         api.build_sdist(tmp_dir)
 
 
 def test_build_sdist_with_bad_path_dep_succeeds(caplog: LogCaptureFixture) -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dep")
     ):
         api.build_sdist(tmp_dir)
@@ -182,7 +182,7 @@ def test_prepare_metadata_for_build_wheel() -> None:
 ==========
 
 """  # noqa: E501
-    with temporary_directory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
+    with TemporaryDirectory() as tmp_dir, cwd(os.path.join(fixtures, "complete")):
         dirname = api.prepare_metadata_for_build_wheel(tmp_dir)
 
         assert dirname == "my_package-1.2.3.dist-info"
@@ -204,7 +204,7 @@ def test_prepare_metadata_for_build_wheel() -> None:
 
 
 def test_prepare_metadata_for_build_wheel_with_bad_path_dev_dep_succeeds() -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dev_dep")
     ):
         api.prepare_metadata_for_build_wheel(tmp_dir)
@@ -213,7 +213,7 @@ def test_prepare_metadata_for_build_wheel_with_bad_path_dev_dep_succeeds() -> No
 def test_prepare_metadata_for_build_wheel_with_bad_path_dep_succeeds(
     caplog: LogCaptureFixture,
 ) -> None:
-    with temporary_directory() as tmp_dir, cwd(
+    with TemporaryDirectory() as tmp_dir, cwd(
         os.path.join(fixtures, "with_bad_path_dep")
     ):
         api.prepare_metadata_for_build_wheel(tmp_dir)
@@ -226,7 +226,7 @@ def test_prepare_metadata_for_build_wheel_with_bad_path_dep_succeeds(
 def test_build_editable_wheel() -> None:
     pkg_dir = Path(fixtures) / "complete"
 
-    with temporary_directory() as tmp_dir, cwd(pkg_dir):
+    with TemporaryDirectory() as tmp_dir, cwd(pkg_dir):
         filename = api.build_editable(tmp_dir)
         wheel_pth = Path(tmp_dir) / filename
 
@@ -246,10 +246,10 @@ def test_build_editable_wheel() -> None:
 def test_build_wheel_with_metadata_directory() -> None:
     pkg_dir = Path(fixtures) / "complete"
 
-    with temporary_directory() as metadata_tmp_dir, cwd(pkg_dir):
+    with TemporaryDirectory() as metadata_tmp_dir, cwd(pkg_dir):
         metadata_directory = api.prepare_metadata_for_build_wheel(metadata_tmp_dir)
 
-        with temporary_directory() as wheel_tmp_dir:
+        with TemporaryDirectory() as wheel_tmp_dir:
             dist_info_path = Path(metadata_tmp_dir) / metadata_directory
             open(dist_info_path / "CUSTOM", "w").close()  # noqa: SIM115
             filename = api.build_wheel(
@@ -273,10 +273,10 @@ def test_build_wheel_with_metadata_directory() -> None:
 def test_build_editable_wheel_with_metadata_directory() -> None:
     pkg_dir = Path(fixtures) / "complete"
 
-    with temporary_directory() as metadata_tmp_dir, cwd(pkg_dir):
+    with TemporaryDirectory() as metadata_tmp_dir, cwd(pkg_dir):
         metadata_directory = api.prepare_metadata_for_build_editable(metadata_tmp_dir)
 
-        with temporary_directory() as wheel_tmp_dir:
+        with TemporaryDirectory() as wheel_tmp_dir:
             dist_info_path = Path(metadata_tmp_dir) / metadata_directory
             open(dist_info_path / "CUSTOM", "w").close()  # noqa: SIM115
             filename = api.build_editable(

tests/utils/test_helpers.py

@@ -1,24 +1,16 @@
 from __future__ import annotations
 
 import os
-import sys
-import tempfile
 
 from pathlib import Path
 from stat import S_IREAD
-from typing import TYPE_CHECKING
+from tempfile import TemporaryDirectory
 
 import pytest
 
-
-if TYPE_CHECKING:
-    from pytest_mock import MockerFixture
-
 from poetry.core.utils.helpers import combine_unicode
 from poetry.core.utils.helpers import parse_requires
 from poetry.core.utils.helpers import readme_content_type
-from poetry.core.utils.helpers import robust_rmtree
-from poetry.core.utils.helpers import temporary_directory
 
 
 def test_parse_requires() -> None:
@@ -101,7 +93,7 @@ def test_utils_helpers_combine_unicode() -> None:
 
 
 def test_utils_helpers_temporary_directory_readonly_file() -> None:
-    with temporary_directory() as temp_dir:
+    with TemporaryDirectory() as temp_dir:
         readonly_filename = os.path.join(temp_dir, "file.txt")
         with open(readonly_filename, "w+") as readonly_file:
             readonly_file.write("Poetry rocks!")
@@ -126,60 +118,3 @@ def test_utils_helpers_readme_content_type(
     readme: str | Path, content_type: str
 ) -> None:
     assert readme_content_type(readme) == content_type
-
-
-def test_temporary_directory_python_3_10_or_newer(mocker: MockerFixture) -> None:
-    mocked_rmtree = mocker.patch("shutil.rmtree")
-    mocked_temp_dir = mocker.patch("tempfile.TemporaryDirectory")
-    mocked_mkdtemp = mocker.patch("tempfile.mkdtemp")
-
-    mocker.patch.object(sys, "version_info", (3, 10))
-    with temporary_directory() as tmp:
-        assert tmp
-
-    assert not mocked_rmtree.called
-    assert not mocked_mkdtemp.called
-    mocked_temp_dir.assert_called_with(ignore_cleanup_errors=True)
-
-
-def test_temporary_directory_python_3_9_or_older(mocker: MockerFixture) -> None:
-    mocked_rmtree = mocker.patch("shutil.rmtree")
-    mocked_temp_dir = mocker.patch("tempfile.TemporaryDirectory")
-    mocked_mkdtemp = mocker.patch("tempfile.mkdtemp")
-
-    mocked_mkdtemp.return_value = "hello from test"
-
-    mocker.patch.object(sys, "version_info", (3, 9))
-    with temporary_directory() as tmp:
-        assert tmp == "hello from test"
-
-    assert mocked_rmtree.called
-    assert mocked_mkdtemp.called
-    assert not mocked_temp_dir.called
-
-
-def test_robust_rmtree(mocker: MockerFixture) -> None:
-    mocked_rmtree = mocker.patch("shutil.rmtree")
-
-    # this should work after an initial exception
-    name = tempfile.mkdtemp()
-    mocked_rmtree.side_effect = [
-        OSError(
-            "Couldn't delete file yet, waiting for references to clear", "mocked path"
-        ),
-        None,
-    ]
-    robust_rmtree(name)
-
-    # this should give up after retrying multiple times
-    mocked_rmtree.side_effect = OSError(
-        "Couldn't delete file yet, this error won't go away after first attempt"
-    )
-    with pytest.raises(OSError):
-        robust_rmtree(name, max_timeout=0.04)
-
-    # clear the side effect (breaks the tear-down otherwise)
-    mocker.stop(mocked_rmtree)
-    # use the real method to remove the temp folder we created for this test
-    robust_rmtree(name)
-    assert not Path(name).exists()