mirror of
https://github.com/jupyter/notebook.git
synced 2025-01-30 12:11:32 +08:00
Improve tests for notebook REST API
This commit is contained in:
Thomas Kluyver
MinRK
parent
6c88abe1d6
commit
9d9adb1186
@ -1,113 +1,160 @@
|
||||
"""Test the notebooks webservice API."""
|
||||
|
||||
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import shutil
|
||||
from zmq.utils import jsonapi
|
||||
|
||||
pjoin = os.path.join
|
||||
|
||||
import requests
|
||||
|
||||
from IPython.html.utils import url_path_join
|
||||
from IPython.html.tests.launchnotebook import NotebookTestBase
|
||||
from IPython.nbformat.current import new_notebook, write
|
||||
from IPython.utils.data import uniq_stable
|
||||
|
||||
class NBAPI(object):
|
||||
"""Wrapper for notebook API calls."""
|
||||
def __init__(self, base_url):
|
||||
self.base_url = base_url
|
||||
|
||||
@property
|
||||
def nb_url(self):
|
||||
return url_path_join(self.base_url, 'api/notebooks')
|
||||
|
||||
def _req(self, verb, path, body=None):
|
||||
response = requests.request(verb,
|
||||
url_path_join(self.base_url, 'api/notebooks', path), data=body)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def list(self, path='/'):
|
||||
return self._req('GET', path)
|
||||
|
||||
def read(self, name, path='/'):
|
||||
return self._req('GET', url_path_join(path, name))
|
||||
|
||||
def create_untitled(self, path='/'):
|
||||
return self._req('POST', path)
|
||||
|
||||
def delete(self, name, path='/'):
|
||||
return self._req('DELETE', url_path_join(path, name))
|
||||
|
||||
def rename(self, name, path, new_name):
|
||||
body = jsonapi.dumps({'name': new_name})
|
||||
return self._req('PATCH', url_path_join(path, name), body)
|
||||
|
||||
class APITest(NotebookTestBase):
|
||||
"""Test the kernels web service API"""
|
||||
dirs_nbs = [('', 'inroot'),
|
||||
('Directory with spaces in', 'inspace'),
|
||||
(u'unicodé', 'innonascii'),
|
||||
('foo', 'a'),
|
||||
('foo', 'b'),
|
||||
('foo', 'name with spaces'),
|
||||
('foo', u'unicodé'),
|
||||
('foo/bar', 'baz'),
|
||||
]
|
||||
|
||||
def notebook_url(self):
|
||||
return url_path_join(super(APITest,self).base_url(), 'api/notebooks')
|
||||
dirs = uniq_stable([d for (d,n) in dirs_nbs])
|
||||
del dirs[0] # remove ''
|
||||
|
||||
def mknb(self, name='', path='/'):
|
||||
url = self.notebook_url() + path
|
||||
return url, requests.post(url)
|
||||
def setUp(self):
|
||||
nbdir = self.notebook_dir.name
|
||||
for d in self.dirs:
|
||||
os.mkdir(pjoin(nbdir, d))
|
||||
|
||||
def delnb(self, name, path='/'):
|
||||
url = self.notebook_url() + path + name
|
||||
r = requests.delete(url)
|
||||
return r.status_code
|
||||
for d, name in self.dirs_nbs:
|
||||
with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w') as f:
|
||||
nb = new_notebook(name=name)
|
||||
write(nb, f, format='ipynb')
|
||||
|
||||
def test_notebook_handler(self):
|
||||
# POST a notebook and test the dict thats returned.
|
||||
#url, nb = self.mknb()
|
||||
url = self.notebook_url()
|
||||
nb = requests.post(url+'/')
|
||||
data = nb.json()
|
||||
status = nb.status_code
|
||||
assert isinstance(data, dict)
|
||||
self.assertIn('name', data)
|
||||
self.assertIn('path', data)
|
||||
self.assertEqual(data['name'], u'Untitled0.ipynb')
|
||||
self.assertEqual(data['path'], u'/')
|
||||
self.nb_api = NBAPI(self.base_url())
|
||||
|
||||
# GET list of notebooks in directory.
|
||||
r = requests.get(url)
|
||||
assert isinstance(r.json(), list)
|
||||
assert isinstance(r.json()[0], dict)
|
||||
|
||||
self.delnb('Untitled0.ipynb')
|
||||
|
||||
# GET with a notebook name.
|
||||
url, nb = self.mknb()
|
||||
data = nb.json()
|
||||
url = self.notebook_url() + '/Untitled0.ipynb'
|
||||
r = requests.get(url)
|
||||
assert isinstance(data, dict)
|
||||
def tearDown(self):
|
||||
nbdir = self.notebook_dir.name
|
||||
|
||||
# PATCH (rename) request.
|
||||
new_name = {'name':'test.ipynb'}
|
||||
r = requests.patch(url, data=jsonapi.dumps(new_name))
|
||||
data = r.json()
|
||||
assert isinstance(data, dict)
|
||||
for dname in ['foo', 'Directory with spaces in', u'unicodé']:
|
||||
shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
|
||||
|
||||
# make sure the patch worked.
|
||||
new_url = self.notebook_url() + '/test.ipynb'
|
||||
r = requests.get(new_url)
|
||||
assert isinstance(r.json(), dict)
|
||||
if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
|
||||
os.unlink(pjoin(nbdir, 'inroot.ipynb'))
|
||||
|
||||
# GET bad (old) notebook name.
|
||||
r = requests.get(url)
|
||||
self.assertEqual(r.status_code, 404)
|
||||
def test_list_notebooks(self):
|
||||
nbs = self.nb_api.list().json()
|
||||
self.assertEqual(len(nbs), 1)
|
||||
self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
|
||||
|
||||
# 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()
|
||||
data2 = nb2.json()
|
||||
assert isinstance(data, dict)
|
||||
assert isinstance(data2, dict)
|
||||
self.assertIn('name', data)
|
||||
self.assertIn('path', data)
|
||||
self.assertEqual(data['name'], u'Untitled0.ipynb')
|
||||
self.assertEqual(data['path'], u'/foo/')
|
||||
self.assertIn('name', data2)
|
||||
self.assertIn('path', data2)
|
||||
self.assertEqual(data2['name'], u'Untitled0.ipynb')
|
||||
self.assertEqual(data2['path'], u'/foo/bar/')
|
||||
nbs = self.nb_api.list('/Directory with spaces in/').json()
|
||||
self.assertEqual(len(nbs), 1)
|
||||
self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
|
||||
|
||||
# GET request on notebooks one and two levels down.
|
||||
r = requests.get(url+'/Untitled0.ipynb')
|
||||
r2 = requests.get(url2+'/Untitled0.ipynb')
|
||||
assert isinstance(r.json(), dict)
|
||||
assert isinstance(r2.json(), dict)
|
||||
nbs = self.nb_api.list(u'/unicodé/').json()
|
||||
self.assertEqual(len(nbs), 1)
|
||||
self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
|
||||
|
||||
# PATCH notebooks that are one and two levels down.
|
||||
new_name = {'name': 'testfoo.ipynb'}
|
||||
r = requests.patch(url+'/Untitled0.ipynb', data=jsonapi.dumps(new_name))
|
||||
r = requests.get(url+'/testfoo.ipynb')
|
||||
data = r.json()
|
||||
assert isinstance(data, dict)
|
||||
self.assertIn('name', data)
|
||||
self.assertEqual(data['name'], 'testfoo.ipynb')
|
||||
r = requests.get(url+'/Untitled0.ipynb')
|
||||
self.assertEqual(r.status_code, 404)
|
||||
|
||||
# DELETE notebooks
|
||||
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)
|
||||
nbs = self.nb_api.list('/foo/bar/').json()
|
||||
self.assertEqual(len(nbs), 1)
|
||||
self.assertEqual(nbs[0]['name'], 'baz.ipynb')
|
||||
|
||||
nbs = self.nb_api.list('foo').json()
|
||||
self.assertEqual(len(nbs), 4)
|
||||
nbnames = set(n['name'] for n in nbs)
|
||||
self.assertEqual(nbnames, {'a.ipynb', 'b.ipynb',
|
||||
'name with spaces.ipynb', u'unicodé.ipynb'})
|
||||
|
||||
def test_get_contents(self):
|
||||
for d, name in self.dirs_nbs:
|
||||
nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
|
||||
self.assertEqual(nb['name'], '%s.ipynb' % name)
|
||||
self.assertIn('content', nb)
|
||||
self.assertIn('metadata', nb['content'])
|
||||
self.assertIsInstance(nb['content']['metadata'], dict)
|
||||
|
||||
# Name that doesn't exist - should be a 404
|
||||
try:
|
||||
self.nb_api.read('q.ipynb', 'foo')
|
||||
except requests.HTTPError as e:
|
||||
self.assertEqual(e.response.status_code, 404)
|
||||
else:
|
||||
assert False, "Reading a non-existent notebook should fail"
|
||||
|
||||
def _check_nb_created(self, resp, name, path):
|
||||
self.assertEqual(resp.headers['Location'].split('/')[-1], name)
|
||||
self.assertEqual(resp.json()['name'], name)
|
||||
assert os.path.isfile(pjoin(self.notebook_dir.name, path, name))
|
||||
|
||||
def test_create_untitled(self):
|
||||
resp = self.nb_api.create_untitled(path='foo')
|
||||
self._check_nb_created(resp, 'Untitled0.ipynb', 'foo')
|
||||
|
||||
# Second time
|
||||
resp = self.nb_api.create_untitled(path='foo')
|
||||
self._check_nb_created(resp, 'Untitled1.ipynb', 'foo')
|
||||
|
||||
# And two directories down
|
||||
resp = self.nb_api.create_untitled(path='foo/bar')
|
||||
self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar'))
|
||||
|
||||
def test_delete(self):
|
||||
for d, name in self.dirs_nbs:
|
||||
resp = self.nb_api.delete('%s.ipynb' % name, d)
|
||||
self.assertEqual(resp.status_code, 204)
|
||||
|
||||
for d in self.dirs + ['/']:
|
||||
nbs = self.nb_api.list(d).json()
|
||||
self.assertEqual(len(nbs), 0)
|
||||
|
||||
def test_rename(self):
|
||||
resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
|
||||
if False:
|
||||
# XXX: Spec says this should be set, but it isn't
|
||||
self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
|
||||
self.assertEqual(resp.json()['name'], 'z.ipynb')
|
||||
assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
|
||||
|
||||
nbs = self.nb_api.list('foo').json()
|
||||
nbnames = set(n['name'] for n in nbs)
|
||||
self.assertIn('z.ipynb', nbnames)
|
||||
self.assertNotIn('a.ipynb', nbnames)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user