Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 28 additions & 5 deletions src/_pytest/pathlib.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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]:
Expand Down
50 changes: 50 additions & 0 deletions testing/test_tmpdir.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from collections.abc import Callable
import dataclasses
import errno
import os
from pathlib import Path
import shutil
Expand Down Expand Up @@ -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
Expand Down
Loading