diff --git a/src/poetry/core/utils/helpers.py b/src/poetry/core/utils/helpers.py index d8b095537..887bb2532 100644 --- a/src/poetry/core/utils/helpers.py +++ b/src/poetry/core/utils/helpers.py @@ -3,7 +3,9 @@ import os import shutil import stat +import sys import tempfile +import time import unicodedata import warnings @@ -40,9 +42,16 @@ def normalize_version(version: str) -> str: @contextmanager def temporary_directory(*args: Any, **kwargs: Any) -> Iterator[str]: - name = tempfile.mkdtemp(*args, **kwargs) - yield name - safe_rmtree(name) + 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]: @@ -90,10 +99,30 @@ def _on_rm_error(func: Any, path: str | Path, exc_info: Any) -> None: func(path) -def safe_rmtree(path: str | Path) -> None: - if Path(path).is_symlink(): - return os.unlink(str(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) diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py index 5baf098be..af54bc979 100644 --- a/tests/utils/test_helpers.py +++ b/tests/utils/test_helpers.py @@ -1,15 +1,23 @@ from __future__ import annotations import os +import sys +import tempfile from pathlib import Path from stat import S_IREAD +from typing import TYPE_CHECKING 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 @@ -118,3 +126,60 @@ 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()