Fix typing on FSNodeEntry

Author: c0llab0rat0r

ipfshttpclient/filescanner.py

@@ -37,6 +37,47 @@
 HAVE_FWALK_BYTES = HAVE_FWALK and sys.version_info >= (3, 7)
 
 
+class FSNodeType(enum.Enum):
+	FILE = enum.auto()
+	DIRECTORY = enum.auto()
+
+
+#XXX: This should be a generic `ty.NamedTuple` subclass, but GH/python/mypy#685 …
+class FSNodeEntry(ty.Generic[AnyStr]):
+	type: FSNodeType
+	path: AnyStr
+	relpath: AnyStr
+	name: AnyStr
+	parentfd: ty.Optional[int]
+
+	def __init__(
+			self,
+			type: FSNodeType,
+			path: AnyStr,
+			relpath: AnyStr,
+			name: AnyStr,
+			parentfd: ty.Optional[int]) -> None:
+		self.type = type
+		self.path = path
+		self.relpath = relpath
+		self.name = name
+		self.parentfd = parentfd
+
+	def __repr__(self) -> str:
+		return (
+			f'FSNodeEntry('
+			f'type={self.type!r}, '
+			f'path={self.path!r}, '
+			f'relpath={self.relpath!r}, '
+			f'name={self.name!r}, '
+			f'parentfd={self.parentfd!r}'
+			f')'
+		)
+
+	def __str__(self) -> str:
+		return str(self.path)
+
+
 class Matcher(ty.Generic[AnyStr], metaclass=abc.ABCMeta):
 	"""Represents a type that can match on file paths and decide whether they
 	should be included in some file scanning/adding operation"""
@@ -466,26 +507,10 @@ def _recursive_matcher_from_spec(spec: match_spec_t[AnyStr], *,
 		raise MatcherSpecInvalidError(spec)
 
 
-if ty.TYPE_CHECKING:
-	from .filescanner_ty import FSNodeType, FSNodeEntry
-else:
-	class FSNodeType(enum.Enum):
-		FILE = enum.auto()
-		DIRECTORY = enum.auto()
-	
-	FSNodeEntry = ty.NamedTuple("FSNodeEntry", [
-		("type", FSNodeType),
-		("path", AnyStr),
-		("relpath", AnyStr),
-		("name", AnyStr),
-		("parentfd", ty.Optional[int])
-	])
-
-
-class walk(ty.Generator[FSNodeEntry, ty.Any, None], ty.Generic[AnyStr]):
+class walk(ty.Generator[FSNodeEntry[AnyStr], ty.Any, None], ty.Generic[AnyStr]):
 	__slots__ = ("_generator", "_close_fd")
 
-	_generator: ty.Generator[FSNodeEntry, None, None]
+	_generator: ty.Generator[FSNodeEntry[AnyStr], ty.Any, None]
 	_close_fd: ty.Optional[int]
 
 	def __init__(
@@ -577,7 +602,7 @@ def __init__(
 	def __iter__(self) -> 'walk[AnyStr]':
 		return self
 
-	def __next__(self) -> FSNodeEntry:
+	def __next__(self) -> FSNodeEntry[AnyStr]:
 		return next(self._generator)
 
 	def __enter__(self) -> 'walk[AnyStr]':
@@ -586,21 +611,21 @@ def __enter__(self) -> 'walk[AnyStr]':
 	def __exit__(self, *a: ty.Any) -> None:
 		self.close()
 
-	def send(self, value: ty.Any) -> FSNodeEntry:
+	def send(self, value: ty.Any) -> FSNodeEntry[AnyStr]:
 		return self._generator.send(value)
 
 	@ty.overload
 	def throw(self, typ: ty.Type[BaseException],  # noqa: E704
 	          val: ty.Union[BaseException, object] = ...,
-	          tb: ty.Optional[types.TracebackType] = ...) -> FSNodeEntry: ...
+	          tb: ty.Optional[types.TracebackType] = ...) -> FSNodeEntry[AnyStr]: ...
 
 	@ty.overload
 	def throw(self, typ: BaseException, val: None = ...,  # noqa: E704
-	          tb: ty.Optional[types.TracebackType] = ...) -> FSNodeEntry: ...
+	          tb: ty.Optional[types.TracebackType] = ...) -> FSNodeEntry[AnyStr]: ...
 
 	def throw(self, typ: ty.Union[ty.Type[BaseException], BaseException],
 	          val: ty.Union[BaseException, object] = None,
-	          tb: ty.Optional[types.TracebackType] = None) -> FSNodeEntry:
+	          tb: ty.Optional[types.TracebackType] = None) -> FSNodeEntry[AnyStr]:
 		try:
 			if isinstance(typ, type):
 				bt = ty.cast(ty.Type[BaseException], typ)  # type: ignore[redundant-cast]
@@ -653,7 +678,11 @@ def _walk_wide(
 			dot: AnyStr,
 			directory: ty.Union[AnyStr, int],
 			follow_symlinks: bool
-	) -> ty.Iterator[ty.Tuple[AnyStr, ty.List[AnyStr], ty.List[AnyStr], ty.Optional[int]]]:
+	) -> ty.Generator[
+			ty.Tuple[AnyStr, ty.List[AnyStr], ty.List[AnyStr], ty.Optional[int]],
+			ty.Any,
+			None
+	]:
 		"""
 		Return a four-part tuple just like os.fwalk does, even if we won't use os.fwalk.
 
@@ -673,7 +702,7 @@ def _walk(
 			matcher: Matcher[AnyStr],
 			follow_symlinks: bool,
 			intermediate_dirs: bool
-	) -> ty.Generator[FSNodeEntry, ty.Any, None]:
+	) -> ty.Generator[FSNodeEntry[AnyStr], ty.Any, None]:
 		separator = self._walk_separator(matcher=matcher, directory_str=directory_str)
 
 		# TODO: Because os.fsencode can return a byte array, we need to refactor how we use 'sep'
@@ -692,12 +721,12 @@ def _walk(
 
 		# Always report the top-level directory even if nothing therein is matched
 		reported_directories.add(utils.maybe_fsencode("", sep))
-		yield FSNodeEntry(  # type: ignore[misc]  # mypy bug: gh/python/mypy#685
-			type     = FSNodeType.DIRECTORY,
-			path     = prefix[:-len(sep)],  # type: ignore[arg-type]
-			relpath  = dot,  # type: ignore[arg-type]
-			name     = dot,  # type: ignore[arg-type]
-			parentfd = None
+		yield FSNodeEntry(
+			type=FSNodeType.DIRECTORY,
+			path=prefix[:-len(sep)],
+			relpath=dot,
+			name=dot,
+			parentfd=None
 		)
 
 		walk_iter = self._walk_wide(dot=dot, directory=directory, follow_symlinks=follow_symlinks)
@@ -729,32 +758,32 @@ def _walk(
 							parent_dirpath = sep.join(parts[0:(end_offset + 1)])
 							if parent_dirpath not in reported_directories:
 								reported_directories.add(parent_dirpath)
-								yield FSNodeEntry(  # type: ignore[misc]  # mypy bug: gh/python/mypy#685
-									type     = FSNodeType.DIRECTORY,
-									path     = (prefix + parent_dirpath),  # type: ignore[arg-type]
-									relpath  = parent_dirpath,  # type: ignore[arg-type]
-									name     = parts[end_offset],  # type: ignore[arg-type]
-									parentfd = None
+								yield FSNodeEntry(
+									type=FSNodeType.DIRECTORY,
+									path=(prefix + parent_dirpath),
+									relpath=parent_dirpath,
+									name=parts[end_offset],
+									parentfd=None
 								)
 						intermediates_reported = True
 
 					# Report the target file or directory
 					if is_dir:
 						reported_directories.add(filepath)
-						yield FSNodeEntry(  # type: ignore[misc]  # mypy bug: gh/python/mypy#685
-							type     = FSNodeType.DIRECTORY,
-							path     = (prefix + filepath),  # type: ignore[arg-type]
-							relpath  = filepath,  # type: ignore[arg-type]
-							name     = filename,  # type: ignore[arg-type]
-							parentfd = dirfd
+						yield FSNodeEntry(
+							type=FSNodeType.DIRECTORY,
+							path=(prefix + filepath),
+							relpath=filepath,
+							name=filename,
+							parentfd=dirfd
 						)
 					else:
-						yield FSNodeEntry(  # type: ignore[misc]  # mypy bug: gh/python/mypy#685
-							type     = FSNodeType.FILE,
-							path     = (prefix + filepath),  # type: ignore[arg-type]
-							relpath  = filepath,  # type: ignore[arg-type]
-							name     = filename,  # type: ignore[arg-type]
-							parentfd = dirfd
+						yield FSNodeEntry(
+							type=FSNodeType.FILE,
+							path=(prefix + filepath),
+							relpath=filepath,
+							name=filename,
+							parentfd=dirfd
 						)
 		finally:
 			# Make sure the file descriptors bound by `os.fwalk` are freed on error

ipfshttpclient/filescanner_ty.pyi

@@ -410,39 +410,39 @@ def __init__(self, directory: ty.Union[ty.AnyStr, utils.PathLike[ty.AnyStr], int
 	def _body(self) -> gen_bytes_t:
 		"""Streams the contents of the selected directory as binary chunks."""
 		try:
-			for type, path, relpath, name, parentfd in self.scanner:
-				relpath_unicode = os.fsdecode(relpath).replace(os.path.sep, "/")
+			for item in self.scanner:
+				relpath_unicode = os.fsdecode(item.relpath).replace(os.path.sep, "/")
 				short_path = self.name + (("/" + relpath_unicode) if relpath_unicode != "." else "")
 
-				if type is filescanner.FSNodeType.FILE:
+				if item.type is filescanner.FSNodeType.FILE:
 					try:
 						# Only regular files and directories can be uploaded
-						if parentfd is not None:
-							stat_data = os.stat(name, dir_fd=parentfd, follow_symlinks=self.follow_symlinks)
+						if item.parentfd is not None:
+							stat_data = os.stat(item.name, dir_fd=item.parentfd, follow_symlinks=self.follow_symlinks)
 						else:
-							stat_data = os.stat(path, follow_symlinks=self.follow_symlinks)
+							stat_data = os.stat(item.path, follow_symlinks=self.follow_symlinks)
 						if not stat.S_ISREG(stat_data.st_mode):
 							continue
 
 						absolute_path: ty.Optional[str] = None
 						if self.abspath is not None:
-							absolute_path = os.fsdecode(os.path.join(self.abspath, relpath))
+							absolute_path = os.fsdecode(os.path.join(self.abspath, item.relpath))
 
-						if parentfd is None:
-							f_path_or_desc: ty.Union[ty.AnyStr, int] = path
+						if item.parentfd is None:
+							f_path_or_desc: ty.Union[ty.AnyStr, int] = item.path
 						else:
-							f_path_or_desc = os.open(name, os.O_RDONLY | os.O_CLOEXEC, dir_fd=parentfd)
+							f_path_or_desc = os.open(item.name, os.O_RDONLY | os.O_CLOEXEC, dir_fd=item.parentfd)
 						# Stream file to client
 						with open(f_path_or_desc, "rb") as file:
 							yield from self._gen_file(short_path, absolute_path, file)
 					except OSError as e:
 						print(e)
 						# File might have disappeared between `os.walk()` and `open()`
 						pass
-				elif type is filescanner.FSNodeType.DIRECTORY:
+				elif item.type is filescanner.FSNodeType.DIRECTORY:
 					# Generate directory as special empty file
 					yield from self._gen_file(short_path, content_type="application/x-directory")
-			
+
 			yield from self._gen_end()
 		finally:
 			self.scanner.close()

ipfshttpclient/multipart.py

@@ -8,10 +8,29 @@
 from datetime import datetime
 from ipfshttpclient import filescanner
 
+from ipfshttpclient.filescanner import FSNodeEntry
+from ipfshttpclient.filescanner import FSNodeType
+
 
 TEST_FILE_DIR: str = os.path.join(os.path.dirname(__file__), "..", "functional")
 
 
+def test_fs_node_entry_as_repr() -> None:
+	entry = FSNodeEntry(type=FSNodeType.FILE, path='b', relpath='c', name='d', parentfd=123)
+
+	assert (
+		repr(entry)
+		==
+		"FSNodeEntry(type=<FSNodeType.FILE: 1>, path='b', relpath='c', name='d', parentfd=123)"
+	)
+
+
+def test_fs_node_entry_as_str() -> None:
+	entry = FSNodeEntry(type=FSNodeType.FILE, path='b', relpath='c', name='d', parentfd=123)
+
+	assert str(entry) == 'b'
+
+
 @pytest.mark.parametrize("pattern,expected,kwargs", [
 	("literal",                  [r"(?![.])(?s:literal)\Z"], {}),
 	(b"literal",                 [br"(?![.])(?s:literal)\Z"], {}),
@@ -191,27 +210,34 @@ def test_walk_instaclose(mocker):
 	close_spy.assert_called_once()
 
 
-@pytest.mark.parametrize("path,pattern,kwargs,expected", [
-	(TEST_FILE_DIR + os.path.sep + "fake_dir_almost_empty" + os.path.sep, None, {}, [
-		(filescanner.FSNodeType.DIRECTORY, ".", "."),
-		(filescanner.FSNodeType.FILE, ".gitignore", ".gitignore"),
+@pytest.mark.parametrize("path,pattern,expected", [
+	(TEST_FILE_DIR + os.path.sep + "fake_dir_almost_empty" + os.path.sep, None, [
+		(FSNodeType.DIRECTORY, ".", "."),
+		(FSNodeType.FILE, ".gitignore", ".gitignore"),
 	]),
-	(TEST_FILE_DIR + os.path.sep + "fake_dir", ["test2", "test3"], {}, [
-		(filescanner.FSNodeType.DIRECTORY, ".", "."),
-		(filescanner.FSNodeType.DIRECTORY, "test2", "test2"),
-		(filescanner.FSNodeType.DIRECTORY, "test3", "test3"),
+	(TEST_FILE_DIR + os.path.sep + "fake_dir", ["test2", "test3"], [
+		(FSNodeType.DIRECTORY, ".", "."),
+		(FSNodeType.DIRECTORY, "test2", "test2"),
+		(FSNodeType.DIRECTORY, "test3", "test3"),
 	]),
 ])
-def test_walk(monkeypatch, path: str, pattern: None, kwargs: ty.Dict[str, bool], expected: ty.List[filescanner.FSNodeEntry]):
-	result = [(e.type, e.relpath, e.name) for e in filescanner.walk(path, pattern, **kwargs)]
-	assert sorted(result, key=lambda r: r[1]) == expected
+def test_walk(
+		monkeypatch,
+		path: str,
+		pattern: ty.Optional[filescanner.match_spec_t[str]],
+		expected: ty.List[filescanner.FSNodeEntry[str]]
+) -> None:
+	def assert_results() -> None:
+		result = [(e.type, e.relpath, e.name) for e in filescanner.walk(path, pattern)]
+		assert sorted(result, key=lambda r: r[1]) == expected
+
+	assert_results()
 
 	# Check again with plain `os.walk` if the current platform supports `os.fwalk`
 	if filescanner.HAVE_FWALK:
 		monkeypatch.setattr(filescanner, "HAVE_FWALK", False)
-		
-		result = [(e.type, e.relpath, e.name) for e in filescanner.walk(path, pattern, **kwargs)]
-		assert sorted(result, key=lambda r: r[1]) == expected
+
+		assert_results()
 
 
 def test_supports_fd():