copy_from in json, not in url param

IPython/html/services/notebooks/handlers.py

@@ -1,273 +1,278 b''
 """Tornado handlers for the notebooks web service.
 
 Authors:
 
 * Brian Granger
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import json
 
 from tornado import 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
 
 #-----------------------------------------------------------------------------
 # Notebook web service handlers
 #-----------------------------------------------------------------------------
 
 
 class NotebookHandler(IPythonHandler):
 
     SUPPORTED_METHODS = (u'GET', u'PUT', u'PATCH', u'POST', u'DELETE')
 
     def notebook_location(self, name, path=''):
         """Return the full URL location of a notebook based.
         
         Parameters
         ----------
         name : unicode
             The base name of the notebook, such as "foo.ipynb".
         path : unicode
             The URL path of the notebook.
         """
         return url_escape(url_path_join(
             self.base_project_url, 'api', 'notebooks', path, name
         ))
 
     def _finish_model(self, model, location=True):
         """Finish a JSON request with a model, setting relevant headers, etc."""
         if location:
             location = self.notebook_location(model['name'], model['path'])
             self.set_header('Location', location)
         self.set_header('Last-Modified', model['last_modified'])
         self.finish(json.dumps(model, default=date_default))
     
     @web.authenticated
     @json_errors
     def get(self, path='', name=None):
         """Return a Notebook or list of notebooks.
 
         * GET with path and no notebook name lists notebooks in a directory
         * GET with path and notebook name returns notebook JSON
         """
         nbm = self.notebook_manager
         # Check to see if a notebook name was given
         if name is None:
             # List notebooks in 'path'
             notebooks = nbm.list_notebooks(path)
             self.finish(json.dumps(notebooks, default=date_default))
             return
         # get and return notebook representation
         model = nbm.get_notebook_model(name, path)
         self._finish_model(model, location=False)
 
     @web.authenticated
     @json_errors
     def patch(self, path='', name=None):
         """PATCH renames a notebook without re-uploading content."""
         nbm = self.notebook_manager
         if name is None:
             raise web.HTTPError(400, u'Notebook name missing')
         model = self.get_json_body()
         if model is None:
             raise web.HTTPError(400, u'JSON body missing')
         model = nbm.update_notebook_model(model, name, path)
         self._finish_model(model)
     
     def _copy_notebook(self, copy_from, path, copy_to=None):
         """Copy a notebook in path, optionally specifying the new name.
         
         Only support copying within the same directory.
         """
         self.log.info(u"Copying notebook from %s/%s to %s/%s",
             path, copy_from,
             path, copy_to or '',
         )
         model = self.notebook_manager.copy_notebook(copy_from, copy_to, path)
         self.set_status(201)
         self._finish_model(model)
     
     def _upload_notebook(self, model, path, name=None):
         """Upload a notebook
         
         If name specified, create it in path/name.
         """
         self.log.info(u"Uploading notebook to %s/%s", path, name or '')
         if name:
             model['name'] = name
         
         model = self.notebook_manager.create_notebook_model(model, path)
         self.set_status(201)
         self._finish_model(model)
     
     def _create_empty_notebook(self, path, name=None):
         """Create an empty notebook in path
         
         If name specified, create it in path/name.
         """
         self.log.info(u"Creating new notebook in %s/%s", path, name or '')
         model = {}
         if name:
             model['name'] = name
         model = self.notebook_manager.create_notebook_model(model, path=path)
         self.set_status(201)
         self._finish_model(model)
     
     def _save_notebook(self, model, path, name):
         """Save an existing notebook."""
         self.log.info(u"Saving notebook at %s/%s", path, name)
         model = self.notebook_manager.save_notebook_model(model, name, path)
         if model['path'] != path.strip('/') or model['name'] != name:
             # a rename happened, set Location header
             location = True
         else:
             location = False
         self._finish_model(model, location)
     
     @web.authenticated
     @json_errors
     def post(self, path='', name=None):
         """Create a new notebook in the specified path.
         
         POST creates new notebooks. The server always decides on the notebook name.
         
         POST /api/notebooks/path : new untitled notebook in path
             If content specified, upload a notebook, otherwise start empty.
         POST /api/notebooks/path?copy=OtherNotebook.ipynb : new copy of OtherNotebook in path
         """
         
-        model = self.get_json_body()
-        copy = self.get_argument("copy", default="")
         if name is not None:
-            raise web.HTTPError(400, "Only POST to directories. Use PUT for full names")
+            raise web.HTTPError(400, "Only POST to directories. Use PUT for full names.")
+        
+        model = self.get_json_body()
         
-        if copy:
-            self._copy_notebook(copy, path)
-        elif model:
-            self._upload_notebook(model, path)
+        if model is not None:
+            copy_from = model.get('copy_from')
+            if copy_from:
+                if model.get('content'):
+                    raise web.HTTPError(400, "Can't upload and copy at the same time.")
+                self._copy_notebook(copy_from, path)
+            else:
+                self._upload_notebook(model, path)
         else:
             self._create_empty_notebook(path)
 
     @web.authenticated
     @json_errors
     def put(self, path='', name=None):
         """Saves the notebook in the location specified by name and path.
         
         PUT /api/notebooks/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.
         PUT /api/notebooks/path/Name.ipynb?copy=OtherNotebook.ipynb : copy OtherNotebook to Name
         
         POST and PUT are basically the same. The only difference:
         
         - with POST, server always picks the name, with PUT the requester does
         """
         if name is None:
             raise web.HTTPError(400, "Only PUT to full names. Use POST for directories.")
+        
         model = self.get_json_body()
-        copy = self.get_argument("copy", default="")
-        if copy:
-            if model is not None:
-                raise web.HTTPError(400)
-            self._copy_notebook(copy, path, name)
-        elif model:
-            if self.notebook_manager.notebook_exists(name, path):
+        if model:
+            copy_from = model.get('copy_from')
+            if copy_from:
+                if model.get('content'):
+                    raise web.HTTPError(400, "Can't upload and copy at the same time.")
+                self._copy_notebook(copy_from, path, name)
+            elif self.notebook_manager.notebook_exists(name, path):
                 self._save_notebook(model, path, name)
             else:
                 self._upload_notebook(model, path, name)
         else:
             self._create_empty_notebook(path, name)
 
     @web.authenticated
     @json_errors
     def delete(self, path='', name=None):
         """delete the notebook in the given notebook path"""
         nbm = self.notebook_manager
         nbm.delete_notebook_model(name, path)
         self.set_status(204)
         self.finish()
 
 
 class NotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('GET', 'POST')
     
     @web.authenticated
     @json_errors
     def get(self, path='', name=None):
         """get lists checkpoints for a notebook"""
         nbm = self.notebook_manager
         checkpoints = nbm.list_checkpoints(name, path)
         data = json.dumps(checkpoints, default=date_default)
         self.finish(data)
     
     @web.authenticated
     @json_errors
     def post(self, path='', name=None):
         """post creates a new checkpoint"""
         nbm = self.notebook_manager
         checkpoint = nbm.create_checkpoint(name, path)
         data = json.dumps(checkpoint, default=date_default)
         location = url_path_join(self.base_project_url, 'api/notebooks',
             path, name, 'checkpoints', checkpoint['id'])
         self.set_header('Location', url_escape(location))
         self.set_status(201)
         self.finish(data)
 
 
 class ModifyNotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('POST', 'DELETE')
     
     @web.authenticated
     @json_errors
     def post(self, path, name, checkpoint_id):
         """post restores a notebook from a checkpoint"""
         nbm = self.notebook_manager
         nbm.restore_checkpoint(checkpoint_id, name, path)
         self.set_status(204)
         self.finish()
     
     @web.authenticated
     @json_errors
     def delete(self, path, name, checkpoint_id):
         """delete clears a checkpoint for a given notebook"""
         nbm = self.notebook_manager
         nbm.delete_checkpoint(checkpoint_id, name, path)
         self.set_status(204)
         self.finish()
         
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 
 _path_regex = r"(?P<path>(?:/.*)*)"
 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
 _notebook_name_regex = r"(?P<name>[^/]+\.ipynb)"
 _notebook_path_regex = "%s/%s" % (_path_regex, _notebook_name_regex)
 
 default_handlers = [
     (r"/api/notebooks%s/checkpoints" % _notebook_path_regex, NotebookCheckpointsHandler),
     (r"/api/notebooks%s/checkpoints/%s" % (_notebook_path_regex, _checkpoint_id_regex),
         ModifyNotebookCheckpointsHandler),
     (r"/api/notebooks%s" % _notebook_path_regex, NotebookHandler),
     (r"/api/notebooks%s" % _path_regex, NotebookHandler),
 ]
 
 
 

IPython/html/services/notebooks/tests/test_notebooks_api.py

@@ -1,295 +1,297 b''
 # coding: utf-8
 """Test the notebooks webservice API."""
 
 import io
+import json
 import os
 import shutil
 from unicodedata import normalize
 
 from zmq.utils import jsonapi
 
 pjoin = os.path.join
 
 import requests
 
 from IPython.html.utils import url_path_join, url_escape
 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
 from IPython.nbformat.current import (new_notebook, write, read, new_worksheet,
                                       new_heading_cell, to_notebook_json)
 from IPython.utils import py3compat
 from IPython.utils.data import uniq_stable
 
 
 class NBAPI(object):
     """Wrapper for notebook API calls."""
     def __init__(self, base_url):
         self.base_url = base_url
 
-    def _req(self, verb, path, body=None, params=None):
+    def _req(self, verb, path, body=None):
         response = requests.request(verb,
                 url_path_join(self.base_url, 'api/notebooks', path),
                 data=body,
-                params=params,
         )
         response.raise_for_status()
         return response
 
     def list(self, path='/'):
         return self._req('GET', path)
 
     def read(self, name, path='/'):
         return self._req('GET', url_path_join(path, name))
 
     def create_untitled(self, path='/'):
         return self._req('POST', path)
 
     def upload_untitled(self, body, path='/'):
         return self._req('POST', path, body)
 
     def copy_untitled(self, copy_from, path='/'):
-        return self._req('POST', path, params={'copy':copy_from})
+        body = json.dumps({'copy_from':copy_from})
+        return self._req('POST', path, body)
 
     def create(self, name, path='/'):
         return self._req('PUT', url_path_join(path, name))
 
     def upload(self, name, body, path='/'):
         return self._req('PUT', url_path_join(path, name), body)
 
     def copy(self, copy_from, copy_to, path='/'):
-        return self._req('PUT', url_path_join(path, copy_to), params={'copy':copy_from})
+        body = json.dumps({'copy_from':copy_from})
+        return self._req('PUT', url_path_join(path, copy_to), body)
 
     def save(self, name, body, path='/'):
         return self._req('PUT', url_path_join(path, name), body)
 
     def delete(self, name, path='/'):
         return self._req('DELETE', url_path_join(path, name))
 
     def rename(self, name, path, new_name):
         body = jsonapi.dumps({'name': new_name})
         return self._req('PATCH', url_path_join(path, name), body)
 
     def get_checkpoints(self, name, path):
         return self._req('GET', url_path_join(path, name, 'checkpoints'))
 
     def new_checkpoint(self, name, path):
         return self._req('POST', url_path_join(path, name, 'checkpoints'))
 
     def restore_checkpoint(self, name, path, checkpoint_id):
         return self._req('POST', url_path_join(path, name, 'checkpoints', checkpoint_id))
 
     def delete_checkpoint(self, name, path, checkpoint_id):
         return self._req('DELETE', url_path_join(path, name, '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'),
                 (u'å b', u'ç d')
                ]
 
     dirs = uniq_stable([d for (d,n) in dirs_nbs])
     del dirs[0]  # remove ''
 
     def setUp(self):
         nbdir = self.notebook_dir.name
 
         for d in self.dirs:
             if not os.path.isdir(pjoin(nbdir, d)):
                 os.mkdir(pjoin(nbdir, d))
 
         for d, name in self.dirs_nbs:
             with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w') as f:
                 nb = new_notebook(name=name)
                 write(nb, f, format='ipynb')
 
         self.nb_api = NBAPI(self.base_url())
 
     def tearDown(self):
         nbdir = self.notebook_dir.name
 
         for dname in ['foo', 'Directory with spaces in', u'unicodé', u'å b']:
             shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
 
         if os.path.isfile(pjoin(nbdir, 'inroot.ipynb')):
             os.unlink(pjoin(nbdir, 'inroot.ipynb'))
 
     def test_list_notebooks(self):
         nbs = self.nb_api.list().json()
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'inroot.ipynb')
 
         nbs = self.nb_api.list('/Directory with spaces in/').json()
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'inspace.ipynb')
 
         nbs = self.nb_api.list(u'/unicodé/').json()
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'innonascii.ipynb')
         self.assertEqual(nbs[0]['path'], u'unicodé')
 
         nbs = self.nb_api.list('/foo/bar/').json()
         self.assertEqual(len(nbs), 1)
         self.assertEqual(nbs[0]['name'], 'baz.ipynb')
         self.assertEqual(nbs[0]['path'], 'foo/bar')
 
         nbs = self.nb_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)
 
     def test_list_nonexistant_dir(self):
         with assert_http_error(404):
             self.nb_api.list('nonexistant')
 
     def test_get_contents(self):
         for d, name in self.dirs_nbs:
             nb = self.nb_api.read('%s.ipynb' % name, d+'/').json()
             self.assertEqual(nb['name'], u'%s.ipynb' % name)
             self.assertIn('content', nb)
             self.assertIn('metadata', nb['content'])
             self.assertIsInstance(nb['content']['metadata'], dict)
 
         # Name that doesn't exist - should be a 404
         with assert_http_error(404):
             self.nb_api.read('q.ipynb', 'foo')
 
     def _check_nb_created(self, resp, name, path):
         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/notebooks', path, name)))
         self.assertEqual(resp.json()['name'], name)
         assert os.path.isfile(pjoin(self.notebook_dir.name, path, name))
 
     def test_create_untitled(self):
         resp = self.nb_api.create_untitled(path=u'å b')
         self._check_nb_created(resp, 'Untitled0.ipynb', u'å b')
 
         # Second time
         resp = self.nb_api.create_untitled(path=u'å b')
         self._check_nb_created(resp, 'Untitled1.ipynb', u'å b')
 
         # And two directories down
         resp = self.nb_api.create_untitled(path='foo/bar')
         self._check_nb_created(resp, 'Untitled0.ipynb', pjoin('foo', 'bar'))
 
     def test_upload_untitled(self):
         nb = new_notebook(name='Upload test')
         nbmodel = {'content': nb}
         resp = self.nb_api.upload_untitled(path=u'å b',
                                               body=jsonapi.dumps(nbmodel))
         self._check_nb_created(resp, 'Untitled0.ipynb', u'å b')
 
     def test_upload(self):
         nb = new_notebook(name=u'ignored')
         nbmodel = {'content': nb}
         resp = self.nb_api.upload(u'Upload tést.ipynb', path=u'å b',
                                               body=jsonapi.dumps(nbmodel))
         self._check_nb_created(resp, u'Upload tést.ipynb', u'å b')
 
     def test_copy_untitled(self):
         resp = self.nb_api.copy_untitled(u'ç d.ipynb', path=u'å b')
         self._check_nb_created(resp, u'ç d-Copy0.ipynb', u'å b')
 
     def test_copy(self):
         resp = self.nb_api.copy(u'ç d.ipynb', u'cøpy.ipynb', path=u'å b')
         self._check_nb_created(resp, u'cøpy.ipynb', u'å b')
 
     def test_delete(self):
         for d, name in self.dirs_nbs:
             resp = self.nb_api.delete('%s.ipynb' % name, d)
             self.assertEqual(resp.status_code, 204)
 
         for d in self.dirs + ['/']:
             nbs = self.nb_api.list(d).json()
             self.assertEqual(len(nbs), 0)
 
     def test_rename(self):
         resp = self.nb_api.rename('a.ipynb', 'foo', 'z.ipynb')
         self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
         self.assertEqual(resp.json()['name'], 'z.ipynb')
         assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
 
         nbs = self.nb_api.list('foo').json()
         nbnames = set(n['name'] for n in nbs)
         self.assertIn('z.ipynb', nbnames)
         self.assertNotIn('a.ipynb', nbnames)
 
     def test_save(self):
         resp = self.nb_api.read('a.ipynb', 'foo')
         nbcontent = jsonapi.loads(resp.text)['content']
         nb = to_notebook_json(nbcontent)
         ws = new_worksheet()
         nb.worksheets = [ws]
         ws.cells.append(new_heading_cell(u'Created by test ³'))
 
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
 
         nbfile = pjoin(self.notebook_dir.name, 'foo', 'a.ipynb')
         with io.open(nbfile, 'r', encoding='utf-8') as f:
             newnb = read(f, format='ipynb')
         self.assertEqual(newnb.worksheets[0].cells[0].source,
                          u'Created by test ³')
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         newnb = to_notebook_json(nbcontent)
         self.assertEqual(newnb.worksheets[0].cells[0].source,
                          u'Created by test ³')
 
         # Save and rename
         nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
         saved = resp.json()
         self.assertEqual(saved['name'], 'a2.ipynb')
         self.assertEqual(saved['path'], 'foo/bar')
         assert os.path.isfile(pjoin(self.notebook_dir.name,'foo','bar','a2.ipynb'))
         assert not os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'a.ipynb'))
         with assert_http_error(404):
             self.nb_api.read('a.ipynb', 'foo')
 
     def test_checkpoints(self):
         resp = self.nb_api.read('a.ipynb', 'foo')
         r = self.nb_api.new_checkpoint('a.ipynb', 'foo')
         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 = jsonapi.loads(resp.text)['content']
         nb = to_notebook_json(nbcontent)
         ws = new_worksheet()
         nb.worksheets = [ws]
         hcell = new_heading_cell('Created by test')
         ws.cells.append(hcell)
         # Save
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb}
         resp = self.nb_api.save('a.ipynb', path='foo', body=jsonapi.dumps(nbmodel))
 
         # List checkpoints
         cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [cp1])
 
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         nb = to_notebook_json(nbcontent)
         self.assertEqual(nb.worksheets[0].cells[0].source, 'Created by test')
 
         # Restore cp1
         r = self.nb_api.restore_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         nbcontent = self.nb_api.read('a.ipynb', 'foo').json()['content']
         nb = to_notebook_json(nbcontent)
         self.assertEqual(nb.worksheets, [])
 
         # Delete cp1
         r = self.nb_api.delete_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         cps = self.nb_api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [])