Ignore errors in temporary directory cleanup

news/11394.bugfix.rst

@@ -0,0 +1 @@
+Ignore errors in temporary directory cleanup (show a warning instead).

src/pip/_internal/utils/misc.py

@@ -11,6 +11,7 @@
 import sys
 import sysconfig
 import urllib.parse
+from functools import partial
 from io import StringIO
 from itertools import filterfalse, tee, zip_longest
 from types import TracebackType
@@ -123,33 +124,66 @@ def get_prog() -> str:
 # Retry every half second for up to 3 seconds
 # Tenacity raises RetryError by default, explicitly raise the original exception
 @retry(reraise=True, stop=stop_after_delay(3), wait=wait_fixed(0.5))
-def rmtree(dir: str, ignore_errors: bool = False) -> None:
+def rmtree(
+ dir: str,
+ ignore_errors: bool = False,
+ onexc: Optional[Callable[[Any, Any, Any], Any]] = None,
+) -> None:
+ if ignore_errors:
+ onexc = _onerror_ignore
+ elif onexc is None:
+ onexc = _onerror_reraise
  if sys.version_info >= (3, 12):
- shutil.rmtree(dir, ignore_errors=ignore_errors, onexc=rmtree_errorhandler)
+ shutil.rmtree(dir, onexc=partial(rmtree_errorhandler, onexc=onexc))
  else:
- shutil.rmtree(dir, ignore_errors=ignore_errors, onerror=rmtree_errorhandler)
+ shutil.rmtree(dir, onerror=partial(rmtree_errorhandler, onexc=onexc))
+
+
+def _onerror_ignore(*_args: Any) -> None:
+ pass
+
+
+def _onerror_reraise(*_args: Any) -> None:
+ raise
 
 
 def rmtree_errorhandler(
- func: Callable[..., Any], path: str, exc_info: Union[ExcInfo, BaseException]
+ func: Callable[..., Any],
+ path: str,
+ exc_info: Union[ExcInfo, BaseException],
+ *,
+ onexc: Callable[..., Any] = _onerror_reraise,
 ) -> None:
- """On Windows, the files in .svn are read-only, so when rmtree() tries to
- remove them, an exception is thrown. We catch that here, remove the
- read-only attribute, and hopefully continue without problems."""
+ """
+ `rmtree` error handler to 'force' a file remove (i.e. like `rm -f`).
+
+ * If a file is readonly then it's write flag is set and operation is
+ retried.
+
+ * `onerror` is the original callback from `rmtree(... onerror=onerror)`
+ that is chained at the end if the "rm -f" still fails.
+ """
  try:
- has_attr_readonly = not (os.stat(path).st_mode & stat.S_IWRITE)
+ st_mode = os.stat(path).st_mode
  except OSError:
  # it's equivalent to os.path.exists
  return
 
- if has_attr_readonly:
+ if not st_mode & stat.S_IWRITE:
  # convert to read/write
- os.chmod(path, stat.S_IWRITE)
- # use the original function to repeat the operation
- func(path)
- return
- else:
- raise
+ try:
+ os.chmod(path, st_mode | stat.S_IWRITE)
+ except OSError:
+ pass
+ else:
+ # use the original function to repeat the operation
+ try:
+ func(path)
+ return
+ except OSError:
+ pass
+
+ onexc(func, path, exc_info)
 
 
 def display_path(path: str) -> str:

src/pip/_internal/utils/temp_dir.py

@@ -3,8 +3,20 @@
 import logging
 import os.path
 import tempfile
+import traceback
 from contextlib import ExitStack, contextmanager
-from typing import Any, Dict, Generator, Optional, TypeVar, Union
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ Generator,
+ List,
+ Optional,
+ Tuple,
+ Type,
+ TypeVar,
+ Union,
+)
 
 from pip._internal.utils.misc import enum, rmtree
 
@@ -106,6 +118,7 @@ def __init__(
  delete: Union[bool, None, _Default] = _default,
  kind: str = "temp",
  globally_managed: bool = False,
+ ignore_cleanup_errors: bool = True,
  ):
  super().__init__()
 
@@ -128,6 +141,7 @@ def __init__(
  self._deleted = False
  self.delete = delete
  self.kind = kind
+ self.ignore_cleanup_errors = ignore_cleanup_errors
 
  if globally_managed:
  assert _tempdir_manager is not None
@@ -170,7 +184,42 @@ def cleanup(self) -> None:
  self._deleted = True
  if not os.path.exists(self._path):
  return
- rmtree(self._path)
+
+ errors: List[BaseException] = []
+
+ def onerror(
+ func: Callable[[str], Any],
+ path: str,
+ exc_info: Tuple[Type[BaseException], BaseException, Any],
+ ) -> None:
+ """Log a warning for a `rmtree` error and continue"""
+ exc_val = "\n".join(traceback.format_exception_only(*exc_info[:2]))
+ exc_val = exc_val.rstrip() # remove trailing new line
+ if func in (os.unlink, os.remove, os.rmdir):
+ logger.debug(
+ "Failed to remove a temporary file '%s' due to %s.\n",
+ path,
+ exc_val,
+ )
+ else:
+ logger.debug("%s failed with %s.", func.__qualname__, exc_val)
+ errors.append(exc_info[1])
+
+ if self.ignore_cleanup_errors:
+ try:
+ # first try with tenacity; retrying to handle ephemeral errors
+ rmtree(self._path, ignore_errors=False)
+ except OSError:
+ # last pass ignore/log all errors
+ rmtree(self._path, onexc=onerror)
+ if errors:
+ logger.warning(
+ "Failed to remove contents in a temporary directory '%s'.\n"
+ "You can safely remove it manually.",
+ self._path,
+ )
+ else:
+ rmtree(self._path)
 
 
 class AdjacentTempDirectory(TempDirectory):

tests/unit/test_utils.py

@@ -257,9 +257,13 @@ def test_rmtree_errorhandler_reraises_error(tmpdir: Path) -> None:
  except RuntimeError:
  # Make sure the handler reraises an exception
  with pytest.raises(RuntimeError, match="test message"):
- # Argument 3 to "rmtree_errorhandler" has incompatible type "None"; expected
- # "Tuple[Type[BaseException], BaseException, TracebackType]"
- rmtree_errorhandler(mock_func, path, None) # type: ignore[arg-type]
+ # Argument 3 to "rmtree_errorhandler" has incompatible type
+ # "Union[Tuple[Type[BaseException], BaseException, TracebackType],
+ # Tuple[None, None, None]]"; expected "Tuple[Type[BaseException],
+ # BaseException, TracebackType]"
+ rmtree_errorhandler(
+ mock_func, path, sys.exc_info() # type: ignore[arg-type]
+ )
 
  mock_func.assert_not_called()
 

tests/unit/test_utils_temp_dir.py

@@ -4,6 +4,7 @@
 import tempfile
 from pathlib import Path
 from typing import Any, Iterator, Optional, Union
+from unittest import mock
 
 import pytest
 
@@ -274,3 +275,25 @@ def test_tempdir_registry_lazy(should_delete: bool) -> None:
  registry.set_delete("test-for-lazy", should_delete)
  assert os.path.exists(path)
  assert os.path.exists(path) == (not should_delete)
+
+
+def test_tempdir_cleanup_ignore_errors() -> None:
+ os_unlink = os.unlink
+
+ # mock os.unlink to fail with EACCES for a specific filename to simulate
+ # how removing a loaded exe/dll behaves.
+ def unlink(name: str, *args: Any, **kwargs: Any) -> None:
+ if "bomb" in name:
+ raise PermissionError(name)
+ else:
+ os_unlink(name)
+
+ with mock.patch("os.unlink", unlink):
+ with TempDirectory(ignore_cleanup_errors=True) as tmp_dir:
+ path = tmp_dir.path
+ with open(os.path.join(path, "bomb"), "a"):
+ pass
+
+ filename = os.path.join(path, "bomb")
+ assert os.path.isfile(filename)
+ os.unlink(filename)