Handle entry inspection errors when walking files

pathspec/util.py

@@ -475,16 +475,25 @@ def _iter_tree_files_next(
 		for node_ent in scan_iter:
 			node_rel = os.path.join(dir_rel, node_ent.name)
 
-			if node_ent.is_dir(follow_symlinks=follow_links):
+			try:
+				is_dir = node_ent.is_dir(follow_symlinks=follow_links)
+				is_file = not is_dir and node_ent.is_file(follow_symlinks=follow_links)
+				is_symlink = not follow_links and not is_dir and not is_file and node_ent.is_symlink()
+			except OSError as e:
+				if on_error is not None:
+					on_error(e)
+				continue
+
+			if is_dir:
 				# Child node is a directory, recurse into it and yield its descendant
 				# files.
 				yield from _iter_tree_files_next(root_full, node_rel, memo, on_error, follow_links)
 
-			elif node_ent.is_file(follow_symlinks=follow_links):
+			elif is_file:
 				# Child node is a file, yield it.
 				yield node_rel
 
-			elif not follow_links and node_ent.is_symlink():
+			elif is_symlink:
 				# Child node is an unfollowed link, yield it.
 				yield node_rel
 

tests/test_01_util.py

@@ -19,6 +19,7 @@
 	ClassVar,
 	Optional)  # Replaced by `X | None` in 3.10.
 from unittest.mock import (
+	MagicMock,
 	patch)
 
 from pathspec.patterns.gitignore.basic import (
@@ -188,6 +189,39 @@ def tearDown(self) -> None:
 		"""
 		shutil.rmtree(self.temp_dir)
 
+	def test_file_inspection_error(self):
+		"""Route errors inspecting an entry through the configured error handler."""
+		for operation in ('is_dir', 'is_file', 'is_symlink'):
+			for handling in ('ignore', 'collect', 'raise'):
+				with self.subTest(operation=operation, handling=handling):
+					error = PermissionError(errno.EACCES, 'Permission denied', 'blocked')
+					blocked = MagicMock(name='blocked entry')
+					blocked.name = 'blocked'
+					blocked.is_dir.return_value = False
+					blocked.is_file.return_value = False
+					getattr(blocked, operation).side_effect = error
+					file = MagicMock(name='readable entry')
+					file.name = 'readable.txt'
+					file.is_dir.return_value = False
+					file.is_file.return_value = True
+					scan = MagicMock()
+					scan.__iter__.return_value = iter([blocked, file])
+					errors = []
+					def on_error(exc):
+						if handling == 'raise':
+							raise exc
+						errors.append(exc)
+					with patch('pathspec.util.os.scandir', return_value=scan):
+						files = iter_tree_files(
+							self.temp_dir, follow_links=False, on_error=None if handling == 'ignore' else on_error)
+						if handling == 'raise':
+							with self.assertRaises(PermissionError) as caught:
+								list(files)
+							self.assertIs(caught.exception, error)
+						else:
+							self.assertEqual(list(files), ['readable.txt'])
+							self.assertEqual(errors, [error] if handling == 'collect' else [])
+
 	def test_01_files_1_entries(self):
 		"""
 		Tests to make sure all files are found.