Fix UPath.rename for absolute paths (#225)

* Fix rename

* use protocol to decide if joinpath needed

* extend rename tests

* upath.core: fix rename protocol comparison if protocols depend on storage_options

* upath: fix SMBPath.rename

* upath: fix rename kwargs for older fsspec versions

* upath: explicitly assert return type in UPath.rename

* upath.implementations.smb: from __future__ import annotations

---------

Co-authored-by: Andreas Poehlmann <ap--@users.noreply.github.com>

upath/core.py

@@ -15,14 +15,17 @@
 from typing import Mapping
 from typing import Sequence
 from typing import TextIO
+from typing import TypedDict
 from typing import TypeVar
 from typing import overload
 from urllib.parse import urlsplit
 
 if sys.version_info >= (3, 11):
     from typing import Self
+    from typing import Unpack
 else:
     from typing_extensions import Self
+    from typing_extensions import Unpack
 
 from fsspec.registry import get_filesystem_class
 from fsspec.spec import AbstractFileSystem
@@ -91,6 +94,11 @@ def _make_instance(cls, args, kwargs):
     return cls(*args, **kwargs)
 
 
+class _UPathRenameParams(TypedDict, total=False):
+    recursive: bool
+    maxdepth: int | None
+
+
 # accessors are deprecated
 _FSSpecAccessor = FSSpecAccessorShim
 
@@ -1005,21 +1013,33 @@ def rmdir(self, recursive: bool = True) -> None:  # fixme: non-standard
     def rename(
         self,
         target: str | os.PathLike[str] | UPath,
-        *,
-        recursive: bool = False,
-        maxdepth: int | None = None,
-        **kwargs: Any,
-    ) -> UPath:  # fixme: non-standard
-        target_: UPath
-        if not isinstance(target, UPath):
-            target_ = self.parent.joinpath(target).resolve()
+        **kwargs: Unpack[_UPathRenameParams],  # note: non-standard compared to pathlib
+    ) -> Self:
+        if isinstance(target, str) and self.storage_options:
+            target = UPath(target, **self.storage_options)
+        target_protocol = get_upath_protocol(target)
+        if target_protocol:
+            if target_protocol != self.protocol:
+                raise ValueError(
+                    f"expected protocol {self.protocol!r}, got: {target_protocol!r}"
+                )
+            if not isinstance(target, UPath):
+                target_ = UPath(target, **self.storage_options)
+            else:
+                target_ = target
+            # avoid calling .resolve for subclasses of UPath
+            if ".." in target_.parts or "." in target_.parts:
+                target_ = target_.resolve()
         else:
-            target_ = target
+            parent = self.parent
+            # avoid calling .resolve for subclasses of UPath
+            if ".." in parent.parts or "." in parent.parts:
+                parent = parent.resolve()
+            target_ = parent.joinpath(os.path.normpath(target))
+        assert isinstance(target_, type(self)), "identical protocols enforced above"
         self.fs.mv(
             self.path,
             target_.path,
-            recursive=recursive,
-            maxdepth=maxdepth,
             **kwargs,
         )
         return target_

upath/implementations/smb.py

@@ -1,8 +1,20 @@
+from __future__ import annotations
+
+import os
+import sys
 import warnings
 
+if sys.version_info >= (3, 11):
+    from typing import Self
+    from typing import Unpack
+else:
+    from typing_extensions import Self
+    from typing_extensions import Unpack
+
 import smbprotocol.exceptions
 
 from upath import UPath
+from upath.core import _UPathRenameParams
 
 
 class SMBPath(UPath):
@@ -29,24 +41,21 @@ def iterdir(self):
         else:
             return super().iterdir()
 
-    def rename(self, target, **kwargs):
-        if "recursive" in kwargs:
+    def rename(
+        self,
+        target: str | os.PathLike[str] | UPath,
+        **kwargs: Unpack[_UPathRenameParams],  # note: non-standard compared to pathlib
+    ) -> Self:
+        if kwargs.pop("recursive", None) is not None:
             warnings.warn(
                 "SMBPath.rename(): recursive is currently ignored.",
                 UserWarning,
                 stacklevel=2,
             )
-        if "maxdepth" in kwargs:
+        if kwargs.pop("maxdepth", None) is not None:
             warnings.warn(
                 "SMBPath.rename(): maxdepth is currently ignored.",
                 UserWarning,
                 stacklevel=2,
             )
-        if not isinstance(target, UPath):
-            target = self.parent.joinpath(target).resolve()
-        self.fs.mv(
-            self.path,
-            target.path,
-            **kwargs,
-        )
-        return target
+        return super().rename(target, **kwargs)

upath/tests/cases.py

@@ -282,6 +282,11 @@ def test_rename(self):
         assert target == moved
         assert not upath.exists()
         assert moved.exists()
+        # reverse with an absolute path as str
+        back = moved.rename(str(upath))
+        assert back == upath
+        assert not moved.exists()
+        assert back.exists()
 
     def test_rename2(self):
         upath = self.path.joinpath("folder1/file2.txt")
@@ -291,6 +296,11 @@ def test_rename2(self):
         assert target_path == moved
         assert not upath.exists()
         assert moved.exists()
+        # reverse with a relative path as UPath
+        back = moved.rename(UPath("file2.txt"))
+        assert back == upath
+        assert not moved.exists()
+        assert back.exists()
 
     def test_replace(self):
         pass