Tests run extensions for posixupath

upath/_stat.py

@@ -307,6 +307,42 @@ def st_birthtime(self) -> int | float:
                 pass
         raise AttributeError("birthtime")
 
+    @property
+    def st_atime_ns(self) -> int:
+        """time of last access in nanoseconds"""
+        try:
+            return int(self._info["atime_ns"])
+        except KeyError:
+            pass
+        atime = self.st_atime
+        if isinstance(atime, float):
+            return int(atime * 1e9)
+        return atime * 1_000_000_000
+
+    @property
+    def st_mtime_ns(self) -> int:
+        """time of last modification in nanoseconds"""
+        try:
+            return int(self._info["mtime_ns"])
+        except KeyError:
+            pass
+        mtime = self.st_mtime
+        if isinstance(mtime, float):
+            return int(mtime * 1e9)
+        return mtime * 1_000_000_000
+
+    @property
+    def st_ctime_ns(self) -> int:
+        """time of last change in nanoseconds"""
+        try:
+            return int(self._info["ctime_ns"])
+        except KeyError:
+            pass
+        ctime = self.st_ctime
+        if isinstance(ctime, float):
+            return int(ctime * 1e9)
+        return ctime * 1_000_000_000
+
     # --- extra fields ------------------------------------------------
 
     def __getattr__(self, item):

upath/core.py

@@ -116,6 +116,13 @@ def _raise_unsupported(cls_name: str, method: str) -> NoReturn:
     raise UnsupportedOperation(f"{cls_name}.{method}() is unsupported")
 
 
+class _IncompatibleProtocolError(TypeError, ValueError):
+    """switch to TypeError for incompatible protocols in a backward compatible way.
+
+    !!! Do not use this exception directly !!!
+    """
+
+
 class _UPathMeta(ABCMeta):
     """metaclass for UPath to customize instance creation
 
@@ -419,11 +426,16 @@ def __new__(
             protocol = storage_options.pop("scheme")
 
         # determine the protocol
-        pth_protocol = get_upath_protocol(
-            args[0] if args else "",
-            protocol=protocol,
-            storage_options=storage_options,
-        )
+        try:
+            pth_protocol = get_upath_protocol(
+                args[0] if args else "",
+                protocol=protocol,
+                storage_options=storage_options,
+            )
+        except ValueError as e:
+            if "incompatible with" in str(e):
+                raise _IncompatibleProtocolError(str(e)) from e
+            raise
         # determine which UPath subclass to dispatch to
         upath_cls: type[UPath] | None
         if cls._protocol_dispatch or cls._protocol_dispatch is None:
@@ -1257,6 +1269,14 @@ def move_into(
             target = self.with_segments(target_dir, name)
         return self.move(target)
 
+    def _copy_from(
+        self,
+        source: ReadablePath,
+        follow_symlinks: bool = True,
+        **kwargs: Any,
+    ) -> None:
+        return super()._copy_from(source, follow_symlinks)
+
     # --- WritablePath attributes -------------------------------------
 
     def symlink_to(

upath/extensions.py

@@ -284,22 +284,37 @@ def is_absolute(self) -> bool:
         return self.__wrapped__.is_absolute()
 
     def __eq__(self, other: object) -> bool:
-        return self.__wrapped__.__eq__(other)
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__eq__(other.__wrapped__)
 
     def __hash__(self) -> int:
         return self.__wrapped__.__hash__()
 
+    def __ne__(self, other: object) -> bool:
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__ne__(other.__wrapped__)
+
     def __lt__(self, other: object) -> bool:
-        return self.__wrapped__.__lt__(other)
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__lt__(other.__wrapped__)
 
     def __le__(self, other: object) -> bool:
-        return self.__wrapped__.__le__(other)
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__le__(other.__wrapped__)
 
     def __gt__(self, other: object) -> bool:
-        return self.__wrapped__.__gt__(other)
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__gt__(other.__wrapped__)
 
     def __ge__(self, other: object) -> bool:
-        return self.__wrapped__.__ge__(other)
+        if not isinstance(other, type(self)):
+            return NotImplemented
+        return self.__wrapped__.__ge__(other.__wrapped__)
 
     def resolve(self, strict: bool = False) -> Self:
         return self._from_upath(self.__wrapped__.resolve(strict=strict))
@@ -313,8 +328,11 @@ def lchmod(self, mode: int) -> None:
     def unlink(self, missing_ok: bool = False) -> None:
         self.__wrapped__.unlink(missing_ok=missing_ok)
 
-    def rmdir(self, recursive: bool = True) -> None:  # fixme: non-standard
-        self.__wrapped__.rmdir(recursive=recursive)
+    def rmdir(self, recursive: bool = UNSET_DEFAULT) -> None:  # fixme: non-standard
+        kwargs: dict[str, Any] = {}
+        if recursive is not UNSET_DEFAULT:
+            kwargs["recursive"] = recursive
+        self.__wrapped__.rmdir(**kwargs)
 
     def rename(
         self,
@@ -324,9 +342,14 @@ def rename(
         maxdepth: int | None = UNSET_DEFAULT,
         **kwargs: Any,
     ) -> Self:
+        if recursive is not UNSET_DEFAULT:
+            kwargs["recursive"] = recursive
+        if maxdepth is not UNSET_DEFAULT:
+            kwargs["maxdepth"] = maxdepth
         return self._from_upath(
             self.__wrapped__.rename(
-                target, recursive=recursive, maxdepth=maxdepth, **kwargs
+                target.__wrapped__ if isinstance(target, ProxyUPath) else target,
+                **kwargs,
             )
         )
 

upath/implementations/local.py

@@ -2,6 +2,7 @@
 
 import os
 import pathlib
+import shutil
 import sys
 import warnings
 from collections.abc import Iterator
@@ -196,6 +197,33 @@ def __vfspath__(self) -> str:
     def __open_reader__(self) -> BinaryIO:
         return self.open("rb")
 
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, UPath):
+            return NotImplemented
+        eq_path = super().__eq__(other)
+        if eq_path is NotImplemented:
+            return NotImplemented
+        return (
+            eq_path
+            and self.protocol == other.protocol
+            and self.storage_options == other.storage_options
+        )
+
+    def __ne__(self, other: object) -> bool:
+        if not isinstance(other, UPath):
+            return NotImplemented
+        ne_path = super().__ne__(other)
+        if ne_path is NotImplemented:
+            return NotImplemented
+        return (
+            ne_path
+            or self.protocol != other.protocol
+            or self.storage_options != other.storage_options
+        )
+
+    def __hash__(self) -> int:
+        return super().__hash__()
+
     if sys.version_info >= (3, 14):
 
         def __open_rb__(self, buffering: int = UNSET_DEFAULT) -> BinaryIO:
@@ -316,6 +344,12 @@ def open(
             **fsspec_kwargs,
         )
 
+    def rmdir(self, recursive: bool = UNSET_DEFAULT) -> None:
+        if recursive is UNSET_DEFAULT or not recursive:
+            return super().rmdir()
+        else:
+            shutil.rmtree(self)
+
     if sys.version_info < (3, 14):  # noqa: C901
 
         @overload
@@ -656,7 +690,10 @@ def chmod(
     if not hasattr(pathlib.Path, "_copy_from"):
 
         def _copy_from(
-            self, source: ReadablePath | LocalPath, follow_symlinks: bool = True
+            self,
+            source: ReadablePath | LocalPath,
+            follow_symlinks: bool = True,
+            preserve_metadata: bool = False,
         ) -> None:
             _copy_from: Any = WritablePath._copy_from.__get__(self)
             _copy_from(source, follow_symlinks=follow_symlinks)

upath/tests/cases.py

@@ -10,8 +10,10 @@
 from fsspec import filesystem
 from packaging.version import Version
 
+from upath import UnsupportedOperation
 from upath import UPath
 from upath._stat import UPathStatResult
+from upath.types import StatResultType
 
 
 class BaseTests:
@@ -29,7 +31,12 @@ def test_home(self):
 
     def test_stat(self):
         stat = self.path.stat()
-        assert isinstance(stat, UPathStatResult)
+
+        # for debugging os.stat_result compatibility
+        attrs = {attr for attr in dir(stat) if attr.startswith("st_")}
+        print(attrs)
+
+        assert isinstance(stat, StatResultType)
         assert len(tuple(stat)) == os.stat_result.n_sequence_fields
 
         with warnings.catch_warnings():
@@ -117,7 +124,12 @@ def test_is_absolute(self):
         assert self.path.is_absolute() is True
 
     def test_is_mount(self):
-        assert self.path.is_mount() is False
+        try:
+            self.path.is_mount()
+        except UnsupportedOperation:
+            pytest.skip(f"is_mount() not supported for {type(self.path).__name__}")
+        else:
+            assert self.path.is_mount() is False
 
     def test_is_symlink(self):
         assert self.path.is_symlink() is False
@@ -175,8 +187,10 @@ def test_parents(self):
         assert p.parents[1].name == self.path.name
 
     def test_lchmod(self):
-        with pytest.raises(NotImplementedError):
-            self.path.lchmod(mode=77)
+        try:
+            self.path.lchmod(mode=0o777)
+        except UnsupportedOperation:
+            pass
 
     def test_lstat(self):
         with pytest.warns(UserWarning, match=r"[A-Za-z]+.stat"):
@@ -540,14 +554,16 @@ def test_hashable(self):
         assert hash(self.path)
 
     def test_storage_options_dont_affect_hash(self):
-        p0 = UPath(str(self.path), test_extra=1, **self.path.storage_options)
-        p1 = UPath(str(self.path), test_extra=2, **self.path.storage_options)
+        cls = type(self.path)
+        p0 = cls(str(self.path), test_extra=1, **self.path.storage_options)
+        p1 = cls(str(self.path), test_extra=2, **self.path.storage_options)
         assert hash(p0) == hash(p1)
 
     def test_eq(self):
-        p0 = UPath(str(self.path), test_extra=1, **self.path.storage_options)
-        p1 = UPath(str(self.path), test_extra=1, **self.path.storage_options)
-        p2 = UPath(str(self.path), test_extra=2, **self.path.storage_options)
+        cls = type(self.path)
+        p0 = cls(str(self.path), test_extra=1, **self.path.storage_options)
+        p1 = cls(str(self.path), test_extra=1, **self.path.storage_options)
+        p2 = cls(str(self.path), test_extra=2, **self.path.storage_options)
         assert p0 == p1
         assert p0 != p2
         assert p1 != p2

upath/tests/test_extensions.py

@@ -1,8 +1,14 @@
+import os
+import sys
+
 import pytest
 
+from upath import UnsupportedOperation
 from upath import UPath
 from upath.extensions import ProxyUPath
 from upath.implementations.local import FilePath
+from upath.implementations.local import PosixUPath
+from upath.implementations.local import WindowsUPath
 from upath.implementations.memory import MemoryPath
 from upath.tests.cases import BaseTests
 
@@ -38,6 +44,64 @@ def test_chmod(self):
         self.path.joinpath("file1.txt").chmod(777)
 
 
+class TestProxyPathlibPath(BaseTests):
+    @pytest.fixture(autouse=True)
+    def path(self, local_testdir):
+        self.path = ProxyUPath(f"{local_testdir}")
+        self.prepare_file_system()
+
+    def test_is_ProxyUPath(self):
+        assert isinstance(self.path, ProxyUPath)
+
+    def test_is_not_PosixUPath_WindowsUPath(self):
+        assert not isinstance(self.path, (PosixUPath, WindowsUPath))
+
+    def test_chmod(self):
+        self.path.joinpath("file1.txt").chmod(777)
+
+    @pytest.mark.skipif(
+        sys.version_info < (3, 12), reason="storage options only handled in 3.12+"
+    )
+    def test_eq(self):
+        super().test_eq()
+
+    def test_group(self):
+        pytest.importorskip("grp")
+        self.path.group()
+
+    def test_owner(self):
+        pytest.importorskip("pwd")
+        self.path.owner()
+
+    def test_readlink(self):
+        try:
+            os.readlink
+        except AttributeError:
+            pytest.skip("os.readlink not available on this platform")
+        with pytest.raises((OSError, UnsupportedOperation)):
+            self.path.readlink()
+
+    def test_protocol(self):
+        assert self.path.protocol == ""
+
+    def test_as_uri(self):
+        assert self.path.as_uri().startswith("file://")
+
+    @pytest.mark.xfail(reason="need to revisit relative path .rename")
+    def test_rename2(self):
+        super().test_rename2()
+
+    def test_lstat(self):
+        st = self.path.lstat()
+        assert st is not None
+
+    def test_relative_to(self):
+        base = self.path
+        child = self.path / "folder1" / "file1.txt"
+        relative = child.relative_to(base)
+        assert str(relative) == f"folder1{os.sep}file1.txt"
+
+
 def test_custom_subclass():
 
     class ReversePath(ProxyUPath):