diff --git a/src/_pytest/pathlib.py b/src/_pytest/pathlib.py index 8eb593d194d..4cafb83a548 100644 --- a/src/_pytest/pathlib.py +++ b/src/_pytest/pathlib.py @@ -6,9 +6,11 @@ import contextlib from enum import Enum from errno import EBADF +from errno import EEXIST from errno import ELOOP from errno import ENOENT from errno import ENOTDIR +from errno import ENOTEMPTY import fnmatch from functools import partial from importlib.machinery import ModuleSpec @@ -119,14 +121,35 @@ def on_rm_rf_error( if isinstance(exc, FileNotFoundError): return False + p = Path(path) + + # If os.unlink/os.remove failed on a directory (e.g. because directory traversal failed + # or d_type could not determine it is a directory), treat it as a directory. + if isinstance(exc, IsADirectoryError) or ( + func in (os.unlink, os.remove) and p.is_dir() + ): + _chmod_rwx(str(p)) + rm_rf(p) + return True + + # If removing a directory fails because it is not empty, restrictive permissions + # on some of its children may have prevented their prior removal. + if func is os.rmdir and getattr(exc, "errno", None) in (ENOTEMPTY, EEXIST): + _chmod_rwx(str(p)) + for root, dirs, files in os.walk(str(p)): + for d in dirs: + _chmod_rwx(os.path.join(root, d)) + for f in files: + _chmod_rwx(os.path.join(root, f)) + rm_rf(p) + return True + if not isinstance(exc, PermissionError): warnings.warn( PytestWarning(f"(rm_rf) error removing {path}\n{type(exc)}: {exc}") ) return False - p = Path(path) - if func in (os.open, os.scandir): # Directory traversal failed (e.g. missing S_IXUSR). Fix permissions # on the path and its parent (bounded by start_path), then remove it @@ -206,6 +229,7 @@ def rm_rf(path: Path) -> None: """Remove the path contents recursively, even if some elements are read-only.""" path = ensure_extended_length_path(path) + _chmod_rwx(str(path)) onerror = partial(on_rm_rf_error, start_path=path) if sys.version_info >= (3, 12): shutil.rmtree(str(path), onexc=onerror) @@ -216,9 +240,8 @@ def rm_rf(path: Path) -> None: def find_prefixed(root: Path, prefix: str) -> Iterator[os.DirEntry[str]]: """Find all elements in root that begin with the prefix, case-insensitive.""" l_prefix = prefix.lower() - for x in os.scandir(root): - if x.name.lower().startswith(l_prefix): - yield x + with os.scandir(root) as it: + yield from [x for x in it if x.name.lower().startswith(l_prefix)] def extract_suffixes(iter: Iterable[os.DirEntry[str]], prefix: str) -> Iterator[str]: diff --git a/testing/test_tmpdir.py b/testing/test_tmpdir.py index 0b33a74b926..1a45cad7f35 100644 --- a/testing/test_tmpdir.py +++ b/testing/test_tmpdir.py @@ -3,6 +3,7 @@ from collections.abc import Callable import dataclasses +import errno import os from pathlib import Path import shutil @@ -757,6 +758,55 @@ def test_on_rm_rf_error_chmod_retry_walks_parents(self, tmp_path: Path) -> None: on_rm_rf_error(os.unlink, str(fn), exc_info, start_path=tmp_path) assert not fn.exists() + def test_on_rm_rf_error_is_a_directory_error(self, tmp_path: Path) -> None: + """When os.unlink is called on a directory (e.g. traversal could not detect it), + on_rm_rf_error removes it via rm_rf and returns True (#15169).""" + subdir = tmp_path / "subdir" + subdir.mkdir() + (subdir / "child.txt").touch() + + with warnings.catch_warnings(record=True) as w: + exc_info = IsADirectoryError() + result = on_rm_rf_error( + os.unlink, str(subdir), exc_info, start_path=tmp_path + ) + assert result is True + assert not subdir.exists() + assert not [x.message for x in w] + + def test_on_rm_rf_error_directory_not_empty(self, tmp_path: Path) -> None: + """When os.rmdir fails with ENOTEMPTY / EEXIST, on_rm_rf_error chmods the tree + and recursively removes it (#15169).""" + parent_dir = tmp_path / "parent_dir" + parent_dir.mkdir() + child_dir = parent_dir / "child_dir" + child_dir.mkdir() + (child_dir / "item.txt").touch() + + with warnings.catch_warnings(record=True) as w: + exc = OSError(errno.ENOTEMPTY, "Directory not empty") + result = on_rm_rf_error(os.rmdir, str(parent_dir), exc, start_path=tmp_path) + assert result is True + assert not parent_dir.exists() + assert not [x.message for x in w] + + def test_find_prefixed_closes_scandir_iterator(self, tmp_path: Path) -> None: + """find_prefixed exhausts and closes the scandir iterator cleanly (#15169).""" + from _pytest.pathlib import find_prefixed + + (tmp_path / "pytest-1").mkdir() + (tmp_path / "pytest-2").mkdir() + (tmp_path / "other-3").mkdir() + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always", ResourceWarning) + items = list(find_prefixed(tmp_path, "pytest-")) + assert len(items) == 2 + resource_warnings = [ + x for x in w if issubclass(x.category, ResourceWarning) + ] + assert len(resource_warnings) == 0 + def attempt_symlink_to(path, to_path): """Try to make a symlink from "path" to "to_path", skipping in case this platform