##// END OF EJS Templates
mv services/notebooks services/contents
MinRK -
Show More
1 NO CONTENT: file renamed from IPython/html/services/notebooks/__init__.py to IPython/html/services/contents/__init__.py
@@ -1,470 +1,470 b''
1 1 """A notebook manager that uses the local file system for storage."""
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 io
7 7 import os
8 8 import glob
9 9 import shutil
10 10
11 11 from tornado import web
12 12
13 13 from .nbmanager import NotebookManager
14 14 from IPython.nbformat import current
15 15 from IPython.utils.path import ensure_dir_exists
16 16 from IPython.utils.traitlets import Unicode, Bool, TraitError
17 17 from IPython.utils.py3compat import getcwd
18 18 from IPython.utils import tz
19 19 from IPython.html.utils import is_hidden, to_os_path
20 20
21 21 def sort_key(item):
22 22 """Case-insensitive sorting."""
23 23 return item['name'].lower()
24 24
25 25 #-----------------------------------------------------------------------------
26 26 # Classes
27 27 #-----------------------------------------------------------------------------
28 28
29 29 class FileNotebookManager(NotebookManager):
30
30
31 31 save_script = Bool(False, config=True,
32 32 help="""Automatically create a Python script when saving the notebook.
33
33
34 34 For easier use of import, %run and %load across notebooks, a
35 35 <notebook-name>.py script will be created next to any
36 36 <notebook-name>.ipynb on each save. This can also be set with the
37 37 short `--script` flag.
38 38 """
39 39 )
40 40 notebook_dir = Unicode(getcwd(), config=True)
41
41
42 42 def _notebook_dir_changed(self, name, old, new):
43 43 """Do a bit of validation of the notebook dir."""
44 44 if not os.path.isabs(new):
45 45 # If we receive a non-absolute path, make it absolute.
46 46 self.notebook_dir = os.path.abspath(new)
47 47 return
48 48 if not os.path.exists(new) or not os.path.isdir(new):
49 49 raise TraitError("notebook dir %r is not a directory" % new)
50
50
51 51 checkpoint_dir = Unicode('.ipynb_checkpoints', config=True,
52 52 help="""The directory name in which to keep notebook checkpoints
53
53
54 54 This is a path relative to the notebook's own directory.
55
55
56 56 By default, it is .ipynb_checkpoints
57 57 """
58 58 )
59
59
60 60 def _copy(self, src, dest):
61 61 """copy src to dest
62
62
63 63 like shutil.copy2, but log errors in copystat
64 64 """
65 65 shutil.copyfile(src, dest)
66 66 try:
67 67 shutil.copystat(src, dest)
68 68 except OSError as e:
69 69 self.log.debug("copystat on %s failed", dest, exc_info=True)
70
70
71 71 def get_notebook_names(self, path=''):
72 72 """List all notebook names in the notebook dir and path."""
73 73 path = path.strip('/')
74 74 if not os.path.isdir(self._get_os_path(path=path)):
75 75 raise web.HTTPError(404, 'Directory not found: ' + path)
76 76 names = glob.glob(self._get_os_path('*'+self.filename_ext, path))
77 77 names = [os.path.basename(name)
78 78 for name in names]
79 79 return names
80 80
81 81 def path_exists(self, path):
82 82 """Does the API-style path (directory) actually exist?
83
83
84 84 Parameters
85 85 ----------
86 86 path : string
87 87 The path to check. This is an API path (`/` separated,
88 88 relative to base notebook-dir).
89
89
90 90 Returns
91 91 -------
92 92 exists : bool
93 93 Whether the path is indeed a directory.
94 94 """
95 95 path = path.strip('/')
96 96 os_path = self._get_os_path(path=path)
97 97 return os.path.isdir(os_path)
98 98
99 99 def is_hidden(self, path):
100 100 """Does the API style path correspond to a hidden directory or file?
101
101
102 102 Parameters
103 103 ----------
104 104 path : string
105 105 The path to check. This is an API path (`/` separated,
106 106 relative to base notebook-dir).
107
107
108 108 Returns
109 109 -------
110 110 exists : bool
111 111 Whether the path is hidden.
112
112
113 113 """
114 114 path = path.strip('/')
115 115 os_path = self._get_os_path(path=path)
116 116 return is_hidden(os_path, self.notebook_dir)
117 117
118 118 def _get_os_path(self, name=None, path=''):
119 119 """Given a notebook name and a URL path, return its file system
120 120 path.
121 121
122 122 Parameters
123 123 ----------
124 124 name : string
125 125 The name of a notebook file with the .ipynb extension
126 126 path : string
127 127 The relative URL path (with '/' as separator) to the named
128 128 notebook.
129 129
130 130 Returns
131 131 -------
132 132 path : string
133 133 A file system path that combines notebook_dir (location where
134 134 server started), the relative path, and the filename with the
135 135 current operating system's url.
136 136 """
137 137 if name is not None:
138 138 path = path + '/' + name
139 139 return to_os_path(path, self.notebook_dir)
140 140
141 141 def notebook_exists(self, name, path=''):
142 142 """Returns a True if the notebook exists. Else, returns False.
143 143
144 144 Parameters
145 145 ----------
146 146 name : string
147 147 The name of the notebook you are checking.
148 148 path : string
149 149 The relative path to the notebook (with '/' as separator)
150 150
151 151 Returns
152 152 -------
153 153 bool
154 154 """
155 155 path = path.strip('/')
156 156 nbpath = self._get_os_path(name, path=path)
157 157 return os.path.isfile(nbpath)
158 158
159 159 # TODO: Remove this after we create the contents web service and directories are
160 160 # no longer listed by the notebook web service.
161 161 def list_dirs(self, path):
162 162 """List the directories for a given API style path."""
163 163 path = path.strip('/')
164 164 os_path = self._get_os_path('', path)
165 165 if not os.path.isdir(os_path):
166 166 raise web.HTTPError(404, u'directory does not exist: %r' % os_path)
167 167 elif is_hidden(os_path, self.notebook_dir):
168 168 self.log.info("Refusing to serve hidden directory, via 404 Error")
169 169 raise web.HTTPError(404, u'directory does not exist: %r' % os_path)
170 170 dir_names = os.listdir(os_path)
171 171 dirs = []
172 172 for name in dir_names:
173 173 os_path = self._get_os_path(name, path)
174 174 if os.path.isdir(os_path) and not is_hidden(os_path, self.notebook_dir)\
175 175 and self.should_list(name):
176 176 try:
177 177 model = self.get_dir_model(name, path)
178 178 except IOError:
179 179 pass
180 180 dirs.append(model)
181 181 dirs = sorted(dirs, key=sort_key)
182 182 return dirs
183 183
184 184 # TODO: Remove this after we create the contents web service and directories are
185 185 # no longer listed by the notebook web service.
186 186 def get_dir_model(self, name, path=''):
187 187 """Get the directory model given a directory name and its API style path"""
188 188 path = path.strip('/')
189 189 os_path = self._get_os_path(name, path)
190 190 if not os.path.isdir(os_path):
191 191 raise IOError('directory does not exist: %r' % os_path)
192 192 info = os.stat(os_path)
193 193 last_modified = tz.utcfromtimestamp(info.st_mtime)
194 194 created = tz.utcfromtimestamp(info.st_ctime)
195 195 # Create the notebook model.
196 196 model ={}
197 197 model['name'] = name
198 198 model['path'] = path
199 199 model['last_modified'] = last_modified
200 200 model['created'] = created
201 201 model['type'] = 'directory'
202 202 return model
203 203
204 204 def list_notebooks(self, path):
205 205 """Returns a list of dictionaries that are the standard model
206 206 for all notebooks in the relative 'path'.
207
207
208 208 Parameters
209 209 ----------
210 210 path : str
211 211 the URL path that describes the relative path for the
212 212 listed notebooks
213
213
214 214 Returns
215 215 -------
216 216 notebooks : list of dicts
217 217 a list of the notebook models without 'content'
218 218 """
219 219 path = path.strip('/')
220 220 notebook_names = self.get_notebook_names(path)
221 221 notebooks = [self.get_notebook(name, path, content=False)
222 222 for name in notebook_names if self.should_list(name)]
223 223 notebooks = sorted(notebooks, key=sort_key)
224 224 return notebooks
225 225
226 226 def get_notebook(self, name, path='', content=True):
227 227 """ Takes a path and name for a notebook and returns its model
228
228
229 229 Parameters
230 230 ----------
231 231 name : str
232 232 the name of the notebook
233 233 path : str
234 234 the URL path that describes the relative path for
235 235 the notebook
236
236
237 237 Returns
238 238 -------
239 239 model : dict
240 the notebook model. If contents=True, returns the 'contents'
240 the notebook model. If contents=True, returns the 'contents'
241 241 dict in the model as well.
242 242 """
243 243 path = path.strip('/')
244 244 if not self.notebook_exists(name=name, path=path):
245 245 raise web.HTTPError(404, u'Notebook does not exist: %s' % name)
246 246 os_path = self._get_os_path(name, path)
247 247 info = os.stat(os_path)
248 248 last_modified = tz.utcfromtimestamp(info.st_mtime)
249 249 created = tz.utcfromtimestamp(info.st_ctime)
250 250 # Create the notebook model.
251 251 model ={}
252 252 model['name'] = name
253 253 model['path'] = path
254 254 model['last_modified'] = last_modified
255 255 model['created'] = created
256 256 model['type'] = 'notebook'
257 257 if content:
258 258 with io.open(os_path, 'r', encoding='utf-8') as f:
259 259 try:
260 260 nb = current.read(f, u'json')
261 261 except Exception as e:
262 262 raise web.HTTPError(400, u"Unreadable Notebook: %s %s" % (os_path, e))
263 263 self.mark_trusted_cells(nb, name, path)
264 264 model['content'] = nb
265 265 return model
266 266
267 267 def save_notebook(self, model, name='', path=''):
268 268 """Save the notebook model and return the model with no content."""
269 269 path = path.strip('/')
270 270
271 271 if 'content' not in model:
272 272 raise web.HTTPError(400, u'No notebook JSON data provided')
273 273
274 274 # One checkpoint should always exist
275 275 if self.notebook_exists(name, path) and not self.list_checkpoints(name, path):
276 276 self.create_checkpoint(name, path)
277 277
278 278 new_path = model.get('path', path).strip('/')
279 279 new_name = model.get('name', name)
280 280
281 281 if path != new_path or name != new_name:
282 282 self.rename_notebook(name, path, new_name, new_path)
283 283
284 284 # Save the notebook file
285 285 os_path = self._get_os_path(new_name, new_path)
286 286 nb = current.to_notebook_json(model['content'])
287
287
288 288 self.check_and_sign(nb, new_name, new_path)
289
289
290 290 if 'name' in nb['metadata']:
291 291 nb['metadata']['name'] = u''
292 292 try:
293 293 self.log.debug("Autosaving notebook %s", os_path)
294 294 with io.open(os_path, 'w', encoding='utf-8') as f:
295 295 current.write(nb, f, u'json')
296 296 except Exception as e:
297 297 raise web.HTTPError(400, u'Unexpected error while autosaving notebook: %s %s' % (os_path, e))
298 298
299 299 # Save .py script as well
300 300 if self.save_script:
301 301 py_path = os.path.splitext(os_path)[0] + '.py'
302 302 self.log.debug("Writing script %s", py_path)
303 303 try:
304 304 with io.open(py_path, 'w', encoding='utf-8') as f:
305 305 current.write(nb, f, u'py')
306 306 except Exception as e:
307 307 raise web.HTTPError(400, u'Unexpected error while saving notebook as script: %s %s' % (py_path, e))
308 308
309 309 model = self.get_notebook(new_name, new_path, content=False)
310 310 return model
311 311
312 312 def update_notebook(self, model, name, path=''):
313 313 """Update the notebook's path and/or name"""
314 314 path = path.strip('/')
315 315 new_name = model.get('name', name)
316 316 new_path = model.get('path', path).strip('/')
317 317 if path != new_path or name != new_name:
318 318 self.rename_notebook(name, path, new_name, new_path)
319 319 model = self.get_notebook(new_name, new_path, content=False)
320 320 return model
321 321
322 322 def delete_notebook(self, name, path=''):
323 323 """Delete notebook by name and path."""
324 324 path = path.strip('/')
325 325 os_path = self._get_os_path(name, path)
326 326 if not os.path.isfile(os_path):
327 327 raise web.HTTPError(404, u'Notebook does not exist: %s' % os_path)
328
328
329 329 # clear checkpoints
330 330 for checkpoint in self.list_checkpoints(name, path):
331 331 checkpoint_id = checkpoint['id']
332 332 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
333 333 if os.path.isfile(cp_path):
334 334 self.log.debug("Unlinking checkpoint %s", cp_path)
335 335 os.unlink(cp_path)
336
336
337 337 self.log.debug("Unlinking notebook %s", os_path)
338 338 os.unlink(os_path)
339 339
340 340 def rename_notebook(self, old_name, old_path, new_name, new_path):
341 341 """Rename a notebook."""
342 342 old_path = old_path.strip('/')
343 343 new_path = new_path.strip('/')
344 344 if new_name == old_name and new_path == old_path:
345 345 return
346
346
347 347 new_os_path = self._get_os_path(new_name, new_path)
348 348 old_os_path = self._get_os_path(old_name, old_path)
349 349
350 350 # Should we proceed with the move?
351 351 if os.path.isfile(new_os_path):
352 352 raise web.HTTPError(409, u'Notebook with name already exists: %s' % new_os_path)
353 353 if self.save_script:
354 354 old_py_path = os.path.splitext(old_os_path)[0] + '.py'
355 355 new_py_path = os.path.splitext(new_os_path)[0] + '.py'
356 356 if os.path.isfile(new_py_path):
357 357 raise web.HTTPError(409, u'Python script with name already exists: %s' % new_py_path)
358 358
359 359 # Move the notebook file
360 360 try:
361 361 shutil.move(old_os_path, new_os_path)
362 362 except Exception as e:
363 363 raise web.HTTPError(500, u'Unknown error renaming notebook: %s %s' % (old_os_path, e))
364 364
365 365 # Move the checkpoints
366 366 old_checkpoints = self.list_checkpoints(old_name, old_path)
367 367 for cp in old_checkpoints:
368 368 checkpoint_id = cp['id']
369 369 old_cp_path = self.get_checkpoint_path(checkpoint_id, old_name, old_path)
370 370 new_cp_path = self.get_checkpoint_path(checkpoint_id, new_name, new_path)
371 371 if os.path.isfile(old_cp_path):
372 372 self.log.debug("Renaming checkpoint %s -> %s", old_cp_path, new_cp_path)
373 373 shutil.move(old_cp_path, new_cp_path)
374 374
375 375 # Move the .py script
376 376 if self.save_script:
377 377 shutil.move(old_py_path, new_py_path)
378
378
379 379 # Checkpoint-related utilities
380
380
381 381 def get_checkpoint_path(self, checkpoint_id, name, path=''):
382 382 """find the path to a checkpoint"""
383 383 path = path.strip('/')
384 384 basename, _ = os.path.splitext(name)
385 385 filename = u"{name}-{checkpoint_id}{ext}".format(
386 386 name=basename,
387 387 checkpoint_id=checkpoint_id,
388 388 ext=self.filename_ext,
389 389 )
390 390 os_path = self._get_os_path(path=path)
391 391 cp_dir = os.path.join(os_path, self.checkpoint_dir)
392 392 ensure_dir_exists(cp_dir)
393 393 cp_path = os.path.join(cp_dir, filename)
394 394 return cp_path
395 395
396 396 def get_checkpoint_model(self, checkpoint_id, name, path=''):
397 397 """construct the info dict for a given checkpoint"""
398 398 path = path.strip('/')
399 399 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
400 400 stats = os.stat(cp_path)
401 401 last_modified = tz.utcfromtimestamp(stats.st_mtime)
402 402 info = dict(
403 403 id = checkpoint_id,
404 404 last_modified = last_modified,
405 405 )
406 406 return info
407
407
408 408 # public checkpoint API
409
409
410 410 def create_checkpoint(self, name, path=''):
411 411 """Create a checkpoint from the current state of a notebook"""
412 412 path = path.strip('/')
413 413 nb_path = self._get_os_path(name, path)
414 414 # only the one checkpoint ID:
415 415 checkpoint_id = u"checkpoint"
416 416 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
417 417 self.log.debug("creating checkpoint for notebook %s", name)
418 418 self._copy(nb_path, cp_path)
419
419
420 420 # return the checkpoint info
421 421 return self.get_checkpoint_model(checkpoint_id, name, path)
422
422
423 423 def list_checkpoints(self, name, path=''):
424 424 """list the checkpoints for a given notebook
425
425
426 426 This notebook manager currently only supports one checkpoint per notebook.
427 427 """
428 428 path = path.strip('/')
429 429 checkpoint_id = "checkpoint"
430 430 os_path = self.get_checkpoint_path(checkpoint_id, name, path)
431 431 if not os.path.exists(os_path):
432 432 return []
433 433 else:
434 434 return [self.get_checkpoint_model(checkpoint_id, name, path)]
435
436
435
436
437 437 def restore_checkpoint(self, checkpoint_id, name, path=''):
438 438 """restore a notebook to a checkpointed state"""
439 439 path = path.strip('/')
440 440 self.log.info("restoring Notebook %s from checkpoint %s", name, checkpoint_id)
441 441 nb_path = self._get_os_path(name, path)
442 442 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
443 443 if not os.path.isfile(cp_path):
444 444 self.log.debug("checkpoint file does not exist: %s", cp_path)
445 445 raise web.HTTPError(404,
446 446 u'Notebook checkpoint does not exist: %s-%s' % (name, checkpoint_id)
447 447 )
448 448 # ensure notebook is readable (never restore from an unreadable notebook)
449 449 with io.open(cp_path, 'r', encoding='utf-8') as f:
450 450 current.read(f, u'json')
451 451 self._copy(cp_path, nb_path)
452 452 self.log.debug("copying %s -> %s", cp_path, nb_path)
453
453
454 454 def delete_checkpoint(self, checkpoint_id, name, path=''):
455 455 """delete a notebook's checkpoint"""
456 456 path = path.strip('/')
457 457 cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
458 458 if not os.path.isfile(cp_path):
459 459 raise web.HTTPError(404,
460 460 u'Notebook checkpoint does not exist: %s%s-%s' % (path, name, checkpoint_id)
461 461 )
462 462 self.log.debug("unlinking %s", cp_path)
463 463 os.unlink(cp_path)
464
464
465 465 def info_string(self):
466 466 return "Serving notebooks from local directory: %s" % self.notebook_dir
467 467
468 468 def get_kernel_path(self, name, path='', model=None):
469 469 """ Return the path to start kernel in """
470 470 return os.path.join(self.notebook_dir, path)
@@ -1,288 +1,287 b''
1 1 """Tornado handlers for the notebooks web service.
2 2
3 3 Authors:
4 4
5 5 * Brian Granger
6 6 """
7 7
8 8 #-----------------------------------------------------------------------------
9 9 # Copyright (C) 2011 The IPython Development Team
10 10 #
11 11 # Distributed under the terms of the BSD License. The full license is in
12 12 # the file COPYING, distributed as part of this software.
13 13 #-----------------------------------------------------------------------------
14 14
15 15 #-----------------------------------------------------------------------------
16 16 # Imports
17 17 #-----------------------------------------------------------------------------
18 18
19 19 import json
20 20
21 21 from tornado import web
22 22
23 23 from IPython.html.utils import url_path_join, url_escape
24 24 from IPython.utils.jsonutil import date_default
25 25
26 26 from IPython.html.base.handlers import (IPythonHandler, json_errors,
27 27 notebook_path_regex, path_regex,
28 28 notebook_name_regex)
29 29
30 30 #-----------------------------------------------------------------------------
31 31 # Notebook web service handlers
32 32 #-----------------------------------------------------------------------------
33 33
34 34
35 35 class NotebookHandler(IPythonHandler):
36 36
37 37 SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
38 38
39 39 def notebook_location(self, name, path=''):
40 40 """Return the full URL location of a notebook based.
41
41
42 42 Parameters
43 43 ----------
44 44 name : unicode
45 45 The base name of the notebook, such as "foo.ipynb".
46 46 path : unicode
47 47 The URL path of the notebook.
48 48 """
49 49 return url_escape(url_path_join(
50 50 self.base_url, 'api', 'notebooks', path, name
51 51 ))
52 52
53 53 def _finish_model(self, model, location=True):
54 54 """Finish a JSON request with a model, setting relevant headers, etc."""
55 55 if location:
56 56 location = self.notebook_location(model['name'], model['path'])
57 57 self.set_header('Location', location)
58 58 self.set_header('Last-Modified', model['last_modified'])
59 59 self.finish(json.dumps(model, default=date_default))
60
60
61 61 @web.authenticated
62 62 @json_errors
63 63 def get(self, path='', name=None):
64 64 """Return a Notebook or list of notebooks.
65 65
66 66 * GET with path and no notebook name lists notebooks in a directory
67 67 * GET with path and notebook name returns notebook JSON
68 68 """
69 69 nbm = self.notebook_manager
70 70 # Check to see if a notebook name was given
71 71 if name is None:
72 72 # TODO: Remove this after we create the contents web service and directories are
73 73 # no longer listed by the notebook web service. This should only handle notebooks
74 74 # and not directories.
75 75 dirs = nbm.list_dirs(path)
76 76 notebooks = []
77 77 index = []
78 78 for nb in nbm.list_notebooks(path):
79 79 if nb['name'].lower() == 'index.ipynb':
80 80 index.append(nb)
81 81 else:
82 82 notebooks.append(nb)
83 83 notebooks = index + dirs + notebooks
84 84 self.finish(json.dumps(notebooks, default=date_default))
85 85 return
86 86 # get and return notebook representation
87 87 model = nbm.get_notebook(name, path)
88 88 self._finish_model(model, location=False)
89 89
90 90 @web.authenticated
91 91 @json_errors
92 92 def patch(self, path='', name=None):
93 93 """PATCH renames a notebook without re-uploading content."""
94 94 nbm = self.notebook_manager
95 95 if name is None:
96 96 raise web.HTTPError(400, u'Notebook name missing')
97 97 model = self.get_json_body()
98 98 if model is None:
99 99 raise web.HTTPError(400, u'JSON body missing')
100 100 model = nbm.update_notebook(model, name, path)
101 101 self._finish_model(model)
102
102
103 103 def _copy_notebook(self, copy_from, path, copy_to=None):
104 104 """Copy a notebook in path, optionally specifying the new name.
105
105
106 106 Only support copying within the same directory.
107 107 """
108 108 self.log.info(u"Copying notebook from %s/%s to %s/%s",
109 109 path, copy_from,
110 110 path, copy_to or '',
111 111 )
112 112 model = self.notebook_manager.copy_notebook(copy_from, copy_to, path)
113 113 self.set_status(201)
114 114 self._finish_model(model)
115
115
116 116 def _upload_notebook(self, model, path, name=None):
117 117 """Upload a notebook
118
118
119 119 If name specified, create it in path/name.
120 120 """
121 121 self.log.info(u"Uploading notebook to %s/%s", path, name or '')
122 122 if name:
123 123 model['name'] = name
124
124
125 125 model = self.notebook_manager.create_notebook(model, path)
126 126 self.set_status(201)
127 127 self._finish_model(model)
128
128
129 129 def _create_empty_notebook(self, path, name=None):
130 130 """Create an empty notebook in path
131
131
132 132 If name specified, create it in path/name.
133 133 """
134 134 self.log.info(u"Creating new notebook in %s/%s", path, name or '')
135 135 model = {}
136 136 if name:
137 137 model['name'] = name
138 138 model = self.notebook_manager.create_notebook(model, path=path)
139 139 self.set_status(201)
140 140 self._finish_model(model)
141
141
142 142 def _save_notebook(self, model, path, name):
143 143 """Save an existing notebook."""
144 144 self.log.info(u"Saving notebook at %s/%s", path, name)
145 145 model = self.notebook_manager.save_notebook(model, name, path)
146 146 if model['path'] != path.strip('/') or model['name'] != name:
147 147 # a rename happened, set Location header
148 148 location = True
149 149 else:
150 150 location = False
151 151 self._finish_model(model, location)
152
152
153 153 @web.authenticated
154 154 @json_errors
155 155 def post(self, path='', name=None):
156 156 """Create a new notebook in the specified path.
157
157
158 158 POST creates new notebooks. The server always decides on the notebook name.
159
159
160 160 POST /api/notebooks/path
161 161 New untitled notebook in path. If content specified, upload a
162 162 notebook, otherwise start empty.
163 163 POST /api/notebooks/path?copy=OtherNotebook.ipynb
164 164 New copy of OtherNotebook in path
165 165 """
166
166
167 167 if name is not None:
168 168 raise web.HTTPError(400, "Only POST to directories. Use PUT for full names.")
169
169
170 170 model = self.get_json_body()
171
171
172 172 if model is not None:
173 173 copy_from = model.get('copy_from')
174 174 if copy_from:
175 175 if model.get('content'):
176 176 raise web.HTTPError(400, "Can't upload and copy at the same time.")
177 177 self._copy_notebook(copy_from, path)
178 178 else:
179 179 self._upload_notebook(model, path)
180 180 else:
181 181 self._create_empty_notebook(path)
182 182
183 183 @web.authenticated
184 184 @json_errors
185 185 def put(self, path='', name=None):
186 186 """Saves the notebook in the location specified by name and path.
187
187
188 188 PUT is very similar to POST, but the requester specifies the name,
189 189 whereas with POST, the server picks the name.
190
190
191 191 PUT /api/notebooks/path/Name.ipynb
192 192 Save notebook at ``path/Name.ipynb``. Notebook structure is specified
193 193 in `content` key of JSON request body. If content is not specified,
194 194 create a new empty notebook.
195 195 PUT /api/notebooks/path/Name.ipynb?copy=OtherNotebook.ipynb
196 196 Copy OtherNotebook to Name
197 197 """
198 198 if name is None:
199 199 raise web.HTTPError(400, "Only PUT to full names. Use POST for directories.")
200
200
201 201 model = self.get_json_body()
202 202 if model:
203 203 copy_from = model.get('copy_from')
204 204 if copy_from:
205 205 if model.get('content'):
206 206 raise web.HTTPError(400, "Can't upload and copy at the same time.")
207 207 self._copy_notebook(copy_from, path, name)
208 208 elif self.notebook_manager.notebook_exists(name, path):
209 209 self._save_notebook(model, path, name)
210 210 else:
211 211 self._upload_notebook(model, path, name)
212 212 else:
213 213 self._create_empty_notebook(path, name)
214 214
215 215 @web.authenticated
216 216 @json_errors
217 217 def delete(self, path='', name=None):
218 218 """delete the notebook in the given notebook path"""
219 219 nbm = self.notebook_manager
220 220 nbm.delete_notebook(name, path)
221 221 self.set_status(204)
222 222 self.finish()
223 223
224 224
225 225 class NotebookCheckpointsHandler(IPythonHandler):
226
226
227 227 SUPPORTED_METHODS = ('GET', 'POST')
228
228
229 229 @web.authenticated
230 230 @json_errors
231 231 def get(self, path='', name=None):
232 232 """get lists checkpoints for a notebook"""
233 233 nbm = self.notebook_manager
234 234 checkpoints = nbm.list_checkpoints(name, path)
235 235 data = json.dumps(checkpoints, default=date_default)
236 236 self.finish(data)
237
237
238 238 @web.authenticated
239 239 @json_errors
240 240 def post(self, path='', name=None):
241 241 """post creates a new checkpoint"""
242 242 nbm = self.notebook_manager
243 243 checkpoint = nbm.create_checkpoint(name, path)
244 244 data = json.dumps(checkpoint, default=date_default)
245 245 location = url_path_join(self.base_url, 'api/notebooks',
246 246 path, name, 'checkpoints', checkpoint['id'])
247 247 self.set_header('Location', url_escape(location))
248 248 self.set_status(201)
249 249 self.finish(data)
250 250
251 251
252 252 class ModifyNotebookCheckpointsHandler(IPythonHandler):
253
253
254 254 SUPPORTED_METHODS = ('POST', 'DELETE')
255
255
256 256 @web.authenticated
257 257 @json_errors
258 258 def post(self, path, name, checkpoint_id):
259 259 """post restores a notebook from a checkpoint"""
260 260 nbm = self.notebook_manager
261 261 nbm.restore_checkpoint(checkpoint_id, name, path)
262 262 self.set_status(204)
263 263 self.finish()
264
264
265 265 @web.authenticated
266 266 @json_errors
267 267 def delete(self, path, name, checkpoint_id):
268 268 """delete clears a checkpoint for a given notebook"""
269 269 nbm = self.notebook_manager
270 270 nbm.delete_checkpoint(checkpoint_id, name, path)
271 271 self.set_status(204)
272 272 self.finish()
273
273
274 274 #-----------------------------------------------------------------------------
275 275 # URL to handler mappings
276 276 #-----------------------------------------------------------------------------
277 277
278 278
279 279 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
280 280
281 281 default_handlers = [
282 282 (r"/api/notebooks%s/checkpoints" % notebook_path_regex, NotebookCheckpointsHandler),
283 283 (r"/api/notebooks%s/checkpoints/%s" % (notebook_path_regex, _checkpoint_id_regex),
284 284 ModifyNotebookCheckpointsHandler),
285 285 (r"/api/notebooks%s" % notebook_path_regex, NotebookHandler),
286 286 (r"/api/notebooks%s" % path_regex, NotebookHandler),
287 287 ]
288
@@ -1,287 +1,287 b''
1 1 """A base class notebook manager.
2 2
3 3 Authors:
4 4
5 5 * Brian Granger
6 6 * Zach Sailer
7 7 """
8 8
9 9 #-----------------------------------------------------------------------------
10 10 # Copyright (C) 2011 The IPython Development Team
11 11 #
12 12 # Distributed under the terms of the BSD License. The full license is in
13 13 # the file COPYING, distributed as part of this software.
14 14 #-----------------------------------------------------------------------------
15 15
16 16 #-----------------------------------------------------------------------------
17 17 # Imports
18 18 #-----------------------------------------------------------------------------
19 19
20 20 from fnmatch import fnmatch
21 21 import itertools
22 22 import os
23 23
24 24 from IPython.config.configurable import LoggingConfigurable
25 25 from IPython.nbformat import current, sign
26 26 from IPython.utils.traitlets import Instance, Unicode, List
27 27
28 28 #-----------------------------------------------------------------------------
29 29 # Classes
30 30 #-----------------------------------------------------------------------------
31 31
32 32 class NotebookManager(LoggingConfigurable):
33 33
34 34 filename_ext = Unicode(u'.ipynb')
35
35
36 36 notary = Instance(sign.NotebookNotary)
37 37 def _notary_default(self):
38 38 return sign.NotebookNotary(parent=self)
39
39
40 40 hide_globs = List(Unicode, [u'__pycache__'], config=True, help="""
41 41 Glob patterns to hide in file and directory listings.
42 42 """)
43 43
44 44 # NotebookManager API part 1: methods that must be
45 45 # implemented in subclasses.
46 46
47 47 def path_exists(self, path):
48 48 """Does the API-style path (directory) actually exist?
49
49
50 50 Override this method in subclasses.
51
51
52 52 Parameters
53 53 ----------
54 54 path : string
55 55 The path to check
56
56
57 57 Returns
58 58 -------
59 59 exists : bool
60 60 Whether the path does indeed exist.
61 61 """
62 62 raise NotImplementedError
63 63
64 64 def is_hidden(self, path):
65 65 """Does the API style path correspond to a hidden directory or file?
66
66
67 67 Parameters
68 68 ----------
69 69 path : string
70 70 The path to check. This is an API path (`/` separated,
71 71 relative to base notebook-dir).
72
72
73 73 Returns
74 74 -------
75 75 exists : bool
76 76 Whether the path is hidden.
77
77
78 78 """
79 79 raise NotImplementedError
80 80
81 81 def notebook_exists(self, name, path=''):
82 82 """Returns a True if the notebook exists. Else, returns False.
83 83
84 84 Parameters
85 85 ----------
86 86 name : string
87 87 The name of the notebook you are checking.
88 88 path : string
89 89 The relative path to the notebook (with '/' as separator)
90 90
91 91 Returns
92 92 -------
93 93 bool
94 94 """
95 95 raise NotImplementedError('must be implemented in a subclass')
96 96
97 97 # TODO: Remove this after we create the contents web service and directories are
98 98 # no longer listed by the notebook web service.
99 99 def list_dirs(self, path):
100 100 """List the directory models for a given API style path."""
101 101 raise NotImplementedError('must be implemented in a subclass')
102 102
103 103 # TODO: Remove this after we create the contents web service and directories are
104 104 # no longer listed by the notebook web service.
105 105 def get_dir_model(self, name, path=''):
106 106 """Get the directory model given a directory name and its API style path.
107
107
108 108 The keys in the model should be:
109 109 * name
110 110 * path
111 111 * last_modified
112 112 * created
113 113 * type='directory'
114 114 """
115 115 raise NotImplementedError('must be implemented in a subclass')
116 116
117 117 def list_notebooks(self, path=''):
118 118 """Return a list of notebook dicts without content.
119 119
120 120 This returns a list of dicts, each of the form::
121 121
122 122 dict(notebook_id=notebook,name=name)
123 123
124 124 This list of dicts should be sorted by name::
125 125
126 126 data = sorted(data, key=lambda item: item['name'])
127 127 """
128 128 raise NotImplementedError('must be implemented in a subclass')
129 129
130 130 def get_notebook(self, name, path='', content=True):
131 131 """Get the notebook model with or without content."""
132 132 raise NotImplementedError('must be implemented in a subclass')
133 133
134 134 def save_notebook(self, model, name, path=''):
135 135 """Save the notebook and return the model with no content."""
136 136 raise NotImplementedError('must be implemented in a subclass')
137 137
138 138 def update_notebook(self, model, name, path=''):
139 139 """Update the notebook and return the model with no content."""
140 140 raise NotImplementedError('must be implemented in a subclass')
141 141
142 142 def delete_notebook(self, name, path=''):
143 143 """Delete notebook by name and path."""
144 144 raise NotImplementedError('must be implemented in a subclass')
145 145
146 146 def create_checkpoint(self, name, path=''):
147 147 """Create a checkpoint of the current state of a notebook
148
148
149 149 Returns a checkpoint_id for the new checkpoint.
150 150 """
151 151 raise NotImplementedError("must be implemented in a subclass")
152
152
153 153 def list_checkpoints(self, name, path=''):
154 154 """Return a list of checkpoints for a given notebook"""
155 155 return []
156
156
157 157 def restore_checkpoint(self, checkpoint_id, name, path=''):
158 158 """Restore a notebook from one of its checkpoints"""
159 159 raise NotImplementedError("must be implemented in a subclass")
160 160
161 161 def delete_checkpoint(self, checkpoint_id, name, path=''):
162 162 """delete a checkpoint for a notebook"""
163 163 raise NotImplementedError("must be implemented in a subclass")
164
164
165 165 def info_string(self):
166 166 return "Serving notebooks"
167 167
168 168 # NotebookManager API part 2: methods that have useable default
169 169 # implementations, but can be overridden in subclasses.
170 170
171 171 def get_kernel_path(self, name, path='', model=None):
172 172 """ Return the path to start kernel in """
173 173 return path
174 174
175 175 def increment_filename(self, basename, path=''):
176 176 """Increment a notebook filename without the .ipynb to make it unique.
177
177
178 178 Parameters
179 179 ----------
180 180 basename : unicode
181 181 The name of a notebook without the ``.ipynb`` file extension.
182 182 path : unicode
183 183 The URL path of the notebooks directory
184 184
185 185 Returns
186 186 -------
187 187 name : unicode
188 188 A notebook name (with the .ipynb extension) that starts
189 189 with basename and does not refer to any existing notebook.
190 190 """
191 191 path = path.strip('/')
192 192 for i in itertools.count():
193 193 name = u'{basename}{i}{ext}'.format(basename=basename, i=i,
194 194 ext=self.filename_ext)
195 195 if not self.notebook_exists(name, path):
196 196 break
197 197 return name
198 198
199 199 def create_notebook(self, model=None, path=''):
200 200 """Create a new notebook and return its model with no content."""
201 201 path = path.strip('/')
202 202 if model is None:
203 203 model = {}
204 204 if 'content' not in model:
205 205 metadata = current.new_metadata(name=u'')
206 206 model['content'] = current.new_notebook(metadata=metadata)
207 207 if 'name' not in model:
208 208 model['name'] = self.increment_filename('Untitled', path)
209
209
210 210 model['path'] = path
211 211 model = self.save_notebook(model, model['name'], model['path'])
212 212 return model
213 213
214 214 def copy_notebook(self, from_name, to_name=None, path=''):
215 215 """Copy an existing notebook and return its new model.
216
216
217 217 If to_name not specified, increment `from_name-Copy#.ipynb`.
218 218 """
219 219 path = path.strip('/')
220 220 model = self.get_notebook(from_name, path)
221 221 if not to_name:
222 222 base = os.path.splitext(from_name)[0] + '-Copy'
223 223 to_name = self.increment_filename(base, path)
224 224 model['name'] = to_name
225 225 model = self.save_notebook(model, to_name, path)
226 226 return model
227
227
228 228 def log_info(self):
229 229 self.log.info(self.info_string())
230 230
231 231 def trust_notebook(self, name, path=''):
232 232 """Explicitly trust a notebook
233
233
234 234 Parameters
235 235 ----------
236 236 name : string
237 237 The filename of the notebook
238 238 path : string
239 239 The notebook's directory
240 240 """
241 241 model = self.get_notebook(name, path)
242 242 nb = model['content']
243 243 self.log.warn("Trusting notebook %s/%s", path, name)
244 244 self.notary.mark_cells(nb, True)
245 245 self.save_notebook(model, name, path)
246
246
247 247 def check_and_sign(self, nb, name, path=''):
248 248 """Check for trusted cells, and sign the notebook.
249
249
250 250 Called as a part of saving notebooks.
251
251
252 252 Parameters
253 253 ----------
254 254 nb : dict
255 255 The notebook structure
256 256 name : string
257 257 The filename of the notebook
258 258 path : string
259 259 The notebook's directory
260 260 """
261 261 if self.notary.check_cells(nb):
262 262 self.notary.sign(nb)
263 263 else:
264 264 self.log.warn("Saving untrusted notebook %s/%s", path, name)
265
265
266 266 def mark_trusted_cells(self, nb, name, path=''):
267 267 """Mark cells as trusted if the notebook signature matches.
268
268
269 269 Called as a part of loading notebooks.
270
270
271 271 Parameters
272 272 ----------
273 273 nb : dict
274 274 The notebook structure
275 275 name : string
276 276 The filename of the notebook
277 277 path : string
278 278 The notebook's directory
279 279 """
280 280 trusted = self.notary.check_signature(nb)
281 281 if not trusted:
282 282 self.log.warn("Notebook %s/%s is not trusted", path, name)
283 283 self.notary.mark_cells(nb, trusted)
284 284
285 285 def should_list(self, name):
286 286 """Should this file/directory name be displayed in a listing?"""
287 287 return not any(fnmatch(name, glob) for glob in self.hide_globs)
1 NO CONTENT: file renamed from IPython/html/services/notebooks/tests/__init__.py to IPython/html/services/contents/tests/__init__.py
@@ -1,320 +1,320 b''
1 1 # coding: utf-8
2 2 """Tests for the notebook manager."""
3 3 from __future__ import print_function
4 4
5 5 import logging
6 6 import os
7 7
8 8 from tornado.web import HTTPError
9 9 from unittest import TestCase
10 10 from tempfile import NamedTemporaryFile
11 11
12 12 from IPython.nbformat import current
13 13
14 14 from IPython.utils.tempdir import TemporaryDirectory
15 15 from IPython.utils.traitlets import TraitError
16 16 from IPython.html.utils import url_path_join
17 17
18 18 from ..filenbmanager import FileNotebookManager
19 19 from ..nbmanager import NotebookManager
20 20
21 21
22 22 class TestFileNotebookManager(TestCase):
23 23
24 24 def test_nb_dir(self):
25 25 with TemporaryDirectory() as td:
26 26 fm = FileNotebookManager(notebook_dir=td)
27 27 self.assertEqual(fm.notebook_dir, td)
28 28
29 29 def test_missing_nb_dir(self):
30 30 with TemporaryDirectory() as td:
31 31 nbdir = os.path.join(td, 'notebook', 'dir', 'is', 'missing')
32 32 self.assertRaises(TraitError, FileNotebookManager, notebook_dir=nbdir)
33 33
34 34 def test_invalid_nb_dir(self):
35 35 with NamedTemporaryFile() as tf:
36 36 self.assertRaises(TraitError, FileNotebookManager, notebook_dir=tf.name)
37 37
38 38 def test_get_os_path(self):
39 39 # full filesystem path should be returned with correct operating system
40 40 # separators.
41 41 with TemporaryDirectory() as td:
42 42 nbdir = td
43 43 fm = FileNotebookManager(notebook_dir=nbdir)
44 44 path = fm._get_os_path('test.ipynb', '/path/to/notebook/')
45 45 rel_path_list = '/path/to/notebook/test.ipynb'.split('/')
46 46 fs_path = os.path.join(fm.notebook_dir, *rel_path_list)
47 47 self.assertEqual(path, fs_path)
48 48
49 49 fm = FileNotebookManager(notebook_dir=nbdir)
50 50 path = fm._get_os_path('test.ipynb')
51 51 fs_path = os.path.join(fm.notebook_dir, 'test.ipynb')
52 52 self.assertEqual(path, fs_path)
53 53
54 54 fm = FileNotebookManager(notebook_dir=nbdir)
55 55 path = fm._get_os_path('test.ipynb', '////')
56 56 fs_path = os.path.join(fm.notebook_dir, 'test.ipynb')
57 57 self.assertEqual(path, fs_path)
58
58
59 59 def test_checkpoint_subdir(self):
60 60 subd = u'sub βˆ‚ir'
61 61 cp_name = 'test-cp.ipynb'
62 62 with TemporaryDirectory() as td:
63 63 nbdir = td
64 64 os.mkdir(os.path.join(td, subd))
65 65 fm = FileNotebookManager(notebook_dir=nbdir)
66 66 cp_dir = fm.get_checkpoint_path('cp', 'test.ipynb', '/')
67 67 cp_subdir = fm.get_checkpoint_path('cp', 'test.ipynb', '/%s/' % subd)
68 68 self.assertNotEqual(cp_dir, cp_subdir)
69 69 self.assertEqual(cp_dir, os.path.join(nbdir, fm.checkpoint_dir, cp_name))
70 70 self.assertEqual(cp_subdir, os.path.join(nbdir, subd, fm.checkpoint_dir, cp_name))
71
71
72 72
73 73 class TestNotebookManager(TestCase):
74
74
75 75 def setUp(self):
76 76 self._temp_dir = TemporaryDirectory()
77 77 self.td = self._temp_dir.name
78 78 self.notebook_manager = FileNotebookManager(
79 79 notebook_dir=self.td,
80 80 log=logging.getLogger()
81 81 )
82
82
83 83 def tearDown(self):
84 84 self._temp_dir.cleanup()
85
85
86 86 def make_dir(self, abs_path, rel_path):
87 87 """make subdirectory, rel_path is the relative path
88 88 to that directory from the location where the server started"""
89 89 os_path = os.path.join(abs_path, rel_path)
90 90 try:
91 91 os.makedirs(os_path)
92 92 except OSError:
93 93 print("Directory already exists: %r" % os_path)
94
94
95 95 def add_code_cell(self, nb):
96 96 output = current.new_output("display_data", output_javascript="alert('hi');")
97 97 cell = current.new_code_cell("print('hi')", outputs=[output])
98 98 if not nb.worksheets:
99 99 nb.worksheets.append(current.new_worksheet())
100 100 nb.worksheets[0].cells.append(cell)
101
101
102 102 def new_notebook(self):
103 103 nbm = self.notebook_manager
104 104 model = nbm.create_notebook()
105 105 name = model['name']
106 106 path = model['path']
107
107
108 108 full_model = nbm.get_notebook(name, path)
109 109 nb = full_model['content']
110 110 self.add_code_cell(nb)
111
111
112 112 nbm.save_notebook(full_model, name, path)
113 113 return nb, name, path
114
114
115 115 def test_create_notebook(self):
116 116 nm = self.notebook_manager
117 117 # Test in root directory
118 118 model = nm.create_notebook()
119 119 assert isinstance(model, dict)
120 120 self.assertIn('name', model)
121 121 self.assertIn('path', model)
122 122 self.assertEqual(model['name'], 'Untitled0.ipynb')
123 123 self.assertEqual(model['path'], '')
124 124
125 125 # Test in sub-directory
126 126 sub_dir = '/foo/'
127 127 self.make_dir(nm.notebook_dir, 'foo')
128 128 model = nm.create_notebook(None, sub_dir)
129 129 assert isinstance(model, dict)
130 130 self.assertIn('name', model)
131 131 self.assertIn('path', model)
132 132 self.assertEqual(model['name'], 'Untitled0.ipynb')
133 133 self.assertEqual(model['path'], sub_dir.strip('/'))
134 134
135 135 def test_get_notebook(self):
136 136 nm = self.notebook_manager
137 137 # Create a notebook
138 138 model = nm.create_notebook()
139 139 name = model['name']
140 140 path = model['path']
141 141
142 142 # Check that we 'get' on the notebook we just created
143 143 model2 = nm.get_notebook(name, path)
144 144 assert isinstance(model2, dict)
145 145 self.assertIn('name', model2)
146 146 self.assertIn('path', model2)
147 147 self.assertEqual(model['name'], name)
148 148 self.assertEqual(model['path'], path)
149 149
150 150 # Test in sub-directory
151 151 sub_dir = '/foo/'
152 152 self.make_dir(nm.notebook_dir, 'foo')
153 153 model = nm.create_notebook(None, sub_dir)
154 154 model2 = nm.get_notebook(name, sub_dir)
155 155 assert isinstance(model2, dict)
156 156 self.assertIn('name', model2)
157 157 self.assertIn('path', model2)
158 158 self.assertIn('content', model2)
159 159 self.assertEqual(model2['name'], 'Untitled0.ipynb')
160 160 self.assertEqual(model2['path'], sub_dir.strip('/'))
161
161
162 162 def test_update_notebook(self):
163 163 nm = self.notebook_manager
164 164 # Create a notebook
165 165 model = nm.create_notebook()
166 166 name = model['name']
167 167 path = model['path']
168 168
169 169 # Change the name in the model for rename
170 170 model['name'] = 'test.ipynb'
171 171 model = nm.update_notebook(model, name, path)
172 172 assert isinstance(model, dict)
173 173 self.assertIn('name', model)
174 174 self.assertIn('path', model)
175 175 self.assertEqual(model['name'], 'test.ipynb')
176 176
177 177 # Make sure the old name is gone
178 178 self.assertRaises(HTTPError, nm.get_notebook, name, path)
179 179
180 180 # Test in sub-directory
181 181 # Create a directory and notebook in that directory
182 182 sub_dir = '/foo/'
183 183 self.make_dir(nm.notebook_dir, 'foo')
184 184 model = nm.create_notebook(None, sub_dir)
185 185 name = model['name']
186 186 path = model['path']
187
187
188 188 # Change the name in the model for rename
189 189 model['name'] = 'test_in_sub.ipynb'
190 190 model = nm.update_notebook(model, name, path)
191 191 assert isinstance(model, dict)
192 192 self.assertIn('name', model)
193 193 self.assertIn('path', model)
194 194 self.assertEqual(model['name'], 'test_in_sub.ipynb')
195 195 self.assertEqual(model['path'], sub_dir.strip('/'))
196
196
197 197 # Make sure the old name is gone
198 198 self.assertRaises(HTTPError, nm.get_notebook, name, path)
199 199
200 200 def test_save_notebook(self):
201 201 nm = self.notebook_manager
202 202 # Create a notebook
203 203 model = nm.create_notebook()
204 204 name = model['name']
205 205 path = model['path']
206 206
207 207 # Get the model with 'content'
208 208 full_model = nm.get_notebook(name, path)
209 209
210 210 # Save the notebook
211 211 model = nm.save_notebook(full_model, name, path)
212 212 assert isinstance(model, dict)
213 213 self.assertIn('name', model)
214 214 self.assertIn('path', model)
215 215 self.assertEqual(model['name'], name)
216 216 self.assertEqual(model['path'], path)
217 217
218 218 # Test in sub-directory
219 219 # Create a directory and notebook in that directory
220 220 sub_dir = '/foo/'
221 221 self.make_dir(nm.notebook_dir, 'foo')
222 222 model = nm.create_notebook(None, sub_dir)
223 223 name = model['name']
224 224 path = model['path']
225 225 model = nm.get_notebook(name, path)
226 226
227 227 # Change the name in the model for rename
228 228 model = nm.save_notebook(model, name, path)
229 229 assert isinstance(model, dict)
230 230 self.assertIn('name', model)
231 231 self.assertIn('path', model)
232 232 self.assertEqual(model['name'], 'Untitled0.ipynb')
233 233 self.assertEqual(model['path'], sub_dir.strip('/'))
234 234
235 235 def test_save_notebook_with_script(self):
236 236 nm = self.notebook_manager
237 237 # Create a notebook
238 238 model = nm.create_notebook()
239 239 nm.save_script = True
240 240 model = nm.create_notebook()
241 241 name = model['name']
242 242 path = model['path']
243 243
244 244 # Get the model with 'content'
245 245 full_model = nm.get_notebook(name, path)
246 246
247 247 # Save the notebook
248 248 model = nm.save_notebook(full_model, name, path)
249 249
250 250 # Check that the script was created
251 251 py_path = os.path.join(nm.notebook_dir, os.path.splitext(name)[0]+'.py')
252 252 assert os.path.exists(py_path), py_path
253 253
254 254 def test_delete_notebook(self):
255 255 nm = self.notebook_manager
256 256 # Create a notebook
257 257 nb, name, path = self.new_notebook()
258
258
259 259 # Delete the notebook
260 260 nm.delete_notebook(name, path)
261
261
262 262 # Check that a 'get' on the deleted notebook raises and error
263 263 self.assertRaises(HTTPError, nm.get_notebook, name, path)
264
264
265 265 def test_copy_notebook(self):
266 266 nm = self.notebook_manager
267 267 path = u'Γ₯ b'
268 268 name = u'nb √.ipynb'
269 269 os.mkdir(os.path.join(nm.notebook_dir, path))
270 270 orig = nm.create_notebook({'name' : name}, path=path)
271
271
272 272 # copy with unspecified name
273 273 copy = nm.copy_notebook(name, path=path)
274 274 self.assertEqual(copy['name'], orig['name'].replace('.ipynb', '-Copy0.ipynb'))
275
275
276 276 # copy with specified name
277 277 copy2 = nm.copy_notebook(name, u'copy 2.ipynb', path=path)
278 278 self.assertEqual(copy2['name'], u'copy 2.ipynb')
279
279
280 280 def test_trust_notebook(self):
281 281 nbm = self.notebook_manager
282 282 nb, name, path = self.new_notebook()
283
283
284 284 untrusted = nbm.get_notebook(name, path)['content']
285 285 assert not nbm.notary.check_cells(untrusted)
286
286
287 287 # print(untrusted)
288 288 nbm.trust_notebook(name, path)
289 289 trusted = nbm.get_notebook(name, path)['content']
290 290 # print(trusted)
291 291 assert nbm.notary.check_cells(trusted)
292
292
293 293 def test_mark_trusted_cells(self):
294 294 nbm = self.notebook_manager
295 295 nb, name, path = self.new_notebook()
296
296
297 297 nbm.mark_trusted_cells(nb, name, path)
298 298 for cell in nb.worksheets[0].cells:
299 299 if cell.cell_type == 'code':
300 300 assert not cell.trusted
301
301
302 302 nbm.trust_notebook(name, path)
303 303 nb = nbm.get_notebook(name, path)['content']
304 304 for cell in nb.worksheets[0].cells:
305 305 if cell.cell_type == 'code':
306 306 assert cell.trusted
307 307
308 308 def test_check_and_sign(self):
309 309 nbm = self.notebook_manager
310 310 nb, name, path = self.new_notebook()
311
311
312 312 nbm.mark_trusted_cells(nb, name, path)
313 313 nbm.check_and_sign(nb, name, path)
314 314 assert not nbm.notary.check_signature(nb)
315
315
316 316 nbm.trust_notebook(name, path)
317 317 nb = nbm.get_notebook(name, path)['content']
318 318 nbm.mark_trusted_cells(nb, name, path)
319 319 nbm.check_and_sign(nb, name, path)
320 320 assert nbm.notary.check_signature(nb)
@@ -1,347 +1,346 b''
1 1 # coding: utf-8
2 2 """Test the notebooks webservice API."""
3 3
4 4 import io
5 5 import json
6 6 import os
7 7 import shutil
8 8 from unicodedata import normalize
9 9
10 10 pjoin = os.path.join
11 11
12 12 import requests
13 13
14 14 from IPython.html.utils import url_path_join, url_escape
15 15 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
16 16 from IPython.nbformat import current
17 17 from IPython.nbformat.current import (new_notebook, write, read, new_worksheet,
18 18 new_heading_cell, to_notebook_json)
19 19 from IPython.nbformat import v2
20 20 from IPython.utils import py3compat
21 21 from IPython.utils.data import uniq_stable
22 22
23 23
24 24 # TODO: Remove this after we create the contents web service and directories are
25 25 # no longer listed by the notebook web service.
26 26 def notebooks_only(nb_list):
27 27 return [nb for nb in nb_list if nb['type']=='notebook']
28 28
29 29 def dirs_only(nb_list):
30 30 return [x for x in nb_list if x['type']=='directory']
31 31
32 32
33 33 class NBAPI(object):
34 34 """Wrapper for notebook 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/notebooks', 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, name, path='/'):
50 50 return self._req('GET', url_path_join(path, name))
51 51
52 52 def create_untitled(self, path='/'):
53 53 return self._req('POST', path)
54 54
55 55 def upload_untitled(self, body, path='/'):
56 56 return self._req('POST', path, body)
57 57
58 58 def copy_untitled(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, name, path='/'):
63 63 return self._req('PUT', url_path_join(path, name))
64 64
65 65 def upload(self, name, body, path='/'):
66 66 return self._req('PUT', url_path_join(path, name), body)
67 67
68 68 def copy(self, copy_from, copy_to, path='/'):
69 69 body = json.dumps({'copy_from':copy_from})
70 70 return self._req('PUT', url_path_join(path, copy_to), body)
71 71
72 72 def save(self, name, body, path='/'):
73 73 return self._req('PUT', url_path_join(path, name), body)
74 74
75 75 def delete(self, name, path='/'):
76 76 return self._req('DELETE', url_path_join(path, name))
77 77
78 78 def rename(self, name, path, new_name):
79 79 body = json.dumps({'name': new_name})
80 80 return self._req('PATCH', url_path_join(path, name), body)
81 81
82 82 def get_checkpoints(self, name, path):
83 83 return self._req('GET', url_path_join(path, name, 'checkpoints'))
84 84
85 85 def new_checkpoint(self, name, path):
86 86 return self._req('POST', url_path_join(path, name, 'checkpoints'))
87 87
88 88 def restore_checkpoint(self, name, path, checkpoint_id):
89 89 return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
90 90
91 91 def delete_checkpoint(self, name, path, checkpoint_id):
92 92 return self._req('DELETE', url_path_join(path, name, 'checkpoints', checkpoint_id))
93 93
94 94 class APITest(NotebookTestBase):
95 95 """Test the kernels web service API"""
96 96 dirs_nbs = [('', 'inroot'),
97 97 ('Directory with spaces in', 'inspace'),
98 98 (u'unicodΓ©', 'innonascii'),
99 99 ('foo', 'a'),
100 100 ('foo', 'b'),
101 101 ('foo', 'name with spaces'),
102 102 ('foo', u'unicodΓ©'),
103 103 ('foo/bar', 'baz'),
104 104 ('ordering', 'A'),
105 105 ('ordering', 'b'),
106 106 ('ordering', 'C'),
107 107 (u'Γ₯ b', u'Γ§ d'),
108 108 ]
109 109 hidden_dirs = ['.hidden', '__pycache__']
110 110
111 111 dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
112 112 del dirs[0] # remove ''
113 113 top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
114 114
115 115 def setUp(self):
116 116 nbdir = self.notebook_dir.name
117 117
118 118 for d in (self.dirs + self.hidden_dirs):
119 119 d.replace('/', os.sep)
120 120 if not os.path.isdir(pjoin(nbdir, d)):
121 121 os.mkdir(pjoin(nbdir, d))
122 122
123 123 for d, name in self.dirs_nbs:
124 124 d = d.replace('/', os.sep)
125 125 with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
126 126 encoding='utf-8') as f:
127 127 nb = new_notebook(name=name)
128 128 write(nb, f, format='ipynb')
129 129
130 130 self.nb_api = NBAPI(self.base_url())
131 131
132 132 def tearDown(self):
133 133 nbdir = self.notebook_dir.name
134 134
135 135 for dname in (list(self.top_level_dirs) + self.hidden_dirs):
136 136 shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
137 137
138 138 if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
139 139 os.unlink(pjoin(nbdir, 'inroot.ipynb'))
140 140
141 141 def test_list_notebooks(self):
142 142 nbs = notebooks_only(self.nb_api.list().json())
143 143 self.assertEqual(len(nbs), 1)
144 144 self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
145 145
146 146 nbs = notebooks_only(self.nb_api.list('/Directory with spaces in/').json())
147 147 self.assertEqual(len(nbs), 1)
148 148 self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
149 149
150 150 nbs = notebooks_only(self.nb_api.list(u'/unicodΓ©/').json())
151 151 self.assertEqual(len(nbs), 1)
152 152 self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
153 153 self.assertEqual(nbs[0]['path'], u'unicodΓ©')
154 154
155 155 nbs = notebooks_only(self.nb_api.list('/foo/bar/').json())
156 156 self.assertEqual(len(nbs), 1)
157 157 self.assertEqual(nbs[0]['name'], 'baz.ipynb')
158 158 self.assertEqual(nbs[0]['path'], 'foo/bar')
159 159
160 160 nbs = notebooks_only(self.nb_api.list('foo').json())
161 161 self.assertEqual(len(nbs), 4)
162 162 nbnames = { normalize('NFC', n['name']) for n in nbs }
163 163 expected = [ u'a.ipynb', u'b.ipynb', u'name with spaces.ipynb', u'unicodΓ©.ipynb']
164 164 expected = { normalize('NFC', name) for name in expected }
165 165 self.assertEqual(nbnames, expected)
166
166
167 167 nbs = notebooks_only(self.nb_api.list('ordering').json())
168 168 nbnames = [n['name'] for n in nbs]
169 169 expected = ['A.ipynb', 'b.ipynb', 'C.ipynb']
170 170 self.assertEqual(nbnames, expected)
171 171
172 172 def test_list_dirs(self):
173 173 dirs = dirs_only(self.nb_api.list().json())
174 174 dir_names = {normalize('NFC', d['name']) for d in dirs}
175 175 self.assertEqual(dir_names, self.top_level_dirs) # Excluding hidden dirs
176 176
177 177 def test_list_nonexistant_dir(self):
178 178 with assert_http_error(404):
179 179 self.nb_api.list('nonexistant')
180 180
181 181 def test_get_contents(self):
182 182 for d, name in self.dirs_nbs:
183 183 nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
184 184 self.assertEqual(nb['name'], u'%s.ipynb' % name)
185 185 self.assertIn('content', nb)
186 186 self.assertIn('metadata', nb['content'])
187 187 self.assertIsInstance(nb['content']['metadata'], dict)
188 188
189 189 # Name that doesn't exist - should be a 404
190 190 with assert_http_error(404):
191 191 self.nb_api.read('q.ipynb', 'foo')
192 192
193 193 def _check_nb_created(self, resp, name, path):
194 194 self.assertEqual(resp.status_code, 201)
195 195 location_header = py3compat.str_to_unicode(resp.headers['Location'])
196 196 self.assertEqual(location_header, url_escape(url_path_join(u'/api/notebooks', path, name)))
197 197 self.assertEqual(resp.json()['name'], name)
198 198 assert os.path.isfile(pjoin(
199 199 self.notebook_dir.name,
200 200 path.replace('/', os.sep),
201 201 name,
202 202 ))
203 203
204 204 def test_create_untitled(self):
205 205 resp = self.nb_api.create_untitled(path=u'Γ₯ b')
206 206 self._check_nb_created(resp, 'Untitled0.ipynb', u'Γ₯ b')
207 207
208 208 # Second time
209 209 resp = self.nb_api.create_untitled(path=u'Γ₯ b')
210 210 self._check_nb_created(resp, 'Untitled1.ipynb', u'Γ₯ b')
211 211
212 212 # And two directories down
213 213 resp = self.nb_api.create_untitled(path='foo/bar')
214 214 self._check_nb_created(resp, 'Untitled0.ipynb', 'foo/bar')
215 215
216 216 def test_upload_untitled(self):
217 217 nb = new_notebook(name='Upload test')
218 218 nbmodel = {'content': nb}
219 219 resp = self.nb_api.upload_untitled(path=u'Γ₯ b',
220 220 body=json.dumps(nbmodel))
221 221 self._check_nb_created(resp, 'Untitled0.ipynb', u'Γ₯ b')
222 222
223 223 def test_upload(self):
224 224 nb = new_notebook(name=u'ignored')
225 225 nbmodel = {'content': nb}
226 226 resp = self.nb_api.upload(u'Upload tΓ©st.ipynb', path=u'Γ₯ b',
227 227 body=json.dumps(nbmodel))
228 228 self._check_nb_created(resp, u'Upload tΓ©st.ipynb', u'Γ₯ b')
229 229
230 230 def test_upload_v2(self):
231 231 nb = v2.new_notebook()
232 232 ws = v2.new_worksheet()
233 233 nb.worksheets.append(ws)
234 234 ws.cells.append(v2.new_code_cell(input='print("hi")'))
235 235 nbmodel = {'content': nb}
236 236 resp = self.nb_api.upload(u'Upload tΓ©st.ipynb', path=u'Γ₯ b',
237 237 body=json.dumps(nbmodel))
238 238 self._check_nb_created(resp, u'Upload tΓ©st.ipynb', u'Γ₯ b')
239 239 resp = self.nb_api.read(u'Upload tΓ©st.ipynb', u'Γ₯ b')
240 240 data = resp.json()
241 241 self.assertEqual(data['content']['nbformat'], current.nbformat)
242 242 self.assertEqual(data['content']['orig_nbformat'], 2)
243 243
244 244 def test_copy_untitled(self):
245 245 resp = self.nb_api.copy_untitled(u'Γ§ d.ipynb', path=u'Γ₯ b')
246 246 self._check_nb_created(resp, u'Γ§ d-Copy0.ipynb', u'Γ₯ b')
247 247
248 248 def test_copy(self):
249 249 resp = self.nb_api.copy(u'Γ§ d.ipynb', u'cΓΈpy.ipynb', path=u'Γ₯ b')
250 250 self._check_nb_created(resp, u'cΓΈpy.ipynb', u'Γ₯ b')
251 251
252 252 def test_delete(self):
253 253 for d, name in self.dirs_nbs:
254 254 resp = self.nb_api.delete('%s.ipynb' % name, d)
255 255 self.assertEqual(resp.status_code, 204)
256 256
257 257 for d in self.dirs + ['/']:
258 258 nbs = notebooks_only(self.nb_api.list(d).json())
259 259 self.assertEqual(len(nbs), 0)
260 260
261 261 def test_rename(self):
262 262 resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
263 263 self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
264 264 self.assertEqual(resp.json()['name'], 'z.ipynb')
265 265 assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
266 266
267 267 nbs = notebooks_only(self.nb_api.list('foo').json())
268 268 nbnames = set(n['name'] for n in nbs)
269 269 self.assertIn('z.ipynb', nbnames)
270 270 self.assertNotIn('a.ipynb', nbnames)
271 271
272 272 def test_rename_existing(self):
273 273 with assert_http_error(409):
274 274 self.nb_api.rename('a.ipynb', 'foo', 'b.ipynb')
275 275
276 276 def test_save(self):
277 277 resp = self.nb_api.read('a.ipynb', 'foo')
278 278 nbcontent = json.loads(resp.text)['content']
279 279 nb = to_notebook_json(nbcontent)
280 280 ws = new_worksheet()
281 281 nb.worksheets = [ws]
282 282 ws.cells.append(new_heading_cell(u'Created by test Β³'))
283 283
284 284 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
285 285 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
286 286
287 287 nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
288 288 with io.open(nbfile, 'r', encoding='utf-8') as f:
289 289 newnb = read(f, format='ipynb')
290 290 self.assertEqual(newnb.worksheets[0].cells[0].source,
291 291 u'Created by test Β³')
292 292 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
293 293 newnb = to_notebook_json(nbcontent)
294 294 self.assertEqual(newnb.worksheets[0].cells[0].source,
295 295 u'Created by test Β³')
296 296
297 297 # Save and rename
298 298 nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
299 299 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
300 300 saved = resp.json()
301 301 self.assertEqual(saved['name'], 'a2.ipynb')
302 302 self.assertEqual(saved['path'], 'foo/bar')
303 303 assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
304 304 assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
305 305 with assert_http_error(404):
306 306 self.nb_api.read('a.ipynb', 'foo')
307 307
308 308 def test_checkpoints(self):
309 309 resp = self.nb_api.read('a.ipynb', 'foo')
310 310 r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
311 311 self.assertEqual(r.status_code, 201)
312 312 cp1 = r.json()
313 313 self.assertEqual(set(cp1), {'id', 'last_modified'})
314 314 self.assertEqual(r.headers['Location'].split('/')[-1], cp1['id'])
315 315
316 316 # Modify it
317 317 nbcontent = json.loads(resp.text)['content']
318 318 nb = to_notebook_json(nbcontent)
319 319 ws = new_worksheet()
320 320 nb.worksheets = [ws]
321 321 hcell = new_heading_cell('Created by test')
322 322 ws.cells.append(hcell)
323 323 # Save
324 324 nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
325 325 resp = self.nb_api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
326 326
327 327 # List checkpoints
328 328 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
329 329 self.assertEqual(cps, [cp1])
330 330
331 331 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
332 332 nb = to_notebook_json(nbcontent)
333 333 self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
334 334
335 335 # Restore cp1
336 336 r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['id'])
337 337 self.assertEqual(r.status_code, 204)
338 338 nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
339 339 nb = to_notebook_json(nbcontent)
340 340 self.assertEqual(nb.worksheets, [])
341 341
342 342 # Delete cp1
343 343 r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['id'])
344 344 self.assertEqual(r.status_code, 204)
345 345 cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
346 346 self.assertEqual(cps, [])
347
General Comments 0
You need to be logged in to leave comments. Login now