Moves ContextManager model fetching to a module.

Moves ContextManager model fetching to a seperate module (`rsconnect_jupyter.managers`) to faciliate isolated unit testing.

rsconnect_jupyter/__init__.py

@@ -1,5 +1,4 @@
 import hashlib
-import inspect
 import json
 import os
 import sys
@@ -30,6 +29,8 @@
 
 from ssl import SSLError
 
+from rsconnect_jupyter.managers import get_model
+
 try:
     from rsconnect_jupyter.version import version as __version__  # noqa
 except ImportError:
@@ -161,11 +162,7 @@ async def post(self, action):
             hide_all_input = data.get("hide_all_input", False)
             hide_tagged_input = data.get("hide_tagged_input", False)
 
-            model = self.contents_manager.get(path=nb_path)
-            if inspect.isawaitable(model):
-                # The default ContentsManager is now async,
-                # but we handle both cases.
-                model = await model
+            model = await get_model(self.contents_manager, nb_path)
 
             if model["type"] != "notebook":
                 # not a notebook

rsconnect_jupyter/managers.py

@@ -0,0 +1,21 @@
+from inspect import isawaitable
+from typing import Union, Awaitable
+
+from jupyter_server.services.contents.manager import ContentsManager
+
+
+async def get_model(manager: ContentsManager, path: str) -> dict:
+    """
+    Gets the model via the ContentsManager.
+
+    If the ContentsManager is async (e.g., AsyncContentsManager), then an await is issued. Otherwise,
+    the model is returned under synchronous expectations.
+
+    :param manager: A Jupyter ContentsManager
+    :param path: The model path
+    :return: The model
+    """
+    model: Union[dict, Awaitable[dict]] = manager.get(path)
+    if isawaitable(model):
+        model = await model
+    return model

setup.cfg

@@ -28,3 +28,8 @@ packages = rsconnect_jupyter
 python_requires = >=3.7
 include_package_data = true
 zip_safe = false
+
+[options.extras_require]
+test =
+    black
+    pytest

tests/test_managers.py

@@ -0,0 +1,30 @@
+from unittest import TestCase
+from unittest.mock import Mock, MagicMock, AsyncMock
+
+from rsconnect_jupyter.managers import ContentsManager, get_model, isawaitable
+
+
+class GetModelTestCase(TestCase):
+    async def test_synchronous(self):
+        model = AsyncMock()
+        manager = MagicMock(spec=ContentsManager)
+        manager.get = Mock(return_value=model)
+        path = "path"
+        spy = Mock(wraps=isawaitable, return_value=False)
+        res = await get_model(manager, path)
+        self.assertEqual(res, model)
+        model.assert_not_awaited()
+        manager.get.assert_called_once_with(path)
+        spy.assert_called_once_with(model)
+
+    async def test_asynchronous(self):
+        model = AsyncMock()
+        manager = MagicMock(spec=ContentsManager)
+        manager.get = Mock(return_value=model)
+        path = "path"
+        spy = Mock(wraps=isawaitable, return_value=True)
+        res = await get_model(manager, path)
+        self.assertEqual(res, model)
+        model.assert_awaited()
+        manager.get.assert_called_once_with(path)
+        spy.assert_called_once_with(model)