Implements is_hidden

Also addresses
jupyter-server/jupyter_server#1224 for our
implementation.

Also does:
- Improves tests by using an actual jupyter server instance instead of
  calling CM methods directly (gains config + handler logic)
- Moves conftest to root, as this is needed by pytest for naming plugins
- FSManager init calls super: this enables the traitlets config system

jupyterfs/tests/conftest.py → conftest.py

@@ -6,6 +6,8 @@
 PLATFORM_INFO = {"darwin": "mac", "linux": "linux", "win32": "windows"}
 PLATFORMS = set(PLATFORM_INFO.keys())
 
+pytest_plugins = ["pytest_jupyter.jupyter_server"]
+
 
 def pytest_configure(config):
     # register the platform markers

jupyterfs/fsmanager.py

@@ -9,8 +9,11 @@
 from datetime import datetime
 from fs import errors, open_fs
 from fs.base import FS
+from fs.errors import NoSysPath, ResourceNotFound
 import fs.path
 import mimetypes
+import pathlib
+import stat
 from tornado import web
 
 import nbformat
@@ -114,7 +117,8 @@ def open_fs(cls, *args, **kwargs):
     def init_fs(cls, pyfs_class, *args, **kwargs):
         return cls(pyfs_class(*args, **kwargs))
 
-    def __init__(self, pyfs, *args, default_writable=True, **kwargs):
+    def __init__(self, pyfs, *args, default_writable=True, parent=None, **kwargs):
+        super().__init__(parent=parent)
         self._default_writable = default_writable
         if isinstance(pyfs, str):
             # pyfs is an opener url
@@ -132,15 +136,61 @@ def __init__(self, pyfs, *args, default_writable=True, **kwargs):
     def _checkpoints_class_default(self):
         return NullCheckpoints
 
+    def _is_path_hidden(self, path):
+        """Does the specific API style path correspond to a hidden node?
+        Args:
+            path (str): The path to check.
+        Returns:
+            hidden (bool): Whether the path is hidden.
+        """
+        # We do not know the OS of the actual FS, so let us be careful
+
+        # We treat entries with leading . in the name as hidden (unix convention)
+        # We can (and should) check this even if the path does not exist
+        if pathlib.PurePosixPath(path).name.startswith("."):
+            return True
+
+        try:
+            info = self._pyfilesystem_instance.getinfo(path, namespaces=("stat",))
+            # Check Windows flag:
+            if info.get("stat", 'st_file_attributes', 0) & stat.FILE_ATTRIBUTE_HIDDEN:
+                return True
+            # Check Mac flag
+            if info.get("stat", 'st_flags', 0) & stat.UF_HIDDEN:
+                return True
+            if info.get('basic', 'is_dir'):
+                # The `access` namespace does not have the facilities for actually checking
+                # whether the current user can read/exec the dir, so we use systempath
+                import os
+                syspath = self._pyfilesystem_instance.getsyspath(path)
+                if not os.access(syspath, os.X_OK | os.R_OK):
+                    return True
+
+        except ResourceNotFound:
+            pass  # if path does not exist (and no leading .), it is not considered hidden
+        except NoSysPath:
+            pass  # if we rely on syspath, and FS does not have it, assume not hidden
+        except Exception:
+            self.log.exception(f"Failed to check if path is hidden: {path!r}")
+        return False
+
     def is_hidden(self, path):
         """Does the API style path correspond to a hidden directory or file?
         Args:
             path (str): The path to check.
         Returns:
-            hidden (bool): Whether the path exists and is hidden.
+            hidden (bool): Whether the path or any of its parents are hidden.
         """
-        # TODO hidden
-        return not self._pyfilesystem_instance.exists(path)
+        ppath = pathlib.PurePosixPath(path)
+        # Path checks are quick, so we do it first to avoid unnecessary stat calls 
+        if any(part.startswith(".") for part in ppath.parts):
+            return True
+        while ppath.parents:
+            if self._is_path_hidden(str(path)):
+                return True
+            ppath = ppath.parent
+        return False
+
 
     def file_exists(self, path):
         """Returns True if the file exists, else returns False.
@@ -226,12 +276,11 @@ def _dir_model(self, path, content=True):
 
         if not self._pyfilesystem_instance.isdir(path):
             raise web.HTTPError(404, four_o_four)
-        # TODO hidden
-        # elif is_hidden(os_path, self.root_dir) and not self.allow_hidden:
-        #     self.log.info("Refusing to serve hidden directory %r, via 404 Error",
-        #         os_path
-        #     )
-        #     raise web.HTTPError(404, four_o_four)
+        elif not self.allow_hidden and self.is_hidden(path):
+            self.log.info("Refusing to serve hidden directory %r, via 404 Error",
+                path
+            )
+            raise web.HTTPError(404, four_o_four)
 
         model = self._base_model(path)
         model["type"] = "directory"
@@ -249,11 +298,10 @@ def _dir_model(self, path, content=True):
                     continue
 
                 if self.should_list(name):
-                    # TODO hidden
-                    # if self.allow_hidden or not is_file_hidden(os_path, stat_res=st):
-                    contents.append(
-                        self.get(path="%s/%s" % (path, name), content=False)
-                    )
+                    if self.allow_hidden or not self._is_path_hidden(name):
+                        contents.append(
+                            self.get(path="%s/%s" % (path, name), content=False)
+                        )
             model["format"] = "json"
         return model
 
@@ -366,9 +414,8 @@ def get(self, path, content=True, type=None, format=None):
 
     def _save_directory(self, path, model):
         """create a directory"""
-        # TODO hidden
-        # if is_hidden(path, self.root_dir) and not self.allow_hidden:
-        #     raise web.HTTPError(400, u'Cannot create hidden directory %r' % path)
+        if not self.allow_hidden and self.is_hidden(path):
+            raise web.HTTPError(400, f'Cannot create directory {path!r}')
         if not self._pyfilesystem_instance.exists(path):
             self._pyfilesystem_instance.makedir(path)
         elif not self._pyfilesystem_instance.isdir(path):

jupyterfs/metamanager.py

@@ -94,7 +94,10 @@ def initResource(self, *resources, options={}):
                     # create new cm
                     default_writable = resource.get("defaultWritable", True)
                     managers[_hash] = FSManager(
-                        urlSubbed, default_writable=default_writable, **self._pyfs_kw
+                        urlSubbed,
+                        default_writable=default_writable,
+                        parent=self,
+                        **self._pyfs_kw
                     )
                     init = True
 

jupyterfs/pathutils.py

@@ -139,12 +139,9 @@ def path_old_new(method_name, returns_model):
 
     def _wrapper(self, old_path, new_path, *args, **kwargs):
         old_prefix, old_mgr, old_mgr_path = _resolve_path(old_path, self._managers)
-        new_prefix, new_mgr, new_mgr_path = _resolve_path(
-            new_path,
-            self._managers,
-        )
+        new_prefix, new_mgr, new_mgr_path = _resolve_path(new_path, self._managers)
         if old_mgr is not new_mgr:
-            # TODO: Consider supporting this via get+delete+save.
+            # TODO: Consider supporting this via get+save+delete.
             raise HTTPError(
                 400,
                 "Can't move files between backends yet ({old} -> {new})".format(

jupyterfs/tests/test_extension.py

@@ -10,8 +10,10 @@
 from unittest.mock import MagicMock
 import tornado.web
 
+import pytest
+
 from jupyterfs.extension import _load_jupyter_server_extension
-from jupyterfs.metamanager import MetaManagerHandler
+from jupyterfs.metamanager import MetaManagerHandler, MetaManager
 
 
 class TestExtension:
@@ -20,12 +22,15 @@ def test_load_jupyter_server_extension(self):
         m = MagicMock()
 
         m.web_app.settings = {}
+        m.contents_manager = MetaManager()
         m.web_app.settings["base_url"] = "/test"
         _load_jupyter_server_extension(m)
 
-    def test_get_handler(self):
-        app = tornado.web.Application()
+    @pytest.mark.asyncio
+    async def test_get_handler(self):
+        contents_manager = MetaManager()
+        app = tornado.web.Application(contents_manager=contents_manager)
         m = MagicMock()
         h = MetaManagerHandler(app, m)
         h._transforms = []
-        h.get()
+        await h.get()