Refactor tests for kernels REST API

IPython/html/services/kernels/tests/test_kernels_api.py

@@ -1,100 +1,113 b''
 """Test the kernels service API."""
 
 
 import os
 import sys
 import json
 
 import requests
 
 from IPython.html.utils import url_path_join
-from IPython.html.tests.launchnotebook import NotebookTestBase
+from IPython.html.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 _req(self, verb, path, body=None):
+        response = requests.request(verb,
+                url_path_join(self.base_url, 'api/kernels', path), data=body)
+
+        if 400 <= response.status_code < 600:
+            try:
+                response.reason = response.json()['message']
+            except:
+                pass
+        response.raise_for_status()
+
+        return response
+
+    def list(self):
+        return self._req('GET', '')
+
+    def get(self, id):
+        return self._req('GET', id)
+
+    def start(self):
+        return self._req('POST', '')
+
+    def shutdown(self, id):
+        return self._req('DELETE', id)
+
+    def interrupt(self, id):
+        return self._req('POST', url_path_join(id, 'interrupt'))
+
+    def restart(self, id):
+        return self._req('POST', url_path_join(id, 'restart'))
 
 class KernelAPITest(NotebookTestBase):
     """Test the kernels web service API"""
+    def setUp(self):
+        self.kern_api = KernelAPI(self.base_url())
 
-    def base_url(self):
-        return url_path_join(super(KernelAPITest,self).base_url(), 'api/kernels')
-
-    def mkkernel(self):
-        r = requests.post(self.base_url())
-        return r.json()
+    def tearDown(self):
+        for k in self.kern_api.list().json():
+            self.kern_api.shutdown(k['id'])
 
     def test__no_kernels(self):
         """Make sure there are no kernels running at the start"""
-        url = self.base_url()
-        r = requests.get(url)
-        self.assertEqual(r.json(), [])
+        kernels = self.kern_api.list().json()
+        self.assertEqual(kernels, [])
 
     def test_main_kernel_handler(self):
         # POST request
-        r = requests.post(self.base_url())
-        data = r.json()
-        status = r.status_code
-        header = r.headers
-        self.assertIn('location', header)
-        self.assertEquals(header['location'], '/api/kernels/' + data['id'])
-        self.assertEquals(status, 201)
-        assert isinstance(data, dict)
+        r = self.kern_api.start()
+        kern1 = r.json()
+        self.assertEquals(r.headers['location'], '/api/kernels/' + kern1['id'])
+        self.assertEquals(r.status_code, 201)
+        self.assertIsInstance(kern1, dict)
 
         # GET request
-        r = requests.get(self.base_url())
-        status = r.status_code
-        self.assertEquals(status, 200)
+        r = self.kern_api.list()
+        self.assertEquals(r.status_code, 200)
         assert isinstance(r.json(), list)
-        self.assertEqual(r.json()[0]['id'], data['id'])
-        
-        # create another kernel and check that they both are added to the 
+        self.assertEqual(r.json()[0]['id'], kern1['id'])
+
+        # create another kernel and check that they both are added to the
         # list of kernels from a GET request
-        data2 = self.mkkernel()
-        assert isinstance(data2, dict)
-        r = requests.get(self.base_url())
+        kern2 = self.kern_api.start().json()
+        assert isinstance(kern2, dict)
+        r = self.kern_api.list()
         kernels = r.json()
-        status = r.status_code
-        self.assertEquals(status, 200)
+        self.assertEquals(r.status_code, 200)
         assert isinstance(kernels, list)
         self.assertEquals(len(kernels), 2)
 
     def test_kernel_handler(self):
         # GET kernel with given id
-        data = self.mkkernel()
-        url = self.base_url() +'/' + data['id']
-        r = requests.get(url)
-        data1 = r.json()
-        status = r.status_code
-        self.assertEquals(status, 200)
-        assert isinstance(data1, dict)
-        self.assertIn('id', data1)
-        self.assertIn('ws_url', data1)
-        self.assertEqual(data1['id'], data['id'])
-        
+        kid = self.kern_api.start().json()['id']
+        r = self.kern_api.get(kid)
+        kern1 = r.json()
+        self.assertEquals(r.status_code, 200)
+        assert isinstance(kern1, dict)
+        self.assertIn('id', kern1)
+        self.assertIn('ws_url', kern1)
+        self.assertEqual(kern1['id'], kid)
+
         # Request a bad kernel id and check that a JSON
         # message is returned!
         bad_id = '111-111-111-111-111'
-        bad_url = self.base_url() + '/' + bad_id
-        r = requests.get(bad_url)
-        status = r.status_code
-        message = r.json()
-        self.assertEquals(status, 404)
-        assert isinstance(message, dict)
-        self.assertIn('message', message)
-        self.assertEquals(message['message'], 'Kernel does not exist: ' + bad_id)
-        
+        with assert_http_error(404, 'Kernel does not exist: ' + bad_id):
+            self.kern_api.get(bad_id)
+
         # DELETE kernel with id
-        r = requests.delete(url)
+        r = self.kern_api.shutdown(kid)
         self.assertEqual(r.status_code, 204)
-        r = requests.get(self.base_url())
-        self.assertEqual(r.json(), [])
-        
+        kernels = self.kern_api.list().json()
+        self.assertEqual(kernels, [])
+
         # Request to delete a non-existent kernel id
         bad_id = '111-111-111-111-111'
-        bad_url = self.base_url() + '/' + bad_id
-        r = requests.delete(bad_url)
-        status = r.status_code
-        message = r.json()
-        self.assertEquals(status, 404)
-        assert isinstance(message, dict)
-        self.assertIn('message', message)
-        self.assertEquals(message['message'], 'Kernel does not exist: ' + bad_id)
-        
  No newline at end of file
+        with assert_http_error(404, 'Kernel does not exist: ' + bad_id):
+            self.kern_api.shutdown(bad_id)

IPython/html/tests/launchnotebook.py

@@ -1,81 +1,83 b''
 """Base class for notebook tests."""
 
 import sys
 import time
 import requests
 from contextlib import contextmanager
 from subprocess import Popen, PIPE
 from unittest import TestCase
 
 from IPython.utils.tempdir import TemporaryDirectory
 
 class NotebookTestBase(TestCase):
     """A base class for tests that need a running notebook.
     
     This creates an empty profile in a temp ipython_dir
     and then starts the notebook server with a separate temp notebook_dir.
     """
 
     port = 12341
 
     @classmethod
     def wait_until_alive(cls):
         """Wait for the server to be alive"""
         url = 'http://localhost:%i/api/notebooks' % cls.port
         while True:
             try:
                 requests.get(url)
             except requests.exceptions.ConnectionError:
                 time.sleep(.1)
             else:
                 break
     
     @classmethod
     def wait_until_dead(cls):
         """Wait for the server to stop getting requests after shutdown"""
         url = 'http://localhost:%i/api/notebooks' % cls.port
         while True:
             try:
                 requests.get(url)
             except requests.exceptions.ConnectionError:
                 break
             else:
                 time.sleep(.1)
     
     @classmethod
     def setup_class(cls):
         cls.ipython_dir = TemporaryDirectory()
         cls.notebook_dir = TemporaryDirectory()
         notebook_args = [
             sys.executable, '-c',
             'from IPython.html.notebookapp import launch_new_instance; launch_new_instance()',
             '--port=%d' % cls.port,
             '--no-browser',
             '--ipython-dir=%s' % cls.ipython_dir.name,
             '--notebook-dir=%s' % cls.notebook_dir.name
         ]        
         cls.notebook = Popen(notebook_args, stdout=PIPE, stderr=PIPE)
         cls.wait_until_alive()
 
     @classmethod
     def teardown_class(cls):
         cls.notebook.terminate()
         cls.ipython_dir.cleanup()
         cls.notebook_dir.cleanup()
         cls.wait_until_dead()
 
     @classmethod
     def base_url(cls):
         return 'http://localhost:%i/' % cls.port
 
 
 @contextmanager
-def assert_http_error(status):
+def assert_http_error(status, msg=None):
     try:
         yield
     except requests.HTTPError as e:
         real_status = e.response.status_code
         assert real_status == status, \
                     "Expected status %d, got %d" % (real_status, status)
+        if msg:
+            assert msg in str(e), e
     else:
         assert False, "Expected HTTP error status"
  No newline at end of file