Skip to content

Commit

Permalink
Fix problem with deleting temporary folders on Windows (python-poetry…
Browse files Browse the repository at this point in the history
  • Loading branch information
eblis committed Apr 30, 2023
1 parent 4736e76 commit f314393
Show file tree
Hide file tree
Showing 2 changed files with 101 additions and 7 deletions.
43 changes: 36 additions & 7 deletions src/poetry/core/utils/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@
import os
import shutil
import stat
import sys
import tempfile
import time
import unicodedata
import warnings

Expand Down Expand Up @@ -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]:
Expand Down Expand Up @@ -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)


Expand Down
65 changes: 65 additions & 0 deletions tests/utils/test_helpers.py
Original file line number Diff line number Diff line change
@@ -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


Expand Down Expand Up @@ -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()

0 comments on commit f314393

Please sign in to comment.