Increase nbconvert and checkpoints coverage (#1066)

Author: blink1073

.github/workflows/python-tests.yml

@@ -31,6 +31,13 @@ jobs:
         uses: actions/checkout@v3
       - name: Base Setup
         uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
+      - name: Install nbconvert dependencies on Linux
+        if: startsWith(runner.os, 'Linux')
+        run: |
+          sudo apt-get update
+          sudo apt-get install texlive-plain-generic inkscape texlive-xetex
+          sudo apt-get install xvfb x11-utils libxkbcommon-x11-0
+          pip install pandoc
       - name: Run the tests
         if: ${{ !startsWith(matrix.python-version, 'pypy') && !startsWith(matrix.os, 'windows') }}
         run: hatch run cov:test -W default || hatch run cov:test -W default --lf

jupyter_server/services/contents/checkpoints.py

@@ -75,13 +75,13 @@ class GenericCheckpointsMixin:
 
     def create_checkpoint(self, contents_mgr, path):
         model = contents_mgr.get(path, content=True)
-        type = model["type"]
-        if type == "notebook":
+        type_ = model["type"]
+        if type_ == "notebook":
             return self.create_notebook_checkpoint(
                 model["content"],
                 path,
             )
-        elif type == "file":
+        elif type_ == "file":
             return self.create_file_checkpoint(
                 model["content"],
                 model["format"],
@@ -92,13 +92,13 @@ def create_checkpoint(self, contents_mgr, path):
 
     def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
         """Restore a checkpoint."""
-        type = contents_mgr.get(path, content=False)["type"]
-        if type == "notebook":
+        type_ = contents_mgr.get(path, content=False)["type"]
+        if type_ == "notebook":
             model = self.get_notebook_checkpoint(checkpoint_id, path)
-        elif type == "file":
+        elif type_ == "file":
             model = self.get_file_checkpoint(checkpoint_id, path)
         else:
-            raise HTTPError(500, "Unexpected type %s" % type)
+            raise HTTPError(500, "Unexpected type %s" % type_)
         contents_mgr.save(model, path)
 
     # Required Methods
@@ -184,30 +184,31 @@ class AsyncGenericCheckpointsMixin(GenericCheckpointsMixin):
 
     async def create_checkpoint(self, contents_mgr, path):
         model = await contents_mgr.get(path, content=True)
-        type = model["type"]
-        if type == "notebook":
+        type_ = model["type"]
+        if type_ == "notebook":
             return await self.create_notebook_checkpoint(
                 model["content"],
                 path,
             )
-        elif type == "file":
+        elif type_ == "file":
             return await self.create_file_checkpoint(
                 model["content"],
                 model["format"],
                 path,
             )
         else:
-            raise HTTPError(500, "Unexpected type %s" % type)
+            raise HTTPError(500, "Unexpected type %s" % type_)
 
     async def restore_checkpoint(self, contents_mgr, checkpoint_id, path):
         """Restore a checkpoint."""
-        type = await contents_mgr.get(path, content=False)["type"]
-        if type == "notebook":
+        content_model = await contents_mgr.get(path, content=False)
+        type_ = content_model["type"]
+        if type_ == "notebook":
             model = await self.get_notebook_checkpoint(checkpoint_id, path)
-        elif type == "file":
+        elif type_ == "file":
             model = await self.get_file_checkpoint(checkpoint_id, path)
         else:
-            raise HTTPError(500, "Unexpected type %s" % type)
+            raise HTTPError(500, "Unexpected type %s" % type_)
         await contents_mgr.save(model, path)
 
     # Required Methods

tests/conftest.py

@@ -1,6 +1,8 @@
 import os
 
 import pytest
+from nbformat import writes
+from nbformat.v4 import new_notebook
 
 from tests.extension.mockextensions.app import MockExtensionApp
 
@@ -84,3 +86,58 @@ def config_file(jp_config_dir):
 def jp_mockextension_cleanup():
     yield
     MockExtensionApp.clear_instance()
+
+
+@pytest.fixture
+def contents_dir(tmp_path, jp_serverapp):
+    return tmp_path / jp_serverapp.root_dir
+
+
+dirs = [
+    ("", "inroot"),
+    ("Directory with spaces in", "inspace"),
+    ("unicodé", "innonascii"),
+    ("foo", "a"),
+    ("foo", "b"),
+    ("foo", "name with spaces"),
+    ("foo", "unicodé"),
+    ("foo/bar", "baz"),
+    ("ordering", "A"),
+    ("ordering", "b"),
+    ("ordering", "C"),
+    ("å b", "ç d"),
+]
+
+
+@pytest.fixture
+def contents(contents_dir):
+    # Create files in temporary directory
+    paths: dict = {"notebooks": [], "textfiles": [], "blobs": [], "contents_dir": contents_dir}
+    for d, name in dirs:
+        p = contents_dir / d
+        p.mkdir(parents=True, exist_ok=True)
+
+        # Create a notebook
+        nb = writes(new_notebook(), version=4)
+        nbname = p.joinpath(f"{name}.ipynb")
+        nbname.write_text(nb, encoding="utf-8")
+        paths["notebooks"].append(nbname.relative_to(contents_dir))
+
+        # Create a text file
+        txt = f"{name} text file"
+        txtname = p.joinpath(f"{name}.txt")
+        txtname.write_text(txt, encoding="utf-8")
+        paths["textfiles"].append(txtname.relative_to(contents_dir))
+
+        # Create a random blob
+        blob = name.encode("utf-8") + b"\xFF"
+        blobname = p.joinpath(f"{name}.blob")
+        blobname.write_bytes(blob)
+        paths["blobs"].append(blobname.relative_to(contents_dir))
+    paths["all"] = list(paths.values())
+    return paths
+
+
+@pytest.fixture
+def folders():
+    return list({item[0] for item in dirs})

tests/services/contents/test_api.py

@@ -7,10 +7,11 @@
 
 import pytest
 import tornado
-from nbformat import from_dict, writes
+from nbformat import from_dict
 from nbformat.v4 import new_markdown_cell, new_notebook
 
 from jupyter_server.utils import url_path_join
+from tests.conftest import dirs
 
 from ...utils import expected_http_error
 
@@ -34,22 +35,6 @@ def dirs_only(dir_model):
     return [x for x in dir_model["content"] if x["type"] == "directory"]
 
 
-dirs = [
-    ("", "inroot"),
-    ("Directory with spaces in", "inspace"),
-    ("unicodé", "innonascii"),
-    ("foo", "a"),
-    ("foo", "b"),
-    ("foo", "name with spaces"),
-    ("foo", "unicodé"),
-    ("foo/bar", "baz"),
-    ("ordering", "A"),
-    ("ordering", "b"),
-    ("ordering", "C"),
-    ("å b", "ç d"),
-]
-
-
 @pytest.fixture(params=["FileContentsManager", "AsyncFileContentsManager"])
 def jp_argv(request):
     return [
@@ -58,49 +43,6 @@ def jp_argv(request):
     ]
 
 
-@pytest.fixture
-def contents_dir(tmp_path, jp_serverapp):
-    return tmp_path / jp_serverapp.root_dir
-
-
-@pytest.fixture
-def contents(contents_dir):
-    # Create files in temporary directory
-    paths: dict = {
-        "notebooks": [],
-        "textfiles": [],
-        "blobs": [],
-    }
-    for d, name in dirs:
-        p = contents_dir / d
-        p.mkdir(parents=True, exist_ok=True)
-
-        # Create a notebook
-        nb = writes(new_notebook(), version=4)
-        nbname = p.joinpath(f"{name}.ipynb")
-        nbname.write_text(nb, encoding="utf-8")
-        paths["notebooks"].append(nbname.relative_to(contents_dir))
-
-        # Create a text file
-        txt = f"{name} text file"
-        txtname = p.joinpath(f"{name}.txt")
-        txtname.write_text(txt, encoding="utf-8")
-        paths["textfiles"].append(txtname.relative_to(contents_dir))
-
-        # Create a random blob
-        blob = name.encode("utf-8") + b"\xFF"
-        blobname = p.joinpath(f"{name}.blob")
-        blobname.write_bytes(blob)
-        paths["blobs"].append(blobname.relative_to(contents_dir))
-    paths["all"] = list(paths.values())
-    return paths
-
-
-@pytest.fixture
-def folders():
-    return list({item[0] for item in dirs})
-
-
 @pytest.mark.parametrize("path,name", dirs)
 async def test_list_notebooks(jp_fetch, contents, path, name):
     response = await jp_fetch(
@@ -853,6 +795,7 @@ async def test_rename_400_hidden(jp_fetch, jp_base_url, contents, contents_dir):
 
 
 async def test_checkpoints_follow_file(jp_fetch, contents):
+
     path = "foo"
     name = "a.ipynb"
 

tests/services/contents/test_checkpoints.py

@@ -0,0 +1,130 @@
+import pytest
+from jupyter_client.utils import ensure_async
+from nbformat import from_dict
+from nbformat.v4 import new_markdown_cell
+
+from jupyter_server.services.contents.filecheckpoints import (
+    AsyncFileCheckpoints,
+    AsyncGenericFileCheckpoints,
+    FileCheckpoints,
+    GenericFileCheckpoints,
+)
+from jupyter_server.services.contents.largefilemanager import (
+    AsyncLargeFileManager,
+    LargeFileManager,
+)
+
+param_pairs = [
+    (LargeFileManager, FileCheckpoints),
+    (LargeFileManager, GenericFileCheckpoints),
+    (AsyncLargeFileManager, AsyncFileCheckpoints),
+    (AsyncLargeFileManager, AsyncGenericFileCheckpoints),
+]
+
+
+@pytest.fixture(params=param_pairs)
+def contents_manager(request, contents):
+    """Returns a LargeFileManager instance."""
+    file_manager, checkpoints_class = request.param
+    root_dir = str(contents["contents_dir"])
+    return file_manager(root_dir=root_dir, checkpoints_class=checkpoints_class)
+
+
+async def test_checkpoints_follow_file(contents_manager):
+    cm: LargeFileManager = contents_manager
+    path = "foo/a.ipynb"
+
+    # Read initial file.
+    model = await ensure_async(cm.get(path))
+
+    # Create a checkpoint of initial state
+    cp1 = await ensure_async(cm.create_checkpoint(path))
+
+    # Modify file and save.
+    nbcontent = model["content"]
+    nb = from_dict(nbcontent)
+    hcell = new_markdown_cell("Created by test")
+    nb.cells.append(hcell)
+    nbmodel = {"content": nb, "type": "notebook"}
+    await ensure_async(cm.save(nbmodel, path))
+
+    # List checkpoints
+    cps = await ensure_async(cm.list_checkpoints(path))
+    assert cps == [cp1]
+
+    model = await ensure_async(cm.get(path))
+    nbcontent = model["content"]
+    nb = from_dict(nbcontent)
+    assert nb.cells[0].source == "Created by test"
+
+
+async def test_nb_checkpoints(contents_manager):
+    cm: LargeFileManager = contents_manager
+    path = "foo/a.ipynb"
+    model = await ensure_async(cm.get(path))
+    cp1 = await ensure_async(cm.create_checkpoint(path))
+    assert set(cp1) == {"id", "last_modified"}
+
+    # Modify it.
+    nbcontent = model["content"]
+    nb = from_dict(nbcontent)
+    hcell = new_markdown_cell("Created by test")
+    nb.cells.append(hcell)
+
+    # Save it.
+    nbmodel = {"content": nb, "type": "notebook"}
+    await ensure_async(cm.save(nbmodel, path))
+
+    # List checkpoints
+    cps = await ensure_async(cm.list_checkpoints(path))
+    assert cps == [cp1]
+
+    nbcontent = await ensure_async(cm.get(path))
+    nb = from_dict(nbcontent["content"])
+    assert nb.cells[0].source == "Created by test"
+
+    # Restore Checkpoint cp1
+    await ensure_async(cm.restore_checkpoint(cp1["id"], path))
+
+    nbcontent = await ensure_async(cm.get(path))
+    nb = from_dict(nbcontent["content"])
+    assert nb.cells == []
+
+    # Delete cp1
+    await ensure_async(cm.delete_checkpoint(cp1["id"], path))
+
+    cps = await ensure_async(cm.list_checkpoints(path))
+    assert cps == []
+
+
+async def test_file_checkpoints(contents_manager):
+    cm: LargeFileManager = contents_manager
+    path = "foo/a.txt"
+    model = await ensure_async(cm.get(path))
+    orig_content = model["content"]
+
+    cp1 = await ensure_async(cm.create_checkpoint(path))
+    assert set(cp1) == {"id", "last_modified"}
+
+    # Modify and save it.
+    model["content"] = new_content = orig_content + "\nsecond line"
+    await ensure_async(cm.save(model, path))
+
+    # List checkpoints
+    cps = await ensure_async(cm.list_checkpoints(path))
+    assert cps == [cp1]
+
+    model = await ensure_async(cm.get(path))
+    assert model["content"] == new_content
+
+    # Restore Checkpoint cp1
+    await ensure_async(cm.restore_checkpoint(cp1["id"], path))
+
+    restored_content = await ensure_async(cm.get(path))
+    assert restored_content["content"] == orig_content
+
+    # Delete cp1
+    await ensure_async(cm.delete_checkpoint(cp1["id"], path))
+
+    cps = await ensure_async(cm.list_checkpoints(path))
+    assert cps == []