Merge pull request #7575 from minrk/content-false

allow requesting contents without body

IPython/html/services/contents/handlers.py

@@ -1,336 +1,342
 """Tornado handlers for the contents web service."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import json
 
 from tornado import gen, web
 
 from IPython.html.utils import url_path_join, url_escape
 from IPython.utils.jsonutil import date_default
 
 from IPython.html.base.handlers import (
     IPythonHandler, json_errors, path_regex,
 )
 
 
 def sort_key(model):
     """key function for case-insensitive sort by name and type"""
     iname = model['name'].lower()
     type_key = {
         'directory' : '0',
         'notebook'  : '1',
         'file'      : '2',
     }.get(model['type'], '9')
     return u'%s%s' % (type_key, iname)
 
 
 def validate_model(model, expect_content):
     """
     Validate a model returned by a ContentsManager method.
 
     If expect_content is True, then we expect non-null entries for 'content'
     and 'format'.
     """
     required_keys = {
         "name",
         "path",
         "type",
         "writable",
         "created",
         "last_modified",
         "mimetype",
         "content",
         "format",
     }
     missing = required_keys - set(model.keys())
     if missing:
         raise web.HTTPError(
             500,
             u"Missing Model Keys: {missing}".format(missing=missing),
         )
 
     maybe_none_keys = ['content', 'format']
     if model['type'] == 'file':
         # mimetype should be populated only for file models
         maybe_none_keys.append('mimetype')
     if expect_content:
         errors = [key for key in maybe_none_keys if model[key] is None]
         if errors:
             raise web.HTTPError(
                 500,
                 u"Keys unexpectedly None: {keys}".format(keys=errors),
             )
     else:
         errors = {
             key: model[key]
             for key in maybe_none_keys
             if model[key] is not None
         }
         if errors:
             raise web.HTTPError(
                 500,
                 u"Keys unexpectedly not None: {keys}".format(keys=errors),
             )
 
 
 class ContentsHandler(IPythonHandler):
 
     SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
 
     def location_url(self, path):
         """Return the full URL location of a file.
 
         Parameters
         ----------
         path : unicode
             The API path of the file, such as "foo/bar.txt".
         """
         return url_escape(url_path_join(
             self.base_url, 'api', 'contents', path
         ))
 
     def _finish_model(self, model, location=True):
         """Finish a JSON request with a model, setting relevant headers, etc."""
         if location:
             location = self.location_url(model['path'])
             self.set_header('Location', location)
         self.set_header('Last-Modified', model['last_modified'])
         self.set_header('Content-Type', 'application/json')
         self.finish(json.dumps(model, default=date_default))
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def get(self, path=''):
         """Return a model for a file or directory.
 
         A directory model contains a list of models (without content)
         of the files and directories it contains.
         """
         path = path or ''
         type = self.get_query_argument('type', default=None)
         if type not in {None, 'directory', 'file', 'notebook'}:
             raise web.HTTPError(400, u'Type %r is invalid' % type)
 
         format = self.get_query_argument('format', default=None)
         if format not in {None, 'text', 'base64'}:
             raise web.HTTPError(400, u'Format %r is invalid' % format)
-
-        model = yield gen.maybe_future(self.contents_manager.get(path=path, type=type, format=format))
-        if model['type'] == 'directory':
+        content = self.get_query_argument('content', default='1')
+        if content not in {'0', '1'}:
+            raise web.HTTPError(400, u'Content %r is invalid' % content)
+        content = int(content)
+        
+        model = yield gen.maybe_future(self.contents_manager.get(
+            path=path, type=type, format=format, content=content,
+        ))
+        if model['type'] == 'directory' and content:
             # group listing by type, then by name (case-insensitive)
             # FIXME: sorting should be done in the frontends
             model['content'].sort(key=sort_key)
-        validate_model(model, expect_content=True)
+        validate_model(model, expect_content=content)
         self._finish_model(model, location=False)
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def patch(self, path=''):
         """PATCH renames a file or directory without re-uploading content."""
         cm = self.contents_manager
         model = self.get_json_body()
         if model is None:
             raise web.HTTPError(400, u'JSON body missing')
         model = yield gen.maybe_future(cm.update(model, path))
         validate_model(model, expect_content=False)
         self._finish_model(model)
     
     @gen.coroutine
     def _copy(self, copy_from, copy_to=None):
         """Copy a file, optionally specifying a target directory."""
         self.log.info(u"Copying {copy_from} to {copy_to}".format(
             copy_from=copy_from,
             copy_to=copy_to or '',
         ))
         model = yield gen.maybe_future(self.contents_manager.copy(copy_from, copy_to))
         self.set_status(201)
         validate_model(model, expect_content=False)
         self._finish_model(model)
 
     @gen.coroutine
     def _upload(self, model, path):
         """Handle upload of a new file to path"""
         self.log.info(u"Uploading file to %s", path)
         model = yield gen.maybe_future(self.contents_manager.new(model, path))
         self.set_status(201)
         validate_model(model, expect_content=False)
         self._finish_model(model)
     
     @gen.coroutine
     def _new_untitled(self, path, type='', ext=''):
         """Create a new, empty untitled entity"""
         self.log.info(u"Creating new %s in %s", type or 'file', path)
         model = yield gen.maybe_future(self.contents_manager.new_untitled(path=path, type=type, ext=ext))
         self.set_status(201)
         validate_model(model, expect_content=False)
         self._finish_model(model)
     
     @gen.coroutine
     def _save(self, model, path):
         """Save an existing file."""
         self.log.info(u"Saving file at %s", path)
         model = yield gen.maybe_future(self.contents_manager.save(model, path))
         validate_model(model, expect_content=False)
         self._finish_model(model)
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def post(self, path=''):
         """Create a new file in the specified path.
 
         POST creates new files. The server always decides on the name.
 
         POST /api/contents/path
           New untitled, empty file or directory.
         POST /api/contents/path
           with body {"copy_from" : "/path/to/OtherNotebook.ipynb"}
           New copy of OtherNotebook in path
         """
 
         cm = self.contents_manager
 
         if cm.file_exists(path):
             raise web.HTTPError(400, "Cannot POST to files, use PUT instead.")
 
         if not cm.dir_exists(path):
             raise web.HTTPError(404, "No such directory: %s" % path)
 
         model = self.get_json_body()
 
         if model is not None:
             copy_from = model.get('copy_from')
             ext = model.get('ext', '')
             type = model.get('type', '')
             if copy_from:
                 yield self._copy(copy_from, path)
             else:
                 yield self._new_untitled(path, type=type, ext=ext)
         else:
             yield self._new_untitled(path)
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def put(self, path=''):
         """Saves the file in the location specified by name and path.
 
         PUT is very similar to POST, but the requester specifies the name,
         whereas with POST, the server picks the name.
 
         PUT /api/contents/path/Name.ipynb
           Save notebook at ``path/Name.ipynb``. Notebook structure is specified
           in `content` key of JSON request body. If content is not specified,
           create a new empty notebook.
         """
         model = self.get_json_body()
         if model:
             if model.get('copy_from'):
                 raise web.HTTPError(400, "Cannot copy with PUT, only POST")
             exists = yield gen.maybe_future(self.contents_manager.file_exists(path))
             if exists:
                 yield gen.maybe_future(self._save(model, path))
             else:
                 yield gen.maybe_future(self._upload(model, path))
         else:
             yield gen.maybe_future(self._new_untitled(path))
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def delete(self, path=''):
         """delete a file in the given path"""
         cm = self.contents_manager
         self.log.warn('delete %s', path)
         yield gen.maybe_future(cm.delete(path))
         self.set_status(204)
         self.finish()
 
 
 class CheckpointsHandler(IPythonHandler):
 
     SUPPORTED_METHODS = ('GET', 'POST')
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def get(self, path=''):
         """get lists checkpoints for a file"""
         cm = self.contents_manager
         checkpoints = yield gen.maybe_future(cm.list_checkpoints(path))
         data = json.dumps(checkpoints, default=date_default)
         self.finish(data)
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def post(self, path=''):
         """post creates a new checkpoint"""
         cm = self.contents_manager
         checkpoint = yield gen.maybe_future(cm.create_checkpoint(path))
         data = json.dumps(checkpoint, default=date_default)
         location = url_path_join(self.base_url, 'api/contents',
             path, 'checkpoints', checkpoint['id'])
         self.set_header('Location', url_escape(location))
         self.set_status(201)
         self.finish(data)
 
 
 class ModifyCheckpointsHandler(IPythonHandler):
 
     SUPPORTED_METHODS = ('POST', 'DELETE')
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def post(self, path, checkpoint_id):
         """post restores a file from a checkpoint"""
         cm = self.contents_manager
         yield gen.maybe_future(cm.restore_checkpoint(checkpoint_id, path))
         self.set_status(204)
         self.finish()
 
     @web.authenticated
     @json_errors
     @gen.coroutine
     def delete(self, path, checkpoint_id):
         """delete clears a checkpoint for a given file"""
         cm = self.contents_manager
         yield gen.maybe_future(cm.delete_checkpoint(checkpoint_id, path))
         self.set_status(204)
         self.finish()
 
 
 class NotebooksRedirectHandler(IPythonHandler):
     """Redirect /api/notebooks to /api/contents"""
     SUPPORTED_METHODS = ('GET', 'PUT', 'PATCH', 'POST', 'DELETE')
 
     def get(self, path):
         self.log.warn("/api/notebooks is deprecated, use /api/contents")
         self.redirect(url_path_join(
             self.base_url,
             'api/contents',
             path
         ))
 
     put = patch = post = delete = get
 
 
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 
 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
 
 default_handlers = [
     (r"/api/contents%s/checkpoints" % path_regex, CheckpointsHandler),
     (r"/api/contents%s/checkpoints/%s" % (path_regex, _checkpoint_id_regex),
         ModifyCheckpointsHandler),
     (r"/api/contents%s" % path_regex, ContentsHandler),
     (r"/api/notebooks/?(.*)", NotebooksRedirectHandler),
 ]

IPython/html/services/contents/tests/test_contents_api.py

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

IPython/html/static/services/contents.js

@@ -1,249 +1,249
 // Copyright (c) IPython Development Team.
 // Distributed under the terms of the Modified BSD License.
 
 define(function(require) {
     "use strict";
 
     var $ = require('jquery');
     var utils = require('base/js/utils');
 
     var Contents = function(options) {
         /**
          * Constructor
          *
          * A contents handles passing file operations
          * to the back-end.  This includes checkpointing
          * with the normal file operations.
          *
          * Parameters:
          *  options: dictionary
          *      Dictionary of keyword arguments.
          *          base_url: string
          */
         this.base_url = options.base_url;
     };
 
     /** Error type */
     Contents.DIRECTORY_NOT_EMPTY_ERROR = 'DirectoryNotEmptyError';
 
     Contents.DirectoryNotEmptyError = function() {
         // Constructor
         //
         // An error representing the result of attempting to delete a non-empty
         // directory.
         this.message = 'A directory must be empty before being deleted.';
     };
     
     Contents.DirectoryNotEmptyError.prototype = Object.create(Error.prototype);
     Contents.DirectoryNotEmptyError.prototype.name =
         Contents.DIRECTORY_NOT_EMPTY_ERROR;
 
 
     Contents.prototype.api_url = function() {
         var url_parts = [this.base_url, 'api/contents'].concat(
                                 Array.prototype.slice.apply(arguments));
         return utils.url_join_encode.apply(null, url_parts);
     };
 
     /**
      * Creates a basic error handler that wraps a jqXHR error as an Error.
      *
      * Takes a callback that accepts an Error, and returns a callback that can
      * be passed directly to $.ajax, which will wrap the error from jQuery
      * as an Error, and pass that to the original callback.
      *
      * @method create_basic_error_handler
      * @param{Function} callback
      * @return{Function}
      */
     Contents.prototype.create_basic_error_handler = function(callback) {
         if (!callback) {
             return utils.log_ajax_error;
         }
         return function(xhr, status, error) {
             callback(utils.wrap_ajax_error(xhr, status, error));
         };
     };
 
     /**
      * File Functions (including notebook operations)
      */
 
     /**
      * Get a file.
      *
-     * Calls success with file JSON model, or error with error.
-     *
      * @method get
      * @param {String} path
      * @param {Object} options
      *    type : 'notebook', 'file', or 'directory'
      *    format: 'text' or 'base64'; only relevant for type: 'file'
+     *    content: true or false; // whether to include the content
      */
     Contents.prototype.get = function (path, options) {
         /**
          * We do the call with settings so we can set cache to false.
          */
         var settings = {
             processData : false,
             cache : false,
             type : "GET",
             dataType : "json",
         };
         var url = this.api_url(path);
         var params = {};
         if (options.type) { params.type = options.type; }
         if (options.format) { params.format = options.format; }
+        if (options.content === false) { params.content = '0'; }
         return utils.promising_ajax(url + '?' + $.param(params), settings);
     };
 
 
     /**
      * Creates a new untitled file or directory in the specified directory path.
      *
      * @method new
      * @param {String} path: the directory in which to create the new file/directory
      * @param {Object} options:
      *      ext: file extension to use
      *      type: model type to create ('notebook', 'file', or 'directory')
      */
     Contents.prototype.new_untitled = function(path, options) {
         var data = JSON.stringify({
           ext: options.ext,
           type: options.type
         });
 
         var settings = {
             processData : false,
             type : "POST",
             data: data,
             dataType : "json",
         };
         return utils.promising_ajax(this.api_url(path), settings);
     };
 
     Contents.prototype.delete = function(path) {
         var settings = {
             processData : false,
             type : "DELETE",
             dataType : "json",
         };
         var url = this.api_url(path);
         return utils.promising_ajax(url, settings).catch(
             // Translate certain errors to more specific ones.
             function(error) {
                 // TODO: update IPEP27 to specify errors more precisely, so
                 // that error types can be detected here with certainty.
                 if (error.xhr.status === 400) {
                     throw new Contents.DirectoryNotEmptyError();
                 }
                 throw error;
             }
         );
     };
 
     Contents.prototype.rename = function(path, new_path) {
         var data = {path: new_path};
         var settings = {
             processData : false,
             type : "PATCH",
             data : JSON.stringify(data),
             dataType: "json",
             contentType: 'application/json',
         };
         var url = this.api_url(path);
         return utils.promising_ajax(url, settings);
     };
 
     Contents.prototype.save = function(path, model) {
         /**
          * We do the call with settings so we can set cache to false.
          */
         var settings = {
             processData : false,
             type : "PUT",
             dataType: "json",
             data : JSON.stringify(model),
             contentType: 'application/json',
         };
         var url = this.api_url(path);
         return utils.promising_ajax(url, settings);
     };
     
     Contents.prototype.copy = function(from_file, to_dir) {
         /**
          * Copy a file into a given directory via POST
          * The server will select the name of the copied file
          */
         var url = this.api_url(to_dir);
         
         var settings = {
             processData : false,
             type: "POST",
             data: JSON.stringify({copy_from: from_file}),
             dataType : "json",
         };
         return utils.promising_ajax(url, settings);
     };
 
     /**
      * Checkpointing Functions
      */
 
     Contents.prototype.create_checkpoint = function(path) {
         var url = this.api_url(path, 'checkpoints');
         var settings = {
             type : "POST",
             dataType : "json",
         };
         return utils.promising_ajax(url, settings);
     };
 
     Contents.prototype.list_checkpoints = function(path) {
         var url = this.api_url(path, 'checkpoints');
         var settings = {
             type : "GET",
             cache: false,
             dataType: "json",
         };
         return utils.promising_ajax(url, settings);
     };
 
     Contents.prototype.restore_checkpoint = function(path, checkpoint_id) {
         var url = this.api_url(path, 'checkpoints', checkpoint_id);
         var settings = {
             type : "POST",
         };
         return utils.promising_ajax(url, settings);
     };
 
     Contents.prototype.delete_checkpoint = function(path, checkpoint_id) {
         var url = this.api_url(path, 'checkpoints', checkpoint_id);
         var settings = {
             type : "DELETE",
         };
         return utils.promising_ajax(url, settings);
     };
 
     /**
      * File management functions
      */
 
     /**
      * List notebooks and directories at a given path
      *
      * On success, load_callback is called with an array of dictionaries
      * representing individual files or directories.  Each dictionary has
      * the keys:
      *     type: "notebook" or "directory"
      *     created: created date
      *     last_modified: last modified dat
      * @method list_notebooks
      * @param {String} path The path to list notebooks in
      */
     Contents.prototype.list_contents = function(path) {
         return this.get(path, {type: 'directory'});
     };
 
     return {'Contents': Contents};
 });