Merge remote-tracking branch 'origin/main' into subclassing

Author: normanrz

upath/core.py

@@ -97,12 +97,6 @@ class UPath(pathlib.Path):
         "home",
         "expanduser",
         "group",
-        "is_mount",
-        "is_symlink",
-        "is_socket",
-        "is_fifo",
-        "is_block_device",
-        "is_char_device",
         "lchmod",
         "lstat",
         "owner",
@@ -112,8 +106,26 @@ class UPath(pathlib.Path):
     def __new__(cls, *args, **kwargs) -> Union["UPath", pathlib.Path]:
         if issubclass(cls, UPath):
             args_list = list(args)
-            url = args_list.pop(0)
-            url = stringify_path(url)
+            first = args_list.pop(0)
+            if isinstance(first, pathlib.PurePath):
+                # Create a (modified) copy, if first arg is a Path object
+                other = first
+                parts = args_list
+                drv, root, parts = other._parse_args(parts)
+                drv, root, parts = other._flavour.join_parsed_parts(
+                    other._drv, other._root, other._parts, drv, root, parts
+                )
+
+                new_kwargs = getattr(other, "_kwargs", {}).copy()
+                new_kwargs.pop("_url", None)
+                new_kwargs.update(kwargs)
+
+                return other.__class__(
+                    other._format_parsed_parts(drv, root, parts),
+                    **new_kwargs,
+                )
+
+            url = stringify_path(first)
             parsed_url = urllib.parse.urlparse(url)
             for key in ["scheme", "netloc"]:
                 val = kwargs.get(key)
@@ -263,6 +275,27 @@ def is_file(self):
             return True
         return False
 
+    def is_mount(self):
+        return False
+
+    def is_symlink(self):
+        info = self._accessor.info(self)
+        if "islink" in info:
+            return info["islink"]
+        return False
+
+    def is_socket(self):
+        return False
+
+    def is_fifo(self):
+        return False
+
+    def is_block_device(self):
+        return False
+
+    def is_char_device(self):
+        return False
+
     def chmod(self, mod):
         raise NotImplementedError
 

upath/tests/cases.py

@@ -62,28 +62,22 @@ def test_is_file(self):
         assert not self.path.is_file()
 
     def test_is_mount(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_mount()
+        assert self.path.is_mount() is False
 
     def test_is_symlink(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_symlink()
+        assert self.path.is_symlink() is False
 
     def test_is_socket(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_socket()
+        assert self.path.is_socket() is False
 
     def test_is_fifo(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_fifo()
+        assert self.path.is_fifo() is False
 
     def test_is_block_device(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_block_device()
+        assert self.path.is_block_device() is False
 
     def test_is_char_device(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_char_device()
+        assert self.path.is_char_device() is False
 
     def test_iterdir(self, local_testdir):
         pl_path = Path(local_testdir)
@@ -259,3 +253,14 @@ def test_child_path(self):
         assert path_a._drv == path_b._drv
         assert path_a._parts == path_b._parts
         assert path_a._url == path_b._url
+
+    def test_copy_path(self):
+        path = self.path
+        copy_path = UPath(path)
+
+        assert type(path) == type(copy_path)
+        assert str(path) == str(copy_path)
+        assert path._drv == copy_path._drv
+        assert path._root == copy_path._root
+        assert path._parts == copy_path._parts
+        assert path.fs.storage_options == copy_path.fs.storage_options

upath/tests/test_core.py

@@ -185,3 +185,51 @@ def test_pickling_child_path():
     assert path._root == recovered_path._root
     assert path._parts == recovered_path._parts
     assert path.fs.storage_options == recovered_path.fs.storage_options
+
+
+def test_copy_path():
+    path = UPath("gcs://bucket/folder", anon=True)
+    copy_path = UPath(path)
+
+    print(type(path), type(copy_path))
+
+    assert type(path) == type(copy_path)
+    assert str(path) == str(copy_path)
+    assert path._drv == copy_path._drv
+    assert path._root == copy_path._root
+    assert path._parts == copy_path._parts
+    assert path.fs.storage_options == copy_path.fs.storage_options
+
+
+def test_copy_path_posix():
+    path = UPath("/tmp/folder")
+    copy_path = UPath(path)
+
+    assert type(path) == type(copy_path) == type(pathlib.Path(""))
+    assert str(path) == str(copy_path)
+    assert path._drv == copy_path._drv
+    assert path._root == copy_path._root
+    assert path._parts == copy_path._parts
+
+
+def test_copy_path_append():
+    path = UPath("/tmp/folder")
+    copy_path = UPath(path, "folder2")
+
+    assert type(path) == type(copy_path) == type(pathlib.Path(""))
+    assert str(path / "folder2") == str(copy_path)
+
+    path = UPath("/tmp/folder")
+    copy_path = UPath(path, "folder2/folder3")
+
+    assert str(path / "folder2" / "folder3") == str(copy_path)
+
+
+def test_copy_path_append_kwargs():
+    path = UPath("gcs://bucket/folder", anon=True)
+    copy_path = UPath(path, anon=False)
+
+    assert type(path) == type(copy_path)
+    assert str(path) == str(copy_path)
+    assert not copy_path._kwargs["anon"]
+    assert path._kwargs["anon"]