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