Add test for kernel filtering

Author: SylvainCorlay

jupyter_server/services/api/tests/test_api.py

@@ -29,4 +29,4 @@ def test_get_status(self):
         assert data['kernels'] == 0
         assert data['last_activity'].endswith('Z')
         assert data['started'].endswith('Z')
-        assert data['started'] == isoformat(self.notebook.web_app.settings['started'])
+        assert data['started'] == isoformat(self.server.web_app.settings['started'])

jupyter_server/services/contents/tests/test_contents_api.py

@@ -681,7 +681,7 @@ def patch_cp_root(self, dirname):
         """
         Temporarily patch the root dir of our checkpoint manager.
         """
-        cpm = self.notebook.contents_manager.checkpoints
+        cpm = self.server.contents_manager.checkpoints
         old_dirname = cpm.root_dir
         cpm.root_dir = dirname
         try:
@@ -717,7 +717,7 @@ class GenericFileCheckpointsAPITest(APITest):
     def test_config_did_something(self):
 
         self.assertIsInstance(
-            self.notebook.contents_manager.checkpoints,
+            self.server.contents_manager.checkpoints,
             GenericFileCheckpoints,
         )
 

jupyter_server/services/kernels/tests/test_kernels_api.py

@@ -3,6 +3,8 @@
 import json
 import time
 
+from traitlets.config import Config
+
 from tornado.httpclient import HTTPRequest
 from tornado.ioloop import IOLoop
 from tornado.websocket import websocket_connect
@@ -184,3 +186,20 @@ def test_connections(self):
                 break
         model = self.kern_api.get(kid).json()
         self.assertEqual(model['connections'], 0)
+
+
+class KernelFilterTest(ServerTestBase):
+    # A special install of ServerTestBase where only `kernel_info_request`
+    # messages are allowed.
+
+    config = Config({
+        'ServerApp': {
+            'MappingKernelManager': {
+                'allowed_message_types': ['kernel_info_request']
+            }
+        }
+    })
+
+    # Sanity check verifying that the configurable was properly set.
+    def test_config(self):
+        self.assertEqual(self.server.kernel_manager.allowed_message_types, ['kernel_info_request'])

jupyter_server/tests/launchserver.py

@@ -59,7 +59,7 @@ def wait_until_alive(cls):
             try:
                 requests.get(url)
             except Exception as e:
-                if not cls.notebook_thread.is_alive():
+                if not cls.server_thread.is_alive():
                     raise RuntimeError("The Jupyter server failed to start")
                 time.sleep(POLL_INTERVAL)
             else:
@@ -70,8 +70,8 @@ def wait_until_alive(cls):
     @classmethod
     def wait_until_dead(cls):
         """Wait for the server process to terminate after shutdown"""
-        cls.notebook_thread.join(timeout=MAX_WAITTIME)
-        if cls.notebook_thread.is_alive():
+        cls.server_thread.join(timeout=MAX_WAITTIME)
+        if cls.server_thread.is_alive():
             raise TimeoutError("Undead Jupyter server")
 
     @classmethod
@@ -110,11 +110,10 @@ def tmp(*parts):
         data_dir = cls.data_dir = tmp('data')
         config_dir = cls.config_dir = tmp('config')
         runtime_dir = cls.runtime_dir = tmp('runtime')
-        cls.root_dir = tmp('notebooks')
+        cls.root_dir = tmp('root_dir')
         cls.env_patch = patch.dict('os.environ', {
             'HOME': cls.home_dir,
             'PYTHONPATH': os.pathsep.join(sys.path),
-            'IPYTHONDIR': pjoin(cls.home_dir, '.ipython'),
             'JUPYTER_NO_CONFIG': '1', # needed in the future
             'JUPYTER_CONFIG_DIR' : config_dir,
             'JUPYTER_DATA_DIR' : data_dir,
@@ -140,7 +139,7 @@ def start_thread():
             if 'asyncio' in sys.modules:
                 import asyncio
                 asyncio.set_event_loop(asyncio.new_event_loop())
-            app = cls.notebook = ServerApp(
+            app = cls.server = ServerApp(
                 port=cls.port,
                 port_retries=0,
                 open_browser=False,
@@ -170,15 +169,15 @@ def start_thread():
                 # set the event, so failure to start doesn't cause a hang
                 started.set()
                 app.session_manager.close()
-        cls.notebook_thread = Thread(target=start_thread)
-        cls.notebook_thread.daemon = True
-        cls.notebook_thread.start()
+        cls.server_thread = Thread(target=start_thread)
+        cls.server_thread.daemon = True
+        cls.server_thread.start()
         started.wait()
         cls.wait_until_alive()
 
     @classmethod
     def teardown_class(cls):
-        cls.notebook.stop()
+        cls.server.stop()
         cls.wait_until_dead()
         cls.env_patch.stop()
         cls.path_patch.stop()

jupyter_server/tests/test_files.py

@@ -57,7 +57,7 @@ def test_hidden_files(self):
                 r = self.request('GET', url_path_join('files', d, foo))
                 self.assertEqual(r.status_code, 404)
 
-        self.notebook.contents_manager.allow_hidden = True
+        self.server.contents_manager.allow_hidden = True
         try:
             for d in not_hidden:
                 path = pjoin(rootdir, d.replace('/', os.sep))
@@ -75,7 +75,7 @@ def test_hidden_files(self):
                     r.raise_for_status()
                     self.assertEqual(r.text, foo)
         finally:
-            self.notebook.contents_manager.allow_hidden = False
+            self.server.contents_manager.allow_hidden = False
 
     def test_contents_manager(self):
         "make sure ContentsManager returns right files (ipynb, bin, txt)."