test_notebooks_api.py
312 lines
| 11.9 KiB
| text/x-python
|
PythonLexer
MinRK
|
r13091 | # coding: utf-8 | ||
Zachary Sailer
|
r13044 | """Test the notebooks webservice API.""" | ||
Zachary Sailer
|
r13041 | |||
Thomas Kluyver
|
r13083 | import io | ||
MinRK
|
r13137 | import json | ||
Zachary Sailer
|
r13041 | import os | ||
Thomas Kluyver
|
r13083 | import shutil | ||
MinRK
|
r13091 | from unicodedata import normalize | ||
Thomas Kluyver
|
r13083 | pjoin = os.path.join | ||
Zachary Sailer
|
r13041 | import requests | ||
MinRK
|
r13132 | from IPython.html.utils import url_path_join, url_escape | ||
Thomas Kluyver
|
r13099 | from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error | ||
MinRK
|
r13141 | from IPython.nbformat import current | ||
Thomas Kluyver
|
r13085 | from IPython.nbformat.current import (new_notebook, write, read, new_worksheet, | ||
new_heading_cell, to_notebook_json) | ||||
MinRK
|
r13141 | from IPython.nbformat import v2 | ||
MinRK
|
r13130 | from IPython.utils import py3compat | ||
Thomas Kluyver
|
r13083 | from IPython.utils.data import uniq_stable | ||
MinRK
|
r13130 | |||
Thomas Kluyver
|
r13083 | class NBAPI(object): | ||
"""Wrapper for notebook API calls.""" | ||||
def __init__(self, base_url): | ||||
self.base_url = base_url | ||||
MinRK
|
r13137 | def _req(self, verb, path, body=None): | ||
Thomas Kluyver
|
r13083 | response = requests.request(verb, | ||
MinRK
|
r13130 | url_path_join(self.base_url, 'api/notebooks', path), | ||
data=body, | ||||
) | ||||
Thomas Kluyver
|
r13083 | 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) | ||||
MinRK
|
r13130 | def upload_untitled(self, body, path='/'): | ||
return self._req('POST', path, body) | ||||
def copy_untitled(self, copy_from, path='/'): | ||||
MinRK
|
r13137 | body = json.dumps({'copy_from':copy_from}) | ||
return self._req('POST', path, body) | ||||
MinRK
|
r13130 | |||
def create(self, name, path='/'): | ||||
return self._req('PUT', url_path_join(path, name)) | ||||
Thomas Kluyver
|
r13084 | def upload(self, name, body, path='/'): | ||
MinRK
|
r13130 | return self._req('PUT', url_path_join(path, name), body) | ||
Thomas Kluyver
|
r13084 | |||
MinRK
|
r13130 | def copy(self, copy_from, copy_to, path='/'): | ||
MinRK
|
r13137 | body = json.dumps({'copy_from':copy_from}) | ||
return self._req('PUT', url_path_join(path, copy_to), body) | ||||
Thomas Kluyver
|
r13086 | |||
Thomas Kluyver
|
r13085 | def save(self, name, body, path='/'): | ||
return self._req('PUT', url_path_join(path, name), body) | ||||
Thomas Kluyver
|
r13083 | def delete(self, name, path='/'): | ||
return self._req('DELETE', url_path_join(path, name)) | ||||
def rename(self, name, path, new_name): | ||||
MinRK
|
r13138 | body = json.dumps({'name': new_name}) | ||
Thomas Kluyver
|
r13083 | return self._req('PATCH', url_path_join(path, name), body) | ||
Zachary Sailer
|
r13041 | |||
Thomas Kluyver
|
r13109 | def get_checkpoints(self, name, path): | ||
return self._req('GET', url_path_join(path, name, 'checkpoints')) | ||||
def new_checkpoint(self, name, path): | ||||
return self._req('POST', url_path_join(path, name, 'checkpoints')) | ||||
def restore_checkpoint(self, name, path, checkpoint_id): | ||||
return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id)) | ||||
def delete_checkpoint(self, name, path, checkpoint_id): | ||||
return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id)) | ||||
Zachary Sailer
|
r13041 | class APITest(NotebookTestBase): | ||
"""Test the kernels web service API""" | ||||
Thomas Kluyver
|
r13083 | 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'), | ||||
MinRK
|
r13130 | (u'å b', u'ç d') | ||
Thomas Kluyver
|
r13083 | ] | ||
dirs = uniq_stable([d for (d,n) in dirs_nbs]) | ||||
del dirs[0] # remove '' | ||||
def setUp(self): | ||||
nbdir = self.notebook_dir.name | ||||
Thomas Kluyver
|
r13084 | |||
Thomas Kluyver
|
r13083 | for d in self.dirs: | ||
MinRK
|
r13130 | if not os.path.isdir(pjoin(nbdir, d)): | ||
os.mkdir(pjoin(nbdir, d)) | ||||
Thomas Kluyver
|
r13083 | |||
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') | ||||
self.nb_api = NBAPI(self.base_url()) | ||||
def tearDown(self): | ||||
nbdir = self.notebook_dir.name | ||||
MinRK
|
r13130 | for dname in ['foo', 'Directory with spaces in', u'unicodé', u'å b']: | ||
Thomas Kluyver
|
r13083 | shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True) | ||
if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')): | ||||
os.unlink(pjoin(nbdir, 'inroot.ipynb')) | ||||
def test_list_notebooks(self): | ||||
nbs = self.nb_api.list().json() | ||||
self.assertEqual(len(nbs), 1) | ||||
self.assertEqual(nbs[0]['name'], 'inroot.ipynb') | ||||
nbs = self.nb_api.list('/Directory with spaces in/').json() | ||||
self.assertEqual(len(nbs), 1) | ||||
self.assertEqual(nbs[0]['name'], 'inspace.ipynb') | ||||
nbs = self.nb_api.list(u'/unicodé/').json() | ||||
self.assertEqual(len(nbs), 1) | ||||
self.assertEqual(nbs[0]['name'], 'innonascii.ipynb') | ||||
MinRK
|
r13130 | self.assertEqual(nbs[0]['path'], u'unicodé') | ||
Thomas Kluyver
|
r13083 | |||
nbs = self.nb_api.list('/foo/bar/').json() | ||||
self.assertEqual(len(nbs), 1) | ||||
self.assertEqual(nbs[0]['name'], 'baz.ipynb') | ||||
MinRK
|
r13130 | self.assertEqual(nbs[0]['path'], 'foo/bar') | ||
Thomas Kluyver
|
r13083 | |||
nbs = self.nb_api.list('foo').json() | ||||
self.assertEqual(len(nbs), 4) | ||||
MinRK
|
r13091 | nbnames = { normalize('NFC', n['name']) for n in nbs } | ||
expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodé.ipynb'] | ||||
expected = { normalize('NFC', name) for name in expected } | ||||
self.assertEqual(nbnames, expected) | ||||
Thomas Kluyver
|
r13083 | |||
Thomas Kluyver
|
r13099 | def test_list_nonexistant_dir(self): | ||
with assert_http_error(404): | ||||
self.nb_api.list('nonexistant') | ||||
Thomas Kluyver
|
r13087 | |||
Thomas Kluyver
|
r13083 | def test_get_contents(self): | ||
for d, name in self.dirs_nbs: | ||||
nb = self.nb_api.read('%s.ipynb' % name, d+'/').json() | ||||
MinRK
|
r13130 | self.assertEqual(nb['name'], u'%s.ipynb' % name) | ||
Thomas Kluyver
|
r13083 | self.assertIn('content', nb) | ||
self.assertIn('metadata', nb['content']) | ||||
self.assertIsInstance(nb['content']['metadata'], dict) | ||||
# Name that doesn't exist - should be a 404 | ||||
Thomas Kluyver
|
r13099 | with assert_http_error(404): | ||
self.nb_api.read('q.ipynb', 'foo') | ||||
Thomas Kluyver
|
r13083 | |||
def _check_nb_created(self, resp, name, path): | ||||
Thomas Kluyver
|
r13084 | self.assertEqual(resp.status_code, 201) | ||
MinRK
|
r13130 | location_header = py3compat.str_to_unicode(resp.headers['Location']) | ||
MinRK
|
r13132 | self.assertEqual(location_header, url_escape(url_path_join(u'/api/notebooks', path, name))) | ||
Thomas Kluyver
|
r13083 | self.assertEqual(resp.json()['name'], name) | ||
assert os.path.isfile(pjoin(self.notebook_dir.name, path, name)) | ||||
def test_create_untitled(self): | ||||
MinRK
|
r13130 | resp = self.nb_api.create_untitled(path=u'Ã¥ b') | ||
self._check_nb_created(resp, 'Untitled0.ipynb', u'Ã¥ b') | ||||
Thomas Kluyver
|
r13083 | |||
# Second time | ||||
MinRK
|
r13130 | resp = self.nb_api.create_untitled(path=u'Ã¥ b') | ||
self._check_nb_created(resp, 'Untitled1.ipynb', u'Ã¥ b') | ||||
Thomas Kluyver
|
r13083 | |||
# And two directories down | ||||
resp = self.nb_api.create_untitled(path='foo/bar') | ||||
self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar')) | ||||
MinRK
|
r13130 | def test_upload_untitled(self): | ||
Thomas Kluyver
|
r13084 | nb = new_notebook(name='Upload test') | ||
Thomas Kluyver
|
r13085 | nbmodel = {'content': nb} | ||
MinRK
|
r13130 | resp = self.nb_api.upload_untitled(path=u'Ã¥ b', | ||
MinRK
|
r13138 | body=json.dumps(nbmodel)) | ||
MinRK
|
r13132 | self._check_nb_created(resp, 'Untitled0.ipynb', u'Ã¥ b') | ||
MinRK
|
r13130 | |||
def test_upload(self): | ||||
nb = new_notebook(name=u'ignored') | ||||
nbmodel = {'content': nb} | ||||
resp = self.nb_api.upload(u'Upload tést.ipynb', path=u'å b', | ||||
MinRK
|
r13138 | body=json.dumps(nbmodel)) | ||
MinRK
|
r13130 | self._check_nb_created(resp, u'Upload tést.ipynb', u'å b') | ||
MinRK
|
r13141 | def test_upload_v2(self): | ||
nb = v2.new_notebook() | ||||
ws = v2.new_worksheet() | ||||
nb.worksheets.append(ws) | ||||
ws.cells.append(v2.new_code_cell(input='print("hi")')) | ||||
nbmodel = {'content': nb} | ||||
resp = self.nb_api.upload(u'Upload tést.ipynb', path=u'å b', | ||||
body=json.dumps(nbmodel)) | ||||
self._check_nb_created(resp, u'Upload tést.ipynb', u'å b') | ||||
resp = self.nb_api.read(u'Upload tést.ipynb', u'å b') | ||||
data = resp.json() | ||||
self.assertEqual(data['content']['nbformat'], current.nbformat) | ||||
self.assertEqual(data['content']['orig_nbformat'], 2) | ||||
MinRK
|
r13130 | def test_copy_untitled(self): | ||
resp = self.nb_api.copy_untitled(u'ç d.ipynb', path=u'å b') | ||||
self._check_nb_created(resp, u'ç d-Copy0.ipynb', u'å b') | ||||
Thomas Kluyver
|
r13084 | |||
Thomas Kluyver
|
r13086 | def test_copy(self): | ||
MinRK
|
r13130 | resp = self.nb_api.copy(u'ç d.ipynb', u'cøpy.ipynb', path=u'å b') | ||
self._check_nb_created(resp, u'cøpy.ipynb', u'å b') | ||||
Thomas Kluyver
|
r13086 | |||
Thomas Kluyver
|
r13083 | 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') | ||||
Thomas Kluyver
|
r13089 | self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb') | ||
Thomas Kluyver
|
r13083 | 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) | ||||
Thomas Kluyver
|
r13085 | |||
def test_save(self): | ||||
resp = self.nb_api.read('a.ipynb', 'foo') | ||||
MinRK
|
r13138 | nbcontent = json.loads(resp.text)['content'] | ||
Thomas Kluyver
|
r13085 | nb = to_notebook_json(nbcontent) | ||
ws = new_worksheet() | ||||
nb.worksheets = [ws] | ||||
Thomas Kluyver
|
r13111 | ws.cells.append(new_heading_cell(u'Created by test ³')) | ||
Thomas Kluyver
|
r13085 | |||
nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb} | ||||
MinRK
|
r13138 | resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel)) | ||
Thomas Kluyver
|
r13085 | |||
nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb') | ||||
Thomas Kluyver
|
r13111 | with io.open(nbfile, 'r', encoding='utf-8') as f: | ||
Thomas Kluyver
|
r13085 | newnb = read(f, format='ipynb') | ||
self.assertEqual(newnb.worksheets[0].cells[0].source, | ||||
Thomas Kluyver
|
r13111 | u'Created by test ³') | ||
nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content'] | ||||
newnb = to_notebook_json(nbcontent) | ||||
self.assertEqual(newnb.worksheets[0].cells[0].source, | ||||
u'Created by test ³') | ||||
Thomas Kluyver
|
r13087 | |||
# Save and rename | ||||
nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb} | ||||
MinRK
|
r13138 | resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel)) | ||
Thomas Kluyver
|
r13087 | saved = resp.json() | ||
self.assertEqual(saved['name'], 'a2.ipynb') | ||||
self.assertEqual(saved['path'], 'foo/bar') | ||||
assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb')) | ||||
assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')) | ||||
Thomas Kluyver
|
r13099 | with assert_http_error(404): | ||
Thomas Kluyver
|
r13109 | self.nb_api.read('a.ipynb', 'foo') | ||
def test_checkpoints(self): | ||||
resp = self.nb_api.read('a.ipynb', 'foo') | ||||
r = self.nb_api.new_checkpoint('a.ipynb', 'foo') | ||||
self.assertEqual(r.status_code, 201) | ||||
cp1 = r.json() | ||||
MinRK
|
r13122 | self.assertEqual(set(cp1), {'id', 'last_modified'}) | ||
self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id']) | ||||
Thomas Kluyver
|
r13109 | |||
# Modify it | ||||
MinRK
|
r13138 | nbcontent = json.loads(resp.text)['content'] | ||
Thomas Kluyver
|
r13109 | nb = to_notebook_json(nbcontent) | ||
ws = new_worksheet() | ||||
nb.worksheets = [ws] | ||||
hcell = new_heading_cell('Created by test') | ||||
ws.cells.append(hcell) | ||||
# Save | ||||
nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb} | ||||
MinRK
|
r13138 | resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel)) | ||
Thomas Kluyver
|
r13109 | |||
# List checkpoints | ||||
cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json() | ||||
self.assertEqual(cps, [cp1]) | ||||
nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content'] | ||||
nb = to_notebook_json(nbcontent) | ||||
self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test') | ||||
# Restore cp1 | ||||
MinRK
|
r13122 | r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['id']) | ||
Thomas Kluyver
|
r13109 | self.assertEqual(r.status_code, 204) | ||
nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content'] | ||||
nb = to_notebook_json(nbcontent) | ||||
self.assertEqual(nb.worksheets, []) | ||||
# Delete cp1 | ||||
MinRK
|
r13122 | r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['id']) | ||
Thomas Kluyver
|
r13109 | self.assertEqual(r.status_code, 204) | ||
cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json() | ||||
self.assertEqual(cps, []) | ||||
MinRK
|
r13141 | |||