Add more tests

jupyter_server/pytest_plugin.py

@@ -523,7 +523,10 @@ def inner(nbpath):
 def jp_server_cleanup(asyncio_loop):
     yield
     app: ServerApp = ServerApp.instance()
-    asyncio_loop.run_until_complete(app._cleanup())
+    try:
+        asyncio_loop.run_until_complete(app._cleanup())
+    except (RuntimeError, SystemExit) as e:
+        print("ignoring cleanup error", e)
     ServerApp.clear_instance()
 
 

tests/base/test_websocket.py

@@ -0,0 +1,62 @@
+"""Test Base Websocket classes"""
+import logging
+import time
+from unittest.mock import MagicMock
+
+import pytest
+from tornado.httpserver import HTTPRequest
+from tornado.httputil import HTTPHeaders
+from tornado.websocket import WebSocketClosedError, WebSocketHandler
+
+from jupyter_server.base.websocket import WebSocketMixin
+from jupyter_server.serverapp import ServerApp
+
+
+class MockHandler(WebSocketMixin, WebSocketHandler):
+    allow_origin = "*"
+    allow_origin_pat = ""
+    log = logging.getLogger()
+
+
+@pytest.fixture
+def mixin(jp_serverapp):
+    app: ServerApp = jp_serverapp
+    headers = HTTPHeaders({"Host": "foo"})
+    request = HTTPRequest("GET", headers=headers)
+    request.connection = MagicMock()
+    return MockHandler(app.web_app, request)
+
+
+def test_web_socket_mixin(mixin):
+    assert mixin.check_origin("foo") is True
+    mixin.allow_origin = ""
+    assert mixin.check_origin("") is False
+    mixin.allow_origin_pat = "foo"
+    assert mixin.check_origin("foo") is True
+    mixin.clear_cookie()
+    assert mixin.get_status() == 200
+
+
+def test_web_socket_mixin_ping(mixin):
+    mixin.ws_connection = MagicMock()
+    mixin.ws_connection.is_closing = lambda: False
+    mixin.send_ping()
+
+
+def test_ping_client_terminated(mixin):
+    mixin.ws_connection = MagicMock()
+    mixin.ws_connection.client_terminated = True
+    mixin.send_ping()
+    with pytest.raises(WebSocketClosedError):
+        mixin.write_message("hello")
+
+
+async def test_ping_client_timeout(mixin):
+    mixin.on_pong("foo")
+    mixin.settings["ws_ping_timeout"] = 0.1
+    time.sleep(0.3)
+    mixin.ws_connection = MagicMock()
+    mixin.ws_connection.is_closing = lambda: False
+    mixin.send_ping()
+    with pytest.raises(WebSocketClosedError):
+        mixin.write_message("hello")

tests/services/contents/test_fileio.py

@@ -1,10 +1,21 @@
+import json
+import logging
 import os
 import stat
 import sys
 
 import pytest
-
-from jupyter_server.services.contents.fileio import atomic_writing
+from nbformat import validate
+from nbformat.v4 import new_notebook
+from tornado.web import HTTPError
+
+from jupyter_server.services.contents.fileio import (
+    AsyncFileManagerMixin,
+    FileManagerMixin,
+    atomic_writing,
+    path_to_intermediate,
+    path_to_invalid,
+)
 
 umask = 0
 
@@ -121,3 +132,51 @@ def test_atomic_writing_newlines(tmp_path):
     with open(path, newline="") as f:
         read = f.read()
     assert read == text
+
+
+def test_path_to_invalid(tmpdir):
+    assert path_to_invalid(tmpdir) == str(tmpdir) + ".invalid"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="test fails on Windows")
+def test_file_manager_mixin(tmpdir):
+    mixin = FileManagerMixin()
+    mixin.log = logging.getLogger()
+    bad_content = tmpdir / "bad_content.ipynb"
+    bad_content.write_text("{}", "utf8")
+    with pytest.raises(HTTPError):
+        mixin._read_notebook(bad_content)
+    other = path_to_intermediate(bad_content)
+    with open(other, "w") as fid:
+        json.dump(new_notebook(), fid)
+    mixin.use_atomic_writing = True
+    nb = mixin._read_notebook(bad_content)
+    validate(nb)
+
+    with pytest.raises(HTTPError):
+        mixin._read_file(tmpdir, "text")
+
+    with pytest.raises(HTTPError):
+        mixin._save_file(tmpdir / "foo", "foo", "bar")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="test fails on Windows")
+async def test_async_file_manager_mixin(tmpdir):
+    mixin = AsyncFileManagerMixin()
+    mixin.log = logging.getLogger()
+    bad_content = tmpdir / "bad_content.ipynb"
+    bad_content.write_text("{}", "utf8")
+    with pytest.raises(HTTPError):
+        await mixin._read_notebook(bad_content)
+    other = path_to_intermediate(bad_content)
+    with open(other, "w") as fid:
+        json.dump(new_notebook(), fid)
+    mixin.use_atomic_writing = True
+    nb = await mixin._read_notebook(bad_content)
+    validate(nb)
+
+    with pytest.raises(HTTPError):
+        await mixin._read_file(tmpdir, "text")
+
+    with pytest.raises(HTTPError):
+        await mixin._save_file(tmpdir / "foo", "foo", "bar")

tests/test_serverapp.py

@@ -437,3 +437,50 @@ def test_server_web_application(jp_serverapp):
             {},
         )
     app.init_handlers([], app.settings)
+
+
+def test_misc(jp_serverapp, tmp_path):
+    app: ServerApp = jp_serverapp
+    assert app.terminals_enabled is True
+    app.extra_args = [str(tmp_path)]
+    app.parse_command_line([])
+
+
+def test_deprecated_props(jp_serverapp):
+    app: ServerApp = jp_serverapp
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        app.cookie_options = dict(foo=1)
+        app.get_secure_cookie_kwargs = dict(bar=1)
+        app.notebook_dir = "foo"
+        app.server_extensions = dict(foo=True)
+        app.kernel_ws_protocol = "foo"
+        app.limit_rate = True
+        app.iopub_msg_rate_limit = 10
+        app.iopub_data_rate_limit = 10
+        app.rate_limit_window = 10
+    with pytest.raises(SystemExit):
+        app.pylab = "foo"
+
+
+def test_signals(jp_serverapp):
+    app: ServerApp = jp_serverapp
+    app.answer_yes = True
+    app._restore_sigint_handler()
+    app._handle_sigint(None, None)
+    app._confirm_exit()
+    app._signal_info(None, None)
+
+
+async def test_shutdown_no_activity(jp_serverapp):
+    app: ServerApp = jp_serverapp
+    app.extension_manager.extensions = {}
+    app.exit = lambda _: None
+    app.shutdown_no_activity()
+    app.shutdown_no_activity_timeout = 1
+    app.init_shutdown_no_activity()
+
+
+def test_running_server_info(jp_serverapp):
+    app: ServerApp = jp_serverapp
+    app.running_server_info(True)