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
2 changes: 2 additions & 0 deletions CHANGES.rst
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ New features:

Bug fixes:

- Honor `on_error` when opening a directory fails during tree traversal.

- `Pull #123`_: Ignore invalid gitignore bracket ranges for `GitIgnoreSpec`.
- `Pull #128`_: Support POSIX character classes (e.g. `[[:alpha:]]`) in gitignore bracket expressions.
- `Issue #129`_ / `Pull #132`_: Fix GitIgnoreSpec re-including files under an excluded directory
Expand Down
2 changes: 2 additions & 0 deletions CHANGES_1.in.rst
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ New features:

Bug fixes:

- Honor `on_error` when opening a directory fails during tree traversal.

- `Pull #123`_: Ignore invalid gitignore bracket ranges for `GitIgnoreSpec`.
- `Pull #128`_: Support POSIX character classes (e.g. `[[:alpha:]]`) in gitignore bracket expressions.
- `Issue #129`_ / `Pull #132`_: Fix GitIgnoreSpec re-including files under an excluded directory
Expand Down
20 changes: 18 additions & 2 deletions pathspec/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -322,7 +322,15 @@ def _iter_tree_entries_next(
else:
raise RecursionError(real_path=dir_real, first_path=memo[dir_real], second_path=dir_rel)

with os.scandir(dir_full) as scan_iter:
try:
scan_iter = os.scandir(dir_full)
except OSError as e:
del memo[dir_real]
if on_error is not None:
on_error(e)
return

with scan_iter:
node_ent: os.DirEntry
for node_ent in scan_iter:
node_rel = os.path.join(dir_rel, node_ent.name)
Expand Down Expand Up @@ -450,7 +458,15 @@ def _iter_tree_files_next(
else:
raise RecursionError(real_path=dir_real, first_path=memo[dir_real], second_path=dir_rel)

with os.scandir(dir_full) as scan_iter:
try:
scan_iter = os.scandir(dir_full)
except OSError as e:
del memo[dir_real]
if on_error is not None:
on_error(e)
return

with scan_iter:
node_ent: os.DirEntry
for node_ent in scan_iter:
node_rel = os.path.join(dir_rel, node_ent.name)
Expand Down
47 changes: 47 additions & 0 deletions tests/test_01_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
from typing import (
ClassVar,
Optional) # Replaced by `X | None` in 3.10.
from unittest.mock import (
patch)

from pathspec.patterns.gitignore.basic import (
GitIgnoreBasicPattern)
Expand Down Expand Up @@ -161,6 +163,51 @@ def tearDown(self) -> None:
"""
shutil.rmtree(self.temp_dir)

def test_scan_missing_directory(self):
"""Directory scan failures honor the error handler."""
missing = os.path.join(self.temp_dir, 'missing')
for walk in (iter_tree_entries, iter_tree_files):
for mode in ('ignore', 'record', 'raise'):
with self.subTest(walk=walk.__name__, mode=mode):
errors = []

def on_error(error):
errors.append(error)
if mode == 'raise':
raise RuntimeError('stop walking') from error

handler = None if mode == 'ignore' else on_error
if mode == 'raise':
with self.assertRaisesRegex(RuntimeError, 'stop walking'):
list(walk(missing, on_error=handler))
else:
self.assertEqual(list(walk(missing, on_error=handler)), [])
if mode != 'ignore':
self.assertEqual(len(errors), 1)
self.assertEqual(errors[0].errno, errno.ENOENT)

def test_scan_inaccessible_child(self):
"""An inaccessible child does not prevent walking readable siblings."""
self.make_dirs(['blocked'])
self.make_files(['kept.txt'])
scandir = os.scandir
error = PermissionError(errno.EACCES, 'permission denied', 'blocked')

def scan(path):
if os.path.basename(path) == 'blocked':
raise error
return scandir(path)

for walk in (iter_tree_entries, iter_tree_files):
for record in (False, True):
with self.subTest(walk=walk.__name__, record=record):
errors = []
with patch('pathspec.util.os.scandir', side_effect=scan):
results = list(walk(self.temp_dir, on_error=errors.append if record else None))
paths = get_paths_from_entries(results) if walk is iter_tree_entries else set(results)
self.assertIn('kept.txt', paths)
self.assertEqual(errors, [error] if record else [])

def test_01_files_1_entries(self):
"""
Tests to make sure all files are found.
Expand Down