Fix NotADirectoryError (#74)

* upath: use NotADirectoryError instead of custom NotDirectoryError

* tests: add test raising NotADirectoryError

* upath.implementations.http: fix listdir behavior

Author: ap--

upath/core.py

@@ -12,8 +12,6 @@
 )
 from fsspec.utils import stringify_path
 
-from upath.errors import NotDirectoryError
-
 
 class _FSSpecAccessor:
     __slots__ = ("_fs",)
@@ -36,7 +34,17 @@ def stat(self, path, **kwargs):
         return self._fs.stat(self._format_path(path), **kwargs)
 
     def listdir(self, path, **kwargs):
-        return self._fs.listdir(self._format_path(path), **kwargs)
+        p_fmt = self._format_path(path)
+        contents = self._fs.listdir(p_fmt, **kwargs)
+        if len(contents) == 0 and not self._fs.isdir(p_fmt):
+            raise NotADirectoryError
+        elif (
+            len(contents) == 1
+            and contents[0]["name"] == p_fmt
+            and contents[0]["type"] == "file"
+        ):
+            raise NotADirectoryError
+        return contents
 
     def glob(self, _path, path_pattern, **kwargs):
         return self._fs.glob(self._format_path(path_pattern), **kwargs)
@@ -326,10 +334,8 @@ def rmdir(self, recursive=True):
         """Add warning if directory not empty
         assert is_dir?
         """
-        try:
-            assert self.is_dir()
-        except AssertionError:
-            raise NotDirectoryError
+        if not self.is_dir():
+            raise NotADirectoryError
         self._accessor.rm(self, recursive=recursive)
 
     def chmod(self, mode, *, follow_symlinks=True):

upath/errors.py

@@ -1,2 +1,13 @@
-class NotDirectoryError(Exception):
-    pass
+import warnings
+
+
+def __getattr__(name):
+    if name == "NotDirectoryError":
+        warnings.warn(
+            "upath.errors.NotDirectoryError is deprecated. "
+            "Use NotADirectoryError instead",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return NotADirectoryError
+    raise AttributeError(name)

upath/implementations/http.py

@@ -25,13 +25,12 @@ def is_file(self):
             return False
 
     def _path_type(self):
-        info = self._accessor.info(self)
-        if (
-            info["type"] == "directory"
-            or next(self.iterdir(), None) is not None
-        ):
+        try:
+            next(self.iterdir())
+        except (StopIteration, NotADirectoryError):
+            return "file"
+        else:
             return "directory"
-        return "file"
 
     def _sub_path(self, name):
         """

upath/tests/cases.py

@@ -334,3 +334,14 @@ def test_repr_after_with_name(self):
     def test_repr_after_with_suffix(self):
         p = self.path.joinpath("file.txt").with_suffix(".zip")
         assert "file.zip" in repr(p)
+
+    def test_rmdir_no_dir(self):
+        p = self.path.joinpath("file1.txt")
+        with pytest.raises(NotADirectoryError):
+            p.rmdir()
+
+    def test_iterdir_no_dir(self):
+        p = self.path.joinpath("file1.txt")
+        assert p.is_file()
+        with pytest.raises(NotADirectoryError):
+            _ = list(p.iterdir())

upath/tests/implementations/test_azure.py

@@ -1,6 +1,5 @@
 import pytest
 from upath import UPath
-from upath.errors import NotDirectoryError
 from upath.implementations.cloud import AzurePath
 
 from ..cases import BaseTests
@@ -36,5 +35,5 @@ def test_rmdir(self):
         new_dir.rmdir()
         assert not new_dir.exists()
 
-        with pytest.raises(NotDirectoryError):
+        with pytest.raises(NotADirectoryError):
             (self.path / "a" / "file.txt").rmdir()

upath/tests/implementations/test_gcs.py

@@ -2,7 +2,6 @@
 
 from upath import UPath
 from upath.implementations.cloud import GCSPath
-from upath.errors import NotDirectoryError
 from ..cases import BaseTests
 from ..utils import skip_on_windows
 
@@ -28,5 +27,5 @@ def test_rmdir(self):
         mock_dir.fs.invalidate_cache()
         mock_dir.rmdir()
         assert not mock_dir.exists()
-        with pytest.raises(NotDirectoryError):
+        with pytest.raises(NotADirectoryError):
             self.path.joinpath("file1.txt").rmdir()

upath/tests/implementations/test_s3.py

@@ -3,7 +3,6 @@
 import pytest  # noqa: F401
 
 from upath import UPath
-from upath.errors import NotDirectoryError
 from upath.implementations.cloud import S3Path
 from ..cases import BaseTests
 
@@ -31,7 +30,7 @@ def test_rmdir(self):
         mock_dir.joinpath("test.txt").touch()
         mock_dir.rmdir()
         assert not mock_dir.exists()
-        with pytest.raises(NotDirectoryError):
+        with pytest.raises(NotADirectoryError):
             self.path.joinpath("file1.txt").rmdir()
 
     def test_relative_to(self):