Added notebooks API tests.

IPython/html/services/sessions/tests/test_api.py

@@ -0,0 +1,32 b''
+"""Test the sessions service API."""
+
+
+import os
+import sys
+import json
+import urllib
+
+import requests
+
+from IPython.html.tests.launchnotebook import NotebookTestBase
+
+'''
+class SessionAPITest(NotebookTestBase):
+    """Test the sessions web service API"""
+
+    def base_url(self):
+        return super(SessionAPITest,self).base_url() + 'api/sessions'
+
+    def test_no_sessions(self):
+        """Make sure there are no sessions running at the start"""
+        url = self.base_url()
+        r = requests.get(url)
+        self.assertEqual(r.json(), [])
+
+    def test_start_session(self):
+        url = self.base_url()
+        param = urllib.urlencode({'notebook_path': 'test.ipynb'})
+        r = requests.post(url, params=param)
+        print r
+        #self.assertNotEqual(r.json(), [])
+'''
  No newline at end of file

IPython/html/services/tests/test_services_api.py

@@ -0,0 +1,124 b''
+"""Test the all of the services API."""
+
+
+import os
+import sys
+import json
+import urllib
+from zmq.utils import jsonapi
+
+import requests
+
+from IPython.html.tests.launchnotebook import NotebookTestBase
+
+class APITest(NotebookTestBase):
+    """Test the kernels web service API"""
+
+    def base_url(self):
+        return super(APITest,self).base_url()
+
+    def notebooks_url(self):
+        return self.base_url() + 'api/notebooks'
+
+    def kernels_url(self):
+        return self.base_url() + 'api/kernels'
+
+    def sessions_url(self):
+        return self.base_url() + 'api/sessions'
+
+    def contents_url(self):
+        return self.contents_url() + 'api/contents'
+
+    def mknb(self, name='', path='/'):
+        url = self.notebooks_url() + path
+        return url, requests.post(url)
+
+    def delnb(self, name, path='/'):
+        url = self.notebooks_url() + path + name
+        r = requests.delete(url)
+        return r.status_code
+
+    def test_no_notebooks(self):
+        url = self.notebooks_url()
+        r = requests.get(url)
+        self.assertEqual(r.json(), [])
+
+    def test_root_notebook_handler(self):
+        # POST a notebook and test the dict thats returned.
+        url, nb = self.mknb()
+        data = nb.json()
+        assert isinstance(data, dict)
+        assert data.has_key("name")
+        assert data.has_key("path")
+        self.assertEqual(data['name'], u'Untitled0.ipynb')
+        self.assertEqual(data['path'], u'/')
+
+        # GET list of notebooks in directory.
+        r = requests.get(url)
+        assert isinstance(r.json(), list)
+        assert isinstance(r.json()[0], dict)
+
+        # GET with a notebook name.
+        url = self.notebooks_url() + '/Untitled0.ipynb'
+        r = requests.get(url)
+        assert isinstance(data, dict)
+        self.assertEqual(r.json(), data)
+
+        # PATCH (rename) request.
+        new_name = {'name':'test.ipynb'}
+        r = requests.patch(url, data=jsonapi.dumps(new_name))
+        data = r.json()
+        assert isinstance(data, dict)
+
+        # make sure the patch worked.
+        new_url = self.notebooks_url() + '/test.ipynb'
+        r = requests.get(new_url)
+        assert isinstance(r.json(), dict)
+        self.assertEqual(r.json(), data)
+
+        # GET bad (old) notebook name.
+        r = requests.get(url)
+        self.assertEqual(r.status_code, 404)
+
+        # POST notebooks to folders one and two levels down.
+        os.makedirs(os.path.join(self.notebook_dir.name, 'foo'))
+        os.makedirs(os.path.join(self.notebook_dir.name, 'foo','bar'))
+        url, nb = self.mknb(path='/foo/')
+        url2, nb2 = self.mknb(path='/foo/bar/')
+        data = nb.json()
+        data2 = nb2.json()
+        assert isinstance(data, dict)
+        assert isinstance(data2, dict)
+        assert data.has_key("name")
+        assert data.has_key("path")
+        self.assertEqual(data['name'], u'Untitled0.ipynb')
+        self.assertEqual(data['path'], u'/foo/')
+        assert data2.has_key("name")
+        assert data2.has_key("path")
+        self.assertEqual(data2['name'], u'Untitled0.ipynb')
+        self.assertEqual(data2['path'], u'/foo/bar/')
+
+        # GET request on notebooks one and two levels down.
+        r = requests.get(url+'Untitled0.ipynb')
+        r2 = requests.get(url2+'Untitled0.ipynb')
+        assert isinstance(r.json(), dict)
+        self.assertEqual(r.json(), data)
+        assert isinstance(r2.json(), dict)
+        self.assertEqual(r2.json(), data2)
+
+        # PATCH notebooks that are one and two levels down.
+        new_name = {'name': 'testfoo.ipynb'}
+        r = requests.patch(url+'Untitled0.ipynb', data=jsonapi.dumps(new_name))
+        r = requests.get(url+'testfoo.ipynb')
+        data = r.json()
+        assert isinstance(data, dict)
+        assert data.has_key('name')
+        self.assertEqual(data['name'], 'testfoo.ipynb')
+        r = requests.get(url+'Untitled0.ipynb')
+        self.assertEqual(r.status_code, 404)
+        
+        # DELETE notebooks
+        r = self.delnb('testfoo.ipynb', '/foo/')
+        r2 = self.delnb('Untitled0.ipynb', '/foo/bar/') 
+        self.assertEqual(r, 204)
+        self.assertEqual(r2, 204)
  No newline at end of file

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

@@ -1,115 +1,117 b''
 """Tests for the content manager."""
 
 import os
 from unittest import TestCase
 from tempfile import NamedTemporaryFile
 
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.utils.traitlets import TraitError
 
 from ..contentmanager import ContentManager
 
 
 class TestContentManager(TestCase):
 
     def test_new_folder(self):
         with TemporaryDirectory() as td:
             # Test that a new directory/folder is created
             cm = ContentManager(content_dir=td)
             name = cm.new_folder(None, '/')
             path = cm.get_os_path(name, '/')
             self.assertTrue(os.path.isdir(path))
             
             # Test that a new directory is created with
             # the name given.
             name = cm.new_folder('foo')
             path = cm.get_os_path(name)
             self.assertTrue(os.path.isdir(path))
             
             # Test that a new directory/folder is created
             # in the '/foo' subdirectory
             name1 = cm.new_folder(None, '/foo/')
             path1 = cm.get_os_path(name1, '/foo/')
             self.assertTrue(os.path.isdir(path1))
 
             # make another file and make sure it incremented
             # the name and does not write over another file.
             name2 = cm.new_folder(None, '/foo/')
             path2 = cm.get_os_path(name, '/foo/')
             self.assertEqual(name2, 'new_folder1')
             
             # Test that an HTTP Error is raised when the user
             # tries to create a new folder with a name that 
             # already exists
             bad_name = 'new_folder1'
             self.assertRaises(HTTPError, cm.new_folder, name=bad_name, path='/foo/')
 
     def test_delete_folder(self):
         with TemporaryDirectory() as td:
             # Create a folder
             cm = ContentManager(content_dir=td)
             name = cm.new_folder('test_folder', '/')
             path = cm.get_os_path(name, '/')
             
             # Raise an exception when trying to delete a 
             # folder that does not exist.
             self.assertRaises(HTTPError, cm.delete_content, name='non_existing_folder', content_path='/')
             
             # Create a subfolder in the folder created above. 
             # *Recall 'name' = 'test_folder' (the new path for
             # subfolder)
             name01 = cm.new_folder(None, name)
             path01 = cm.get_os_path(name01, name)
             # Try to delete a subfolder that does not exist.
             self.assertRaises(HTTPError, cm.delete_content, name='non_existing_folder', content_path='/')
             # Delete the created subfolder
             cm.delete_content(name01, name)
             self.assertFalse(os.path.isdir(path01))
             
             # Delete the created folder
             cm.delete_content(name, '/')
             self.assertFalse(os.path.isdir(path))
             
             self.assertRaises(HTTPError, cm.delete_content, name=None, content_path='/')
             self.assertRaises(HTTPError, cm.delete_content, name='/', content_path='/')
             
             
     def test_get_content_names(self):
         with TemporaryDirectory() as td:
             # Create a few folders and subfolders
             cm = ContentManager(content_dir=td)
             name1 = cm.new_folder('fold1', '/')
             name2 = cm.new_folder('fold2', '/')
             name3 = cm.new_folder('fold3', '/')
             name01 = cm.new_folder('fold01', 'fold1')
             name02 = cm.new_folder('fold02', 'fold1')
             name03 = cm.new_folder('fold03', 'fold1')
             
             # List the names in the root folder
             names = cm.get_content_names('/')
             expected = ['fold1', 'fold2', 'fold3']
             self.assertEqual(set(names), set(expected))
             
             # List the names in the subfolder 'fold1'.
             names = cm.get_content_names('fold1')
             expected = ['fold01', 'fold02', 'fold03']
             self.assertEqual(set(names), set(expected))
 
     def test_content_model(self):
         with TemporaryDirectory() as td:
             # Create a few folders and subfolders
             cm = ContentManager(content_dir=td)
             name1 = cm.new_folder('fold1', '/')
             name2 = cm.new_folder('fold2', '/')
             name01 = cm.new_folder('fold01', 'fold1')
             name02 = cm.new_folder('fold02', 'fold1')
             
             # Check to see if the correct model and list of 
             # model dicts are returned for root directory
             # and subdirectory.
             contents = cm.list_contents('/')
             contents1 = cm.list_contents('fold1')
-            self.assertEqual(type(contents), type(list()))
-            self.assertEqual(type(contents[0]), type(dict()))
+            assert isinstance(contents, list)
+            assert isinstance(contents[0], dict)
+            assert contents[0].has_key('path')
+            assert contents[0].has_key('path')
             self.assertEqual(contents[0]['path'], '/')
             self.assertEqual(contents1[0]['path'], 'fold1')

IPython/html/services/notebooks/filenbmanager.py

@@ -1,373 +1,347 b''
 """A notebook manager that uses the local file system for storage.
 
 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 datetime
 import io
 import os
 import glob
 import shutil
 
 from unicodedata import normalize
 
 from tornado import web
 
 from .nbmanager import NotebookManager
 from IPython.nbformat import current
 from IPython.utils.traitlets import Unicode, Dict, Bool, TraitError
 from IPython.utils import tz
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class FileNotebookManager(NotebookManager):
     
     save_script = Bool(False, config=True,
         help="""Automatically create a Python script when saving the notebook.
         
         For easier use of import, %run and %load across notebooks, a
         <notebook-name>.py script will be created next to any
         <notebook-name>.ipynb on each save.  This can also be set with the
         short `--script` flag.
         """
     )
     
     checkpoint_dir = Unicode(config=True,
         help="""The location in which to keep notebook checkpoints
         
         By default, it is notebook-dir/.ipynb_checkpoints
         """
     )
     def _checkpoint_dir_default(self):
         return os.path.join(self.notebook_dir, '.ipynb_checkpoints')
     
     def _checkpoint_dir_changed(self, name, old, new):
         """do a bit of validation of the checkpoint dir"""
         if not os.path.isabs(new):
             # If we receive a non-absolute path, make it absolute.
             abs_new = os.path.abspath(new)
             self.checkpoint_dir = abs_new
             return
         if os.path.exists(new) and not os.path.isdir(new):
             raise TraitError("checkpoint dir %r is not a directory" % new)
         if not os.path.exists(new):
             self.log.info("Creating checkpoint dir %s", new)
             try:
                 os.mkdir(new)
             except:
                 raise TraitError("Couldn't create checkpoint dir %r" % new)
     
     filename_ext = Unicode(u'.ipynb')
 
     
     def get_notebook_names(self, path):
         """List all notebook names in the notebook dir."""
         names = glob.glob(self.get_os_path('*'+self.filename_ext, path))
         names = [os.path.basename(name)
                  for name in names]
         return names
         
     def list_notebooks(self, path):
         """List all notebooks in the notebook dir."""
         notebook_names = self.get_notebook_names(path)
         notebooks = []
         for name in notebook_names:
             model = self.notebook_model(name, path, content=False)
             notebooks.append(model)
         return notebooks
 
     def change_notebook(self, data, notebook_name, notebook_path='/'):
         """Changes notebook"""
         changes = data.keys()
-        response = 200
         for change in changes:
             full_path = self.get_os_path(notebook_name, notebook_path)
             if change == "name":
                 new_path = self.get_os_path(data['name'], notebook_path)
-                if not os.path.isfile(new_path):
+                try:
                     os.rename(full_path,
                         self.get_os_path(data['name'], notebook_path))
                     notebook_name = data['name']
-                else:
-                    response = 409
+                except OSError as e:
+                    raise web.HTTPError(409, u'Notebook name already exists.')
             if change == "path":
                 new_path = self.get_os_path(data['name'], data['path'])
                 stutil.move(full_path, new_path)
                 notebook_path = data['path']
             if change == "content":
                 self.save_notebook(data, notebook_name, notebook_path)
         model = self.notebook_model(notebook_name, notebook_path)
-        return model, response
+        return model
 
     def notebook_exists(self, name, path):
         """Returns a True if the notebook exists. Else, returns False.
         
         Parameters
         ----------
         name : string
             The name of the notebook you are checking.
         path : string
             The relative path to the notebook (with '/' as separator)
         
         Returns
         -------
         bool
         """
         path = self.get_os_path(name, path)
         return os.path.isfile(path)
 
-    def get_os_path(self, fname, path='/'):
-        """Given a notebook name and a server URL path, return its file system
-        path.
-        
-        Parameters
-        ----------
-        fname : string
-            The name of a notebook file with the .ipynb extension
-        path : string
-            The relative URL path (with '/' as separator) to the named
-            notebook.
-            
-        Returns
-        -------
-        path : string
-            A file system path that combines notebook_dir (location where
-            server started), the relative path, and the filename with the
-            current operating system's url.
-        """
-        parts = path.split('/')
-        parts = [p for p in parts if p != ''] # remove duplicate splits
-        parts += [fname]
-        path = os.path.join(self.notebook_dir, *parts)
-        return path
-
     def read_notebook_object_from_path(self, path):
         """read a notebook object from a path"""
         info = os.stat(path)
         last_modified = tz.utcfromtimestamp(info.st_mtime)
         with open(path,'r') as f:
             s = f.read()
             try:
                 # v1 and v2 and json in the .ipynb files.
                 nb = current.reads(s, u'json')
             except ValueError as e:
                 msg = u"Unreadable Notebook: %s" % e
                 raise web.HTTPError(400, msg, reason=msg)
         return last_modified, nb
     
     def read_notebook_object(self, notebook_name, notebook_path='/'):
         """Get the Notebook representation of a notebook by notebook_name."""
         path = self.get_os_path(notebook_name, notebook_path)
         if not os.path.isfile(path):
             raise web.HTTPError(404, u'Notebook does not exist: %s' % notebook_name)
         last_modified, nb = self.read_notebook_object_from_path(path)
         # Always use the filename as the notebook name.
         # Eventually we will get rid of the notebook name in the metadata
         # but for now, that name is just an empty string. Until the notebooks
         # web service knows about names in URLs we still pass the name
         # back to the web app using the metadata though.
         nb.metadata.name = os.path.splitext(os.path.basename(path))[0]
         return last_modified, nb
     
     def write_notebook_object(self, nb, notebook_name=None, notebook_path='/', new_name= None):
         """Save an existing notebook object by notebook_name."""
         if new_name == None:
             try:
                 new_name = normalize('NFC', nb.metadata.name)
             except AttributeError:
                 raise web.HTTPError(400, u'Missing notebook name')
 
         new_path = notebook_path
         old_name = notebook_name
         old_checkpoints = self.list_checkpoints(old_name)
         
         path = self.get_os_path(new_name, new_path)
         
         # Right before we save the notebook, we write an empty string as the
         # notebook name in the metadata. This is to prepare for removing
         # this attribute entirely post 1.0. The web app still uses the metadata
         # name for now.
         nb.metadata.name = u''
 
         try:
             self.log.debug("Autosaving notebook %s", path)
             with open(path,'w') as f:
                 current.write(nb, f, u'json')
         except Exception as e:
             raise web.HTTPError(400, u'Unexpected error while autosaving notebook: %s' % e)
 
         # save .py script as well
         if self.save_script:
             pypath = os.path.splitext(path)[0] + '.py'
             self.log.debug("Writing script %s", pypath)
             try:
                 with io.open(pypath,'w', encoding='utf-8') as f:
                     current.write(nb, f, u'py')
             except Exception as e:
                 raise web.HTTPError(400, u'Unexpected error while saving notebook as script: %s' % e)
         
         if old_name != None:
             # remove old files if the name changed
             if old_name != new_name:            
                 # remove renamed original, if it exists
                 old_path = self.get_os_path(old_name, notebook_path)
                 if os.path.isfile(old_path):
                     self.log.debug("unlinking notebook %s", old_path)
                     os.unlink(old_path)
             
                 # cleanup old script, if it exists
                 if self.save_script:
                     old_pypath = os.path.splitext(old_path)[0] + '.py'
                     if os.path.isfile(old_pypath):
                         self.log.debug("unlinking script %s", old_pypath)
                         os.unlink(old_pypath)
                 
                 # rename checkpoints to follow file
                 for cp in old_checkpoints:
                     checkpoint_id = cp['checkpoint_id']
                     old_cp_path = self.get_checkpoint_path_by_name(old_name, checkpoint_id)
                     new_cp_path = self.get_checkpoint_path_by_name(new_name, checkpoint_id)
                     if os.path.isfile(old_cp_path):
                         self.log.debug("renaming checkpoint %s -> %s", old_cp_path, new_cp_path)
                         os.rename(old_cp_path, new_cp_path)
             
         return new_name
             
     def delete_notebook(self, notebook_name, notebook_path):
         """Delete notebook by notebook_name."""
         nb_path = self.get_os_path(notebook_name, notebook_path)
         if not os.path.isfile(nb_path):
             raise web.HTTPError(404, u'Notebook does not exist: %s' % notebook_name)
         
         # clear checkpoints
         for checkpoint in self.list_checkpoints(notebook_name):
             checkpoint_id = checkpoint['checkpoint_id']
             path = self.get_checkpoint_path(notebook_name, checkpoint_id)
             self.log.debug(path)
             if os.path.isfile(path):
                 self.log.debug("unlinking checkpoint %s", path)
                 os.unlink(path)
         
         self.log.debug("unlinking notebook %s", nb_path)
         os.unlink(nb_path)
 
     def increment_filename(self, basename, notebook_path='/'):
         """Return a non-used filename of the form basename<int>.
         
         This searches through the filenames (basename0, basename1, ...)
         until is find one that is not already being used. It is used to
         create Untitled and Copy names that are unique.
         """
         i = 0
         while True:
             name = u'%s%i.ipynb' % (basename,i)
             path = self.get_os_path(name, notebook_path)
             if not os.path.isfile(path):
                 break
             else:
                 i = i+1
         return name
     
     # Checkpoint-related utilities
     
     def get_checkpoint_path_by_name(self, name, checkpoint_id, notebook_path='/'):
         """Return a full path to a notebook checkpoint, given its name and checkpoint id."""
         filename = u"{name}-{checkpoint_id}{ext}".format(
             name=name,
             checkpoint_id=checkpoint_id,
             ext=self.filename_ext,
         )
         if notebook_path ==None:
             path = os.path.join(self.checkpoint_dir, filename)
         else:
             path = os.path.join(notebook_path, self.checkpoint_dir, filename)
         return path
     
     def get_checkpoint_path(self, notebook_name, checkpoint_id, notebook_path='/'):
         """find the path to a checkpoint"""
         name = notebook_name
         return self.get_checkpoint_path_by_name(name, checkpoint_id, notebook_path)
     
     def get_checkpoint_info(self, notebook_name, checkpoint_id, notebook_path='/'):
         """construct the info dict for a given checkpoint"""
         path = self.get_checkpoint_path(notebook_name, checkpoint_id, notebook_path)
         stats = os.stat(path)
         last_modified = tz.utcfromtimestamp(stats.st_mtime)
         info = dict(
             checkpoint_id = checkpoint_id,
             last_modified = last_modified,
         )
         
         return info
         
     # public checkpoint API
     
     def create_checkpoint(self, notebook_name, notebook_path='/'):
         """Create a checkpoint from the current state of a notebook"""
         nb_path = self.get_os_path(notebook_name, notebook_path)
         # only the one checkpoint ID:
         checkpoint_id = u"checkpoint"
         cp_path = self.get_checkpoint_path(notebook_name, checkpoint_id, notebook_path)
         self.log.debug("creating checkpoint for notebook %s", notebook_name)
         if not os.path.exists(self.checkpoint_dir):
             os.mkdir(self.checkpoint_dir)
         shutil.copy2(nb_path, cp_path)
         
         # return the checkpoint info
         return self.get_checkpoint_info(notebook_name, checkpoint_id, notebook_path)
     
     def list_checkpoints(self, notebook_name, notebook_path='/'):
         """list the checkpoints for a given notebook
         
         This notebook manager currently only supports one checkpoint per notebook.
         """
         checkpoint_id = "checkpoint"
         path = self.get_checkpoint_path(notebook_name, checkpoint_id, notebook_path)
         if not os.path.exists(path):
             return []
         else:
             return [self.get_checkpoint_info(notebook_name, checkpoint_id, notebook_path)]
         
     
     def restore_checkpoint(self, notebook_name, checkpoint_id, notebook_path='/'):
         """restore a notebook to a checkpointed state"""
         self.log.info("restoring Notebook %s from checkpoint %s", notebook_name, checkpoint_id)
         nb_path = self.get_os_path(notebook_name, notebook_path)
         cp_path = self.get_checkpoint_path(notebook_name, checkpoint_id, notebook_path)
         if not os.path.isfile(cp_path):
             self.log.debug("checkpoint file does not exist: %s", cp_path)
             raise web.HTTPError(404,
                 u'Notebook checkpoint does not exist: %s-%s' % (notebook_name, checkpoint_id)
             )
         # ensure notebook is readable (never restore from an unreadable notebook)
         last_modified, nb = self.read_notebook_object_from_path(cp_path)
         shutil.copy2(cp_path, nb_path)
         self.log.debug("copying %s -> %s", cp_path, nb_path)
     
     def delete_checkpoint(self, notebook_name, checkpoint_id, notebook_path='/'):
         """delete a notebook's checkpoint"""
         path = self.get_checkpoint_path(notebook_name, checkpoint_id, notebook_path)
         if not os.path.isfile(path):
             raise web.HTTPError(404,
                 u'Notebook checkpoint does not exist: %s-%s' % (notebook_name, checkpoint_id)
             )
         self.log.debug("unlinking %s", path)
         os.unlink(path)
     
     def info_string(self):
         return "Serving notebooks from local directory: %s" % self.notebook_dir

IPython/html/services/notebooks/handlers.py

@@ -1,226 +1,216 b''
 """Tornado handlers for the notebooks web service.
 
 Authors:
 
 * Brian Granger
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-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
 #-----------------------------------------------------------------------------
 
 from tornado import web
 
 from zmq.utils import jsonapi
 
 from IPython.utils.jsonutil import date_default
 
 from ...base.handlers import IPythonHandler
 
 #-----------------------------------------------------------------------------
 # Notebook web service handlers
 #-----------------------------------------------------------------------------
 
 
 class NotebookRootHandler(IPythonHandler):
 
     @web.authenticated
     def get(self):
         """get returns a list of notebooks from the location
         where the server was started."""
         nbm = self.notebook_manager
-        notebooks = nbm.list_notebooks("")
+        notebooks = nbm.list_notebooks("/")
         self.finish(jsonapi.dumps(notebooks))
 
     @web.authenticated
     def post(self):
         """post creates a notebooks in the directory where the
         server was started"""
         nbm = self.notebook_manager
         body = self.request.body.strip()
         format = self.get_argument('format', default='json')
         name = self.get_argument('name', default=None)
         if body:
             fname = nbm.save_new_notebook(body, notebook_path='/', name=name, format=format)
         else:
             fname = nbm.new_notebook(notebook_path='/')
         self.set_header('Location', nbm.notebook_dir + fname)
         model = nbm.notebook_model(fname)
-        self.set_header('Location', '{0}api/notebooks/{1}'.format(self.base_project_url, notebook_name))
+        self.set_header('Location', '{0}api/notebooks/{1}'.format(self.base_project_url, fname))
         self.finish(jsonapi.dumps(model))
 
-
-class NotebookRootRedirect(IPythonHandler):
-
-    @web.authenticated
-    def get(self):
-        """get redirects to not include trailing backslash"""
-        self.redirect("/api/notebooks")
-
-
 class NotebookHandler(IPythonHandler):
 
     SUPPORTED_METHODS = ('GET', 'PUT', 'PATCH', 'POST','DELETE')
 
     @web.authenticated
     def get(self, notebook_path):
         """get checks if a notebook is not named, an returns a list of notebooks
         in the notebook path given. If a name is given, return 
         the notebook representation"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         
         # Check to see if a notebook name was given
         if name is None:
             # List notebooks in 'notebook_path'
             notebooks = nbm.list_notebooks(path)
             self.finish(jsonapi.dumps(notebooks))
         else:
             # get and return notebook representation
             format = self.get_argument('format', default='json')
             download = self.get_argument('download', default='False')
             model = nbm.notebook_model(name,path)
             last_mod, representation, name = nbm.get_notebook(name, path, format)
             self.set_header('Last-Modified', last_mod)
             
             if download == 'True':
                 if format == u'json':
                     self.set_header('Content-Type', 'application/json')
                     self.set_header('Content-Disposition','attachment; filename="%s.ipynb"' % name)
                     self.finish(representation)
                 elif format == u'py':
                     self.set_header('Content-Type', 'application/x-python')
                     self.set_header('Content-Disposition','attachment; filename="%s.py"' % name)
                     self.finish(representation)
             else:
                 self.finish(jsonapi.dumps(model))
 
     @web.authenticated
     def patch(self, notebook_path):
         """patch is currently used strictly for notebook renaming.
         Changes the notebook name to the name given in data."""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         data = jsonapi.loads(self.request.body)
-        model, response = nbm.change_notebook(data, name, path)
-        self.set_status(response)
+        model = nbm.change_notebook(data, name, path)
         self.finish(jsonapi.dumps(model))
 
     @web.authenticated
     def post(self,notebook_path):
         """Create a new notebook in the location given by 'notebook_path'."""
         nbm = self.notebook_manager
         fname, path = nbm.named_notebook_path(notebook_path)
         body = self.request.body.strip()
         format = self.get_argument('format', default='json')
         name = self.get_argument('name', default=None)
         if body:
             fname = nbm.save_new_notebook(body, notebook_path=path, name=name, format=format)
         else:
             fname = nbm.new_notebook(notebook_path=path)
         self.set_header('Location', nbm.notebook_dir + path + fname)
         model = nbm.notebook_model(fname, path)
         self.finish(jsonapi.dumps(model))
 
     @web.authenticated
     def put(self, notebook_path):
         """saves the notebook in the location given by 'notebook_path'."""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         format = self.get_argument('format', default='json')
         nbm.save_notebook(self.request.body, notebook_path=path, name=name, format=format)
         model = nbm.notebook_model(name, path)
         self.set_status(204)
         self.finish(jsonapi.dumps(model))
 
     @web.authenticated
     def delete(self, notebook_path):
         """delete rmoves the notebook in the given notebook path"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         nbm.delete_notebook(name, path)
         self.set_status(204)
         self.finish()
 
 
 class NotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('GET', 'POST')
     
     @web.authenticated
     def get(self, notebook_path):
         """get lists checkpoints for a notebook"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         checkpoints = nbm.list_checkpoints(name, path)
         data = jsonapi.dumps(checkpoints, default=date_default)
         self.finish(data)
     
     @web.authenticated
     def post(self, notebook_path):
         """post creates a new checkpoint"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         checkpoint = nbm.create_checkpoint(name, path)
         data = jsonapi.dumps(checkpoint, default=date_default)
         if path == None:
             self.set_header('Location', '{0}notebooks/{1}/checkpoints/{2}'.format(
                 self.base_project_url, name, checkpoint['checkpoint_id']
                 ))
         else:
             self.set_header('Location', '{0}notebooks/{1}/{2}/checkpoints/{3}'.format(
                 self.base_project_url, path, name, checkpoint['checkpoint_id']
                 ))
         self.finish(data)
 
 
 class ModifyNotebookCheckpointsHandler(IPythonHandler):
     
     SUPPORTED_METHODS = ('POST', 'DELETE')
     
     @web.authenticated
     def post(self, notebook_path, checkpoint_id):
         """post restores a notebook from a checkpoint"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         nbm.restore_checkpoint(name, checkpoint_id, path)
         self.set_status(204)
         self.finish()
     
     @web.authenticated
     def delete(self, notebook_path, checkpoint_id):
         """delete clears a checkpoint for a given notebook"""
         nbm = self.notebook_manager
         name, path = nbm.named_notebook_path(notebook_path)
         nbm.delete_checkpoint(name, checkpoint_id, path)
         self.set_status(204)
         self.finish()
         
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 
 _notebook_path_regex = r"(?P<notebook_path>.+)"
 _checkpoint_id_regex = r"(?P<checkpoint_id>[\w-]+)"
 
 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" % _notebook_path_regex, NotebookHandler),
-    (r"api/notebooks/", NotebookRootRedirect),
+    (r"api/notebooks/", NotebookRootHandler),
     (r"api/notebooks", NotebookRootHandler),
 ]
 
 
 
 

IPython/html/services/notebooks/nbmanager.py

@@ -1,263 +1,289 b''
 """A base class notebook manager.
 
 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 os
 import uuid
 
 from tornado import web
 from urllib import quote, unquote
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.nbformat import current
 from IPython.utils.traitlets import List, Dict, Unicode, TraitError
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class NotebookManager(LoggingConfigurable):
 
     # Todo:
     # The notebook_dir attribute is used to mean a couple of different things:
     # 1. Where the notebooks are stored if FileNotebookManager is used.
     # 2. The cwd of the kernel for a project.
     # Right now we use this attribute in a number of different places and
     # we are going to have to disentangle all of this.
     notebook_dir = Unicode(os.getcwdu(), config=True, help="""
             The directory to use for notebooks.
             """)
 
     def named_notebook_path(self, notebook_path):
         """Given a notebook_path name, returns a (name, path) tuple, where
         name is a .ipynb file, and path is the directory for the file, which
         *always* starts *and* ends with a '/' character.
 
         Parameters
         ----------
         notebook_path : string
             A path that may be a .ipynb name or a directory
 
         Returns
         -------
         name : string or None
             the filename of the notebook, or None if not a .ipynb extension
         path : string
             the path to the directory which contains the notebook
         """
         names = notebook_path.split('/')
         names = [n for n in names if n != ''] # remove duplicate splits
 
         names = [''] + names
 
         if names and names[-1].endswith(".ipynb"):
             name = names[-1]
             path = "/".join(names[:-1]) + '/'
         else:
             name = None
             path = "/".join(names) + '/'
         return name, path
 
+    def get_os_path(self, fname=None, path='/'):
+        """Given a notebook name and a server URL path, return its file system
+        path.
+
+        Parameters
+        ----------
+        fname : string
+            The name of a notebook file with the .ipynb extension
+        path : string
+            The relative URL path (with '/' as separator) to the named
+            notebook.
+
+        Returns
+        -------
+        path : string
+            A file system path that combines notebook_dir (location where
+            server started), the relative path, and the filename with the
+            current operating system's url.
+        """
+        parts = path.split('/')
+        parts = [p for p in parts if p != ''] # remove duplicate splits
+        if fname is not None:
+            parts += [fname]
+        path = os.path.join(self.notebook_dir, *parts)
+        return path
+
     def url_encode(self, path):
         """Returns the path with all special characters URL encoded"""
         parts = os.path.split(path)
         return os.path.join(*[quote(p) for p in parts])
 
     def url_decode(self, path):
         """Returns the URL with special characters decoded"""
         parts = os.path.split(path)
         return os.path.join(*[unquote(p) for p in parts])
 
     def _notebook_dir_changed(self, new):
         """do a bit of validation of the notebook dir"""
         if not os.path.isabs(new):
             # If we receive a non-absolute path, make it absolute.
             abs_new = os.path.abspath(new)
             #self.notebook_dir = os.path.dirname(abs_new)
             return
         if os.path.exists(new) and not os.path.isdir(new):
             raise TraitError("notebook dir %r is not a directory" % new)
         if not os.path.exists(new):
             self.log.info("Creating notebook dir %s", new)
             try:
                 os.mkdir(new)
             except:
                 raise TraitError("Couldn't create notebook dir %r" % new)
                 
     allowed_formats = List([u'json',u'py'])
 
     def add_new_folder(self, path=None):
         new_path = os.path.join(self.notebook_dir, path)
         if not os.path.exists(new_path):
             os.makedirs(new_path) 
         else:
             raise web.HTTPError(409, u'Directory already exists or creation permission not allowed.')
 
     def load_notebook_names(self, path):
         """Load the notebook names into memory.
 
         This should be called once immediately after the notebook manager
         is created to load the existing notebooks into the mapping in
         memory.
         """
         self.list_notebooks(path)
 
     def list_notebooks(self):
         """List all notebooks.
 
         This returns a list of dicts, each of the form::
 
             dict(notebook_id=notebook,name=name)
 
         This list of dicts should be sorted by name::
 
             data = sorted(data, key=lambda item: item['name'])
         """
         raise NotImplementedError('must be implemented in a subclass')
 
     def notebook_model(self, notebook_name, notebook_path='/', content=True):
         """ Creates the standard notebook model """
         last_modified, contents = self.read_notebook_object(notebook_name, notebook_path)
         model = {"name": notebook_name, 
                     "path": notebook_path,
                     "last_modified (UTC)": last_modified.ctime()}
-        if content == True:
+        if content is True:
             model['content'] = contents
         return model
 
     def get_notebook(self, notebook_name, notebook_path='/', format=u'json'):
         """Get the representation of a notebook in format by notebook_name."""
         format = unicode(format)
         if format not in self.allowed_formats:
             raise web.HTTPError(415, u'Invalid notebook format: %s' % format)
         kwargs = {}
         last_mod, nb = self.read_notebook_object(notebook_name, notebook_path)
         if format == 'json':
             # don't split lines for sending over the wire, because it
             # should match the Python in-memory format.
             kwargs['split_lines'] = False
         representation = current.writes(nb, format, **kwargs)
         name = nb.metadata.get('name', 'notebook')
         return last_mod, representation, name
 
     def read_notebook_object(self, notebook_name, notebook_path='/'):
         """Get the object representation of a notebook by notebook_id."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def save_new_notebook(self, data, notebook_path='/', name=None, format=u'json'):
         """Save a new notebook and return its name.
 
         If a name is passed in, it overrides any values in the notebook data
         and the value in the data is updated to use that value.
         """
         if format not in self.allowed_formats:
             raise web.HTTPError(415, u'Invalid notebook format: %s' % format)
 
         try:
             nb = current.reads(data.decode('utf-8'), format)
         except:
             raise web.HTTPError(400, u'Invalid JSON data')
 
         if name is None:
            try:
                name = nb.metadata.name
            except AttributeError:
                raise web.HTTPError(400, u'Missing notebook name')
         nb.metadata.name = name
 
         notebook_name = self.write_notebook_object(nb, notebook_path=notebook_path)
         return notebook_name
 
     def save_notebook(self, data, notebook_path='/', name=None, new_name=None, format=u'json'):
         """Save an existing notebook by notebook_name."""
         if format not in self.allowed_formats:
             raise web.HTTPError(415, u'Invalid notebook format: %s' % format)
 
         try:
             nb = current.reads(data.decode('utf-8'), format)
         except:
             raise web.HTTPError(400, u'Invalid JSON data')
 
         if name is not None:
             nb.metadata.name = name
         self.write_notebook_object(nb, name, notebook_path, new_name)
 
-    def write_notebook_object(self, nb, notebook_name=None, notebook_path='/', new_name=None):
+    def write_notebook_object(self, nb, notebook_name='/', notebook_path='/', new_name=None):
         """Write a notebook object and return its notebook_name.
 
         If notebook_name is None, this method should create a new notebook_name.
         If notebook_name is not None, this method should check to make sure it
         exists and is valid.
         """
         raise NotImplementedError('must be implemented in a subclass')
 
     def delete_notebook(self, notebook_name, notebook_path):
         """Delete notebook by notebook_id."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def increment_filename(self, name):
         """Increment a filename to make it unique.
 
         This exists for notebook stores that must have unique names. When a notebook
         is created or copied this method constructs a unique filename, typically
         by appending an integer to the name.
         """
         return name
 
     def new_notebook(self, notebook_path='/'):
         """Create a new notebook and return its notebook_name."""
         name = self.increment_filename('Untitled', notebook_path)
         metadata = current.new_metadata(name=name)
         nb = current.new_notebook(metadata=metadata)
         notebook_name = self.write_notebook_object(nb, notebook_path=notebook_path)
         return notebook_name
 
     def copy_notebook(self, name, path='/'):
         """Copy an existing notebook and return its new notebook_name."""
         last_mod, nb = self.read_notebook_object(name, path)
         name = nb.metadata.name + '-Copy'
         name = self.increment_filename(name, path)
         nb.metadata.name = name
         notebook_name = self.write_notebook_object(nb, notebook_path = path)
         return notebook_name  
     
     # Checkpoint-related
     
     def create_checkpoint(self, notebook_name, notebook_path='/'):
         """Create a checkpoint of the current state of a notebook
         
         Returns a checkpoint_id for the new checkpoint.
         """
         raise NotImplementedError("must be implemented in a subclass")
     
     def list_checkpoints(self, notebook_name, notebook_path='/'):
         """Return a list of checkpoints for a given notebook"""
         return []
     
     def restore_checkpoint(self, notebook_name, checkpoint_id, notebook_path='/'):
         """Restore a notebook from one of its checkpoints"""
         raise NotImplementedError("must be implemented in a subclass")
 
     def delete_checkpoint(self, notebook_name, checkpoint_id, notebook_path='/'):
         """delete a checkpoint for a notebook"""
         raise NotImplementedError("must be implemented in a subclass")
     
     def log_info(self):
         self.log.info(self.info_string())
     
     def info_string(self):
         return "Serving notebooks"

IPython/html/tests/launchnotebook.py

@@ -1,40 +1,41 b''
 """Base class for notebook tests."""
 
 import sys
 import time
 from subprocess import Popen, PIPE
 from unittest import TestCase
 
 from IPython.utils.tempdir import TemporaryDirectory
 
 
 class NotebookTestBase(TestCase):
     """A base class for tests that need a running notebook.
     
     This creates an empty profile in a temp ipython_dir
     and then starts the notebook server with a separate temp notebook_dir.
     """
 
     port = 12342
 
     def setUp(self):
         self.ipython_dir = TemporaryDirectory()
         self.notebook_dir = TemporaryDirectory()
         notebook_args = [
             sys.executable, '-c',
             'from IPython.html.notebookapp import launch_new_instance; launch_new_instance()',
             '--port=%d' % self.port,
             '--no-browser',
             '--ipython-dir=%s' % self.ipython_dir.name,
             '--notebook-dir=%s' % self.notebook_dir.name
         ]
+        #self.notebook = Popen(notebook_args)
         self.notebook = Popen(notebook_args, stdout=PIPE, stderr=PIPE)
         time.sleep(3.0)
 
     def tearDown(self):
         self.notebook.terminate()
         self.ipython_dir.cleanup()
         self.notebook_dir.cleanup()
 
     def base_url(self):
         return 'http://localhost:%i/' % self.port