Test kernel specs REST API

And fix kernel not found producing a 404 code.
This commit is contained in:
Thomas Kluyver 2014-05-08 17:32:47 -07:00
parent 02fa348c69
commit 1dabf3c547
2 changed files with 104 additions and 14 deletions

View File

@ -3,12 +3,11 @@
# Copyright (c) IPython Development Team. # Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License. # Distributed under the terms of the Modified BSD License.
import logging
from tornado import web from tornado import web
from zmq.utils import jsonapi from zmq.utils import jsonapi
from ...base.handlers import IPythonHandler, json_errors, path_regex from ...base.handlers import IPythonHandler, json_errors
class MainKernelSpecHandler(IPythonHandler): class MainKernelSpecHandler(IPythonHandler):
@ -34,7 +33,10 @@ class KernelSpecHandler(IPythonHandler):
@json_errors @json_errors
def get(self, kernel_name): def get(self, kernel_name):
ksm = self.kernel_spec_manager ksm = self.kernel_spec_manager
kernelspec = ksm.get_kernel_spec(kernel_name) try:
kernelspec = ksm.get_kernel_spec(kernel_name)
except KeyError:
raise web.HTTPError(404, u'Kernel spec %s not found' % kernel_name)
self.set_header("Content-Type", 'application/json') self.set_header("Content-Type", 'application/json')
self.finish(kernelspec.to_json()) self.finish(kernelspec.to_json())
@ -47,24 +49,18 @@ class KernelSpecResourceHandler(web.StaticFileHandler, IPythonHandler):
def get(self, kernel_name, path, include_body=True): def get(self, kernel_name, path, include_body=True):
ksm = self.kernel_spec_manager ksm = self.kernel_spec_manager
self.root = ksm.get_kernel_spec(kernel_name).resource_dir try:
self.log.warn("Set root: %s", self.root) self.root = ksm.get_kernel_spec(kernel_name).resource_dir
except KeyError:
raise web.HTTPError(404, u'Kernel spec %s not found' % kernel_name)
self.log.debug("Serving kernel resource from: %s", self.root)
return web.StaticFileHandler.get(self, path, include_body=include_body) return web.StaticFileHandler.get(self, path, include_body=include_body)
# @classmethod
# def get_absolute_path(cls, root, path):
# res = web.StaticFileHandler.get_absolute_path(cls, root, path)
# self.log.warn("Full path: %s", res)
# return res
def head(self, kernel_name, path): def head(self, kernel_name, path):
self.get(kernel_name, path, include_body=False) self.get(kernel_name, path, include_body=False)
#-----------------------------------------------------------------------------
# URL to handler mappings # URL to handler mappings
#-----------------------------------------------------------------------------
_kernel_name_regex = r"(?P<kernel_name>\w+)" _kernel_name_regex = r"(?P<kernel_name>\w+)"

View File

@ -0,0 +1,94 @@
# coding: utf-8
"""Test the kernel specs webservice API."""
import errno
import io
import json
import os
pjoin = os.path.join
import requests
from IPython.html.utils import url_path_join
from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
# Copied from IPython.kernel.tests.test_kernelspec so updating that doesn't
# break these tests
sample_kernel_json = {'argv':['cat', '{connection_file}'],
'display_name':'Test kernel',
'language':'bash',
}
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 _req(self, verb, path, body=None):
response = requests.request(verb,
url_path_join(self.base_url, 'api/kernelspecs', path),
data=body,
)
response.raise_for_status()
return response
def list(self):
return self._req('GET', '')
def kernel_spec_info(self, name):
return self._req('GET', name)
def kernel_resource(self, name, path):
return self._req('GET', url_path_join(name, path))
class APITest(NotebookTestBase):
"""Test the kernelspec web service API"""
def setUp(self):
ipydir = self.ipython_dir.name
sample_kernel_dir = pjoin(ipydir, 'kernels', 'sample')
try:
os.makedirs(sample_kernel_dir)
except OSError as e:
if e.errno != errno.EEXIST:
raise
with open(pjoin(sample_kernel_dir, 'kernel.json'), 'w') as f:
json.dump(sample_kernel_json, f)
with io.open(pjoin(sample_kernel_dir, 'resource.txt'), 'w',
encoding='utf-8') as f:
f.write(some_resource)
self.ks_api = KernelSpecAPI(self.base_url())
def test_list_kernelspecs(self):
specs = self.ks_api.list().json()
assert isinstance(specs, list)
# 2: the sample kernelspec created in setUp, and the native Python kernel
self.assertEqual(len(specs), 2)
assert any(s == {'name': 'sample', 'display_name': 'Test kernel'}
for s in specs), specs
def test_get_kernelspec(self):
spec = self.ks_api.kernel_spec_info('Sample').json() # Case insensitive
self.assertEqual(spec['language'], 'bash')
def test_get_nonexistant_kernelspec(self):
with assert_http_error(404):
self.ks_api.kernel_spec_info('nonexistant')
def test_get_kernel_resource_file(self):
res = self.ks_api.kernel_resource('sAmple', 'resource.txt')
self.assertEqual(res.text, some_resource)
def test_get_nonexistant_resource(self):
with assert_http_error(404):
self.ks_api.kernel_resource('nonexistant', 'resource.txt')
with assert_http_error(404):
self.ks_api.kernel_resource('sample', 'nonexistant.txt')