mv ws_fetch kernelspecs and contents_manager fixtures to pytest_plugin for 3rd party usage

Author: echarles

jupyter_server/pytest_plugin.py

@@ -15,6 +15,7 @@
 from jupyter_server.extension import serverextension
 from jupyter_server.serverapp import ServerApp
 from jupyter_server.utils import url_path_join
+from jupyter_server.services.contents.filemanager import FileContentsManager
 
 import nbformat
 
@@ -62,6 +63,11 @@ def mkdir(tmp_path, *parts):
 env_config_path = pytest.fixture(
     lambda tmp_path: mkdir(tmp_path, "env", "etc", "jupyter")
 )
+some_resource = u"The very model of a modern major general"
+sample_kernel_json = {
+    'argv':['cat', '{connection_file}'],
+    'display_name':'Test kernel',
+}
 argv = pytest.fixture(lambda: [])
 
 
@@ -199,6 +205,52 @@ def client_fetch(*parts, headers={}, params={}, **kwargs):
     return client_fetch
 
 
+@pytest.fixture
+def ws_fetch(auth_header, http_port):
+    """websocket fetch fixture that handles auth, base_url, and path"""
+    def client_fetch(*parts, headers={}, params={}, **kwargs):
+        # Handle URL strings
+        path = url_escape(url_path_join(*parts), plus=False)
+        urlparts = urllib.parse.urlparse('ws://localhost:{}'.format(http_port))
+        urlparts = urlparts._replace(
+            path=path,
+            query=urllib.parse.urlencode(params)
+        )
+        url = urlparts.geturl()
+        # Add auth keys to header
+        headers.update(auth_header)
+        # Make request.
+        req = tornado.httpclient.HTTPRequest(
+            url,
+            headers=auth_header,
+            connect_timeout=120
+        )
+        return tornado.websocket.websocket_connect(req)
+    return client_fetch
+
+
+@pytest.fixture
+def kernelspecs(data_dir):
+    spec_names = ['sample', 'sample 2']
+    for name in spec_names:
+        sample_kernel_dir = data_dir.joinpath('kernels', name)
+        sample_kernel_dir.mkdir(parents=True)
+        # Create kernel json file
+        sample_kernel_file = sample_kernel_dir.joinpath('kernel.json')
+        sample_kernel_file.write_text(json.dumps(sample_kernel_json))
+        # Create resources text
+        sample_kernel_resources = sample_kernel_dir.joinpath('resource.txt')
+        sample_kernel_resources.write_text(some_resource)
+
+
+# contents_manager_atomic = pytest.fixture(lambda tmp_path: FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=True))
+# contents_manager_nonatomic = pytest.fixture(lambda tmp_path: FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=False))
+
+@pytest.fixture(params=[True, False])
+def contents_manager(request, tmp_path):
+    return FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=request.param)
+
+
 @pytest.fixture
 def create_notebook(root_dir):
     """Create a notebook in the test's home directory."""

tests/services/contents/test_manager.py

@@ -15,14 +15,6 @@
 
 # -------------- Functions ----------------------------
 
-# contents_manager_atomic = pytest.fixture(lambda tmp_path: FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=True))
-# contents_manager_nonatomic = pytest.fixture(lambda tmp_path: FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=False))
-
-@pytest.fixture(params=[True, False])
-def contents_manager(request, tmp_path):
-    return FileContentsManager(root_dir=str(tmp_path), use_atomic_writing=request.param)
-
-
 def _make_dir(contents_manager, api_path):
     """
     Make a directory.

tests/services/kernels/test_api.py

@@ -13,30 +13,6 @@
 from ...utils import expected_http_error
 
 
-@pytest.fixture
-def ws_fetch(auth_header, http_port):
-    """fetch fixture that handles auth, base_url, and path"""
-    def client_fetch(*parts, headers={}, params={}, **kwargs):
-        # Handle URL strings
-        path = url_escape(url_path_join(*parts), plus=False)
-        urlparts = urllib.parse.urlparse('ws://localhost:{}'.format(http_port))
-        urlparts = urlparts._replace(
-            path=path,
-            query=urllib.parse.urlencode(params)
-        )
-        url = urlparts.geturl()
-        # Add auth keys to header
-        headers.update(auth_header)
-        # Make request.
-        req = tornado.httpclient.HTTPRequest(
-            url,
-            headers=auth_header,
-            connect_timeout=120
-        )
-        return tornado.websocket.websocket_connect(req)
-    return client_fetch
-
-
 async def test_no_kernels(fetch):
     r = await fetch(
         'api', 'kernels',

tests/services/kernelspecs/test_api.py

@@ -3,32 +3,13 @@
 
 import tornado
 
+from jupyter_server.pytest_plugin import some_resource
+
 from jupyter_client.kernelspec import NATIVE_KERNEL_NAME
 
 from ...utils import expected_http_error
 
 
-sample_kernel_json = {
-    'argv':['cat', '{connection_file}'],
-    'display_name':'Test kernel',
-}
-some_resource = u"The very model of a modern major general"
-
-
-@pytest.fixture
-def kernelspecs(data_dir):
-    spec_names = ['sample', 'sample 2']
-    for name in spec_names:
-        sample_kernel_dir = data_dir.joinpath('kernels', name)
-        sample_kernel_dir.mkdir(parents=True)
-        # Create kernel json file
-        sample_kernel_file = sample_kernel_dir.joinpath('kernel.json')
-        sample_kernel_file.write_text(json.dumps(sample_kernel_json))
-        # Create resources text
-        sample_kernel_resources = sample_kernel_dir.joinpath('resource.txt')
-        sample_kernel_resources.write_text(some_resource)
-
-
 async def test_list_kernelspecs_bad(fetch, kernelspecs, data_dir):
     bad_kernel_dir = data_dir.joinpath(data_dir, 'kernels', 'bad')
     bad_kernel_dir.mkdir(parents=True)