##// END OF EJS Templates
remove copy via PUT...
Min RK -
Show More
@@ -1,273 +1,262 b''
1 """Tornado handlers for the contents web service."""
1 """Tornado handlers for the contents web service."""
2
2
3 # Copyright (c) IPython Development Team.
3 # Copyright (c) IPython Development Team.
4 # Distributed under the terms of the Modified BSD License.
4 # Distributed under the terms of the Modified BSD License.
5
5
6 import json
6 import json
7
7
8 from tornado import web
8 from tornado import web
9
9
10 from IPython.html.utils import url_path_join, url_escape
10 from IPython.html.utils import url_path_join, url_escape
11 from IPython.utils.jsonutil import date_default
11 from IPython.utils.jsonutil import date_default
12
12
13 from IPython.html.base.handlers import (
13 from IPython.html.base.handlers import (
14 IPythonHandler, json_errors, path_regex,
14 IPythonHandler, json_errors, path_regex,
15 )
15 )
16
16
17
17
18 def sort_key(model):
18 def sort_key(model):
19 """key function for case-insensitive sort by name and type"""
19 """key function for case-insensitive sort by name and type"""
20 iname = model['name'].lower()
20 iname = model['name'].lower()
21 type_key = {
21 type_key = {
22 'directory' : '0',
22 'directory' : '0',
23 'notebook' : '1',
23 'notebook' : '1',
24 'file' : '2',
24 'file' : '2',
25 }.get(model['type'], '9')
25 }.get(model['type'], '9')
26 return u'%s%s' % (type_key, iname)
26 return u'%s%s' % (type_key, iname)
27
27
28 class ContentsHandler(IPythonHandler):
28 class ContentsHandler(IPythonHandler):
29
29
30 SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
30 SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
31
31
32 def location_url(self, path):
32 def location_url(self, path):
33 """Return the full URL location of a file.
33 """Return the full URL location of a file.
34
34
35 Parameters
35 Parameters
36 ----------
36 ----------
37 path : unicode
37 path : unicode
38 The API path of the file, such as "foo/bar.txt".
38 The API path of the file, such as "foo/bar.txt".
39 """
39 """
40 return url_escape(url_path_join(
40 return url_escape(url_path_join(
41 self.base_url, 'api', 'contents', path
41 self.base_url, 'api', 'contents', path
42 ))
42 ))
43
43
44 def _finish_model(self, model, location=True):
44 def _finish_model(self, model, location=True):
45 """Finish a JSON request with a model, setting relevant headers, etc."""
45 """Finish a JSON request with a model, setting relevant headers, etc."""
46 if location:
46 if location:
47 location = self.location_url(model['path'])
47 location = self.location_url(model['path'])
48 self.set_header('Location', location)
48 self.set_header('Location', location)
49 self.set_header('Last-Modified', model['last_modified'])
49 self.set_header('Last-Modified', model['last_modified'])
50 self.finish(json.dumps(model, default=date_default))
50 self.finish(json.dumps(model, default=date_default))
51
51
52 @web.authenticated
52 @web.authenticated
53 @json_errors
53 @json_errors
54 def get(self, path=''):
54 def get(self, path=''):
55 """Return a model for a file or directory.
55 """Return a model for a file or directory.
56
56
57 A directory model contains a list of models (without content)
57 A directory model contains a list of models (without content)
58 of the files and directories it contains.
58 of the files and directories it contains.
59 """
59 """
60 path = path or ''
60 path = path or ''
61 model = self.contents_manager.get_model(path=path)
61 model = self.contents_manager.get_model(path=path)
62 if model['type'] == 'directory':
62 if model['type'] == 'directory':
63 # group listing by type, then by name (case-insensitive)
63 # group listing by type, then by name (case-insensitive)
64 # FIXME: sorting should be done in the frontends
64 # FIXME: sorting should be done in the frontends
65 model['content'].sort(key=sort_key)
65 model['content'].sort(key=sort_key)
66 self._finish_model(model, location=False)
66 self._finish_model(model, location=False)
67
67
68 @web.authenticated
68 @web.authenticated
69 @json_errors
69 @json_errors
70 def patch(self, path=''):
70 def patch(self, path=''):
71 """PATCH renames a file or directory without re-uploading content."""
71 """PATCH renames a file or directory without re-uploading content."""
72 cm = self.contents_manager
72 cm = self.contents_manager
73 model = self.get_json_body()
73 model = self.get_json_body()
74 if model is None:
74 if model is None:
75 raise web.HTTPError(400, u'JSON body missing')
75 raise web.HTTPError(400, u'JSON body missing')
76 print('before', model)
76 print('before', model)
77 model = cm.update(model, path)
77 model = cm.update(model, path)
78 print('after', model)
78 print('after', model)
79 self._finish_model(model)
79 self._finish_model(model)
80
80
81 def _copy(self, copy_from, copy_to=None):
81 def _copy(self, copy_from, copy_to=None):
82 """Copy a file, optionally specifying the new path.
82 """Copy a file, optionally specifying the new path.
83 """
83 """
84 self.log.info(u"Copying {copy_from} to {copy_to}".format(
84 self.log.info(u"Copying {copy_from} to {copy_to}".format(
85 copy_from=copy_from,
85 copy_from=copy_from,
86 copy_to=copy_to or '',
86 copy_to=copy_to or '',
87 ))
87 ))
88 model = self.contents_manager.copy(copy_from, copy_to)
88 model = self.contents_manager.copy(copy_from, copy_to)
89 self.set_status(201)
89 self.set_status(201)
90 self._finish_model(model)
90 self._finish_model(model)
91
91
92 def _upload(self, model, path):
92 def _upload(self, model, path):
93 """Handle upload of a new file to path"""
93 """Handle upload of a new file to path"""
94 self.log.info(u"Uploading file to %s", path)
94 self.log.info(u"Uploading file to %s", path)
95 model = self.contents_manager.create_file(model, path)
95 model = self.contents_manager.create_file(model, path)
96 self.set_status(201)
96 self.set_status(201)
97 self._finish_model(model)
97 self._finish_model(model)
98
98
99 def _create_empty_file(self, path, ext='.ipynb'):
99 def _create_empty_file(self, path, ext='.ipynb'):
100 """Create an empty file in path
100 """Create an empty file in path
101
101
102 If name specified, create it in path.
102 If name specified, create it in path.
103 """
103 """
104 self.log.info(u"Creating new file in %s", path)
104 self.log.info(u"Creating new file in %s", path)
105 model = self.contents_manager.create_file(path=path, ext=ext)
105 model = self.contents_manager.create_file(path=path, ext=ext)
106 self.set_status(201)
106 self.set_status(201)
107 self._finish_model(model)
107 self._finish_model(model)
108
108
109 def _save(self, model, path):
109 def _save(self, model, path):
110 """Save an existing file."""
110 """Save an existing file."""
111 self.log.info(u"Saving file at %s", path)
111 self.log.info(u"Saving file at %s", path)
112 model = self.contents_manager.save(model, path)
112 model = self.contents_manager.save(model, path)
113 self._finish_model(model)
113 self._finish_model(model)
114
114
115 @web.authenticated
115 @web.authenticated
116 @json_errors
116 @json_errors
117 def post(self, path=''):
117 def post(self, path=''):
118 """Create a new file or directory in the specified path.
118 """Create a new file or directory in the specified path.
119
119
120 POST creates new files or directories. The server always decides on the name.
120 POST creates new files or directories. The server always decides on the name.
121
121
122 POST /api/contents/path
122 POST /api/contents/path
123 New untitled, empty file or directory.
123 New untitled, empty file or directory.
124 POST /api/contents/path
124 POST /api/contents/path
125 with body {"copy_from" : "/path/to/OtherNotebook.ipynb"}
125 with body {"copy_from" : "/path/to/OtherNotebook.ipynb"}
126 New copy of OtherNotebook in path
126 New copy of OtherNotebook in path
127 """
127 """
128
128
129 cm = self.contents_manager
129 cm = self.contents_manager
130
130
131 if cm.file_exists(path):
131 if cm.file_exists(path):
132 raise web.HTTPError(400, "Cannot POST to existing files, use PUT instead.")
132 raise web.HTTPError(400, "Cannot POST to existing files, use PUT instead.")
133
133
134 if not cm.dir_exists(path):
134 if not cm.dir_exists(path):
135 raise web.HTTPError(404, "No such directory: %s" % path)
135 raise web.HTTPError(404, "No such directory: %s" % path)
136
136
137 model = self.get_json_body()
137 model = self.get_json_body()
138
138
139 if model is not None:
139 if model is not None:
140 copy_from = model.get('copy_from')
140 copy_from = model.get('copy_from')
141 ext = model.get('ext', '.ipynb')
141 ext = model.get('ext', '.ipynb')
142 if copy_from:
142 if copy_from:
143 self._copy(copy_from, path)
143 self._copy(copy_from, path)
144 else:
144 else:
145 self._create_empty_file(path, ext=ext)
145 self._create_empty_file(path, ext=ext)
146 else:
146 else:
147 self._create_empty_file(path)
147 self._create_empty_file(path)
148
148
149 @web.authenticated
149 @web.authenticated
150 @json_errors
150 @json_errors
151 def put(self, path=''):
151 def put(self, path=''):
152 """Saves the file in the location specified by name and path.
152 """Saves the file in the location specified by name and path.
153
153
154 PUT is very similar to POST, but the requester specifies the name,
154 PUT is very similar to POST, but the requester specifies the name,
155 whereas with POST, the server picks the name.
155 whereas with POST, the server picks the name.
156
156
157 PUT /api/contents/path/Name.ipynb
157 PUT /api/contents/path/Name.ipynb
158 Save notebook at ``path/Name.ipynb``. Notebook structure is specified
158 Save notebook at ``path/Name.ipynb``. Notebook structure is specified
159 in `content` key of JSON request body. If content is not specified,
159 in `content` key of JSON request body. If content is not specified,
160 create a new empty notebook.
160 create a new empty notebook.
161 PUT /api/contents/path/Name.ipynb
162 with JSON body::
163
164 {
165 "copy_from" : "[path/to/]OtherNotebook.ipynb"
166 }
167
168 Copy OtherNotebook to Name
169 """
161 """
170 model = self.get_json_body()
162 model = self.get_json_body()
171 if model:
163 if model:
172 copy_from = model.get('copy_from')
164 if model.get('copy_from'):
173 if copy_from:
165 raise web.HTTPError(400, "Cannot copy with PUT, only POST")
174 if model.get('content'):
166 if self.contents_manager.file_exists(path):
175 raise web.HTTPError(400, "Can't upload and copy at the same time.")
176 self._copy(copy_from, path)
177 elif self.contents_manager.file_exists(path):
178 self._save(model, path)
167 self._save(model, path)
179 else:
168 else:
180 self._upload(model, path)
169 self._upload(model, path)
181 else:
170 else:
182 self._create_empty_file(path)
171 self._create_empty_file(path)
183
172
184 @web.authenticated
173 @web.authenticated
185 @json_errors
174 @json_errors
186 def delete(self, path=''):
175 def delete(self, path=''):
187 """delete a file in the given path"""
176 """delete a file in the given path"""
188 cm = self.contents_manager
177 cm = self.contents_manager
189 self.log.warn('delete %s', path)
178 self.log.warn('delete %s', path)
190 cm.delete(path)
179 cm.delete(path)
191 self.set_status(204)
180 self.set_status(204)
192 self.finish()
181 self.finish()
193
182
194
183
195 class CheckpointsHandler(IPythonHandler):
184 class CheckpointsHandler(IPythonHandler):
196
185
197 SUPPORTED_METHODS = ('GET', 'POST')
186 SUPPORTED_METHODS = ('GET', 'POST')
198
187
199 @web.authenticated
188 @web.authenticated
200 @json_errors
189 @json_errors
201 def get(self, path=''):
190 def get(self, path=''):
202 """get lists checkpoints for a file"""
191 """get lists checkpoints for a file"""
203 cm = self.contents_manager
192 cm = self.contents_manager
204 checkpoints = cm.list_checkpoints(path)
193 checkpoints = cm.list_checkpoints(path)
205 data = json.dumps(checkpoints, default=date_default)
194 data = json.dumps(checkpoints, default=date_default)
206 self.finish(data)
195 self.finish(data)
207
196
208 @web.authenticated
197 @web.authenticated
209 @json_errors
198 @json_errors
210 def post(self, path=''):
199 def post(self, path=''):
211 """post creates a new checkpoint"""
200 """post creates a new checkpoint"""
212 cm = self.contents_manager
201 cm = self.contents_manager
213 checkpoint = cm.create_checkpoint(path)
202 checkpoint = cm.create_checkpoint(path)
214 data = json.dumps(checkpoint, default=date_default)
203 data = json.dumps(checkpoint, default=date_default)
215 location = url_path_join(self.base_url, 'api/contents',
204 location = url_path_join(self.base_url, 'api/contents',
216 path, 'checkpoints', checkpoint['id'])
205 path, 'checkpoints', checkpoint['id'])
217 self.set_header('Location', url_escape(location))
206 self.set_header('Location', url_escape(location))
218 self.set_status(201)
207 self.set_status(201)
219 self.finish(data)
208 self.finish(data)
220
209
221
210
222 class ModifyCheckpointsHandler(IPythonHandler):
211 class ModifyCheckpointsHandler(IPythonHandler):
223
212
224 SUPPORTED_METHODS = ('POST', 'DELETE')
213 SUPPORTED_METHODS = ('POST', 'DELETE')
225
214
226 @web.authenticated
215 @web.authenticated
227 @json_errors
216 @json_errors
228 def post(self, path, checkpoint_id):
217 def post(self, path, checkpoint_id):
229 """post restores a file from a checkpoint"""
218 """post restores a file from a checkpoint"""
230 cm = self.contents_manager
219 cm = self.contents_manager
231 cm.restore_checkpoint(checkpoint_id, path)
220 cm.restore_checkpoint(checkpoint_id, path)
232 self.set_status(204)
221 self.set_status(204)
233 self.finish()
222 self.finish()
234
223
235 @web.authenticated
224 @web.authenticated
236 @json_errors
225 @json_errors
237 def delete(self, path, checkpoint_id):
226 def delete(self, path, checkpoint_id):
238 """delete clears a checkpoint for a given file"""
227 """delete clears a checkpoint for a given file"""
239 cm = self.contents_manager
228 cm = self.contents_manager
240 cm.delete_checkpoint(checkpoint_id, path)
229 cm.delete_checkpoint(checkpoint_id, path)
241 self.set_status(204)
230 self.set_status(204)
242 self.finish()
231 self.finish()
243
232
244
233
245 class NotebooksRedirectHandler(IPythonHandler):
234 class NotebooksRedirectHandler(IPythonHandler):
246 """Redirect /api/notebooks to /api/contents"""
235 """Redirect /api/notebooks to /api/contents"""
247 SUPPORTED_METHODS = ('GET', 'PUT', 'PATCH', 'POST', 'DELETE')
236 SUPPORTED_METHODS = ('GET', 'PUT', 'PATCH', 'POST', 'DELETE')
248
237
249 def get(self, path):
238 def get(self, path):
250 self.log.warn("/api/notebooks is deprecated, use /api/contents")
239 self.log.warn("/api/notebooks is deprecated, use /api/contents")
251 self.redirect(url_path_join(
240 self.redirect(url_path_join(
252 self.base_url,
241 self.base_url,
253 'api/contents',
242 'api/contents',
254 path
243 path
255 ))
244 ))
256
245
257 put = patch = post = delete = get
246 put = patch = post = delete = get
258
247
259
248
260 #-----------------------------------------------------------------------------
249 #-----------------------------------------------------------------------------
261 # URL to handler mappings
250 # URL to handler mappings
262 #-----------------------------------------------------------------------------
251 #-----------------------------------------------------------------------------
263
252
264
253
265 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
254 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
266
255
267 default_handlers = [
256 default_handlers = [
268 (r"/api/contents%s/checkpoints" % path_regex, CheckpointsHandler),
257 (r"/api/contents%s/checkpoints" % path_regex, CheckpointsHandler),
269 (r"/api/contents%s/checkpoints/%s" % (path_regex, _checkpoint_id_regex),
258 (r"/api/contents%s/checkpoints/%s" % (path_regex, _checkpoint_id_regex),
270 ModifyCheckpointsHandler),
259 ModifyCheckpointsHandler),
271 (r"/api/contents%s" % path_regex, ContentsHandler),
260 (r"/api/contents%s" % path_regex, ContentsHandler),
272 (r"/api/notebooks/?(.*)", NotebooksRedirectHandler),
261 (r"/api/notebooks/?(.*)", NotebooksRedirectHandler),
273 ]
262 ]
@@ -1,474 +1,474 b''
1 # coding: utf-8
1 # coding: utf-8
2 """Test the contents webservice API."""
2 """Test the contents webservice API."""
3
3
4 import base64
4 import base64
5 import io
5 import io
6 import json
6 import json
7 import os
7 import os
8 import shutil
8 import shutil
9 from unicodedata import normalize
9 from unicodedata import normalize
10
10
11 pjoin = os.path.join
11 pjoin = os.path.join
12
12
13 import requests
13 import requests
14
14
15 from IPython.html.utils import url_path_join, url_escape
15 from IPython.html.utils import url_path_join, url_escape
16 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
16 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
17 from IPython.nbformat import read, write, from_dict
17 from IPython.nbformat import read, write, from_dict
18 from IPython.nbformat.v4 import (
18 from IPython.nbformat.v4 import (
19 new_notebook, new_markdown_cell,
19 new_notebook, new_markdown_cell,
20 )
20 )
21 from IPython.nbformat import v2
21 from IPython.nbformat import v2
22 from IPython.utils import py3compat
22 from IPython.utils import py3compat
23 from IPython.utils.data import uniq_stable
23 from IPython.utils.data import uniq_stable
24
24
25
25
26 def notebooks_only(dir_model):
26 def notebooks_only(dir_model):
27 return [nb for nb in dir_model['content'] if nb['type']=='notebook']
27 return [nb for nb in dir_model['content'] if nb['type']=='notebook']
28
28
29 def dirs_only(dir_model):
29 def dirs_only(dir_model):
30 return [x for x in dir_model['content'] if x['type']=='directory']
30 return [x for x in dir_model['content'] if x['type']=='directory']
31
31
32
32
33 class API(object):
33 class API(object):
34 """Wrapper for contents API calls."""
34 """Wrapper for contents API calls."""
35 def __init__(self, base_url):
35 def __init__(self, base_url):
36 self.base_url = base_url
36 self.base_url = base_url
37
37
38 def _req(self, verb, path, body=None):
38 def _req(self, verb, path, body=None):
39 response = requests.request(verb,
39 response = requests.request(verb,
40 url_path_join(self.base_url, 'api/contents', path),
40 url_path_join(self.base_url, 'api/contents', path),
41 data=body,
41 data=body,
42 )
42 )
43 response.raise_for_status()
43 response.raise_for_status()
44 return response
44 return response
45
45
46 def list(self, path='/'):
46 def list(self, path='/'):
47 return self._req('GET', path)
47 return self._req('GET', path)
48
48
49 def read(self, path):
49 def read(self, path):
50 return self._req('GET', path)
50 return self._req('GET', path)
51
51
52 def create_untitled(self, path='/', ext=None):
52 def create_untitled(self, path='/', ext=None):
53 body = None
53 body = None
54 if ext:
54 if ext:
55 body = json.dumps({'ext': ext})
55 body = json.dumps({'ext': ext})
56 return self._req('POST', path, body)
56 return self._req('POST', path, body)
57
57
58 def copy_untitled(self, copy_from, path='/'):
58 def copy(self, copy_from, path='/'):
59 body = json.dumps({'copy_from':copy_from})
59 body = json.dumps({'copy_from':copy_from})
60 return self._req('POST', path, body)
60 return self._req('POST', path, body)
61
61
62 def create(self, path='/'):
62 def create(self, path='/'):
63 return self._req('PUT', path)
63 return self._req('PUT', path)
64
64
65 def upload(self, path, body):
65 def upload(self, path, body):
66 return self._req('PUT', path, body)
66 return self._req('PUT', path, body)
67
67
68 def mkdir(self, path='/'):
68 def mkdir(self, path='/'):
69 return self._req('PUT', path, json.dumps({'type': 'directory'}))
69 return self._req('PUT', path, json.dumps({'type': 'directory'}))
70
70
71 def copy(self, copy_from, path):
71 def copy_put(self, copy_from, path='/'):
72 body = json.dumps({'copy_from':copy_from})
72 body = json.dumps({'copy_from':copy_from})
73 return self._req('PUT', path, body)
73 return self._req('PUT', path, body)
74
74
75 def save(self, path, body):
75 def save(self, path, body):
76 return self._req('PUT', path, body)
76 return self._req('PUT', path, body)
77
77
78 def delete(self, path='/'):
78 def delete(self, path='/'):
79 return self._req('DELETE', path)
79 return self._req('DELETE', path)
80
80
81 def rename(self, path, new_path):
81 def rename(self, path, new_path):
82 body = json.dumps({'path': new_path})
82 body = json.dumps({'path': new_path})
83 return self._req('PATCH', path, body)
83 return self._req('PATCH', path, body)
84
84
85 def get_checkpoints(self, path):
85 def get_checkpoints(self, path):
86 return self._req('GET', url_path_join(path, 'checkpoints'))
86 return self._req('GET', url_path_join(path, 'checkpoints'))
87
87
88 def new_checkpoint(self, path):
88 def new_checkpoint(self, path):
89 return self._req('POST', url_path_join(path, 'checkpoints'))
89 return self._req('POST', url_path_join(path, 'checkpoints'))
90
90
91 def restore_checkpoint(self, path, checkpoint_id):
91 def restore_checkpoint(self, path, checkpoint_id):
92 return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id))
92 return self._req('POST', url_path_join(path, 'checkpoints', checkpoint_id))
93
93
94 def delete_checkpoint(self, path, checkpoint_id):
94 def delete_checkpoint(self, path, checkpoint_id):
95 return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id))
95 return self._req('DELETE', url_path_join(path, 'checkpoints', checkpoint_id))
96
96
97 class APITest(NotebookTestBase):
97 class APITest(NotebookTestBase):
98 """Test the kernels web service API"""
98 """Test the kernels web service API"""
99 dirs_nbs = [('', 'inroot'),
99 dirs_nbs = [('', 'inroot'),
100 ('Directory with spaces in', 'inspace'),
100 ('Directory with spaces in', 'inspace'),
101 (u'unicodΓ©', 'innonascii'),
101 (u'unicodΓ©', 'innonascii'),
102 ('foo', 'a'),
102 ('foo', 'a'),
103 ('foo', 'b'),
103 ('foo', 'b'),
104 ('foo', 'name with spaces'),
104 ('foo', 'name with spaces'),
105 ('foo', u'unicodΓ©'),
105 ('foo', u'unicodΓ©'),
106 ('foo/bar', 'baz'),
106 ('foo/bar', 'baz'),
107 ('ordering', 'A'),
107 ('ordering', 'A'),
108 ('ordering', 'b'),
108 ('ordering', 'b'),
109 ('ordering', 'C'),
109 ('ordering', 'C'),
110 (u'Γ₯ b', u'Γ§ d'),
110 (u'Γ₯ b', u'Γ§ d'),
111 ]
111 ]
112 hidden_dirs = ['.hidden', '__pycache__']
112 hidden_dirs = ['.hidden', '__pycache__']
113
113
114 dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
114 dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
115 del dirs[0] # remove ''
115 del dirs[0] # remove ''
116 top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
116 top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
117
117
118 @staticmethod
118 @staticmethod
119 def _blob_for_name(name):
119 def _blob_for_name(name):
120 return name.encode('utf-8') + b'\xFF'
120 return name.encode('utf-8') + b'\xFF'
121
121
122 @staticmethod
122 @staticmethod
123 def _txt_for_name(name):
123 def _txt_for_name(name):
124 return u'%s text file' % name
124 return u'%s text file' % name
125
125
126 def setUp(self):
126 def setUp(self):
127 nbdir = self.notebook_dir.name
127 nbdir = self.notebook_dir.name
128 self.blob = os.urandom(100)
128 self.blob = os.urandom(100)
129 self.b64_blob = base64.encodestring(self.blob).decode('ascii')
129 self.b64_blob = base64.encodestring(self.blob).decode('ascii')
130
130
131 for d in (self.dirs + self.hidden_dirs):
131 for d in (self.dirs + self.hidden_dirs):
132 d.replace('/', os.sep)
132 d.replace('/', os.sep)
133 if not os.path.isdir(pjoin(nbdir, d)):
133 if not os.path.isdir(pjoin(nbdir, d)):
134 os.mkdir(pjoin(nbdir, d))
134 os.mkdir(pjoin(nbdir, d))
135
135
136 for d, name in self.dirs_nbs:
136 for d, name in self.dirs_nbs:
137 d = d.replace('/', os.sep)
137 d = d.replace('/', os.sep)
138 # create a notebook
138 # create a notebook
139 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
139 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
140 encoding='utf-8') as f:
140 encoding='utf-8') as f:
141 nb = new_notebook()
141 nb = new_notebook()
142 write(nb, f, version=4)
142 write(nb, f, version=4)
143
143
144 # create a text file
144 # create a text file
145 with io.open(pjoin(nbdir, d, '%s.txt' % name), 'w',
145 with io.open(pjoin(nbdir, d, '%s.txt' % name), 'w',
146 encoding='utf-8') as f:
146 encoding='utf-8') as f:
147 f.write(self._txt_for_name(name))
147 f.write(self._txt_for_name(name))
148
148
149 # create a binary file
149 # create a binary file
150 with io.open(pjoin(nbdir, d, '%s.blob' % name), 'wb') as f:
150 with io.open(pjoin(nbdir, d, '%s.blob' % name), 'wb') as f:
151 f.write(self._blob_for_name(name))
151 f.write(self._blob_for_name(name))
152
152
153 self.api = API(self.base_url())
153 self.api = API(self.base_url())
154
154
155 def tearDown(self):
155 def tearDown(self):
156 nbdir = self.notebook_dir.name
156 nbdir = self.notebook_dir.name
157
157
158 for dname in (list(self.top_level_dirs) + self.hidden_dirs):
158 for dname in (list(self.top_level_dirs) + self.hidden_dirs):
159 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
159 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
160
160
161 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
161 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
162 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
162 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
163
163
164 def test_list_notebooks(self):
164 def test_list_notebooks(self):
165 nbs = notebooks_only(self.api.list().json())
165 nbs = notebooks_only(self.api.list().json())
166 self.assertEqual(len(nbs), 1)
166 self.assertEqual(len(nbs), 1)
167 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
167 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
168
168
169 nbs = notebooks_only(self.api.list('/Directory with spaces in/').json())
169 nbs = notebooks_only(self.api.list('/Directory with spaces in/').json())
170 self.assertEqual(len(nbs), 1)
170 self.assertEqual(len(nbs), 1)
171 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
171 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
172
172
173 nbs = notebooks_only(self.api.list(u'/unicodΓ©/').json())
173 nbs = notebooks_only(self.api.list(u'/unicodΓ©/').json())
174 self.assertEqual(len(nbs), 1)
174 self.assertEqual(len(nbs), 1)
175 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
175 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
176 self.assertEqual(nbs[0]['path'], u'unicodΓ©/innonascii.ipynb')
176 self.assertEqual(nbs[0]['path'], u'unicodΓ©/innonascii.ipynb')
177
177
178 nbs = notebooks_only(self.api.list('/foo/bar/').json())
178 nbs = notebooks_only(self.api.list('/foo/bar/').json())
179 self.assertEqual(len(nbs), 1)
179 self.assertEqual(len(nbs), 1)
180 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
180 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
181 self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb')
181 self.assertEqual(nbs[0]['path'], 'foo/bar/baz.ipynb')
182
182
183 nbs = notebooks_only(self.api.list('foo').json())
183 nbs = notebooks_only(self.api.list('foo').json())
184 self.assertEqual(len(nbs), 4)
184 self.assertEqual(len(nbs), 4)
185 nbnames = { normalize('NFC', n['name']) for n in nbs }
185 nbnames = { normalize('NFC', n['name']) for n in nbs }
186 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
186 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
187 expected = { normalize('NFC', name) for name in expected }
187 expected = { normalize('NFC', name) for name in expected }
188 self.assertEqual(nbnames, expected)
188 self.assertEqual(nbnames, expected)
189
189
190 nbs = notebooks_only(self.api.list('ordering').json())
190 nbs = notebooks_only(self.api.list('ordering').json())
191 nbnames = [n['name'] for n in nbs]
191 nbnames = [n['name'] for n in nbs]
192 expected = ['A.ipynb', 'b.ipynb', 'C.ipynb']
192 expected = ['A.ipynb', 'b.ipynb', 'C.ipynb']
193 self.assertEqual(nbnames, expected)
193 self.assertEqual(nbnames, expected)
194
194
195 def test_list_dirs(self):
195 def test_list_dirs(self):
196 dirs = dirs_only(self.api.list().json())
196 dirs = dirs_only(self.api.list().json())
197 dir_names = {normalize('NFC', d['name']) for d in dirs}
197 dir_names = {normalize('NFC', d['name']) for d in dirs}
198 self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
198 self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
199
199
200 def test_list_nonexistant_dir(self):
200 def test_list_nonexistant_dir(self):
201 with assert_http_error(404):
201 with assert_http_error(404):
202 self.api.list('nonexistant')
202 self.api.list('nonexistant')
203
203
204 def test_get_nb_contents(self):
204 def test_get_nb_contents(self):
205 for d, name in self.dirs_nbs:
205 for d, name in self.dirs_nbs:
206 path = url_path_join(d, name + '.ipynb')
206 path = url_path_join(d, name + '.ipynb')
207 nb = self.api.read(path).json()
207 nb = self.api.read(path).json()
208 self.assertEqual(nb['name'], u'%s.ipynb' % name)
208 self.assertEqual(nb['name'], u'%s.ipynb' % name)
209 self.assertEqual(nb['path'], path)
209 self.assertEqual(nb['path'], path)
210 self.assertEqual(nb['type'], 'notebook')
210 self.assertEqual(nb['type'], 'notebook')
211 self.assertIn('content', nb)
211 self.assertIn('content', nb)
212 self.assertEqual(nb['format'], 'json')
212 self.assertEqual(nb['format'], 'json')
213 self.assertIn('content', nb)
213 self.assertIn('content', nb)
214 self.assertIn('metadata', nb['content'])
214 self.assertIn('metadata', nb['content'])
215 self.assertIsInstance(nb['content']['metadata'], dict)
215 self.assertIsInstance(nb['content']['metadata'], dict)
216
216
217 def test_get_contents_no_such_file(self):
217 def test_get_contents_no_such_file(self):
218 # Name that doesn't exist - should be a 404
218 # Name that doesn't exist - should be a 404
219 with assert_http_error(404):
219 with assert_http_error(404):
220 self.api.read('foo/q.ipynb')
220 self.api.read('foo/q.ipynb')
221
221
222 def test_get_text_file_contents(self):
222 def test_get_text_file_contents(self):
223 for d, name in self.dirs_nbs:
223 for d, name in self.dirs_nbs:
224 path = url_path_join(d, name + '.txt')
224 path = url_path_join(d, name + '.txt')
225 model = self.api.read(path).json()
225 model = self.api.read(path).json()
226 self.assertEqual(model['name'], u'%s.txt' % name)
226 self.assertEqual(model['name'], u'%s.txt' % name)
227 self.assertEqual(model['path'], path)
227 self.assertEqual(model['path'], path)
228 self.assertIn('content', model)
228 self.assertIn('content', model)
229 self.assertEqual(model['format'], 'text')
229 self.assertEqual(model['format'], 'text')
230 self.assertEqual(model['type'], 'file')
230 self.assertEqual(model['type'], 'file')
231 self.assertEqual(model['content'], self._txt_for_name(name))
231 self.assertEqual(model['content'], self._txt_for_name(name))
232
232
233 # Name that doesn't exist - should be a 404
233 # Name that doesn't exist - should be a 404
234 with assert_http_error(404):
234 with assert_http_error(404):
235 self.api.read('foo/q.txt')
235 self.api.read('foo/q.txt')
236
236
237 def test_get_binary_file_contents(self):
237 def test_get_binary_file_contents(self):
238 for d, name in self.dirs_nbs:
238 for d, name in self.dirs_nbs:
239 path = url_path_join(d, name + '.blob')
239 path = url_path_join(d, name + '.blob')
240 model = self.api.read(path).json()
240 model = self.api.read(path).json()
241 self.assertEqual(model['name'], u'%s.blob' % name)
241 self.assertEqual(model['name'], u'%s.blob' % name)
242 self.assertEqual(model['path'], path)
242 self.assertEqual(model['path'], path)
243 self.assertIn('content', model)
243 self.assertIn('content', model)
244 self.assertEqual(model['format'], 'base64')
244 self.assertEqual(model['format'], 'base64')
245 self.assertEqual(model['type'], 'file')
245 self.assertEqual(model['type'], 'file')
246 b64_data = base64.encodestring(self._blob_for_name(name)).decode('ascii')
246 b64_data = base64.encodestring(self._blob_for_name(name)).decode('ascii')
247 self.assertEqual(model['content'], b64_data)
247 self.assertEqual(model['content'], b64_data)
248
248
249 # Name that doesn't exist - should be a 404
249 # Name that doesn't exist - should be a 404
250 with assert_http_error(404):
250 with assert_http_error(404):
251 self.api.read('foo/q.txt')
251 self.api.read('foo/q.txt')
252
252
253 def _check_created(self, resp, path, type='notebook'):
253 def _check_created(self, resp, path, type='notebook'):
254 self.assertEqual(resp.status_code, 201)
254 self.assertEqual(resp.status_code, 201)
255 location_header = py3compat.str_to_unicode(resp.headers['Location'])
255 location_header = py3compat.str_to_unicode(resp.headers['Location'])
256 self.assertEqual(location_header, url_escape(url_path_join(u'/api/contents', path)))
256 self.assertEqual(location_header, url_escape(url_path_join(u'/api/contents', path)))
257 rjson = resp.json()
257 rjson = resp.json()
258 self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
258 self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
259 self.assertEqual(rjson['path'], path)
259 self.assertEqual(rjson['path'], path)
260 self.assertEqual(rjson['type'], type)
260 self.assertEqual(rjson['type'], type)
261 isright = os.path.isdir if type == 'directory' else os.path.isfile
261 isright = os.path.isdir if type == 'directory' else os.path.isfile
262 assert isright(pjoin(
262 assert isright(pjoin(
263 self.notebook_dir.name,
263 self.notebook_dir.name,
264 path.replace('/', os.sep),
264 path.replace('/', os.sep),
265 ))
265 ))
266
266
267 def test_create_untitled(self):
267 def test_create_untitled(self):
268 resp = self.api.create_untitled(path=u'Γ₯ b')
268 resp = self.api.create_untitled(path=u'Γ₯ b')
269 self._check_created(resp, u'Γ₯ b/Untitled0.ipynb')
269 self._check_created(resp, u'Γ₯ b/Untitled0.ipynb')
270
270
271 # Second time
271 # Second time
272 resp = self.api.create_untitled(path=u'Γ₯ b')
272 resp = self.api.create_untitled(path=u'Γ₯ b')
273 self._check_created(resp, u'Γ₯ b/Untitled1.ipynb')
273 self._check_created(resp, u'Γ₯ b/Untitled1.ipynb')
274
274
275 # And two directories down
275 # And two directories down
276 resp = self.api.create_untitled(path='foo/bar')
276 resp = self.api.create_untitled(path='foo/bar')
277 self._check_created(resp, 'foo/bar/Untitled0.ipynb')
277 self._check_created(resp, 'foo/bar/Untitled0.ipynb')
278
278
279 def test_create_untitled_txt(self):
279 def test_create_untitled_txt(self):
280 resp = self.api.create_untitled(path='foo/bar', ext='.txt')
280 resp = self.api.create_untitled(path='foo/bar', ext='.txt')
281 self._check_created(resp, 'foo/bar/untitled0.txt', type='file')
281 self._check_created(resp, 'foo/bar/untitled0.txt', type='file')
282
282
283 resp = self.api.read(path='foo/bar/untitled0.txt')
283 resp = self.api.read(path='foo/bar/untitled0.txt')
284 model = resp.json()
284 model = resp.json()
285 self.assertEqual(model['type'], 'file')
285 self.assertEqual(model['type'], 'file')
286 self.assertEqual(model['format'], 'text')
286 self.assertEqual(model['format'], 'text')
287 self.assertEqual(model['content'], '')
287 self.assertEqual(model['content'], '')
288
288
289 def test_upload(self):
289 def test_upload(self):
290 nb = new_notebook()
290 nb = new_notebook()
291 nbmodel = {'content': nb, 'type': 'notebook'}
291 nbmodel = {'content': nb, 'type': 'notebook'}
292 path = u'Γ₯ b/Upload tΓ©st.ipynb'
292 path = u'Γ₯ b/Upload tΓ©st.ipynb'
293 resp = self.api.upload(path, body=json.dumps(nbmodel))
293 resp = self.api.upload(path, body=json.dumps(nbmodel))
294 self._check_created(resp, path)
294 self._check_created(resp, path)
295
295
296 def test_mkdir(self):
296 def test_mkdir(self):
297 path = u'Γ₯ b/New βˆ‚ir'
297 path = u'Γ₯ b/New βˆ‚ir'
298 resp = self.api.mkdir(path)
298 resp = self.api.mkdir(path)
299 self._check_created(resp, path, type='directory')
299 self._check_created(resp, path, type='directory')
300
300
301 def test_mkdir_hidden_400(self):
301 def test_mkdir_hidden_400(self):
302 with assert_http_error(400):
302 with assert_http_error(400):
303 resp = self.api.mkdir(u'Γ₯ b/.hidden')
303 resp = self.api.mkdir(u'Γ₯ b/.hidden')
304
304
305 def test_upload_txt(self):
305 def test_upload_txt(self):
306 body = u'ΓΌnicode tΓ©xt'
306 body = u'ΓΌnicode tΓ©xt'
307 model = {
307 model = {
308 'content' : body,
308 'content' : body,
309 'format' : 'text',
309 'format' : 'text',
310 'type' : 'file',
310 'type' : 'file',
311 }
311 }
312 path = u'Γ₯ b/Upload tΓ©st.txt'
312 path = u'Γ₯ b/Upload tΓ©st.txt'
313 resp = self.api.upload(path, body=json.dumps(model))
313 resp = self.api.upload(path, body=json.dumps(model))
314
314
315 # check roundtrip
315 # check roundtrip
316 resp = self.api.read(path)
316 resp = self.api.read(path)
317 model = resp.json()
317 model = resp.json()
318 self.assertEqual(model['type'], 'file')
318 self.assertEqual(model['type'], 'file')
319 self.assertEqual(model['format'], 'text')
319 self.assertEqual(model['format'], 'text')
320 self.assertEqual(model['content'], body)
320 self.assertEqual(model['content'], body)
321
321
322 def test_upload_b64(self):
322 def test_upload_b64(self):
323 body = b'\xFFblob'
323 body = b'\xFFblob'
324 b64body = base64.encodestring(body).decode('ascii')
324 b64body = base64.encodestring(body).decode('ascii')
325 model = {
325 model = {
326 'content' : b64body,
326 'content' : b64body,
327 'format' : 'base64',
327 'format' : 'base64',
328 'type' : 'file',
328 'type' : 'file',
329 }
329 }
330 path = u'Γ₯ b/Upload tΓ©st.blob'
330 path = u'Γ₯ b/Upload tΓ©st.blob'
331 resp = self.api.upload(path, body=json.dumps(model))
331 resp = self.api.upload(path, body=json.dumps(model))
332
332
333 # check roundtrip
333 # check roundtrip
334 resp = self.api.read(path)
334 resp = self.api.read(path)
335 model = resp.json()
335 model = resp.json()
336 self.assertEqual(model['type'], 'file')
336 self.assertEqual(model['type'], 'file')
337 self.assertEqual(model['path'], path)
337 self.assertEqual(model['path'], path)
338 self.assertEqual(model['format'], 'base64')
338 self.assertEqual(model['format'], 'base64')
339 decoded = base64.decodestring(model['content'].encode('ascii'))
339 decoded = base64.decodestring(model['content'].encode('ascii'))
340 self.assertEqual(decoded, body)
340 self.assertEqual(decoded, body)
341
341
342 def test_upload_v2(self):
342 def test_upload_v2(self):
343 nb = v2.new_notebook()
343 nb = v2.new_notebook()
344 ws = v2.new_worksheet()
344 ws = v2.new_worksheet()
345 nb.worksheets.append(ws)
345 nb.worksheets.append(ws)
346 ws.cells.append(v2.new_code_cell(input='print("hi")'))
346 ws.cells.append(v2.new_code_cell(input='print("hi")'))
347 nbmodel = {'content': nb, 'type': 'notebook'}
347 nbmodel = {'content': nb, 'type': 'notebook'}
348 path = u'Γ₯ b/Upload tΓ©st.ipynb'
348 path = u'Γ₯ b/Upload tΓ©st.ipynb'
349 resp = self.api.upload(path, body=json.dumps(nbmodel))
349 resp = self.api.upload(path, body=json.dumps(nbmodel))
350 self._check_created(resp, path)
350 self._check_created(resp, path)
351 resp = self.api.read(path)
351 resp = self.api.read(path)
352 data = resp.json()
352 data = resp.json()
353 self.assertEqual(data['content']['nbformat'], 4)
353 self.assertEqual(data['content']['nbformat'], 4)
354
354
355 def test_copy_untitled(self):
355 def test_copy(self):
356 resp = self.api.copy_untitled(u'Γ₯ b/Γ§ d.ipynb', u'unicodΓ©')
356 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'unicodΓ©')
357 self._check_created(resp, u'unicodΓ©/Γ§ d-Copy0.ipynb')
357 self._check_created(resp, u'unicodΓ©/Γ§ d-Copy0.ipynb')
358
358
359 resp = self.api.copy_untitled(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
359 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b')
360 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy0.ipynb')
360 self._check_created(resp, u'Γ₯ b/Γ§ d-Copy0.ipynb')
361
361
362 def test_copy(self):
363 resp = self.api.copy(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b/cΓΈpy.ipynb')
364 self._check_created(resp, u'Γ₯ b/cΓΈpy.ipynb')
365
366 def test_copy_path(self):
362 def test_copy_path(self):
367 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b/cΓΈpyfoo.ipynb')
363 resp = self.api.copy(u'foo/a.ipynb', u'Γ₯ b')
368 self._check_created(resp, u'Γ₯ b/cΓΈpyfoo.ipynb')
364 self._check_created(resp, u'Γ₯ b/a-Copy0.ipynb')
365
366 def test_copy_put_400(self):
367 with assert_http_error(400):
368 resp = self.api.copy_put(u'Γ₯ b/Γ§ d.ipynb', u'Γ₯ b/cΓΈpy.ipynb')
369
369
370 def test_copy_dir_400(self):
370 def test_copy_dir_400(self):
371 # can't copy directories
371 # can't copy directories
372 with assert_http_error(400):
372 with assert_http_error(400):
373 resp = self.api.copy(u'Γ₯ b', u'Γ₯ c')
373 resp = self.api.copy(u'Γ₯ b', u'foo')
374
374
375 def test_delete(self):
375 def test_delete(self):
376 for d, name in self.dirs_nbs:
376 for d, name in self.dirs_nbs:
377 print('%r, %r' % (d, name))
377 print('%r, %r' % (d, name))
378 resp = self.api.delete(url_path_join(d, name + '.ipynb'))
378 resp = self.api.delete(url_path_join(d, name + '.ipynb'))
379 self.assertEqual(resp.status_code, 204)
379 self.assertEqual(resp.status_code, 204)
380
380
381 for d in self.dirs + ['/']:
381 for d in self.dirs + ['/']:
382 nbs = notebooks_only(self.api.list(d).json())
382 nbs = notebooks_only(self.api.list(d).json())
383 print('------')
383 print('------')
384 print(d)
384 print(d)
385 print(nbs)
385 print(nbs)
386 self.assertEqual(nbs, [])
386 self.assertEqual(nbs, [])
387
387
388 def test_delete_dirs(self):
388 def test_delete_dirs(self):
389 # depth-first delete everything, so we don't try to delete empty directories
389 # depth-first delete everything, so we don't try to delete empty directories
390 for name in sorted(self.dirs + ['/'], key=len, reverse=True):
390 for name in sorted(self.dirs + ['/'], key=len, reverse=True):
391 listing = self.api.list(name).json()['content']
391 listing = self.api.list(name).json()['content']
392 for model in listing:
392 for model in listing:
393 self.api.delete(model['path'])
393 self.api.delete(model['path'])
394 listing = self.api.list('/').json()['content']
394 listing = self.api.list('/').json()['content']
395 self.assertEqual(listing, [])
395 self.assertEqual(listing, [])
396
396
397 def test_delete_non_empty_dir(self):
397 def test_delete_non_empty_dir(self):
398 """delete non-empty dir raises 400"""
398 """delete non-empty dir raises 400"""
399 with assert_http_error(400):
399 with assert_http_error(400):
400 self.api.delete(u'Γ₯ b')
400 self.api.delete(u'Γ₯ b')
401
401
402 def test_rename(self):
402 def test_rename(self):
403 resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
403 resp = self.api.rename('foo/a.ipynb', 'foo/z.ipynb')
404 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
404 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
405 self.assertEqual(resp.json()['name'], 'z.ipynb')
405 self.assertEqual(resp.json()['name'], 'z.ipynb')
406 self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
406 self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
407 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
407 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
408
408
409 nbs = notebooks_only(self.api.list('foo').json())
409 nbs = notebooks_only(self.api.list('foo').json())
410 nbnames = set(n['name'] for n in nbs)
410 nbnames = set(n['name'] for n in nbs)
411 self.assertIn('z.ipynb', nbnames)
411 self.assertIn('z.ipynb', nbnames)
412 self.assertNotIn('a.ipynb', nbnames)
412 self.assertNotIn('a.ipynb', nbnames)
413
413
414 def test_rename_existing(self):
414 def test_rename_existing(self):
415 with assert_http_error(409):
415 with assert_http_error(409):
416 self.api.rename('foo/a.ipynb', 'foo/b.ipynb')
416 self.api.rename('foo/a.ipynb', 'foo/b.ipynb')
417
417
418 def test_save(self):
418 def test_save(self):
419 resp = self.api.read('foo/a.ipynb')
419 resp = self.api.read('foo/a.ipynb')
420 nbcontent = json.loads(resp.text)['content']
420 nbcontent = json.loads(resp.text)['content']
421 nb = from_dict(nbcontent)
421 nb = from_dict(nbcontent)
422 nb.cells.append(new_markdown_cell(u'Created by test Β³'))
422 nb.cells.append(new_markdown_cell(u'Created by test Β³'))
423
423
424 nbmodel= {'content': nb, 'type': 'notebook'}
424 nbmodel= {'content': nb, 'type': 'notebook'}
425 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
425 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
426
426
427 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
427 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
428 with io.open(nbfile, 'r', encoding='utf-8') as f:
428 with io.open(nbfile, 'r', encoding='utf-8') as f:
429 newnb = read(f, as_version=4)
429 newnb = read(f, as_version=4)
430 self.assertEqual(newnb.cells[0].source,
430 self.assertEqual(newnb.cells[0].source,
431 u'Created by test Β³')
431 u'Created by test Β³')
432 nbcontent = self.api.read('foo/a.ipynb').json()['content']
432 nbcontent = self.api.read('foo/a.ipynb').json()['content']
433 newnb = from_dict(nbcontent)
433 newnb = from_dict(nbcontent)
434 self.assertEqual(newnb.cells[0].source,
434 self.assertEqual(newnb.cells[0].source,
435 u'Created by test Β³')
435 u'Created by test Β³')
436
436
437
437
438 def test_checkpoints(self):
438 def test_checkpoints(self):
439 resp = self.api.read('foo/a.ipynb')
439 resp = self.api.read('foo/a.ipynb')
440 r = self.api.new_checkpoint('foo/a.ipynb')
440 r = self.api.new_checkpoint('foo/a.ipynb')
441 self.assertEqual(r.status_code, 201)
441 self.assertEqual(r.status_code, 201)
442 cp1 = r.json()
442 cp1 = r.json()
443 self.assertEqual(set(cp1), {'id', 'last_modified'})
443 self.assertEqual(set(cp1), {'id', 'last_modified'})
444 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
444 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
445
445
446 # Modify it
446 # Modify it
447 nbcontent = json.loads(resp.text)['content']
447 nbcontent = json.loads(resp.text)['content']
448 nb = from_dict(nbcontent)
448 nb = from_dict(nbcontent)
449 hcell = new_markdown_cell('Created by test')
449 hcell = new_markdown_cell('Created by test')
450 nb.cells.append(hcell)
450 nb.cells.append(hcell)
451 # Save
451 # Save
452 nbmodel= {'content': nb, 'type': 'notebook'}
452 nbmodel= {'content': nb, 'type': 'notebook'}
453 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
453 resp = self.api.save('foo/a.ipynb', body=json.dumps(nbmodel))
454
454
455 # List checkpoints
455 # List checkpoints
456 cps = self.api.get_checkpoints('foo/a.ipynb').json()
456 cps = self.api.get_checkpoints('foo/a.ipynb').json()
457 self.assertEqual(cps, [cp1])
457 self.assertEqual(cps, [cp1])
458
458
459 nbcontent = self.api.read('foo/a.ipynb').json()['content']
459 nbcontent = self.api.read('foo/a.ipynb').json()['content']
460 nb = from_dict(nbcontent)
460 nb = from_dict(nbcontent)
461 self.assertEqual(nb.cells[0].source, 'Created by test')
461 self.assertEqual(nb.cells[0].source, 'Created by test')
462
462
463 # Restore cp1
463 # Restore cp1
464 r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id'])
464 r = self.api.restore_checkpoint('foo/a.ipynb', cp1['id'])
465 self.assertEqual(r.status_code, 204)
465 self.assertEqual(r.status_code, 204)
466 nbcontent = self.api.read('foo/a.ipynb').json()['content']
466 nbcontent = self.api.read('foo/a.ipynb').json()['content']
467 nb = from_dict(nbcontent)
467 nb = from_dict(nbcontent)
468 self.assertEqual(nb.cells, [])
468 self.assertEqual(nb.cells, [])
469
469
470 # Delete cp1
470 # Delete cp1
471 r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id'])
471 r = self.api.delete_checkpoint('foo/a.ipynb', cp1['id'])
472 self.assertEqual(r.status_code, 204)
472 self.assertEqual(r.status_code, 204)
473 cps = self.api.get_checkpoints('foo/a.ipynb').json()
473 cps = self.api.get_checkpoints('foo/a.ipynb').json()
474 self.assertEqual(cps, [])
474 self.assertEqual(cps, [])
General Comments 0
You need to be logged in to leave comments. Login now