run Python tests with a token

exercises token auth

Author: minrk

notebook/nbconvert/tests/test_nbconvert_handlers.py

@@ -20,12 +20,12 @@
 
 class NbconvertAPI(object):
     """Wrapper for nbconvert API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None, params=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'nbconvert', path),
+        response = self.request(verb,
+                url_path_join('nbconvert', path),
                 data=body, params=params,
         )
         response.raise_for_status()
@@ -69,7 +69,7 @@ def setUp(self):
                      encoding='utf-8') as f:
             write(nb, f, version=4)
 
-        self.nbconvert_api = NbconvertAPI(self.base_url())
+        self.nbconvert_api = NbconvertAPI(self.request)
 
     def tearDown(self):
         nbdir = self.notebook_dir.name
@@ -109,8 +109,7 @@ def test_from_file_zip(self):
 
     @onlyif_cmds_exist('pandoc')
     def test_from_post(self):
-        nbmodel_url = url_path_join(self.base_url(), 'api/contents/foo/testnb.ipynb')
-        nbmodel = requests.get(nbmodel_url).json()
+        nbmodel = self.request('GET', 'api/contents/foo/testnb.ipynb').json()
 
         r = self.nbconvert_api.from_post(format='html', nbmodel=nbmodel)
         self.assertEqual(r.status_code, 200)
@@ -124,8 +123,7 @@ def test_from_post(self):
 
     @onlyif_cmds_exist('pandoc')
     def test_from_post_zip(self):
-        nbmodel_url = url_path_join(self.base_url(), 'api/contents/foo/testnb.ipynb')
-        nbmodel = requests.get(nbmodel_url).json()
+        nbmodel = self.request('GET', 'api/contents/foo/testnb.ipynb').json()
 
         r = self.nbconvert_api.from_post(format='latex', nbmodel=nbmodel)
         self.assertIn(u'application/zip', r.headers['Content-Type'])

notebook/services/config/tests/test_config_api.py

@@ -11,12 +11,12 @@
 
 class ConfigAPI(object):
     """Wrapper for notebook API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, section, body=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'api/config', section),
+        response = self.request(verb,
+                url_path_join('api/config', section),
                 data=body,
         )
         response.raise_for_status()
@@ -34,7 +34,7 @@ def modify(self, section, values):
 class APITest(NotebookTestBase):
     """Test the config web service API"""
     def setUp(self):
-        self.config_api = ConfigAPI(self.base_url())
+        self.config_api = ConfigAPI(self.request)
 
     def test_create_retrieve_config(self):
         sample = {'foo': 'bar', 'baz': 73}

notebook/services/contents/tests/test_contents_api.py

@@ -44,12 +44,12 @@ def dirs_only(dir_model):
 
 class API(object):
     """Wrapper for contents API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None, params=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'api/contents', path),
+        response = self.request(verb,
+                url_path_join('api/contents', path),
                 data=body, params=params,
         )
         response.raise_for_status()
@@ -209,7 +209,7 @@ def setUp(self):
             blob = self._blob_for_name(name)
             self.make_blob(u'{}/{}.blob'.format(d, name), blob)
 
-        self.api = API(self.base_url())
+        self.api = API(self.request)
 
     def tearDown(self):
         for dname in (list(self.top_level_dirs) + self.hidden_dirs):

notebook/services/kernels/tests/test_kernels_api.py

@@ -10,12 +10,12 @@
 
 class KernelAPI(object):
     """Wrapper for kernel REST API requests"""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'api/kernels', path), data=body)
+        response = self.request(verb,
+                url_path_join('api/kernels', path), data=body)
 
         if 400 <= response.status_code < 600:
             try:
@@ -48,7 +48,7 @@ def restart(self, id):
 class KernelAPITest(NotebookTestBase):
     """Test the kernels web service API"""
     def setUp(self):
-        self.kern_api = KernelAPI(self.base_url())
+        self.kern_api = KernelAPI(self.request)
 
     def tearDown(self):
         for k in self.kern_api.list().json():

notebook/services/kernelspecs/tests/test_kernelspecs_api.py

@@ -26,12 +26,12 @@
 
 class KernelSpecAPI(object):
     """Wrapper for notebook API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, path),
+        response = self.request(verb,
+                path,
                 data=body,
         )
         response.raise_for_status()
@@ -52,7 +52,7 @@ class APITest(NotebookTestBase):
     def setUp(self):
         self.create_spec('sample')
         self.create_spec('sample 2')
-        self.ks_api = KernelSpecAPI(self.base_url())
+        self.ks_api = KernelSpecAPI(self.request)
 
     def create_spec(self, name):
         sample_kernel_dir = pjoin(self.data_dir.name, 'kernels', name)

notebook/services/nbconvert/tests/test_nbconvert_api.py

@@ -5,12 +5,12 @@
 
 class NbconvertAPI(object):
     """Wrapper for nbconvert API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None, params=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'api/nbconvert', path),
+        response = self.request(verb,
+                url_path_join('api/nbconvert', path),
                 data=body, params=params,
         )
         response.raise_for_status()
@@ -21,7 +21,7 @@ def list_formats(self):
 
 class APITest(NotebookTestBase):
     def setUp(self):
-        self.nbconvert_api = NbconvertAPI(self.base_url())
+        self.nbconvert_api = NbconvertAPI(self.request)
 
     def test_list_formats(self):
         formats = self.nbconvert_api.list_formats().json()

notebook/services/sessions/tests/test_sessions_api.py

@@ -17,12 +17,12 @@
 
 class SessionAPI(object):
     """Wrapper for notebook API calls."""
-    def __init__(self, base_url):
-        self.base_url = base_url
+    def __init__(self, request):
+        self.request = request
 
     def _req(self, verb, path, body=None):
-        response = requests.request(verb,
-                url_path_join(self.base_url, 'api/sessions', path), data=body)
+        response = self.request(verb,
+                url_path_join('api/sessions', path), data=body)
 
         if 400 <= response.status_code < 600:
             try:
@@ -92,7 +92,7 @@ def setUp(self):
             nb = new_notebook()
             write(nb, f, version=4)
 
-        self.sess_api = SessionAPI(self.base_url())
+        self.sess_api = SessionAPI(self.request)
 
     def tearDown(self):
         for session in self.sess_api.list().json():
@@ -152,7 +152,7 @@ def test_create_deprecated(self):
 
     def test_create_with_kernel_id(self):
         # create a new kernel
-        r = requests.post(url_path_join(self.base_url(), 'api/kernels'))
+        r = self.request('POST', 'api/kernels')
         r.raise_for_status()
         kernel = r.json()
 
@@ -220,7 +220,7 @@ def test_modify_kernel_name(self):
         self.assertNotEqual(after['kernel']['id'], before['kernel']['id'])
 
         # check kernel list, to be sure previous kernel was cleaned up
-        r = requests.get(url_path_join(self.base_url(), 'api/kernels'))
+        r = self.request('GET', 'api/kernels')
         r.raise_for_status()
         kernel_list = r.json()
         self.assertEqual(kernel_list, [after['kernel']])
@@ -230,7 +230,7 @@ def test_modify_kernel_id(self):
         sid = before['id']
 
         # create a new kernel
-        r = requests.post(url_path_join(self.base_url(), 'api/kernels'))
+        r = self.request('POST', 'api/kernels')
         r.raise_for_status()
         kernel = r.json()
 
@@ -243,7 +243,7 @@ def test_modify_kernel_id(self):
         self.assertEqual(after['kernel']['id'], kernel['id'])
 
         # check kernel list, to be sure previous kernel was cleaned up
-        r = requests.get(url_path_join(self.base_url(), 'api/kernels'))
+        r = self.request('GET', 'api/kernels')
         r.raise_for_status()
         kernel_list = r.json()
         self.assertEqual(kernel_list, [kernel])

notebook/tests/launchnotebook.py

@@ -2,8 +2,8 @@
 
 from __future__ import print_function
 
+from binascii import hexlify
 import os
-import sys
 import time
 import requests
 from contextlib import contextmanager
@@ -22,6 +22,7 @@
 
 import jupyter_core.paths
 from ..notebookapp import NotebookApp
+from ..utils import url_path_join
 from ipython_genutils.tempdir import TemporaryDirectory
 
 MAX_WAITTIME = 30   # seconds to wait for notebook server to start
@@ -67,6 +68,20 @@ def wait_until_dead(cls):
         cls.notebook_thread.join(timeout=MAX_WAITTIME)
         if cls.notebook_thread.is_alive():
             raise TimeoutError("Undead notebook server")
+    
+    @classmethod
+    def request(self, verb, path, **kwargs):
+        """Send a request to my server
+        
+        with authentication and everything.
+        """
+        headers = kwargs.setdefault('headers', {})
+        # kwargs.setdefault('allow_redirects', False)
+        headers.setdefault('Authorization', 'token %s' % self.token)
+        response = requests.request(verb,
+            url_path_join(self.base_url(), path),
+            **kwargs)
+        return response
 
     @classmethod
     def setup_class(cls):
@@ -84,6 +99,7 @@ def setup_class(cls):
         cls.data_dir = data_dir
         cls.runtime_dir = TemporaryDirectory()
         cls.notebook_dir = TemporaryDirectory()
+        cls.token = hexlify(os.urandom(4)).decode('ascii')
 
         started = Event()
         def start_thread():
@@ -97,7 +113,7 @@ def start_thread():
                 notebook_dir=cls.notebook_dir.name,
                 base_url=cls.url_prefix,
                 config=cls.config,
-                token='',
+                token=cls.token,
             )
             # don't register signal handler during tests
             app.init_signal = lambda : None