test_contents_api.py
687 lines
| 24.2 KiB
| text/x-python
|
PythonLexer
|
r13091 | # coding: utf-8 | ||
|
r17524 | """Test the contents webservice API.""" | ||
|
r13041 | |||
|
r17525 | import base64 | ||
|
r19748 | from contextlib import contextmanager | ||
|
r13083 | import io | ||
|
r13137 | import json | ||
|
r13041 | import os | ||
|
r13083 | import shutil | ||
|
r13091 | from unicodedata import normalize | ||
|
r13083 | pjoin = os.path.join | ||
|
r13041 | import requests | ||
|
r19839 | from ..filecheckpoints import GenericFileCheckpoints | ||
|
r19838 | |||
from IPython.config import Config | ||||
|
r19604 | from IPython.html.utils import url_path_join, url_escape, to_os_path | ||
|
r13099 | from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error | ||
|
r18607 | from IPython.nbformat import read, write, from_dict | ||
from IPython.nbformat.v4 import ( | ||||
new_notebook, new_markdown_cell, | ||||
) | ||||
|
r13141 | from IPython.nbformat import v2 | ||
|
r13130 | from IPython.utils import py3compat | ||
|
r13083 | from IPython.utils.data import uniq_stable | ||
|
r19748 | from IPython.utils.tempdir import TemporaryDirectory | ||
|
r13083 | |||
|
r13130 | |||
|
r17525 | def notebooks_only(dir_model): | ||
return [nb for nb in dir_model['content'] if nb['type']=='notebook'] | ||||
|
r15079 | |||
|
r17525 | def dirs_only(dir_model): | ||
return [x for x in dir_model['content'] if x['type']=='directory'] | ||||
|
r15522 | |||
|
r15079 | |||
|
r17524 | class API(object): | ||
"""Wrapper for contents API calls.""" | ||||
|
r13083 | def __init__(self, base_url): | ||
self.base_url = base_url | ||||
|
r18790 | def _req(self, verb, path, body=None, params=None): | ||
|
r13083 | response = requests.request(verb, | ||
|
r17524 | url_path_join(self.base_url, 'api/contents', path), | ||
|
r18795 | data=body, params=params, | ||
|
r13130 | ) | ||
|
r13083 | response.raise_for_status() | ||
return response | ||||
def list(self, path='/'): | ||||
return self._req('GET', path) | ||||
|
r20134 | def read(self, path, type=None, format=None, content=None): | ||
|
r18790 | params = {} | ||
|
r19391 | if type is not None: | ||
params['type'] = type | ||||
|
r18788 | if format is not None: | ||
|
r18790 | params['format'] = format | ||
|
r20134 | if content == False: | ||
params['content'] = '0' | ||||
|
r18790 | return self._req('GET', path, params=params) | ||
|
r13083 | |||
|
r18759 | def create_untitled(self, path='/', ext='.ipynb'): | ||
|
r17527 | body = None | ||
if ext: | ||||
body = json.dumps({'ext': ext}) | ||||
return self._req('POST', path, body) | ||||
|
r13083 | |||
|
r18758 | def mkdir_untitled(self, path='/'): | ||
return self._req('POST', path, json.dumps({'type': 'directory'})) | ||||
|
r18750 | def copy(self, copy_from, path='/'): | ||
|
r13137 | body = json.dumps({'copy_from':copy_from}) | ||
return self._req('POST', path, body) | ||||
|
r13130 | |||
|
r18749 | def create(self, path='/'): | ||
return self._req('PUT', path) | ||||
|
r13130 | |||
|
r18749 | def upload(self, path, body): | ||
return self._req('PUT', path, body) | ||||
|
r13084 | |||
|
r18749 | def mkdir(self, path='/'): | ||
return self._req('PUT', path, json.dumps({'type': 'directory'})) | ||||
|
r17530 | |||
|
r18750 | def copy_put(self, copy_from, path='/'): | ||
|
r13137 | body = json.dumps({'copy_from':copy_from}) | ||
|
r18749 | return self._req('PUT', path, body) | ||
|
r13086 | |||
|
r18749 | def save(self, path, body): | ||
return self._req('PUT', path, body) | ||||
|
r13085 | |||
|
r18749 | def delete(self, path='/'): | ||
return self._req('DELETE', path) | ||||
|
r13083 | |||
|
r18749 | def rename(self, path, new_path): | ||
body = json.dumps({'path': new_path}) | ||||
return self._req('PATCH', path, body) | ||||
|
r13041 | |||
|
r18749 | def get_checkpoints(self, path): | ||
return self._req('GET', url_path_join(path, 'checkpoints')) | ||||
|
r13109 | |||
|
r18749 | def new_checkpoint(self, path): | ||
return self._req('POST', url_path_join(path, 'checkpoints')) | ||||
|
r13109 | |||
|
r18749 | def restore_checkpoint(self, path, checkpoint_id): | ||
return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id)) | ||||
|
r13109 | |||
|
r18749 | def delete_checkpoint(self, path, checkpoint_id): | ||
return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id)) | ||||
|
r13109 | |||
|
r13041 | class APITest(NotebookTestBase): | ||
"""Test the kernels web service API""" | ||||
|
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'), | ||||
|
r15523 | ('ordering', 'A'), | ||
('ordering', 'b'), | ||||
('ordering', 'C'), | ||||
(u'å b', u'ç d'), | ||||
|
r13083 | ] | ||
|
r15522 | hidden_dirs = ['.hidden', '__pycache__'] | ||
|
r13083 | |||
|
r19617 | # Don't include root dir. | ||
dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs[1:]]) | ||||
|
r15623 | top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs} | ||
|
r13083 | |||
|
r17525 | @staticmethod | ||
def _blob_for_name(name): | ||||
return name.encode('utf-8') + b'\xFF' | ||||
@staticmethod | ||||
def _txt_for_name(name): | ||||
return u'%s text file' % name | ||||
|
r19604 | |||
def to_os_path(self, api_path): | ||||
return to_os_path(api_path, root=self.notebook_dir.name) | ||||
def make_dir(self, api_path): | ||||
"""Create a directory at api_path""" | ||||
os_path = self.to_os_path(api_path) | ||||
try: | ||||
os.makedirs(os_path) | ||||
except OSError: | ||||
print("Directory already exists: %r" % os_path) | ||||
|
r19613 | |||
|
r19604 | def make_txt(self, api_path, txt): | ||
"""Make a text file at a given api_path""" | ||||
os_path = self.to_os_path(api_path) | ||||
with io.open(os_path, 'w', encoding='utf-8') as f: | ||||
f.write(txt) | ||||
def make_blob(self, api_path, blob): | ||||
"""Make a binary file at a given api_path""" | ||||
os_path = self.to_os_path(api_path) | ||||
with io.open(os_path, 'wb') as f: | ||||
f.write(blob) | ||||
def make_nb(self, api_path, nb): | ||||
"""Make a notebook file at a given api_path""" | ||||
os_path = self.to_os_path(api_path) | ||||
with io.open(os_path, 'w', encoding='utf-8') as f: | ||||
write(nb, f, version=4) | ||||
|
r19613 | |||
def delete_dir(self, api_path): | ||||
"""Delete a directory at api_path, removing any contents.""" | ||||
os_path = self.to_os_path(api_path) | ||||
shutil.rmtree(os_path, ignore_errors=True) | ||||
def delete_file(self, api_path): | ||||
"""Delete a file at the given path if it exists.""" | ||||
if self.isfile(api_path): | ||||
os.unlink(self.to_os_path(api_path)) | ||||
|
r19604 | |||
def isfile(self, api_path): | ||||
return os.path.isfile(self.to_os_path(api_path)) | ||||
def isdir(self, api_path): | ||||
return os.path.isdir(self.to_os_path(api_path)) | ||||
|
r13083 | def setUp(self): | ||
|
r17525 | |||
|
r15522 | for d in (self.dirs + self.hidden_dirs): | ||
|
r19604 | self.make_dir(d) | ||
|
r13083 | |||
for d, name in self.dirs_nbs: | ||||
|
r17525 | # create a notebook | ||
|
r19604 | nb = new_notebook() | ||
self.make_nb(u'{}/{}.ipynb'.format(d, name), nb) | ||||
|
r17525 | # create a text file | ||
|
r19604 | txt = self._txt_for_name(name) | ||
self.make_txt(u'{}/{}.txt'.format(d, name), txt) | ||||
|
r17525 | # create a binary file | ||
|
r19604 | blob = self._blob_for_name(name) | ||
self.make_blob(u'{}/{}.blob'.format(d, name), blob) | ||||
|
r17525 | |||
|
r17524 | self.api = API(self.base_url()) | ||
|
r13083 | |||
def tearDown(self): | ||||
|
r15522 | for dname in (list(self.top_level_dirs) + self.hidden_dirs): | ||
|
r19613 | self.delete_dir(dname) | ||
self.delete_file('inroot.ipynb') | ||||
|
r13083 | |||
def test_list_notebooks(self): | ||||
|
r17524 | nbs = notebooks_only(self.api.list().json()) | ||
|
r13083 | self.assertEqual(len(nbs), 1) | ||
self.assertEqual(nbs[0]['name'], 'inroot.ipynb') | ||||
|
r17524 | nbs = notebooks_only(self.api.list('/Directory with spaces in/').json()) | ||
|
r13083 | self.assertEqual(len(nbs), 1) | ||
self.assertEqual(nbs[0]['name'], 'inspace.ipynb') | ||||
|
r17524 | nbs = notebooks_only(self.api.list(u'/unicodé/').json()) | ||
|
r13083 | self.assertEqual(len(nbs), 1) | ||
self.assertEqual(nbs[0]['name'], 'innonascii.ipynb') | ||||
|
r18749 | self.assertEqual(nbs[0]['path'], u'unicodé/innonascii.ipynb') | ||
|
r13083 | |||
|
r17524 | nbs = notebooks_only(self.api.list('/foo/bar/').json()) | ||
|
r13083 | self.assertEqual(len(nbs), 1) | ||
self.assertEqual(nbs[0]['name'], 'baz.ipynb') | ||||
|
r18749 | self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb') | ||
|
r13083 | |||
|
r17524 | nbs = notebooks_only(self.api.list('foo').json()) | ||
|
r13083 | self.assertEqual(len(nbs), 4) | ||
|
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) | ||||
|
r17523 | |||
|
r17524 | nbs = notebooks_only(self.api.list('ordering').json()) | ||
|
r15523 | nbnames = [n['name'] for n in nbs] | ||
expected = ['A.ipynb', 'b.ipynb', 'C.ipynb'] | ||||
self.assertEqual(nbnames, expected) | ||||
|
r13083 | |||
|
r15522 | def test_list_dirs(self): | ||
|
r17524 | dirs = dirs_only(self.api.list().json()) | ||
|
r15623 | dir_names = {normalize('NFC', d['name']) for d in dirs} | ||
|
r15522 | self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs | ||
|
r20134 | def test_get_dir_no_content(self): | ||
for d in self.dirs: | ||||
model = self.api.read(d, content=False).json() | ||||
self.assertEqual(model['path'], d) | ||||
self.assertEqual(model['type'], 'directory') | ||||
self.assertIn('content', model) | ||||
self.assertEqual(model['content'], None) | ||||
|
r13099 | def test_list_nonexistant_dir(self): | ||
with assert_http_error(404): | ||||
|
r17524 | self.api.list('nonexistant') | ||
|
r13087 | |||
|
r17525 | def test_get_nb_contents(self): | ||
|
r13083 | for d, name in self.dirs_nbs: | ||
|
r18749 | path = url_path_join(d, name + '.ipynb') | ||
nb = self.api.read(path).json() | ||||
|
r13130 | self.assertEqual(nb['name'], u'%s.ipynb' % name) | ||
|
r18749 | self.assertEqual(nb['path'], path) | ||
|
r17525 | self.assertEqual(nb['type'], 'notebook') | ||
self.assertIn('content', nb) | ||||
self.assertEqual(nb['format'], 'json') | ||||
|
r13083 | self.assertIn('metadata', nb['content']) | ||
self.assertIsInstance(nb['content']['metadata'], dict) | ||||
|
r20134 | def test_get_nb_no_content(self): | ||
for d, name in self.dirs_nbs: | ||||
path = url_path_join(d, name + '.ipynb') | ||||
nb = self.api.read(path, content=False).json() | ||||
self.assertEqual(nb['name'], u'%s.ipynb' % name) | ||||
self.assertEqual(nb['path'], path) | ||||
self.assertEqual(nb['type'], 'notebook') | ||||
self.assertIn('content', nb) | ||||
self.assertEqual(nb['content'], None) | ||||
|
r17525 | def test_get_contents_no_such_file(self): | ||
|
r13083 | # Name that doesn't exist - should be a 404 | ||
|
r13099 | with assert_http_error(404): | ||
|
r18749 | self.api.read('foo/q.ipynb') | ||
|
r13083 | |||
|
r17525 | def test_get_text_file_contents(self): | ||
for d, name in self.dirs_nbs: | ||||
|
r18749 | path = url_path_join(d, name + '.txt') | ||
model = self.api.read(path).json() | ||||
|
r17525 | self.assertEqual(model['name'], u'%s.txt' % name) | ||
|
r18749 | self.assertEqual(model['path'], path) | ||
|
r17525 | self.assertIn('content', model) | ||
self.assertEqual(model['format'], 'text') | ||||
self.assertEqual(model['type'], 'file') | ||||
self.assertEqual(model['content'], self._txt_for_name(name)) | ||||
# Name that doesn't exist - should be a 404 | ||||
with assert_http_error(404): | ||||
|
r18749 | self.api.read('foo/q.txt') | ||
|
r17525 | |||
|
r18788 | # Specifying format=text should fail on a non-UTF-8 file | ||
with assert_http_error(400): | ||||
|
r19391 | self.api.read('foo/bar/baz.blob', type='file', format='text') | ||
|
r18788 | |||
|
r17525 | def test_get_binary_file_contents(self): | ||
for d, name in self.dirs_nbs: | ||||
|
r18749 | path = url_path_join(d, name + '.blob') | ||
model = self.api.read(path).json() | ||||
|
r17525 | self.assertEqual(model['name'], u'%s.blob' % name) | ||
|
r18749 | self.assertEqual(model['path'], path) | ||
|
r17525 | self.assertIn('content', model) | ||
self.assertEqual(model['format'], 'base64') | ||||
self.assertEqual(model['type'], 'file') | ||||
|
r19615 | self.assertEqual( | ||
|
r19616 | base64.decodestring(model['content'].encode('ascii')), | ||
|
r19615 | self._blob_for_name(name), | ||
) | ||||
|
r17525 | |||
# Name that doesn't exist - should be a 404 | ||||
with assert_http_error(404): | ||||
|
r18749 | self.api.read('foo/q.txt') | ||
|
r17525 | |||
|
r18781 | def test_get_bad_type(self): | ||
with assert_http_error(400): | ||||
|
r19391 | self.api.read(u'unicodé', type='file') # this is a directory | ||
|
r18781 | |||
with assert_http_error(400): | ||||
|
r19391 | self.api.read(u'unicodé/innonascii.ipynb', type='directory') | ||
|
r18781 | |||
|
r18749 | def _check_created(self, resp, path, type='notebook'): | ||
|
r13084 | self.assertEqual(resp.status_code, 201) | ||
|
r13130 | location_header = py3compat.str_to_unicode(resp.headers['Location']) | ||
|
r18749 | self.assertEqual(location_header, url_escape(url_path_join(u'/api/contents', path))) | ||
|
r17528 | rjson = resp.json() | ||
|
r18749 | self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1]) | ||
|
r17528 | self.assertEqual(rjson['path'], path) | ||
self.assertEqual(rjson['type'], type) | ||||
|
r19604 | isright = self.isdir if type == 'directory' else self.isfile | ||
assert isright(path) | ||||
|
r13083 | |||
def test_create_untitled(self): | ||||
|
r17524 | resp = self.api.create_untitled(path=u'Ã¥ b') | ||
|
r18813 | self._check_created(resp, u'Ã¥ b/Untitled.ipynb') | ||
|
r13083 | |||
# Second time | ||||
|
r17524 | resp = self.api.create_untitled(path=u'Ã¥ b') | ||
|
r18749 | self._check_created(resp, u'Ã¥ b/Untitled1.ipynb') | ||
|
r13083 | |||
# And two directories down | ||||
|
r17524 | resp = self.api.create_untitled(path='foo/bar') | ||
|
r18813 | self._check_created(resp, 'foo/bar/Untitled.ipynb') | ||
|
r13083 | |||
|
r17527 | def test_create_untitled_txt(self): | ||
resp = self.api.create_untitled(path='foo/bar', ext='.txt') | ||||
|
r18813 | self._check_created(resp, 'foo/bar/untitled.txt', type='file') | ||
|
r17527 | |||
|
r18813 | resp = self.api.read(path='foo/bar/untitled.txt') | ||
|
r17527 | model = resp.json() | ||
self.assertEqual(model['type'], 'file') | ||||
self.assertEqual(model['format'], 'text') | ||||
self.assertEqual(model['content'], '') | ||||
|
r13130 | def test_upload(self): | ||
|
r18584 | nb = new_notebook() | ||
|
r17527 | nbmodel = {'content': nb, 'type': 'notebook'} | ||
|
r18749 | path = u'å b/Upload tést.ipynb' | ||
resp = self.api.upload(path, body=json.dumps(nbmodel)) | ||||
self._check_created(resp, path) | ||||
|
r17528 | |||
|
r18758 | def test_mkdir_untitled(self): | ||
resp = self.api.mkdir_untitled(path=u'Ã¥ b') | ||||
|
r18813 | self._check_created(resp, u'Ã¥ b/Untitled Folder', type='directory') | ||
|
r18758 | |||
# Second time | ||||
resp = self.api.mkdir_untitled(path=u'Ã¥ b') | ||||
|
r18813 | self._check_created(resp, u'Ã¥ b/Untitled Folder 1', type='directory') | ||
|
r18758 | |||
# And two directories down | ||||
resp = self.api.mkdir_untitled(path='foo/bar') | ||||
|
r18813 | self._check_created(resp, 'foo/bar/Untitled Folder', type='directory') | ||
|
r18758 | |||
|
r17528 | def test_mkdir(self): | ||
|
r18749 | path = u'å b/New ∂ir' | ||
resp = self.api.mkdir(path) | ||||
self._check_created(resp, path, type='directory') | ||||
|
r13130 | |||
|
r17537 | def test_mkdir_hidden_400(self): | ||
with assert_http_error(400): | ||||
|
r18749 | resp = self.api.mkdir(u'Ã¥ b/.hidden') | ||
|
r17537 | |||
|
r17527 | def test_upload_txt(self): | ||
body = u'ünicode téxt' | ||||
model = { | ||||
'content' : body, | ||||
'format' : 'text', | ||||
'type' : 'file', | ||||
} | ||||
|
r18749 | path = u'å b/Upload tést.txt' | ||
resp = self.api.upload(path, body=json.dumps(model)) | ||||
|
r17527 | |||
# check roundtrip | ||||
|
r18749 | resp = self.api.read(path) | ||
|
r17527 | model = resp.json() | ||
self.assertEqual(model['type'], 'file') | ||||
self.assertEqual(model['format'], 'text') | ||||
self.assertEqual(model['content'], body) | ||||
def test_upload_b64(self): | ||||
body = b'\xFFblob' | ||||
b64body = base64.encodestring(body).decode('ascii') | ||||
model = { | ||||
'content' : b64body, | ||||
'format' : 'base64', | ||||
'type' : 'file', | ||||
} | ||||
|
r18749 | path = u'å b/Upload tést.blob' | ||
resp = self.api.upload(path, body=json.dumps(model)) | ||||
|
r17527 | |||
# check roundtrip | ||||
|
r18749 | resp = self.api.read(path) | ||
|
r17527 | model = resp.json() | ||
self.assertEqual(model['type'], 'file') | ||||
|
r18749 | self.assertEqual(model['path'], path) | ||
|
r17527 | self.assertEqual(model['format'], 'base64') | ||
decoded = base64.decodestring(model['content'].encode('ascii')) | ||||
self.assertEqual(decoded, body) | ||||
|
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")')) | ||||
|
r17527 | nbmodel = {'content': nb, 'type': 'notebook'} | ||
|
r18749 | path = u'å b/Upload tést.ipynb' | ||
resp = self.api.upload(path, body=json.dumps(nbmodel)) | ||||
self._check_created(resp, path) | ||||
resp = self.api.read(path) | ||||
|
r13141 | data = resp.json() | ||
|
r18607 | self.assertEqual(data['content']['nbformat'], 4) | ||
|
r13141 | |||
|
r18750 | def test_copy(self): | ||
|
r18813 | resp = self.api.copy(u'å b/ç d.ipynb', u'å b') | ||
self._check_created(resp, u'å b/ç d-Copy1.ipynb') | ||||
|
r18749 | |||
|
r18750 | resp = self.api.copy(u'å b/ç d.ipynb', u'å b') | ||
|
r18813 | self._check_created(resp, u'å b/ç d-Copy2.ipynb') | ||
def test_copy_copy(self): | ||||
resp = self.api.copy(u'å b/ç d.ipynb', u'å b') | ||||
self._check_created(resp, u'å b/ç d-Copy1.ipynb') | ||||
resp = self.api.copy(u'å b/ç d-Copy1.ipynb', u'å b') | ||||
self._check_created(resp, u'å b/ç d-Copy2.ipynb') | ||||
|
r17535 | def test_copy_path(self): | ||
|
r18750 | resp = self.api.copy(u'foo/a.ipynb', u'Ã¥ b') | ||
|
r18813 | self._check_created(resp, u'Ã¥ b/a.ipynb') | ||
resp = self.api.copy(u'foo/a.ipynb', u'Ã¥ b') | ||||
self._check_created(resp, u'Ã¥ b/a-Copy1.ipynb') | ||||
|
r18750 | |||
def test_copy_put_400(self): | ||||
with assert_http_error(400): | ||||
resp = self.api.copy_put(u'å b/ç d.ipynb', u'å b/cøpy.ipynb') | ||||
|
r17535 | |||
|
r17530 | def test_copy_dir_400(self): | ||
# can't copy directories | ||||
with assert_http_error(400): | ||||
|
r18750 | resp = self.api.copy(u'Ã¥ b', u'foo') | ||
|
r17530 | |||
|
r13083 | def test_delete(self): | ||
for d, name in self.dirs_nbs: | ||||
|
r18749 | print('%r, %r' % (d, name)) | ||
resp = self.api.delete(url_path_join(d, name + '.ipynb')) | ||||
|
r13083 | self.assertEqual(resp.status_code, 204) | ||
for d in self.dirs + ['/']: | ||||
|
r17524 | nbs = notebooks_only(self.api.list(d).json()) | ||
|
r18749 | print('------') | ||
print(d) | ||||
print(nbs) | ||||
self.assertEqual(nbs, []) | ||||
|
r13083 | |||
|
r17530 | def test_delete_dirs(self): | ||
# depth-first delete everything, so we don't try to delete empty directories | ||||
for name in sorted(self.dirs + ['/'], key=len, reverse=True): | ||||
listing = self.api.list(name).json()['content'] | ||||
for model in listing: | ||||
|
r18749 | self.api.delete(model['path']) | ||
|
r17530 | listing = self.api.list('/').json()['content'] | ||
self.assertEqual(listing, []) | ||||
def test_delete_non_empty_dir(self): | ||||
"""delete non-empty dir raises 400""" | ||||
with assert_http_error(400): | ||||
self.api.delete(u'Ã¥ b') | ||||
|
r13083 | def test_rename(self): | ||
|
r18749 | resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb') | ||
|
r13089 | self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb') | ||
|
r13083 | self.assertEqual(resp.json()['name'], 'z.ipynb') | ||
|
r18749 | self.assertEqual(resp.json()['path'], 'foo/z.ipynb') | ||
|
r19604 | assert self.isfile('foo/z.ipynb') | ||
|
r13083 | |||
|
r17524 | nbs = notebooks_only(self.api.list('foo').json()) | ||
|
r13083 | nbnames = set(n['name'] for n in nbs) | ||
self.assertIn('z.ipynb', nbnames) | ||||
self.assertNotIn('a.ipynb', nbnames) | ||||
|
r13085 | |||
|
r20878 | def test_checkpoints_follow_file(self): | ||
# Read initial file state | ||||
orig = self.api.read('foo/a.ipynb') | ||||
# Create a checkpoint of initial state | ||||
r = self.api.new_checkpoint('foo/a.ipynb') | ||||
cp1 = r.json() | ||||
# Modify file and save | ||||
nbcontent = json.loads(orig.text)['content'] | ||||
nb = from_dict(nbcontent) | ||||
hcell = new_markdown_cell('Created by test') | ||||
nb.cells.append(hcell) | ||||
nbmodel = {'content': nb, 'type': 'notebook'} | ||||
self.api.save('foo/a.ipynb', body=json.dumps(nbmodel)) | ||||
# Rename the file. | ||||
self.api.rename('foo/a.ipynb', 'foo/z.ipynb') | ||||
# Looking for checkpoints in the old location should yield no results. | ||||
self.assertEqual(self.api.get_checkpoints('foo/a.ipynb').json(), []) | ||||
# Looking for checkpoints in the new location should work. | ||||
cps = self.api.get_checkpoints('foo/z.ipynb').json() | ||||
self.assertEqual(cps, [cp1]) | ||||
# Delete the file. The checkpoint should be deleted as well. | ||||
self.api.delete('foo/z.ipynb') | ||||
cps = self.api.get_checkpoints('foo/z.ipynb').json() | ||||
self.assertEqual(cps, []) | ||||
|
r13710 | def test_rename_existing(self): | ||
with assert_http_error(409): | ||||
|
r18749 | self.api.rename('foo/a.ipynb', 'foo/b.ipynb') | ||
|
r13710 | |||
|
r13085 | def test_save(self): | ||
|
r18749 | resp = self.api.read('foo/a.ipynb') | ||
|
r13138 | nbcontent = json.loads(resp.text)['content'] | ||
|
r18601 | nb = from_dict(nbcontent) | ||
|
r18596 | nb.cells.append(new_markdown_cell(u'Created by test ³')) | ||
|
r13085 | |||
|
r20878 | nbmodel = {'content': nb, 'type': 'notebook'} | ||
|
r18749 | resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel)) | ||
|
r13085 | |||
|
r18749 | nbcontent = self.api.read('foo/a.ipynb').json()['content'] | ||
|
r18601 | newnb = from_dict(nbcontent) | ||
|
r18584 | self.assertEqual(newnb.cells[0].source, | ||
|
r13111 | u'Created by test ³') | ||
|
r13087 | |||
|
r13109 | def test_checkpoints(self): | ||
|
r18749 | resp = self.api.read('foo/a.ipynb') | ||
r = self.api.new_checkpoint('foo/a.ipynb') | ||||
|
r13109 | self.assertEqual(r.status_code, 201) | ||
cp1 = r.json() | ||||
|
r13122 | self.assertEqual(set(cp1), {'id', 'last_modified'}) | ||
self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id']) | ||||
|
r13109 | |||
# Modify it | ||||
|
r13138 | nbcontent = json.loads(resp.text)['content'] | ||
|
r18601 | nb = from_dict(nbcontent) | ||
|
r18596 | hcell = new_markdown_cell('Created by test') | ||
|
r18584 | nb.cells.append(hcell) | ||
|
r13109 | # Save | ||
|
r18749 | nbmodel= {'content': nb, 'type': 'notebook'} | ||
resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel)) | ||||
|
r13109 | |||
# List checkpoints | ||||
|
r18749 | cps = self.api.get_checkpoints('foo/a.ipynb').json() | ||
|
r13109 | self.assertEqual(cps, [cp1]) | ||
|
r18749 | nbcontent = self.api.read('foo/a.ipynb').json()['content'] | ||
|
r18601 | nb = from_dict(nbcontent) | ||
|
r18584 | self.assertEqual(nb.cells[0].source, 'Created by test') | ||
|
r13109 | |||
# Restore cp1 | ||||
|
r18749 | r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id']) | ||
|
r13109 | self.assertEqual(r.status_code, 204) | ||
|
r18749 | nbcontent = self.api.read('foo/a.ipynb').json()['content'] | ||
|
r18601 | nb = from_dict(nbcontent) | ||
|
r18584 | self.assertEqual(nb.cells, []) | ||
|
r13109 | |||
# Delete cp1 | ||||
|
r18749 | r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id']) | ||
|
r13109 | self.assertEqual(r.status_code, 204) | ||
|
r18749 | cps = self.api.get_checkpoints('foo/a.ipynb').json() | ||
|
r13109 | self.assertEqual(cps, []) | ||
|
r19748 | |||
|
r19786 | def test_file_checkpoints(self): | ||
""" | ||||
Test checkpointing of non-notebook files. | ||||
""" | ||||
filename = 'foo/a.txt' | ||||
resp = self.api.read(filename) | ||||
orig_content = json.loads(resp.text)['content'] | ||||
# Create a checkpoint. | ||||
r = self.api.new_checkpoint(filename) | ||||
self.assertEqual(r.status_code, 201) | ||||
cp1 = r.json() | ||||
self.assertEqual(set(cp1), {'id', 'last_modified'}) | ||||
self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id']) | ||||
# Modify the file and save. | ||||
new_content = orig_content + '\nsecond line' | ||||
model = { | ||||
'content': new_content, | ||||
'type': 'file', | ||||
'format': 'text', | ||||
} | ||||
resp = self.api.save(filename, body=json.dumps(model)) | ||||
# List checkpoints | ||||
cps = self.api.get_checkpoints(filename).json() | ||||
self.assertEqual(cps, [cp1]) | ||||
content = self.api.read(filename).json()['content'] | ||||
self.assertEqual(content, new_content) | ||||
# Restore cp1 | ||||
r = self.api.restore_checkpoint(filename, cp1['id']) | ||||
self.assertEqual(r.status_code, 204) | ||||
restored_content = self.api.read(filename).json()['content'] | ||||
self.assertEqual(restored_content, orig_content) | ||||
# Delete cp1 | ||||
r = self.api.delete_checkpoint(filename, cp1['id']) | ||||
self.assertEqual(r.status_code, 204) | ||||
cps = self.api.get_checkpoints(filename).json() | ||||
self.assertEqual(cps, []) | ||||
|
r19748 | @contextmanager | ||
def patch_cp_root(self, dirname): | ||||
""" | ||||
Temporarily patch the root dir of our checkpoint manager. | ||||
""" | ||||
|
r19839 | cpm = self.notebook.contents_manager.checkpoints | ||
|
r19748 | old_dirname = cpm.root_dir | ||
cpm.root_dir = dirname | ||||
try: | ||||
yield | ||||
finally: | ||||
cpm.root_dir = old_dirname | ||||
def test_checkpoints_separate_root(self): | ||||
""" | ||||
|
r19839 | Test that FileCheckpoints functions correctly even when it's | ||
|
r19748 | using a different root dir from FileContentsManager. This also keeps | ||
the implementation honest for use with ContentsManagers that don't map | ||||
models to the filesystem | ||||
|
r19786 | Override this method to a no-op when testing other managers. | ||
""" | ||||
|
r19748 | with TemporaryDirectory() as td: | ||
with self.patch_cp_root(td): | ||||
self.test_checkpoints() | ||||
|
r19786 | |||
with TemporaryDirectory() as td: | ||||
with self.patch_cp_root(td): | ||||
self.test_file_checkpoints() | ||||
|
r19828 | |||
|
r19838 | class GenericFileCheckpointsAPITest(APITest): | ||
""" | ||||
|
r19839 | Run the tests from APITest with GenericFileCheckpoints. | ||
|
r19838 | """ | ||
config = Config() | ||||
|
r19839 | config.FileContentsManager.checkpoints_class = GenericFileCheckpoints | ||
def test_config_did_something(self): | ||||
self.assertIsInstance( | ||||
self.notebook.contents_manager.checkpoints, | ||||
GenericFileCheckpoints, | ||||
) | ||||