lets paths function the same in upath as fsspec

Author: andrewfulton9

upath/core.py

@@ -16,7 +16,7 @@ def __new__(cls, *args, **kwargs):
             for key in ["scheme", "netloc"]:
                 val = kwargs.get(key)
                 if val:
-                    parsed_url._replace(**{key: val})
+                    parsed_url = parsed_url._replace(**{key: val})
             # treat as local filesystem, return PosixPath or WindowsPath
             impls = list(registry) + list(known_implementations.keys())
             if not parsed_url.scheme or parsed_url.scheme not in impls:

upath/implementations/memory.py

@@ -4,6 +4,13 @@
 class _MemoryAccessor(_FSSpecAccessor):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        self._fs.root_marker = ""
+
+    def _format_path(self, s):
+        """If the filesystem backend doesn't have a root_marker, strip the
+        leading slash of a path
+        """
+        return s
 
 
 class MemoryPath(UniversalPath):

upath/tests/conftest.py

@@ -34,6 +34,7 @@ def pytest_collection_modifyitems(config, items):
 
 class DummyTestFS(LocalFileSystem):
     protocol = "mock"
+    root_marker = '/'
 
 
 @pytest.fixture(scope="session")

upath/tests/implementations/test_hdfs.py

@@ -4,11 +4,11 @@
 
 from upath import UPath
 from upath.implementations.hdfs import HDFSPath
-from upath.tests.test_core import TestUpath
+from upath.tests.cases import BaseTests
 
 
 @pytest.mark.hdfs
-class TestUPathHDFS(TestUpath):
+class TestUPathHDFS(BaseTests):
     @pytest.fixture(autouse=True)
     def path(self, local_testdir, hdfs):
         host, user, port = hdfs
@@ -21,3 +21,6 @@ def test_is_HDFSPath(self):
     def test_chmod(self):
         # todo
         pass
+
+    def test_fsspec_compat(self):
+        pass

upath/tests/implementations/test_memory.py

@@ -3,10 +3,11 @@
 
 from upath import UPath
 from upath.implementations.memory import MemoryPath
-from upath.tests.test_core import TestUpath
+from upath.tests.cases import BaseTests
 
 
-class TestMemoryPath(TestUpath):
+class TestMemoryPath(BaseTests):
+
     @pytest.fixture(autouse=True)
     def path(self, local_testdir):
         path = f"memory:/{local_testdir}"

upath/tests/implementations/test_s3.py

@@ -2,19 +2,24 @@
 """
 import pytest  # noqa: F401
 
+from s3fs import S3FileSystem
+
 from upath import UPath
 from upath.errors import NotDirectoryError
 from upath.implementations.s3 import S3Path
-from upath.tests.test_core import TestUpath
+from upath.tests.cases import BaseTests
 
 
-class TestUPathS3(TestUpath):
+class TestUPathS3(BaseTests):
+    
     @pytest.fixture(autouse=True)
     def path(self, local_testdir, s3):
         anon, s3so = s3
         path = f"s3:/{local_testdir}"
-        print(path)
         self.path = UPath(path, anon=anon, **s3so)
+        self.anon = anon
+        self.s3so = s3so
+
 
     def test_is_S3Path(self):
         assert isinstance(self.path, S3Path)
@@ -62,3 +67,23 @@ def test_glob(self, pathlib_base):
         assert all(
             map(lambda m: m.path in [str(p)[4:] for p in path_glob], mock_glob)
         )
+
+    def test_fsspec_compat(self):
+        fs = self.path.fs
+        scheme = self.path._url.scheme
+        content = b"a,b,c\n1,2,3\n4,5,6"
+
+        p1 = f"{scheme}:///tmp/output1.csv"
+        upath1 = UPath(p1, anon=self.anon, **self.s3so)
+        upath1.write_bytes(content)
+        with fs.open(p1) as f:
+            assert f.read() == content
+        upath1.unlink()
+
+        # write with fsspec, read with upath
+        p2 = f"{scheme}:///tmp/output2.csv"
+        with fs.open(p2, "wb") as f:
+            f.write(content)
+        upath2 = UPath(p2, anon=self.anon, **self.s3so)
+        assert upath2.read_bytes() == content
+        upath2.unlink()

upath/tests/test_core.py

@@ -4,8 +4,10 @@
 import warnings
 
 import pytest
+from fsspec.registry import get_filesystem_class
 
 from upath import UPath
+from upath.tests.cases import BaseTests
 
 
 @pytest.mark.skipif(
@@ -31,212 +33,18 @@ def test_UPath_warning():
         assert "mock" in str(w[-1].message)
 
 
-class TestUpath:
+
+class TestUpath(BaseTests):
+
     @pytest.fixture(autouse=True)
     def path(self, local_testdir):
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
             self.path = UPath(f"mock:{local_testdir}")
 
-    def test_cwd(self):
-        with pytest.raises(NotImplementedError):
-            self.path.cwd()
-
-    def test_home(self):
-        with pytest.raises(NotImplementedError):
-            self.path.home()
-
-    def test_stat(self):
-        stat = self.path.stat()
-        assert stat
-
-    def test_chmod(self):
-        with pytest.raises(NotImplementedError):
-            self.path.joinpath("file1.txt").chmod(777)
-
-    @pytest.mark.parametrize(
-        "url, expected", [("file1.txt", True), ("fakefile.txt", False)]
-    )
-    def test_exists(self, url, expected):
-        path = self.path.joinpath(url)
-        assert path.exists() == expected
-
-    def test_expanduser(self):
-        with pytest.raises(NotImplementedError):
-            self.path.expanduser()
-
-    def test_glob(self, pathlib_base):
-        mock_glob = list(self.path.glob("**.txt"))
-        path_glob = list(pathlib_base.glob("**/*.txt"))
-
-        assert len(mock_glob) == len(path_glob)
-        assert all(
-            map(
-                lambda m: m.path
-                in [str(p).replace("\\", "/") for p in path_glob],
-                mock_glob,
-            )
-        )
-
-    def test_group(self):
-        with pytest.raises(NotImplementedError):
-            self.path.group()
-
-    def test_is_dir(self):
-        assert self.path.is_dir()
-
-        path = self.path.joinpath("file1.txt")
-        assert not path.is_dir()
-
-    def test_is_file(self):
-        path = self.path.joinpath("file1.txt")
-        assert path.is_file()
-        assert not self.path.is_file()
-
-    def test_is_mount(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_mount()
-
-    def test_is_symlink(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_symlink()
-
-    def test_is_socket(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_socket()
-
-    def test_is_fifo(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_fifo()
-
-    def test_is_block_device(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_block_device()
-
-    def test_is_char_device(self):
-        with pytest.raises(NotImplementedError):
-            self.path.is_char_device()
-
-    def test_iterdir(self, local_testdir):
-        pl_path = Path(local_testdir)
-
-        up_iter = list(self.path.iterdir())
-        pl_iter = list(pl_path.iterdir())
-
-        assert len(up_iter) == len(pl_iter)
-        pnames = [p.name for p in pl_iter]
-        assert all(map(lambda x: x.name in pnames, up_iter))
-
-    def test_lchmod(self):
-        with pytest.raises(NotImplementedError):
-            self.path.lchmod(mode=77)
-
-    def test_lstat(self):
-        with pytest.raises(NotImplementedError):
-            self.path.lstat()
-
-    def test_mkdir(self):
-        new_dir = self.path.joinpath("new_dir")
-        new_dir.mkdir()
-        assert new_dir.exists()
-
-    def test_open(self):
+    def test_fsspec_compat(self):
         pass
 
-    def test_owner(self):
-        with pytest.raises(NotImplementedError):
-            self.path.owner()
-
-    def test_read_bytes(self, pathlib_base):
-        mock = self.path.joinpath("file2.txt")
-        pl = pathlib_base.joinpath("file2.txt")
-        assert mock.read_bytes() == pl.read_bytes()
-
-    def test_read_text(self, local_testdir):
-        upath = self.path.joinpath("file1.txt")
-        assert (
-            upath.read_text()
-            == Path(local_testdir).joinpath("file1.txt").read_text()
-        )
-
-    def test_readlink(self):
-        with pytest.raises(NotImplementedError):
-            self.path.readlink()
-
-    @pytest.mark.xfail
-    def test_rename(self):
-        # need to impliment
-        raise False
-
-    def test_replace(self):
-        pass
-
-    def test_resolve(self):
-        pass
-
-    def test_rglob(self):
-        pass
-
-    def test_samefile(self):
-        pass
-
-    def test_symlink_to(self):
-        pass
-
-    def test_touch_unlink(self):
-        path = self.path.joinpath("test_touch.txt")
-        path.touch()
-        assert path.exists()
-        path.unlink()
-        assert not path.exists()
-
-        # should raise FileNotFoundError since file is missing
-        with pytest.raises(FileNotFoundError):
-            path.unlink()
-
-        # file doesn't exists, but missing_ok is True
-        path.unlink(missing_ok=True)
-
-    def test_link_to(self):
-        pass
-
-    def test_write_bytes(self, pathlib_base):
-        fn = "test_write_bytes.txt"
-        s = b"hello_world"
-        path = self.path.joinpath(fn)
-        path.write_bytes(s)
-        assert path.read_bytes() == s
-
-    def test_write_text(self, pathlib_base):
-        fn = "test_write_text.txt"
-        s = "hello_world"
-        path = self.path.joinpath(fn)
-        path.write_text(s)
-        assert path.read_text() == s
-
-    def prepare_file_system(self):
-        self.make_top_folder()
-        self.make_test_files()
-
-    def make_top_folder(self):
-        self.path.mkdir(parents=True, exist_ok=True)
-
-    def make_test_files(self):
-        folder1 = self.path.joinpath("folder1")
-        folder1.mkdir(exist_ok=True)
-        folder1_files = ["file1.txt", "file2.txt"]
-        for f in folder1_files:
-            p = folder1.joinpath(f)
-            p.touch()
-            p.write_text(f)
-
-        file1 = self.path.joinpath("file1.txt")
-        file1.touch()
-        file1.write_text("hello world")
-        file2 = self.path.joinpath("file2.txt")
-        file2.touch()
-        file2.write_bytes(b"hello world")
-
 
 @pytest.mark.hdfs
 def test_multiple_backend_paths(local_testdir, s3, hdfs):

upath/universal_path.py

@@ -46,13 +46,9 @@ def _transform_arg_paths(self, args, kwargs):
         return args, kwargs
 
     def _format_path(self, s):
-        """If the filesystem backend doesn't have a root_marker, strip the
-        leading slash of a path
+        """placeholder method for subclassed filesystems
         """
-        if not self._fs.root_marker:
-            return s.lstrip("/")
-        else:
-            return s
+        return s
 
     def __getattribute__(self, item):
         class_attrs = ["_url", "_fs", "__class__"]