add tests for session api

Zachary Sailer 13 years ago committed by MinRK
parent 8ce586a241
commit e9ac1c2d37

@ -41,8 +41,7 @@ class MainKernelHandler(IPythonHandler):
@web.authenticated
def post(self):
km = self.kernel_manager
nbm = self.notebook_manager
kernel_id = km.start_kernel(cwd=nbm.notebook_dir)
kernel_id = km.start_kernel()
model = km.kernel_model(kernel_id, self.ws_url)
self.set_header('Location', '{0}kernels/{1}'.format(self.base_kernel_url, kernel_id))
self.finish(jsonapi.dumps(model))

@ -21,3 +21,16 @@ class KernelAPITest(NotebookTestBase):
url = self.base_url()
r = requests.get(url)
assert r.json() == []
def test_main_kernel_handler(self):
# POST request
r = requests.post(self.base_url())
data = r.json()
assert isinstance(data, dict)
# GET request
r = requests.get(self.base_url())
assert isinstance(r.json(), list)
self.assertEqual(r.json()[0], data['id'])

@ -98,11 +98,11 @@ class FileNotebookManager(NotebookManager):
full_path = self.get_os_path(notebook_name, notebook_path)
if change == "name":
new_path = self.get_os_path(data['name'], notebook_path)
try:
if not os.path.isfile(new_path):
os.rename(full_path,
self.get_os_path(data['name'], notebook_path))
notebook_name = data['name']
except OSError as e:
else:
raise web.HTTPError(409, u'Notebook name already exists.')
if change == "path":
new_path = self.get_os_path(data['name'], data['path'])

@ -44,6 +44,7 @@ class NotebookRootHandler(IPythonHandler):
"""post creates a notebooks in the directory where the
server was started"""
nbm = self.notebook_manager
self.log.info(nbm.notebook_dir)
body = self.request.body.strip()
format = self.get_argument('format', default='json')
name = self.get_argument('name', default=None)

@ -1,10 +1,9 @@
"""Test the all of the services API."""
"""Test the notebooks webservice API."""
import os
import sys
import json
import urllib
from zmq.utils import jsonapi
import requests
@ -14,29 +13,23 @@ from IPython.html.tests.launchnotebook import NotebookTestBase
class APITest(NotebookTestBase):
"""Test the kernels web service API"""
def base_url(self):
return super(APITest,self).base_url()
def notebooks_url(self):
return self.base_url() + 'api/notebooks'
def notebook_url(self):
return super(APITest,self).base_url() + 'api/notebooks'
def mknb(self, name='', path='/'):
url = self.notebooks_url() + path
url = self.notebook_url() + path
return url, requests.post(url)
def delnb(self, name, path='/'):
url = self.notebooks_url() + path + name
url = self.notebook_url() + path + name
r = requests.delete(url)
return r.status_code
def test_no_notebooks(self):
url = self.notebooks_url()
r = requests.get(url)
self.assertEqual(r.json(), [])
def test_notebook_root_handler(self):
# POST a notebook and test the dict thats returned.
url, nb = self.mknb()
#url, nb = self.mknb()
url = self.notebook_url()
nb = requests.post(url)
data = nb.json()
assert isinstance(data, dict)
assert data.has_key("name")
@ -48,12 +41,14 @@ class APITest(NotebookTestBase):
r = requests.get(url)
assert isinstance(r.json(), list)
assert isinstance(r.json()[0], dict)
self.delnb('Untitled0.ipynb')
def test_notebook_handler(self):
# GET with a notebook name.
url, nb = self.mknb()
data = nb.json()
url = self.notebooks_url() + '/Untitled0.ipynb'
url = self.notebook_url() + '/Untitled0.ipynb'
r = requests.get(url)
assert isinstance(data, dict)
self.assertEqual(r.json(), data)
@ -65,7 +60,7 @@ class APITest(NotebookTestBase):
assert isinstance(data, dict)
# make sure the patch worked.
new_url = self.notebooks_url() + '/test.ipynb'
new_url = self.notebook_url() + '/test.ipynb'
r = requests.get(new_url)
assert isinstance(r.json(), dict)
self.assertEqual(r.json(), data)
@ -77,6 +72,7 @@ class APITest(NotebookTestBase):
# POST notebooks to folders one and two levels down.
os.makedirs(os.path.join(self.notebook_dir.name, 'foo'))
os.makedirs(os.path.join(self.notebook_dir.name, 'foo','bar'))
assert os.path.isdir(os.path.join(self.notebook_dir.name, 'foo'))
url, nb = self.mknb(path='/foo/')
url2, nb2 = self.mknb(path='/foo/bar/')
data = nb.json()
@ -112,7 +108,9 @@ class APITest(NotebookTestBase):
self.assertEqual(r.status_code, 404)
# DELETE notebooks
r = self.delnb('testfoo.ipynb', '/foo/')
r2 = self.delnb('Untitled0.ipynb', '/foo/bar/')
self.assertEqual(r, 204)
self.assertEqual(r2, 204)
r0 = self.delnb('test.ipynb')
r1 = self.delnb('testfoo.ipynb', '/foo/')
r2 = self.delnb('Untitled0.ipynb', '/foo/bar/')
self.assertEqual(r0, 204)
self.assertEqual(r1, 204)
self.assertEqual(r2, 204)

@ -80,8 +80,8 @@ class SessionHandler(IPythonHandler):
sm = self.session_manager
nbm = self.notebook_manager
km = self.kernel_manager
notebook_path = self.request.body
name, path = nbm.named_notebook_path(notebook_path)
data = jsonapi.loads(self.request.body)
name, path = nbm.named_notebook_path(data['notebook_path'])
sm.update_session(session_id, name=name)
model = sm.get_session(id=session_id)
self.finish(jsonapi.dumps(model))
@ -106,7 +106,9 @@ class SessionHandler(IPythonHandler):
_session_id_regex = r"(?P<session_id>\w+-\w+-\w+-\w+-\w+)"
default_handlers = [
(r"api/sessions/%s/" % _session_id_regex, SessionHandler),
(r"api/sessions/%s" % _session_id_regex, SessionHandler),
(r"api/sessions/", SessionRootHandler),
(r"api/sessions", SessionRootHandler)
]

@ -0,0 +1,95 @@
"""Test the sessions web service API."""
import os
import sys
import json
from zmq.utils import jsonapi
import requests
from IPython.html.tests.launchnotebook import NotebookTestBase
class SessionAPITest(NotebookTestBase):
"""Test the sessions web service API"""
def notebook_url(self):
return super(SessionAPITest,self).base_url() + 'api/notebooks'
def session_url(self):
return super(SessionAPITest,self).base_url() + 'api/sessions'
def mknb(self, name='', path='/'):
url = self.notebook_url() + path
return url, requests.post(url)
def delnb(self, name, path='/'):
url = self.notebook_url() + path + name
r = requests.delete(url)
return r.status_code
def test_no_sessions(self):
"""Make sure there are no sessions running at the start"""
url = self.session_url()
r = requests.get(url)
self.assertEqual(r.json(), [])
def test_session_root_handler(self):
# POST a session
url, nb = self.mknb()
notebook = nb.json()
param = {'notebook_path': notebook['path'] + notebook['name']}
r = requests.post(self.session_url(), params=param)
data = r.json()
assert isinstance(data, dict)
assert data.has_key('name')
self.assertEqual(data['name'], notebook['name'])
# GET sessions
r = requests.get(self.session_url())
assert isinstance(r.json(), list)
assert isinstance(r.json()[0], dict)
self.assertEqual(r.json()[0]['id'], data['id'])
# Clean up
self.delnb('Untitled0.ipynb')
sess_url = self.session_url() +'/'+data['id']
r = requests.delete(sess_url)
self.assertEqual(r.status_code, 204)
def test_session_handler(self):
# Create a session
url, nb = self.mknb()
notebook = nb.json()
param = {'notebook_path': notebook['path'] + notebook['name']}
r = requests.post(self.session_url(), params=param)
session = r.json()
# GET a session
sess_url = self.session_url() + '/' + session['id']
r = requests.get(sess_url)
assert isinstance(r.json(), dict)
self.assertEqual(r.json(), session)
# PATCH a session
data = {'notebook_path': 'test.ipynb'}
r = requests.patch(sess_url, data=jsonapi.dumps(data))
# Patching the notebook webservice too (just for consistency)
requests.patch(self.notebook_url() + '/Untitled0.ipynb',
data=jsonapi.dumps({'name':'test.ipynb'}))
assert isinstance(r.json(), dict)
assert r.json().has_key('name')
assert r.json().has_key('id')
self.assertEqual(r.json()['name'], 'test.ipynb')
self.assertEqual(r.json()['id'], session['id'])
# DELETE a session
r = requests.delete(sess_url)
self.assertEqual(r.status_code, 204)
r = requests.get(self.session_url())
assert r.json() == []
# Clean up
r = self.delnb('test.ipynb')
assert r == 204

@ -32,11 +32,12 @@ var IPython = (function (IPython) {
Session.prototype.notebook_rename = function (notebook_path) {
this.notebook_path = notebook_path;
name = {'notebook_path': notebook_path}
var settings = {
processData : false,
cache : false,
type : "PATCH",
data: notebook_path,
data: JSON.stringify(name),
dataType : "json",
};
var url = this._baseProjectUrl + 'api/sessions/' + this.session_id;

@ -15,7 +15,7 @@ class NotebookTestBase(TestCase):
and then starts the notebook server with a separate temp notebook_dir.
"""
port = 12342
port = 1234
def setUp(self):
self.ipython_dir = TemporaryDirectory()
@ -27,8 +27,7 @@ class NotebookTestBase(TestCase):
'--no-browser',
'--ipython-dir=%s' % self.ipython_dir.name,
'--notebook-dir=%s' % self.notebook_dir.name
]
#self.notebook = Popen(notebook_args)
]
self.notebook = Popen(notebook_args, stdout=PIPE, stderr=PIPE)
time.sleep(3.0)
@ -36,6 +35,7 @@ class NotebookTestBase(TestCase):
self.notebook.terminate()
self.ipython_dir.cleanup()
self.notebook_dir.cleanup()
time.sleep(3.0)
def base_url(self):
return 'http://localhost:%i/' % self.port

Loading…
Cancel
Save