Ensure terminal cwd exists

jupyter_server/terminal/api_handlers.py

@@ -1,5 +1,5 @@
 import json
-import os
+from pathlib import Path
 
 from tornado import web
 
@@ -30,11 +30,23 @@ def post(self):
         # if cwd is a relative path, it should be relative to the root_dir,
         # but if we pass it as relative, it will we be considered as relative to
         # the path jupyter_server was started in
-        if "cwd" in data.keys():
-            if not os.path.isabs(data["cwd"]):
-                cwd = data["cwd"]
-                cwd = os.path.join(self.settings["server_root_dir"], cwd)
-                data["cwd"] = cwd
+        if "cwd" in data:
+            cwd = Path(data["cwd"])
+            if not cwd.resolve().exists():
+                cwd = Path(self.settings["server_root_dir"]).expanduser() / cwd
+                if not cwd.resolve().exists():
+                    cwd = None
+
+            if cwd is None:
+                server_root_dir = self.settings["server_root_dir"]
+                self.log.debug(
+                    f"Failed to find requested terminal cwd: {data.get('cwd')}\n"
+                    f"  It was not found within the server root neither: {server_root_dir}."
+                )
+                del data["cwd"]
+            else:
+                self.log.debug(f"Opening terminal in: {cwd.resolve()!s}")
+                data["cwd"] = str(cwd.resolve())
 
         model = self.terminal_manager.create(**data)
         self.finish(json.dumps(model))

tests/test_terminal.py

@@ -2,6 +2,7 @@
 import json
 import os
 import shutil
+import sys
 
 import pytest
 from tornado.httpclient import HTTPClientError
@@ -18,6 +19,16 @@ def terminal_path(tmp_path):
     shutil.rmtree(str(subdir), ignore_errors=True)
 
 
+@pytest.fixture
+def terminal_root_dir(jp_root_dir):
+    subdir = jp_root_dir.joinpath("terminal_path")
+    subdir.mkdir()
+
+    yield subdir
+
+    shutil.rmtree(str(subdir), ignore_errors=True)
+
+
 CULL_TIMEOUT = 10
 CULL_INTERVAL = 3
 
@@ -137,6 +148,78 @@ async def test_terminal_create_with_cwd(
     await jp_cleanup_subprocesses()
 
 
+async def test_terminal_create_with_relative_cwd(
+    jp_fetch, jp_ws_fetch, jp_root_dir, terminal_root_dir, jp_cleanup_subprocesses
+):
+    resp = await jp_fetch(
+        "api",
+        "terminals",
+        method="POST",
+        body=json.dumps({"cwd": str(terminal_root_dir.relative_to(jp_root_dir))}),
+        allow_nonstandard_methods=True,
+    )
+
+    data = json.loads(resp.body.decode())
+    term_name = data["name"]
+
+    ws = await jp_ws_fetch("terminals", "websocket", term_name)
+
+    ws.write_message(json.dumps(["stdin", "pwd\r\n"]))
+
+    message_stdout = ""
+    while True:
+        try:
+            message = await asyncio.wait_for(ws.read_message(), timeout=5.0)
+        except asyncio.TimeoutError:
+            break
+
+        message = json.loads(message)
+
+        if message[0] == "stdout":
+            message_stdout += message[1]
+
+    ws.close()
+
+    expected = terminal_root_dir.name if sys.platform == "win32" else str(terminal_root_dir)
+    assert expected in message_stdout
+    await jp_cleanup_subprocesses()
+
+
+async def test_terminal_create_with_bad_cwd(jp_fetch, jp_ws_fetch, jp_cleanup_subprocesses):
+    non_existing_path = "/tmp/path/to/nowhere"
+    resp = await jp_fetch(
+        "api",
+        "terminals",
+        method="POST",
+        body=json.dumps({"cwd": non_existing_path}),
+        allow_nonstandard_methods=True,
+    )
+
+    data = json.loads(resp.body.decode())
+    term_name = data["name"]
+
+    ws = await jp_ws_fetch("terminals", "websocket", term_name)
+
+    ws.write_message(json.dumps(["stdin", "pwd\r\n"]))
+
+    message_stdout = ""
+    while True:
+        try:
+            message = await asyncio.wait_for(ws.read_message(), timeout=5.0)
+        except asyncio.TimeoutError:
+            break
+
+        message = json.loads(message)
+
+        if message[0] == "stdout":
+            message_stdout += message[1]
+
+    ws.close()
+
+    assert non_existing_path not in message_stdout
+    await jp_cleanup_subprocesses()
+
+
 async def test_culling_config(jp_server_config, jp_configurable_serverapp):
     terminal_mgr_config = jp_configurable_serverapp().config.ServerApp.TerminalManager
     assert terminal_mgr_config.cull_inactive_timeout == CULL_TIMEOUT