tests passing for hdfs and s3

Author: andrewfulton9

upath/core.py

@@ -3,7 +3,10 @@
 from pathlib import *
 import urllib
 import re
+import asyncio
 
+import fsspec
+from fsspec.asyn import AsyncFileSystem
 from fsspec.core import url_to_fs
 from fsspec.registry import filesystem, get_filesystem_class
 
@@ -47,8 +50,6 @@ def wrapper(*args, **kwargs):
                     if func.__name__ == 'mkdir':
                         args = args[:1]
 
-            print('args:', args)
-            print('kwargs:', kwargs)
             return func(*args, **kwargs)
         return wrapper
 
@@ -80,8 +81,10 @@ class PureUniversalPath(PurePath):
 class UPath(pathlib.Path):
 
     def __new__(cls, *args, **kwargs):
+        print('new')
         if cls is UPath:
             new_args = list(args)
+            print(new_args)
             first_arg = new_args.pop(0)
             parsed_url = urllib.parse.urlparse(first_arg)
             for key in ['scheme', 'netloc']:
@@ -92,13 +95,12 @@ def __new__(cls, *args, **kwargs):
                 cls = WindowsPath if os.name == 'nt' else PosixPath
             else:
                 cls = UniversalPath
-                cls._url = parsed_url
-                #kwargs['_url'] = parsed_url
-                cls._kwargs = kwargs
+                # cls._url = parsed_url
+                kwargs['_url'] = parsed_url
                 new_args.insert(0, parsed_url.path)
                 args = tuple(new_args)
 
-        self = cls._from_parts(args, init=False)
+        self = cls._from_parts_init(args, init=False)
         if not self._flavour.is_supported:
             raise NotImplementedError("cannot instantiate %r on your system"
                                       % (cls.__name__,))
@@ -109,68 +111,50 @@ def __new__(cls, *args, **kwargs):
         return self
 
 
-class UniversalPath(Path, PureUniversalPath):
+class UniversalPath(UPath, PureUniversalPath):
 
-    __slots__ = ('_url', '_kwargs', '_closed')
+    __slots__ = ('_url', '_kwargs', '_closed', 'fs')
 
     not_implemented = ['cwd', 'home', 'expanduser', 'group', 'is_mount',
                        'is_symlink', 'is_socket', 'is_fifo', 'is_block_device',
-                       'is_char_device', 'lchmod', 'lstat', 'owner', 'readlink',
-    ]
-    
+                       'is_char_device', 'lchmod', 'lstat', 'owner', 'readlink']
 
     def _init(self, *args, template=None, **kwargs):
         self._closed = False
+        if not kwargs:
+            kwargs = dict(**self._kwargs)
+        else:
+            self._kwargs = dict(**kwargs)
+        self._url = kwargs.pop('_url') if kwargs.get('_url') else None
+
         if not self._root:
             if not self._parts:
                 self._root = '/'
             elif self._parts[0] == '/':
                 self._root = self._parts.pop(0)
         if getattr(self, '_str', None):
             delattr(self, '_str')
-
         if template is not None:
             self._accessor = template._accessor
         else:
-            self._accessor = _FSSpecAccessor(self._url, *args, **self._kwargs)
-
-    @classmethod
-    def _parse_args(cls, args):
-        # This is useful when you don't want to create an instance, just
-        # canonicalize some constructor arguments.
-        parts = []
-        for a in args:
-            if isinstance(a, PurePath):
-                parts += a._parts
-            else:
-                a = os.fspath(a)
-                if isinstance(a, str):
-                    # Force-cast str subclasses to str (issue #21127)
-                    parts.append(str(a))
-                else:
-                    raise TypeError(
-                        "argument should be a str object or an os.PathLike "
-                        "object returning str, not %r"
-                        % type(a))
-        return cls._flavour.parse_parts(parts)
+            self._accessor = _FSSpecAccessor(self._url, *args, **kwargs)
+        self.fs = self._accessor._fs
 
     def __getattribute__(self, item):
         if item == '__class__':
             return UniversalPath
-        not_implemented = getattr(UniversalPath, 'not_implemented')
-        if item in not_implemented:
+        if item in getattr(UniversalPath, 'not_implemented'):
             raise NotImplementedError(f'UniversalPath has no attribute {item}')
         else:
             return super().__getattribute__(item)
 
-    @classmethod
-    def _format_parsed_parts(cls, drv, root, parts):
+    def _format_parsed_parts(self, drv, root, parts):
         join_parts = parts[1:] if parts[0] == '/' else parts
         if (drv or root):
-            path = drv + root + cls._flavour.join(join_parts)
+            path = drv + root + self._flavour.join(join_parts)
         else:
-            path = cls._flavour.join(join_parts)
-        scheme, netloc = cls._url.scheme, cls._url.netloc
+            path = self._flavour.join(join_parts)
+        scheme, netloc = self._url.scheme, self._url.netloc
         scheme = scheme + ':'
         netloc = '//' + netloc if netloc else ''
         formatted = scheme + netloc + path
@@ -230,7 +214,6 @@ def is_dir(self):
 
     def is_file(self):
         info = self._accessor.info(self)
-        print(info)
         if info['type'] == 'file':
             return True
         return False
@@ -251,15 +234,26 @@ def touch(self, trunicate=True, **kwargs):
         self._accessor.touch(self, trunicate=trunicate, **kwargs)
 
     def unlink(self, missing_ok=False):
+        async def async_unlink():
+            async def rm():
+                try:
+                    await self._accessor.rm_file(self)
+                except:
+                    await self._accessor.rm(self, recursive=False)
+                    
+                        
+            coro = rm()
+            done, pending = await asyncio.wait({coro})
+            if coro is done:
+                return
+
+        # print('exists:', self.exists())
         if not self.exists():
             if not missing_ok:
                 raise FileNotFoundError
             else:
                 return
-        try:
-            self._accessor.rm_file(self)
-        except:
-            self._accessor.rm(self, recursive=False)
+        asyncio.run(async_unlink())
 
     def rmdir(self, recursive=True):
         '''Add warning if directory not empty
@@ -271,9 +265,30 @@ def rmdir(self, recursive=True):
             raise NotDirectoryError
         self._accessor.rm(self, recursive=recursive)
 
-    # def mkdir(self, mode=0o777, parents=False, exist_ok=False):
-    #     super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
-
-        
-
+    @classmethod
+    def _from_parts_init(cls, args, init=False):
+        print(args)
+        print(init)
+        return super()._from_parts(args, init=init)
+
+    def _from_parts(self, args, init=True):
+        # We need to call _parse_args on the instance, so as to get the
+        # right flavour.
+        obj = object.__new__(UniversalPath)
+        drv, root, parts = self._parse_args(args)
+        obj._drv = drv
+        obj._root = root
+        obj._parts = parts
+        if init:
+            obj._init(**self._kwargs)
+        return obj
+
+    def _from_parsed_parts(self, drv, root, parts, init=True):
+        obj = object.__new__(UniversalPath)
+        obj._drv = drv
+        obj._root = root
+        obj._parts = parts
+        if init:
+            obj._init(**self._kwargs)
+        return obj
 

upath/tests/conftest.py

@@ -87,16 +87,12 @@ def hdfs(htcluster, tempdir, local_testdir):
     host, user, port = '0.0.0.0', 'hdfs', 9000
     #hdfs = filesystem('hdfs', host=host, user=user, port=port, driver='libhdfs3')
     hdfs = pyarrow.hdfs.connect(host='0.0.0.0', port=9000, user=user)
-    print(tempdir)
     hdfs.mkdir(tempdir, create_parents=True)
     for x in Path(local_testdir).glob('**/*'):
-        print(x)
         if x.is_file():
             text = x.read_text().encode('utf8')
-            print(text)
             if not hdfs.exists(str(x.parent)):
                 hdfs.mkdir(str(x.parent), create_parents=True)
-            print(hdfs.exists(str(x.parent)))
             with hdfs.open(str(x), 'wb') as f:
                 f.write(text)
         else:
@@ -105,8 +101,8 @@ def hdfs(htcluster, tempdir, local_testdir):
     yield host, user, port
 
 
-@pytest.fixture()
-def s3(tempdir, local_testdir):
+@pytest.fixture(scope='session')
+def s3_server():
     # writable local S3 system
     if "BOTO_CONFIG" not in os.environ:  # pragma: no cover
         os.environ["BOTO_CONFIG"] = "/dev/null"
@@ -115,7 +111,7 @@ def s3(tempdir, local_testdir):
     if "AWS_SECRET_ACCESS_KEY" not in os.environ:  # pragma: no cover
         os.environ["AWS_SECRET_ACCESS_KEY"] = "bar"
     requests = pytest.importorskip("requests")
-    s3fs = pytest.importorskip("s3fs")
+    
     pytest.importorskip("moto")
 
     port = 5555
@@ -136,20 +132,24 @@ def s3(tempdir, local_testdir):
     anon = False
     s3so = dict(client_kwargs={'endpoint_url': endpoint_uri},
                 use_listings_cache=False)
+    
+    yield anon, s3so
+    proc.terminate()
+    proc.wait()
+
+@pytest.fixture
+def s3(s3_server, tempdir, local_testdir):
+    s3fs = pytest.importorskip("s3fs")
+    anon, s3so = s3_server
     s3 = s3fs.S3FileSystem(anon=False, **s3so)
     s3.mkdir(tempdir, create_parents=True)
     for x in Path(local_testdir).glob('**/*'):
-        print(x)
         if x.is_file():
             text = x.read_text().encode('utf8')
-            print(text)
             if not s3.exists(str(x.parent)):
                 s3.mkdir(str(x.parent), create_parents=True)
-            print(s3.exists(str(x.parent)))
             with s3.open(str(x), 'wb') as f:
                 f.write(text)
         else:
-            s3.mkdir(str(x))    
+            s3.mkdir(str(x))
     yield anon, s3so
-    proc.terminate()
-    proc.wait()

upath/tests/test_core.py

@@ -116,6 +116,7 @@ def test_lstat(self):
     def test_mkdir(self):
         new_dir = self.path.joinpath('new_dir')
         new_dir.mkdir()
+        print(new_dir._accessor.info(new_dir))
         assert new_dir.exists()
 
     def test_open(self):
@@ -152,15 +153,7 @@ def test_resolve(self):
     def test_rglob(self):
         pass
 
-    def test_rmdir(self, local_testdir):
-        dirname = 'rmdir_test'
-        mock_dir = self.path.joinpath(dirname)
-        mock_dir.mkdir()
-        mock_dir.rmdir()
-        assert not mock_dir.exists()
 
-        with pytest.raises(NotDirectoryError):
-            self.path.joinpath('file1.txt').rmdir()
 
     def test_samefile(self):
         pass
@@ -205,9 +198,7 @@ class TestUPathHDFS(TestUpath):
     @pytest.fixture(autouse=True)
     def path(self, local_testdir, hdfs):
         host, user, port = hdfs
-        print(local_testdir)
         path = f'hdfs:{local_testdir}'
-        print(path)
         self.path = UPath(path,
                           host=host,
                           user=user,
@@ -225,3 +216,52 @@ def path(self, local_testdir, s3):
         anon, s3so = s3
         path = f's3:{local_testdir}'
         self.path = UPath(path, anon=anon, **s3so)
+
+    def test_chmod(self):
+        # todo
+        pass
+
+    def test_mkdir(self):
+        new_dir = self.path.joinpath('new_dir')
+        #new_dir.mkdir()
+        # mkdir doesnt really do anything. A directory only exists in s3
+        # if some file or something is written to it
+        f = new_dir.joinpath('test.txt').touch()
+        assert new_dir.exists()
+
+    def test_rmdir(self, local_testdir):
+        dirname = 'rmdir_test'
+        mock_dir = self.path.joinpath(dirname)
+        f = mock_dir.joinpath('test.txt').touch()
+        mock_dir.rmdir()
+        assert not mock_dir.exists()
+        with pytest.raises(NotDirectoryError):
+            self.path.joinpath('file1.txt').rmdir()
+
+    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_multiple_backend_paths(local_testdir, s3, hdfs):
+    anon, s3so = s3
+    path = f's3:{local_testdir}'
+    s3_path = UPath(path, anon=anon, **s3so)
+    assert s3_path.joinpath('text.txt')._url.scheme == 's3'
+    host, user, port = hdfs
+    path = f'hdfs:{local_testdir}'
+    hdfs_path = UPath(path,
+                      host=host,
+                      user=user,
+                      port=port)
+    assert s3_path.joinpath('text1.txt')._url.scheme == 's3'
+