Test/proxy upath copy from 546

upath/extensions.py

@@ -24,6 +24,7 @@
 from upath.core import UPath
 from upath.types import UNSET_DEFAULT
 from upath.types import JoinablePathLike
+from upath.types import OnNameCollisionFunc
 from upath.types import PathInfo
 from upath.types import ReadablePath
 from upath.types import ReadablePathLike
@@ -519,10 +520,14 @@ def write_text(
             data, encoding=encoding, errors=errors, newline=newline
         )
 
-    def _copy_from(
-        self, source: ReadablePath | Self, follow_symlinks: bool = True
-    ) -> None:
-        self.__wrapped__._copy_from(source, follow_symlinks=follow_symlinks)  # type: ignore  # noqa: E501
+    def _copy_from( 
+        self, 
+        source: ReadablePath, 
+        follow_symlinks: bool = True, 
+        on_name_collision: OnNameCollisionFunc | None = None, 
+        **kwargs: Any, 
+    ) -> None: 
+        self.__wrapped__._copy_from(source, follow_symlinks=follow_symlinks, on_name_collision=on_name_collision, **kwargs)  # type: ignore  # noqa: E501
 
     @property
     def anchor(self) -> str:

upath/tests/test_extensions.py

@@ -245,3 +245,33 @@ class MyProxyPath(ProxyUPath):
     # ProxyUPath wraps the underlying path, so it should also raise TypeError
     with pytest.raises(TypeError, match=r".*incompatible with"):
         MyProxyPath(uri, protocol=protocol)
+
+
+def test_proxy_upath_copy_from_local(tmp_path, s3_fixture       ):
+    """Test thats ProxyPath can accept extra kwargs in _copy_from() without breaking on Python 3.14.
+
+    Regression test for https://github.com/fsspec/universal_pathlib/issues/546
+    """
+    bucket, anon, s3so = s3_fixture
+
+    class MyProxyPath(ProxyUPath):
+        def __init__(self, path, **kwargs):
+            self._lazy = UPath(path, **kwargs)
+
+        @property
+        def __wrapped__(self):
+            return self._lazy
+
+        @classmethod
+        def _from_upath(cls, upath, /):
+            obj = object.__new__(cls)
+            obj._lazy = upath
+            return obj
+
+    source = UPath(tmp_path / "test.txt")
+    source.write_text("hello")
+
+    destination = MyProxyPath(f"{bucket}/test_copy_from.txt", anon=anon, **s3so)
+    source.move(destination)
+
+    assert destination.read_text() == "hello"