diff --git a/jupyterfs/tests/conftest.py b/conftest.py
similarity index 93%
rename from jupyterfs/tests/conftest.py
rename to conftest.py
index b4eda17d..203a123e 100644
--- a/jupyterfs/tests/conftest.py
+++ b/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
diff --git a/jupyterfs/fsmanager.py b/jupyterfs/fsmanager.py
index 2d0faa9b..62763644 100644
--- a/jupyterfs/fsmanager.py
+++ b/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):
diff --git a/jupyterfs/metamanager.py b/jupyterfs/metamanager.py
index 3041ee55..728c94c8 100644
--- a/jupyterfs/metamanager.py
+++ b/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
 
diff --git a/jupyterfs/pathutils.py b/jupyterfs/pathutils.py
index f8bdda14..2b8324f8 100644
--- a/jupyterfs/pathutils.py
+++ b/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(
diff --git a/jupyterfs/tests/test_extension.py b/jupyterfs/tests/test_extension.py
index 51c3d7f1..6f9f3d26 100644
--- a/jupyterfs/tests/test_extension.py
+++ b/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:
@@ -19,12 +21,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()
diff --git a/jupyterfs/tests/test_fsmanager.py b/jupyterfs/tests/test_fsmanager.py
index b26403a1..57cc0562 100644
--- a/jupyterfs/tests/test_fsmanager.py
+++ b/jupyterfs/tests/test_fsmanager.py
@@ -5,13 +5,18 @@
 # This file is part of the jupyter-fs library, distributed under the terms of
 # the Apache License 2.0.  The full license can be found in the LICENSE file.
 
-from pathlib import Path
+from contextlib import nullcontext
+import json
+from pathlib import Path, PurePosixPath
 import pytest
 import os
 import shutil
 import socket
 import sys
 
+from jupyter_core.utils import ensure_async
+import tornado.web
+
 from jupyterfs.fsmanager import FSManager
 from .utils import s3, samba
 
@@ -42,37 +47,123 @@
     "writable": True,
 }
 
+configs = [
+    {
+        "ServerApp": {
+            "jpserver_extensions": {
+                "jupyterfs.extension": True
+            },
+            "contents_manager_class": "jupyterfs.metamanager.MetaManager"
+        },
+        "ContentsManager": {
+            "allow_hidden": True
+        }
+    },
+    {
+        "ServerApp": {
+            "jpserver_extensions": {
+                "jupyterfs.extension": True
+            },
+            "contents_manager_class": "jupyterfs.metamanager.MetaManager"
+        },
+        "ContentsManager": {
+            "allow_hidden": False
+        }
+    },
+]
+
+class ContentsClient:
+    def __init__(self, jp_fetch):
+        self.fetch = jp_fetch
+
+    async def set_resources(self, resources):
+        rep = await self.fetch(
+            "/jupyterfs/resources",
+            method='POST',
+            body=json.dumps({
+                'options': {},
+                'resources': resources,
+            })
+        )
+        return json.loads(rep.body)
+
+    async def mkdir(self, path, parents=False):
+        if parents:
+            pp = PurePosixPath(path)
+            for i, p in enumerate(pp.parts):
+                await self.mkdir("/".join(pp.parts[:i]))
+            return
+        rep = await self.fetch(
+            f"/api/contents/{path.strip('/')}",
+            method='PUT',
+            body=json.dumps({'type': 'directory'}),
+        )
+        return json.loads(rep.body)
+
+    async def save(self, path, model):
+        rep = await self.fetch(
+            f"/api/contents/{path}",
+            method='PUT',
+            body=json.dumps(model),
+        )
+        return json.loads(rep.body)
+
+    async def get(self, path):
+        rep = await self.fetch(f"/api/contents/{path}", raise_error=True)
+        return json.loads(rep.body)
+
 
 class _TestBase:
     """Contains tests universal to all PyFilesystemContentsManager flavors"""
 
-    def _createContentsManager(self):
+    @pytest.fixture
+    def resource_uri(self):
         raise NotImplementedError
 
-    def testWriteRead(self):
-        cm = self._createContentsManager()
+    @pytest.mark.parametrize("jp_server_config", configs)
+    async def test_write_read(self, jp_fetch, resource_uri, jp_server_config):
+        allow_hidden = jp_server_config['ContentsManager']['allow_hidden']
 
-        fpaths = [
-            "" + test_fname,
-            "root0/" + test_fname,
-            "root1/leaf1/" + test_fname,
-        ]
+        cc = ContentsClient(jp_fetch)
 
-        # set up dir structure
-        cm._save_directory("root0", None)
-        cm._save_directory("root1", None)
-        cm._save_directory("root1/leaf1", None)
+        resources = await cc.set_resources([{'url': resource_uri}])
+        drive = resources[0]['drive']
 
-        # save to root and tips
-        cm.save(_test_file_model, fpaths[0])
-        cm.save(_test_file_model, fpaths[1])
-        cm.save(_test_file_model, fpaths[2])
+        fpaths = [
+            f"{drive}:{test_fname}",
+            f"{drive}:root0/{test_fname}",
+            f"{drive}:root1/leaf1/{test_fname}",
+        ]
 
-        # read and check
-        assert test_content == cm.get(fpaths[0])["content"]
-        assert test_content == cm.get(fpaths[1])["content"]
-        assert test_content == cm.get(fpaths[2])["content"]
+        hidden_paths = [
+            f"{drive}:root1/leaf1/.hidden.txt",
+            f"{drive}:root1/.leaf1/also_hidden.txt",
+        ]
 
+        # set up dir structure
+        await cc.mkdir(f"{drive}:root0")
+        await cc.mkdir(f"{drive}:root1")
+        await cc.mkdir(f"{drive}:root1/leaf1")
+        if allow_hidden:
+            await cc.mkdir(f"{drive}:root1/.leaf1")
+
+        for p in fpaths:
+            # save to root and tips
+            await cc.save(p, _test_file_model)
+            # read and check
+            assert test_content == (await cc.get(p))["content"]
+
+        for p in hidden_paths:
+            ctx = nullcontext() if allow_hidden else pytest.raises(tornado.httpclient.HTTPClientError)
+            with ctx as c:
+                # save to root and tips
+                await cc.save(p, _test_file_model)
+                # read and check
+                assert test_content == (await cc.get(p))["content"]
+
+            if not allow_hidden:
+                assert c.value.code == 400
+                
 
 class Test_FSManager_osfs(_TestBase):
     """No extra setup required for this test suite"""
@@ -90,12 +181,12 @@ def setup_method(self, method):
     def teardown_method(self, method):
         shutil.rmtree(self._test_dir, ignore_errors=True)
 
-    def _createContentsManager(self):
-        uri = "osfs://{local_dir}".format(local_dir=self._test_dir)
-
-        return FSManager.open_fs(uri)
+    @pytest.fixture
+    def resource_uri(self, tmp_path):
+        yield f"osfs://{tmp_path}"
 
 
+@pytest.mark.skipif(not s3.has_docker_env(), reason="docker env required")
 class Test_FSManager_s3(_TestBase):
     """Tests on an instance of s3proxy running in a docker
     Manual startup of equivalent docker:
@@ -126,7 +217,8 @@ def setup_method(self, method):
     def teardown_method(self, method):
         self._rootDirUtil.delete()
 
-    def _createContentsManager(self):
+    @pytest.fixture
+    def resource_uri(self):
         uri = "s3://{id}:{key}@{bucket}?endpoint_url={url}:{port}".format(
             id=s3.aws_access_key_id,
             key=s3.aws_secret_access_key,
@@ -134,8 +226,7 @@ def _createContentsManager(self):
             url=test_url_s3.strip("/"),
             port=test_port_s3,
         )
-
-        return FSManager.open_fs(uri)
+        yield uri
 
 
 @pytest.mark.darwin
@@ -182,7 +273,8 @@ def teardown_method(self, method):
         # delete any existing root
         self._rootDirUtil.delete()
 
-    def _createContentsManager(self):
+    @pytest.fixture
+    def resource_uri(self):
         uri = "smb://{username}:{passwd}@{host}/{share}?name-port={name_port}".format(
             username=samba.smb_user,
             passwd=samba.smb_passwd,
@@ -190,10 +282,7 @@ def _createContentsManager(self):
             name_port=test_name_port_smb_docker_share,
             share=test_dir,
         )
-
-        cm = FSManager.open_fs(uri)
-        assert cm.dir_exists(".")
-        return cm
+        yield uri
 
 
 @pytest.mark.win32
@@ -221,7 +310,8 @@ def teardown_method(self, method):
         # delete any existing root
         self._rootDirUtil.delete()
 
-    def _createContentsManager(self):
+    @pytest.fixture
+    def resource_uri(self):
         kwargs = dict(
             direct_tcp=test_direct_tcp_smb_os_share,
             host=test_host_smb_os_share,
@@ -239,8 +329,4 @@ def _createContentsManager(self):
             uri = "smb://{username}:{passwd}@{host}/{share}?hostname={hostname}&direct-tcp={direct_tcp}".format(
                 **kwargs
             )
-
-        cm = FSManager.open_fs(uri)
-
-        assert cm.dir_exists(".")
-        return cm
+        yield uri
diff --git a/jupyterfs/tests/utils/s3.py b/jupyterfs/tests/utils/s3.py
index a7104f64..cc251a39 100755
--- a/jupyterfs/tests/utils/s3.py
+++ b/jupyterfs/tests/utils/s3.py
@@ -20,6 +20,13 @@
 aws_secret_access_key = "s3_local"
 
 
+def has_docker_env():
+    try: 
+        docker.from_env(version="auto")
+    except Exception:
+        return False
+    return True
+
 def startServer(s3_port=9000):
     ports = {"80/tcp": s3_port}
 
@@ -65,7 +72,6 @@ def exitHandler():
 
     return container, exitHandler
 
-
 class RootDirUtil:
     def __init__(
         self,
diff --git a/setup.py b/setup.py
index 0f0f5f20..2ef902f1 100644
--- a/setup.py
+++ b/setup.py
@@ -71,6 +71,8 @@
     "pysmb",
     "pytest",
     "pytest-cov",
+    "pytest-asyncio",
+    "pytest-jupyter[server]",
 ]
 
 dev_requires = test_requires + [