run Python tests with a token

exercises token auth

notebook/nbconvert/tests/test_nbconvert_handlers.py

@@ -26,12 +26,12 @@ except ImportError: #PY2
 
 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()
@@ -84,7 +84,7 @@ class APITest(NotebookTestBase):
                      encoding='utf-8') as f:
             write(nb, f, version=4)
 
-        self.nbconvert_api = NbconvertAPI(self.base_url())
+        self.nbconvert_api = NbconvertAPI(self.request)
 
     @onlyif_cmds_exist('pandoc')
     def test_from_file(self):
@@ -118,8 +118,7 @@ class APITest(NotebookTestBase):
 
     @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)
@@ -133,8 +132,7 @@ class APITest(NotebookTestBase):
 
     @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 @@ from notebook.tests.launchnotebook import NotebookTestBase
 
 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 @@ class ConfigAPI(object):
 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

@@ -50,12 +50,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()
@@ -220,7 +220,7 @@ class APITest(NotebookTestBase):
             self.make_blob(blobname, blob)
             self.addCleanup(partial(self.delete_file, blobname))
 
-        self.api = API(self.base_url())
+        self.api = API(self.request)
 
     def test_list_notebooks(self):
         nbs = notebooks_only(self.api.list().json())

notebook/services/kernels/tests/test_kernels_api.py

@@ -10,12 +10,12 @@ from notebook.tests.launchnotebook import NotebookTestBase, assert_http_error
 
 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 @@ class KernelAPI(object):
 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 @@ some_resource = u"The very model of a modern major general"
 
 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 @@ from notebook.tests.launchnotebook import NotebookTestBase
 
 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 @@ class NbconvertAPI(object):
 
 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

@@ -18,12 +18,12 @@ from nbformat import write
 
 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:
@@ -95,7 +95,7 @@ class SessionAPITest(NotebookTestBase):
             nb = new_notebook()
             write(nb, f, version=4)
 
-        self.sess_api = SessionAPI(self.base_url())
+        self.sess_api = SessionAPI(self.request)
 
         @self.addCleanup
         def cleanup_sessions():
@@ -154,7 +154,7 @@ class SessionAPITest(NotebookTestBase):
 
     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()
 
@@ -222,7 +222,7 @@ class SessionAPITest(NotebookTestBase):
         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']])
@@ -232,7 +232,7 @@ class SessionAPITest(NotebookTestBase):
         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()
 
@@ -245,7 +245,7 @@ class SessionAPITest(NotebookTestBase):
         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
@@ -23,6 +23,7 @@ import zmq
 import jupyter_core.paths
 from traitlets.config import Config
 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
@@ -68,6 +69,20 @@ class NotebookTestBase(TestCase):
         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):
@@ -86,9 +101,12 @@ class NotebookTestBase(TestCase):
         cls.data_dir = data_dir
         cls.runtime_dir = TemporaryDirectory()
         cls.notebook_dir = TemporaryDirectory()
+
         config = cls.config or Config()
         config.NotebookNotary.db_file = ':memory:'
 
+        cls.token = hexlify(os.urandom(4)).decode('ascii')
+
         started = Event()
         def start_thread():
             app = cls.notebook = NotebookApp(
@@ -102,7 +120,7 @@ class NotebookTestBase(TestCase):
                 base_url=cls.url_prefix,
                 config=config,
                 allow_root=True,
-                token='',
+                token=cls.token,
             )
             # don't register signal handler during tests
             app.init_signal = lambda : None

notebook/tests/test_files.py

@@ -41,27 +41,25 @@ class FilesTest(NotebookTestBase):
                 f.write('foo')
             with open(pjoin(path, '.foo'), 'w') as f:
                 f.write('.foo')
-        url = self.base_url()
         
         for d in not_hidden:
             path = pjoin(nbdir, d.replace('/', os.sep))
-            r = requests.get(url_path_join(url, 'files', d, 'foo'))
+            r = self.request('GET', url_path_join('files', d, 'foo'))
             r.raise_for_status()
             self.assertEqual(r.text, 'foo')
-            r = requests.get(url_path_join(url, 'files', d, '.foo'))
+            r = self.request('GET', url_path_join('files', d, '.foo'))
             self.assertEqual(r.status_code, 404)
             
         for d in hidden:
             path = pjoin(nbdir, d.replace('/', os.sep))
             for foo in ('foo', '.foo'):
-                r = requests.get(url_path_join(url, 'files', d, foo))
+                r = self.request('GET', url_path_join('files', d, foo))
                 self.assertEqual(r.status_code, 404)
     
     def test_contents_manager(self):
         "make sure ContentsManager returns right files (ipynb, bin, txt)."
 
         nbdir = self.notebook_dir.name
-        base = self.base_url()
 
         nb = new_notebook(
             cells=[
@@ -84,35 +82,34 @@ class FilesTest(NotebookTestBase):
             f.write(u'foobar')
             f.close()
 
-        r = requests.get(url_path_join(base, 'files', 'testnb.ipynb'))
+        r = self.request('GET', 'files/testnb.ipynb')
         self.assertEqual(r.status_code, 200)
         self.assertIn('print(2*6)', r.text)
         json.loads(r.text)
 
-        r = requests.get(url_path_join(base, 'files', 'test.bin'))
+        r = self.request('GET', 'files/test.bin')
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.headers['content-type'], 'application/octet-stream')
         self.assertEqual(r.content[:1], b'\xff')
         self.assertEqual(len(r.content), 6)
 
-        r = requests.get(url_path_join(base, 'files', 'test.txt'))
+        r = self.request('GET', 'files/test.txt')
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.headers['content-type'], 'text/plain')
         self.assertEqual(r.text, 'foobar')
     
     def test_download(self):
         nbdir = self.notebook_dir.name
-        base = self.base_url()
         
         text = 'hello'
         with open(pjoin(nbdir, 'test.txt'), 'w') as f:
             f.write(text)
         
-        r = requests.get(url_path_join(base, 'files', 'test.txt'))
+        r = self.request('GET', 'files/test.txt')
         disposition = r.headers.get('Content-Disposition', '')
         self.assertNotIn('attachment', disposition)
 
-        r = requests.get(url_path_join(base, 'files', 'test.txt') + '?download=1')
+        r = self.request('GET', 'files/test.txt?download=1')
         disposition = r.headers.get('Content-Disposition', '')
         self.assertIn('attachment', disposition)
         self.assertIn('filename="test.txt"', disposition)
@@ -120,7 +117,6 @@ class FilesTest(NotebookTestBase):
     def test_old_files_redirect(self):
         """pre-2.0 'files/' prefixed links are properly redirected"""
         nbdir = self.notebook_dir.name
-        base = self.base_url()
         
         os.mkdir(pjoin(nbdir, 'files'))
         os.makedirs(pjoin(nbdir, 'sub', 'files'))
@@ -134,19 +130,19 @@ class FilesTest(NotebookTestBase):
                 f.write(prefix + '/f2')
             with open(pjoin(nbdir, prefix, 'f3.txt'), 'w') as f:
                 f.write(prefix + '/f3')
-            
-            url = url_path_join(base, 'notebooks', prefix, 'files', 'f1.txt')
-            r = requests.get(url)
+
+            url = url_path_join('notebooks', prefix, 'files', 'f1.txt')
+            r = self.request('GET', url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/files/f1')
 
-            url = url_path_join(base, 'notebooks', prefix, 'files', 'f2.txt')
-            r = requests.get(url)
+            url = url_path_join('notebooks', prefix, 'files', 'f2.txt')
+            r = self.request('GET', url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/files/f2')
 
-            url = url_path_join(base, 'notebooks', prefix, 'files', 'f3.txt')
-            r = requests.get(url)
+            url = url_path_join('notebooks', prefix, 'files', 'f3.txt')
+            r = self.request('GET', url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/f3')
 

notebook/tree/tests/test_tree_handler.py

@@ -25,8 +25,8 @@ class TreeTest(NotebookTestBase):
         self.base_url()
 
     def test_redirect(self):
-        r = requests.get(url_path_join(self.base_url(), 'tree/foo/bar.ipynb'))
+        r = self.request('GET', 'tree/foo/bar.ipynb')
         self.assertEqual(r.url, self.base_url() + 'notebooks/foo/bar.ipynb')
 
-        r = requests.get(url_path_join(self.base_url(), 'tree/foo/baz.txt'))
+        r = self.request('GET', 'tree/foo/baz.txt')
         self.assertEqual(r.url, url_path_join(self.base_url(), 'files/foo/baz.txt'))