Make preferred_dir content manager trait (#983)

* fix terminal test on Windows

* Make preferred_dir content manager trait

Move the preferred_dir trait to content manager, but keep the old one for backwards compatibility. The new location should also cause the value to be read later in the init, allowing us to avoid the deferred validation. Also fixes an issue with escaping the root dir on Windows if an absolute path is passed.

* Remove leading "/" from preferred_dir

Also validates set value to strip any leading slashes

* Allow filemanager to have OS path set

Will transparently convert to API path in validator.

* fix mypy

* fix ruff

Author: vidartf

jupyter_server/serverapp.py

@@ -1647,22 +1647,11 @@ def _normalize_dir(self, value):
             value = os.path.abspath(value)
         return value
 
-    # Because the validation of preferred_dir depends on root_dir and validation
-    # occurs when the trait is loaded, there are times when we should defer the
-    # validation of preferred_dir (e.g., when preferred_dir is defined via CLI
-    # and root_dir is defined via a config file).
-    _defer_preferred_dir_validation = False
-
     @validate("root_dir")
     def _root_dir_validate(self, proposal):
         value = self._normalize_dir(proposal["value"])
         if not os.path.isdir(value):
             raise TraitError(trans.gettext("No such directory: '%r'") % value)
-
-        if self._defer_preferred_dir_validation:
-            # If we're here, then preferred_dir is configured on the CLI and
-            # root_dir is configured in a file
-            self._preferred_dir_validation(self.preferred_dir, value)
         return value
 
     preferred_dir = Unicode(
@@ -1679,39 +1668,8 @@ def _preferred_dir_validate(self, proposal):
         value = self._normalize_dir(proposal["value"])
         if not os.path.isdir(value):
             raise TraitError(trans.gettext("No such preferred dir: '%r'") % value)
-
-        # Before we validate against root_dir, check if this trait is defined on the CLI
-        # and root_dir is not.  If that's the case, we'll defer it's further validation
-        # until root_dir is validated or the server is starting (the latter occurs when
-        # the default root_dir (cwd) is used).
-        cli_config = self.cli_config.get("ServerApp", {})
-        if "preferred_dir" in cli_config and "root_dir" not in cli_config:
-            self._defer_preferred_dir_validation = True
-
-        if not self._defer_preferred_dir_validation:  # Validate now
-            self._preferred_dir_validation(value, self.root_dir)
         return value
 
-    def _preferred_dir_validation(self, preferred_dir: str, root_dir: str) -> None:
-        """Validate preferred dir relative to root_dir - preferred_dir must be equal or a subdir of root_dir"""
-        if not preferred_dir.startswith(root_dir):
-            raise TraitError(
-                trans.gettext(
-                    "preferred_dir must be equal or a subdir of root_dir. preferred_dir: '%r' root_dir: '%r'"
-                )
-                % (preferred_dir, root_dir)
-            )
-        self._defer_preferred_dir_validation = False
-
-    @observe("root_dir")
-    def _root_dir_changed(self, change):
-        self._root_dir_set = True
-        if not self.preferred_dir.startswith(change["new"]):
-            self.log.warning(
-                trans.gettext("Value of preferred_dir updated to use value of root_dir")
-            )
-            self.preferred_dir = change["new"]
-
     @observe("server_extensions")
     def _update_server_extensions(self, change):
         self.log.warning(_i18n("server_extensions is deprecated, use jpserver_extensions"))
@@ -1893,6 +1851,9 @@ def init_configurables(self):
             parent=self,
             log=self.log,
         )
+        # Trigger a default/validation here explicitly while we still support the
+        # deprecated trait on ServerApp (FIXME remove when deprecation finalized)
+        self.contents_manager.preferred_dir
         self.session_manager = self.session_manager_class(
             parent=self,
             log=self.log,
@@ -2508,10 +2469,6 @@ def initialize(
         # Parse command line, load ServerApp config files,
         # and update ServerApp config.
         super().initialize(argv=argv)
-        if self._defer_preferred_dir_validation:
-            # If we're here, then preferred_dir is configured on the CLI and
-            # root_dir has the default value (cwd)
-            self._preferred_dir_validation(self.preferred_dir, self.root_dir)
         if self._dispatching:
             return
         # initialize io loop as early as possible,

jupyter_server/services/contents/fileio.py

@@ -254,6 +254,9 @@ def _get_os_path(self, path):
         404: if path is outside root
         """
         root = os.path.abspath(self.root_dir)  # type:ignore
+        # to_os_path is not safe if path starts with a drive, since os.path.join discards first part
+        if os.path.splitdrive(path)[0]:
+            raise HTTPError(404, "%s is not a relative API path" % path)
         os_path = to_os_path(path, root)
         if not (os.path.abspath(os_path) + os.path.sep).startswith(root):
             raise HTTPError(404, "%s is outside root contents directory" % path)

jupyter_server/services/contents/filemanager.py

@@ -7,7 +7,9 @@
 import shutil
 import stat
 import sys
+import warnings
 from datetime import datetime
+from pathlib import Path
 
 import nbformat
 from anyio.to_thread import run_sync
@@ -19,6 +21,7 @@
 from jupyter_server import _tz as tz
 from jupyter_server.base.handlers import AuthenticatedFileHandler
 from jupyter_server.transutils import _i18n
+from jupyter_server.utils import to_api_path
 
 from .filecheckpoints import AsyncFileCheckpoints, FileCheckpoints
 from .fileio import AsyncFileManagerMixin, FileManagerMixin
@@ -55,6 +58,34 @@ def _validate_root_dir(self, proposal):
             raise TraitError("%r is not a directory" % value)
         return value
 
+    @default("preferred_dir")
+    def _default_preferred_dir(self):
+        try:
+            value = self.parent.preferred_dir
+            if value == self.parent.root_dir:
+                value = None
+        except AttributeError:
+            pass
+        else:
+            if value is not None:
+                warnings.warn(
+                    "ServerApp.preferred_dir config is deprecated in jupyter-server 2.0. Use FileContentsManager.preferred_dir instead",
+                    FutureWarning,
+                    stacklevel=3,
+                )
+                try:
+                    path = Path(value)
+                    return path.relative_to(self.root_dir).as_posix()
+                except ValueError:
+                    raise TraitError("%s is outside root contents directory" % value) from None
+        return ""
+
+    @validate("preferred_dir")
+    def _validate_preferred_dir(self, proposal):
+        # It should be safe to pass an API path through this method:
+        proposal["value"] = to_api_path(proposal["value"], self.root_dir)
+        return super()._validate_preferred_dir(proposal)
+
     @default("checkpoints_class")
     def _checkpoints_class_default(self):
         return FileCheckpoints

jupyter_server/services/contents/manager.py

@@ -7,6 +7,7 @@
 import warnings
 from fnmatch import fnmatch
 
+from jupyter_client.utils import run_sync
 from jupyter_events import EventLogger
 from nbformat import ValidationError, sign
 from nbformat import validate as validate_nb
@@ -64,10 +65,30 @@ def emit(self, data):
 
     root_dir = Unicode("/", config=True)
 
+    preferred_dir = Unicode(
+        "",
+        config=True,
+        help=_i18n(
+            "Preferred starting directory to use for notebooks. This is an API path (`/` separated, relative to root dir)"
+        ),
+    )
+
+    @validate("preferred_dir")
+    def _validate_preferred_dir(self, proposal):
+        value = proposal["value"].strip("/")
+        try:
+            dir_exists = run_sync(self.dir_exists)(value)
+        except HTTPError as e:
+            raise TraitError(e.log_message) from e
+        if not dir_exists:
+            raise TraitError(_i18n("Preferred directory not found: %r") % value)
+        return value
+
     allow_hidden = Bool(False, config=True, help="Allow access to hidden files")
 
     notary = Instance(sign.NotebookNotary)
 
+    @default("notary")
     def _notary_default(self):
         return sign.NotebookNotary(parent=self)
 

tests/test_gateway.py

@@ -427,7 +427,7 @@ async def test_gateway_session_lifecycle(init_gateway, jp_root_dir, jp_fetch, cu
     # Validate session lifecycle functions; create and delete.
 
     # create
-    session_id, kernel_id = await create_session(jp_root_dir, jp_fetch, "kspec_foo")
+    session_id, kernel_id = await create_session(jp_fetch, "kspec_foo")
 
     # ensure kernel still considered running
     assert await is_session_active(jp_fetch, session_id) is True
@@ -622,12 +622,12 @@ async def is_session_active(jp_fetch, session_id):
         return False
 
 
-async def create_session(root_dir, jp_fetch, kernel_name):
+async def create_session(jp_fetch, kernel_name):
     """Creates a session for a kernel.  The session is created against the server
     which then uses the gateway for kernel management.
     """
     with mocked_gateway:
-        nb_path = root_dir / "testgw.ipynb"
+        nb_path = "/testgw.ipynb"
         body = json.dumps(
             {"path": str(nb_path), "type": "notebook", "kernel": {"name": kernel_name}}
         )