don't use nbformat.current in IPython.html

use top-level nbformat.read/write, v4 directly for compose

IPython/html/nbconvert/handlers.py

@@ -1,148 +1,148
 """Tornado handlers for nbconvert."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import io
 import os
 import zipfile
 
 from tornado import web
 
 from ..base.handlers import (
     IPythonHandler, FilesRedirectHandler,
     notebook_path_regex, path_regex,
 )
-from IPython.nbformat.current import from_dict
+from IPython.nbformat import from_dict
 
 from IPython.utils.py3compat import cast_bytes
 
 def find_resource_files(output_files_dir):
     files = []
     for dirpath, dirnames, filenames in os.walk(output_files_dir):
         files.extend([os.path.join(dirpath, f) for f in filenames])
     return files
 
 def respond_zip(handler, name, output, resources):
     """Zip up the output and resource files and respond with the zip file.
 
     Returns True if it has served a zip file, False if there are no resource
     files, in which case we serve the plain output file.
     """
     # Check if we have resource files we need to zip
     output_files = resources.get('outputs', None)
     if not output_files:
         return False
 
     # Headers
     zip_filename = os.path.splitext(name)[0] + '.zip'
     handler.set_header('Content-Disposition',
                        'attachment; filename="%s"' % zip_filename)
     handler.set_header('Content-Type', 'application/zip')
 
     # Prepare the zip file
     buffer = io.BytesIO()
     zipf = zipfile.ZipFile(buffer, mode='w', compression=zipfile.ZIP_DEFLATED)
     output_filename = os.path.splitext(name)[0] + '.' + resources['output_extension']
     zipf.writestr(output_filename, cast_bytes(output, 'utf-8'))
     for filename, data in output_files.items():
         zipf.writestr(os.path.basename(filename), data)
     zipf.close()
 
     handler.finish(buffer.getvalue())
     return True
 
 def get_exporter(format, **kwargs):
     """get an exporter, raising appropriate errors"""
     # if this fails, will raise 500
     try:
         from IPython.nbconvert.exporters.export import exporter_map
     except ImportError as e:
         raise web.HTTPError(500, "Could not import nbconvert: %s" % e)
     
     try:
         Exporter = exporter_map[format]
     except KeyError:
         # should this be 400?
         raise web.HTTPError(404, u"No exporter for format: %s" % format)
     
     try:
         return Exporter(**kwargs)
     except Exception as e:
         raise web.HTTPError(500, "Could not construct Exporter: %s" % e)
 
 class NbconvertFileHandler(IPythonHandler):
 
     SUPPORTED_METHODS = ('GET',)
     
     @web.authenticated
     def get(self, format, path='', name=None):
         
         exporter = get_exporter(format, config=self.config, log=self.log)
         
         path = path.strip('/')
         model = self.contents_manager.get_model(name=name, path=path)
 
         self.set_header('Last-Modified', model['last_modified'])
         
         try:
             output, resources = exporter.from_notebook_node(model['content'])
         except Exception as e:
             raise web.HTTPError(500, "nbconvert failed: %s" % e)
 
         if respond_zip(self, name, output, resources):
             return
 
         # Force download if requested
         if self.get_argument('download', 'false').lower() == 'true':
             filename = os.path.splitext(name)[0] + '.' + resources['output_extension']
             self.set_header('Content-Disposition',
                                'attachment; filename="%s"' % filename)
 
         # MIME type
         if exporter.output_mimetype:
             self.set_header('Content-Type',
                             '%s; charset=utf-8' % exporter.output_mimetype)
 
         self.finish(output)
 
 class NbconvertPostHandler(IPythonHandler):
     SUPPORTED_METHODS = ('POST',)
 
     @web.authenticated 
     def post(self, format):
         exporter = get_exporter(format, config=self.config)
         
         model = self.get_json_body()
         name = model.get('name', 'notebook.ipynb')
         nbnode = from_dict(model['content'])
         
         try:
             output, resources = exporter.from_notebook_node(nbnode)
         except Exception as e:
             raise web.HTTPError(500, "nbconvert failed: %s" % e)
 
         if respond_zip(self, name, output, resources):
             return
 
         # MIME type
         if exporter.output_mimetype:
             self.set_header('Content-Type',
                             '%s; charset=utf-8' % exporter.output_mimetype)
 
         self.finish(output)
 
 
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 _format_regex = r"(?P<format>\w+)"
 
 
 default_handlers = [
     (r"/nbconvert/%s%s" % (_format_regex, notebook_path_regex),
          NbconvertFileHandler),
     (r"/nbconvert/%s" % _format_regex, NbconvertPostHandler),
     (r"/nbconvert/html%s" % path_regex, FilesRedirectHandler),
 ]

IPython/html/nbconvert/tests/test_nbconvert_handlers.py

@@ -1,131 +1,132
 # coding: utf-8
 import base64
 import io
 import json
 import os
 from os.path import join as pjoin
 import shutil
 
 import requests
 
 from IPython.html.utils import url_path_join
 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
-from IPython.nbformat.current import (new_notebook, write,
-                                      new_markdown_cell, new_code_cell,
-                                      new_output)
+from IPython.nbformat import write
+from IPython.nbformat.v4 import (
+    new_notebook, new_markdown_cell, new_code_cell, new_output,
+)
 
 from IPython.testing.decorators import onlyif_cmds_exist
 
 
 class NbconvertAPI(object):
     """Wrapper for nbconvert 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, 'nbconvert', path),
                 data=body, params=params,
         )
         response.raise_for_status()
         return response
 
     def from_file(self, format, path, name, download=False):
         return self._req('GET', url_path_join(format, path, name),
                          params={'download':download})
 
     def from_post(self, format, nbmodel):
         body = json.dumps(nbmodel)
         return self._req('POST', format, body)
 
     def list_formats(self):
         return self._req('GET', '')
 
 png_green_pixel = base64.encodestring(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00'
 b'\x00\x00\x01\x00\x00x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDAT'
 b'\x08\xd7c\x90\xfb\xcf\x00\x00\x02\\\x01\x1e.~d\x87\x00\x00\x00\x00IEND\xaeB`\x82'
 ).decode('ascii')
 
 class APITest(NotebookTestBase):
     def setUp(self):
         nbdir = self.notebook_dir.name
         
         if not os.path.isdir(pjoin(nbdir, 'foo')):
             os.mkdir(pjoin(nbdir, 'foo'))
         
         nb = new_notebook()
         
         nb.cells.append(new_markdown_cell(u'Created by test ³'))
         cc1 = new_code_cell(source=u'print(2*6)')
         cc1.outputs.append(new_output(output_type="stream", text=u'12'))
         cc1.outputs.append(new_output(output_type="execute_result",
             data={'image/png' : png_green_pixel},
             execution_count=1,
         ))
         nb.cells.append(cc1)
         
         with io.open(pjoin(nbdir, 'foo', 'testnb.ipynb'), 'w',
                      encoding='utf-8') as f:
-            write(nb, f)
+            write(f, nb, version=4)
 
         self.nbconvert_api = NbconvertAPI(self.base_url())
 
     def tearDown(self):
         nbdir = self.notebook_dir.name
 
         for dname in ['foo']:
             shutil.rmtree(pjoin(nbdir, dname), ignore_errors=True)
     
     @onlyif_cmds_exist('pandoc')
     def test_from_file(self):
         r = self.nbconvert_api.from_file('html', 'foo', 'testnb.ipynb')
         self.assertEqual(r.status_code, 200)
         self.assertIn(u'text/html', r.headers['Content-Type'])
         self.assertIn(u'Created by test', r.text)
         self.assertIn(u'print', r.text)
 
         r = self.nbconvert_api.from_file('python', 'foo', 'testnb.ipynb')
         self.assertIn(u'text/x-python', r.headers['Content-Type'])
         self.assertIn(u'print(2*6)', r.text)
 
     @onlyif_cmds_exist('pandoc')
     def test_from_file_404(self):
         with assert_http_error(404):
             self.nbconvert_api.from_file('html', 'foo', 'thisdoesntexist.ipynb')
 
     @onlyif_cmds_exist('pandoc')
     def test_from_file_download(self):
         r = self.nbconvert_api.from_file('python', 'foo', 'testnb.ipynb', download=True)
         content_disposition = r.headers['Content-Disposition']
         self.assertIn('attachment', content_disposition)
         self.assertIn('testnb.py', content_disposition)
 
     @onlyif_cmds_exist('pandoc')
     def test_from_file_zip(self):
         r = self.nbconvert_api.from_file('latex', 'foo', 'testnb.ipynb', download=True)
         self.assertIn(u'application/zip', r.headers['Content-Type'])
         self.assertIn(u'.zip', r.headers['Content-Disposition'])
 
     @onlyif_cmds_exist('pandoc')
     def test_from_post(self):
         nbmodel_url = url_path_join(self.base_url(), 'api/contents/foo/testnb.ipynb')
         nbmodel = requests.get(nbmodel_url).json()
         
         r = self.nbconvert_api.from_post(format='html', nbmodel=nbmodel)
         self.assertEqual(r.status_code, 200)
         self.assertIn(u'text/html', r.headers['Content-Type'])
         self.assertIn(u'Created by test', r.text)
         self.assertIn(u'print', r.text)
         
         r = self.nbconvert_api.from_post(format='python', nbmodel=nbmodel)
         self.assertIn(u'text/x-python', r.headers['Content-Type'])
         self.assertIn(u'print(2*6)', r.text)
 
     @onlyif_cmds_exist('pandoc')
     def test_from_post_zip(self):
         nbmodel_url = url_path_join(self.base_url(), 'api/contents/foo/testnb.ipynb')
         nbmodel = requests.get(nbmodel_url).json()
 
         r = self.nbconvert_api.from_post(format='latex', nbmodel=nbmodel)
         self.assertIn(u'application/zip', r.headers['Content-Type'])
         self.assertIn(u'.zip', r.headers['Content-Disposition'])

IPython/html/services/contents/filemanager.py

@@ -1,545 +1,545
 """A contents manager that uses the local file system for storage."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import base64
 import io
 import os
 import glob
 import shutil
 
 from tornado import web
 
 from .manager import ContentsManager
-from IPython.nbformat import current
+from IPython import nbformat
 from IPython.utils.io import atomic_writing
 from IPython.utils.path import ensure_dir_exists
 from IPython.utils.traitlets import Unicode, Bool, TraitError
 from IPython.utils.py3compat import getcwd
 from IPython.utils import tz
 from IPython.html.utils import is_hidden, to_os_path, url_path_join
 
 
 class FileContentsManager(ContentsManager):
 
     root_dir = Unicode(getcwd(), config=True)
 
     save_script = Bool(False, config=True, help='DEPRECATED, IGNORED')
     def _save_script_changed(self):
         self.log.warn("""
         Automatically saving notebooks as scripts has been removed.
         Use `ipython nbconvert --to python [notebook]` instead.
         """)
 
     def _root_dir_changed(self, name, old, new):
         """Do a bit of validation of the root_dir."""
         if not os.path.isabs(new):
             # If we receive a non-absolute path, make it absolute.
             self.root_dir = os.path.abspath(new)
             return
         if not os.path.isdir(new):
             raise TraitError("%r is not a directory" % new)
 
     checkpoint_dir = Unicode('.ipynb_checkpoints', config=True,
         help="""The directory name in which to keep file checkpoints
 
         This is a path relative to the file's own directory.
 
         By default, it is .ipynb_checkpoints
         """
     )
 
     def _copy(self, src, dest):
         """copy src to dest
 
         like shutil.copy2, but log errors in copystat
         """
         shutil.copyfile(src, dest)
         try:
             shutil.copystat(src, dest)
         except OSError as e:
             self.log.debug("copystat on %s failed", dest, exc_info=True)
 
     def _get_os_path(self, name=None, path=''):
         """Given a filename and API path, return its file system
         path.
 
         Parameters
         ----------
         name : string
             A filename
         path : string
             The relative API path to the named file.
 
         Returns
         -------
         path : string
             API path to be evaluated relative to root_dir.
         """
         if name is not None:
             path = url_path_join(path, name)
         return to_os_path(path, self.root_dir)
 
     def path_exists(self, path):
         """Does the API-style path refer to an extant directory?
 
         API-style wrapper for os.path.isdir
 
         Parameters
         ----------
         path : string
             The path to check. This is an API path (`/` separated,
             relative to root_dir).
 
         Returns
         -------
         exists : bool
             Whether the path is indeed a directory.
         """
         path = path.strip('/')
         os_path = self._get_os_path(path=path)
         return os.path.isdir(os_path)
 
     def is_hidden(self, path):
         """Does the API style path correspond to a hidden directory or file?
 
         Parameters
         ----------
         path : string
             The path to check. This is an API path (`/` separated,
             relative to root_dir).
 
         Returns
         -------
         exists : bool
             Whether the path is hidden.
 
         """
         path = path.strip('/')
         os_path = self._get_os_path(path=path)
         return is_hidden(os_path, self.root_dir)
 
     def file_exists(self, name, path=''):
         """Returns True if the file exists, else returns False.
 
         API-style wrapper for os.path.isfile
 
         Parameters
         ----------
         name : string
             The name of the file you are checking.
         path : string
             The relative path to the file's directory (with '/' as separator)
 
         Returns
         -------
         exists : bool
             Whether the file exists.
         """
         path = path.strip('/')
         nbpath = self._get_os_path(name, path=path)
         return os.path.isfile(nbpath)
 
     def exists(self, name=None, path=''):
         """Returns True if the path [and name] exists, else returns False.
 
         API-style wrapper for os.path.exists
 
         Parameters
         ----------
         name : string
             The name of the file you are checking.
         path : string
             The relative path to the file's directory (with '/' as separator)
 
         Returns
         -------
         exists : bool
             Whether the target exists.
         """
         path = path.strip('/')
         os_path = self._get_os_path(name, path=path)
         return os.path.exists(os_path)
 
     def _base_model(self, name, path=''):
         """Build the common base of a contents model"""
         os_path = self._get_os_path(name, path)
         info = os.stat(os_path)
         last_modified = tz.utcfromtimestamp(info.st_mtime)
         created = tz.utcfromtimestamp(info.st_ctime)
         # Create the base model.
         model = {}
         model['name'] = name
         model['path'] = path
         model['last_modified'] = last_modified
         model['created'] = created
         model['content'] = None
         model['format'] = None
         model['message'] = None
         return model
 
     def _dir_model(self, name, path='', content=True):
         """Build a model for a directory
 
         if content is requested, will include a listing of the directory
         """
         os_path = self._get_os_path(name, path)
 
         four_o_four = u'directory does not exist: %r' % os_path
 
         if not os.path.isdir(os_path):
             raise web.HTTPError(404, four_o_four)
         elif is_hidden(os_path, self.root_dir):
             self.log.info("Refusing to serve hidden directory %r, via 404 Error",
                 os_path
             )
             raise web.HTTPError(404, four_o_four)
 
         if name is None:
             if '/' in path:
                 path, name = path.rsplit('/', 1)
             else:
                 name = ''
         model = self._base_model(name, path)
         model['type'] = 'directory'
         dir_path = u'{}/{}'.format(path, name)
         if content:
             model['content'] = contents = []
             for os_path in glob.glob(self._get_os_path('*', dir_path)):
                 name = os.path.basename(os_path)
                 # skip over broken symlinks in listing
                 if not os.path.exists(os_path):
                     self.log.warn("%s doesn't exist", os_path)
                     continue
                 if self.should_list(name) and not is_hidden(os_path, self.root_dir):
                     contents.append(self.get_model(name=name, path=dir_path, content=False))
 
             model['format'] = 'json'
 
         return model
 
     def _file_model(self, name, path='', content=True):
         """Build a model for a file
 
         if content is requested, include the file contents.
         UTF-8 text files will be unicode, binary files will be base64-encoded.
         """
         model = self._base_model(name, path)
         model['type'] = 'file'
         if content:
             os_path = self._get_os_path(name, path)
             with io.open(os_path, 'rb') as f:
                 bcontent = f.read()
             try:
                 model['content'] = bcontent.decode('utf8')
             except UnicodeError as e:
                 model['content'] = base64.encodestring(bcontent).decode('ascii')
                 model['format'] = 'base64'
             else:
                 model['format'] = 'text'
         return model
 
 
     def _notebook_model(self, name, path='', content=True):
         """Build a notebook model
 
         if content is requested, the notebook content will be populated
         as a JSON structure (not double-serialized)
         """
         model = self._base_model(name, path)
         model['type'] = 'notebook'
         if content:
             os_path = self._get_os_path(name, path)
             with io.open(os_path, 'r', encoding='utf-8') as f:
                 try:
-                    nb = current.read(f, u'json')
+                    nb = nbformat.read(f, as_version=4)
                 except Exception as e:
                     raise web.HTTPError(400, u"Unreadable Notebook: %s %r" % (os_path, e))
             self.mark_trusted_cells(nb, name, path)
             model['content'] = nb
             model['format'] = 'json'
             self.validate_notebook_model(model)
         return model
 
     def get_model(self, name, path='', content=True):
         """ Takes a path and name for an entity and returns its model
 
         Parameters
         ----------
         name : str
             the name of the target
         path : str
             the API path that describes the relative path for the target
 
         Returns
         -------
         model : dict
             the contents model. If content=True, returns the contents
             of the file or directory as well.
         """
         path = path.strip('/')
 
         if not self.exists(name=name, path=path):
             raise web.HTTPError(404, u'No such file or directory: %s/%s' % (path, name))
 
         os_path = self._get_os_path(name, path)
         if os.path.isdir(os_path):
             model = self._dir_model(name, path, content)
         elif name.endswith('.ipynb'):
             model = self._notebook_model(name, path, content)
         else:
             model = self._file_model(name, path, content)
         return model
 
     def _save_notebook(self, os_path, model, name='', path=''):
         """save a notebook file"""
         # Save the notebook file
-        nb = current.from_dict(model['content'])
+        nb = nbformat.from_dict(model['content'])
 
         self.check_and_sign(nb, name, path)
 
         if 'name' in nb['metadata']:
             nb['metadata']['name'] = u''
 
         with atomic_writing(os_path, encoding='utf-8') as f:
-            current.write(nb, f, version=nb.nbformat)
+            nbformat.write(f, nb, version=nbformat.NO_CONVERT)
 
     def _save_file(self, os_path, model, name='', path=''):
         """save a non-notebook file"""
         fmt = model.get('format', None)
         if fmt not in {'text', 'base64'}:
             raise web.HTTPError(400, "Must specify format of file contents as 'text' or 'base64'")
         try:
             content = model['content']
             if fmt == 'text':
                 bcontent = content.encode('utf8')
             else:
                 b64_bytes = content.encode('ascii')
                 bcontent = base64.decodestring(b64_bytes)
         except Exception as e:
             raise web.HTTPError(400, u'Encoding error saving %s: %s' % (os_path, e))
         with atomic_writing(os_path, text=False) as f:
             f.write(bcontent)
 
     def _save_directory(self, os_path, model, name='', path=''):
         """create a directory"""
         if is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, u'Cannot create hidden directory %r' % os_path)
         if not os.path.exists(os_path):
             os.mkdir(os_path)
         elif not os.path.isdir(os_path):
             raise web.HTTPError(400, u'Not a directory: %s' % (os_path))
         else:
             self.log.debug("Directory %r already exists", os_path)
 
     def save(self, model, name='', path=''):
         """Save the file model and return the model with no content."""
         path = path.strip('/')
 
         if 'type' not in model:
             raise web.HTTPError(400, u'No file type provided')
         if 'content' not in model and model['type'] != 'directory':
             raise web.HTTPError(400, u'No file content provided')
 
         # One checkpoint should always exist
         if self.file_exists(name, path) and not self.list_checkpoints(name, path):
             self.create_checkpoint(name, path)
 
         new_path = model.get('path', path).strip('/')
         new_name = model.get('name', name)
 
         if path != new_path or name != new_name:
             self.rename(name, path, new_name, new_path)
 
         os_path = self._get_os_path(new_name, new_path)
         self.log.debug("Saving %s", os_path)
         try:
             if model['type'] == 'notebook':
                 self._save_notebook(os_path, model, new_name, new_path)
             elif model['type'] == 'file':
                 self._save_file(os_path, model, new_name, new_path)
             elif model['type'] == 'directory':
                 self._save_directory(os_path, model, new_name, new_path)
             else:
                 raise web.HTTPError(400, "Unhandled contents type: %s" % model['type'])
         except web.HTTPError:
             raise
         except Exception as e:
             raise web.HTTPError(400, u'Unexpected error while saving file: %s %s' % (os_path, e))
 
         validation_message = None
         if model['type'] == 'notebook':
             self.validate_notebook_model(model)
             validation_message = model.get('message', None)
 
         model = self.get_model(new_name, new_path, content=False)
         if validation_message:
             model['message'] = validation_message
         return model
 
     def update(self, model, name, path=''):
         """Update the file's path and/or name
 
         For use in PATCH requests, to enable renaming a file without
         re-uploading its contents. Only used for renaming at the moment.
         """
         path = path.strip('/')
         new_name = model.get('name', name)
         new_path = model.get('path', path).strip('/')
         if path != new_path or name != new_name:
             self.rename(name, path, new_name, new_path)
         model = self.get_model(new_name, new_path, content=False)
         return model
 
     def delete(self, name, path=''):
         """Delete file by name and path."""
         path = path.strip('/')
         os_path = self._get_os_path(name, path)
         rm = os.unlink
         if os.path.isdir(os_path):
             listing = os.listdir(os_path)
             # don't delete non-empty directories (checkpoints dir doesn't count)
             if listing and listing != [self.checkpoint_dir]:
                 raise web.HTTPError(400, u'Directory %s not empty' % os_path)
         elif not os.path.isfile(os_path):
             raise web.HTTPError(404, u'File does not exist: %s' % os_path)
 
         # clear checkpoints
         for checkpoint in self.list_checkpoints(name, path):
             checkpoint_id = checkpoint['id']
             cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
             if os.path.isfile(cp_path):
                 self.log.debug("Unlinking checkpoint %s", cp_path)
                 os.unlink(cp_path)
 
         if os.path.isdir(os_path):
             self.log.debug("Removing directory %s", os_path)
             shutil.rmtree(os_path)
         else:
             self.log.debug("Unlinking file %s", os_path)
             rm(os_path)
 
     def rename(self, old_name, old_path, new_name, new_path):
         """Rename a file."""
         old_path = old_path.strip('/')
         new_path = new_path.strip('/')
         if new_name == old_name and new_path == old_path:
             return
 
         new_os_path = self._get_os_path(new_name, new_path)
         old_os_path = self._get_os_path(old_name, old_path)
 
         # Should we proceed with the move?
         if os.path.isfile(new_os_path):
             raise web.HTTPError(409, u'File with name already exists: %s' % new_os_path)
 
         # Move the file
         try:
             shutil.move(old_os_path, new_os_path)
         except Exception as e:
             raise web.HTTPError(500, u'Unknown error renaming file: %s %s' % (old_os_path, e))
 
         # Move the checkpoints
         old_checkpoints = self.list_checkpoints(old_name, old_path)
         for cp in old_checkpoints:
             checkpoint_id = cp['id']
             old_cp_path = self.get_checkpoint_path(checkpoint_id, old_name, old_path)
             new_cp_path = self.get_checkpoint_path(checkpoint_id, new_name, new_path)
             if os.path.isfile(old_cp_path):
                 self.log.debug("Renaming checkpoint %s -> %s", old_cp_path, new_cp_path)
                 shutil.move(old_cp_path, new_cp_path)
 
     # Checkpoint-related utilities
 
     def get_checkpoint_path(self, checkpoint_id, name, path=''):
         """find the path to a checkpoint"""
         path = path.strip('/')
         basename, ext = os.path.splitext(name)
         filename = u"{name}-{checkpoint_id}{ext}".format(
             name=basename,
             checkpoint_id=checkpoint_id,
             ext=ext,
         )
         os_path = self._get_os_path(path=path)
         cp_dir = os.path.join(os_path, self.checkpoint_dir)
         ensure_dir_exists(cp_dir)
         cp_path = os.path.join(cp_dir, filename)
         return cp_path
 
     def get_checkpoint_model(self, checkpoint_id, name, path=''):
         """construct the info dict for a given checkpoint"""
         path = path.strip('/')
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         stats = os.stat(cp_path)
         last_modified = tz.utcfromtimestamp(stats.st_mtime)
         info = dict(
             id = checkpoint_id,
             last_modified = last_modified,
         )
         return info
 
     # public checkpoint API
 
     def create_checkpoint(self, name, path=''):
         """Create a checkpoint from the current state of a file"""
         path = path.strip('/')
         src_path = self._get_os_path(name, path)
         # only the one checkpoint ID:
         checkpoint_id = u"checkpoint"
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         self.log.debug("creating checkpoint for %s", name)
         self._copy(src_path, cp_path)
 
         # return the checkpoint info
         return self.get_checkpoint_model(checkpoint_id, name, path)
 
     def list_checkpoints(self, name, path=''):
         """list the checkpoints for a given file
 
         This contents manager currently only supports one checkpoint per file.
         """
         path = path.strip('/')
         checkpoint_id = "checkpoint"
         os_path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.exists(os_path):
             return []
         else:
             return [self.get_checkpoint_model(checkpoint_id, name, path)]
 
 
     def restore_checkpoint(self, checkpoint_id, name, path=''):
         """restore a file to a checkpointed state"""
         path = path.strip('/')
         self.log.info("restoring %s from checkpoint %s", name, checkpoint_id)
         nb_path = self._get_os_path(name, path)
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.isfile(cp_path):
             self.log.debug("checkpoint file does not exist: %s", cp_path)
             raise web.HTTPError(404,
                 u'checkpoint does not exist: %s-%s' % (name, checkpoint_id)
             )
         # ensure notebook is readable (never restore from an unreadable notebook)
         if cp_path.endswith('.ipynb'):
             with io.open(cp_path, 'r', encoding='utf-8') as f:
-                current.read(f, u'json')
+                nbformat.read(f, as_version=4)
         self._copy(cp_path, nb_path)
         self.log.debug("copying %s -> %s", cp_path, nb_path)
 
     def delete_checkpoint(self, checkpoint_id, name, path=''):
         """delete a file's checkpoint"""
         path = path.strip('/')
         cp_path = self.get_checkpoint_path(checkpoint_id, name, path)
         if not os.path.isfile(cp_path):
             raise web.HTTPError(404,
                 u'Checkpoint does not exist: %s%s-%s' % (path, name, checkpoint_id)
             )
         self.log.debug("unlinking %s", cp_path)
         os.unlink(cp_path)
 
     def info_string(self):
         return "Serving notebooks from local directory: %s" % self.root_dir
 
     def get_kernel_path(self, name, path='', model=None):
         """Return the initial working dir a kernel associated with a given notebook"""
         return os.path.join(self.root_dir, path)

IPython/html/services/contents/manager.py

@@ -1,345 +1,347
 """A base class for contents managers."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from fnmatch import fnmatch
 import itertools
 import json
 import os
 
 from tornado.web import HTTPError
 
 from IPython.config.configurable import LoggingConfigurable
-from IPython.nbformat import current, sign
+from IPython.nbformat import sign, validate, ValidationError
+from IPython.nbformat.v4 import new_notebook
 from IPython.utils.traitlets import Instance, Unicode, List
 
 
 class ContentsManager(LoggingConfigurable):
     """Base class for serving files and directories.
 
     This serves any text or binary file,
     as well as directories,
     with special handling for JSON notebook documents.
 
     Most APIs take a path argument,
     which is always an API-style unicode path,
     and always refers to a directory.
 
     - unicode, not url-escaped
     - '/'-separated
     - leading and trailing '/' will be stripped
     - if unspecified, path defaults to '',
       indicating the root path.
 
     name is also unicode, and refers to a specfic target:
 
     - unicode, not url-escaped
     - must not contain '/'
     - It refers to an individual filename
     - It may refer to a directory name,
       in the case of listing or creating directories.
 
     """
 
     notary = Instance(sign.NotebookNotary)
     def _notary_default(self):
         return sign.NotebookNotary(parent=self)
 
     hide_globs = List(Unicode, [
             u'__pycache__', '*.pyc', '*.pyo',
             '.DS_Store', '*.so', '*.dylib', '*~',
         ], config=True, help="""
         Glob patterns to hide in file and directory listings.
     """)
 
     untitled_notebook = Unicode("Untitled", config=True,
         help="The base name used when creating untitled notebooks."
     )
 
     untitled_file = Unicode("untitled", config=True,
         help="The base name used when creating untitled files."
     )
 
     untitled_directory = Unicode("Untitled Folder", config=True,
         help="The base name used when creating untitled directories."
     )
 
     # ContentsManager API part 1: methods that must be
     # implemented in subclasses.
 
     def path_exists(self, path):
         """Does the API-style path (directory) actually exist?
 
         Like os.path.isdir
 
         Override this method in subclasses.
 
         Parameters
         ----------
         path : string
             The path to check
 
         Returns
         -------
         exists : bool
             Whether the path does indeed exist.
         """
         raise NotImplementedError
 
     def is_hidden(self, path):
         """Does the API style path correspond to a hidden directory or file?
 
         Parameters
         ----------
         path : string
             The path to check. This is an API path (`/` separated,
             relative to root dir).
 
         Returns
         -------
         hidden : bool
             Whether the path is hidden.
 
         """
         raise NotImplementedError
 
     def file_exists(self, name, path=''):
         """Does a file exist at the given name and path?
 
         Like os.path.isfile
 
         Override this method in subclasses.
 
         Parameters
         ----------
         name : string
             The name of the file you are checking.
         path : string
             The relative path to the file's directory (with '/' as separator)
 
         Returns
         -------
         exists : bool
             Whether the file exists.
         """
         raise NotImplementedError('must be implemented in a subclass')
 
     def exists(self, name, path=''):
         """Does a file or directory exist at the given name and path?
 
         Like os.path.exists
 
         Parameters
         ----------
         name : string
             The name of the file you are checking.
         path : string
             The relative path to the file's directory (with '/' as separator)
 
         Returns
         -------
         exists : bool
             Whether the target exists.
         """
         return self.file_exists(name, path) or self.path_exists("%s/%s" % (path, name))
 
     def get_model(self, name, path='', content=True):
         """Get the model of a file or directory with or without content."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def save(self, model, name, path=''):
         """Save the file or directory and return the model with no content."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def update(self, model, name, path=''):
         """Update the file or directory and return the model with no content.
 
         For use in PATCH requests, to enable renaming a file without
         re-uploading its contents. Only used for renaming at the moment.
         """
         raise NotImplementedError('must be implemented in a subclass')
 
     def delete(self, name, path=''):
         """Delete file or directory by name and path."""
         raise NotImplementedError('must be implemented in a subclass')
 
     def create_checkpoint(self, name, path=''):
         """Create a checkpoint of the current state of a file
 
         Returns a checkpoint_id for the new checkpoint.
         """
         raise NotImplementedError("must be implemented in a subclass")
 
     def list_checkpoints(self, name, path=''):
         """Return a list of checkpoints for a given file"""
         return []
 
     def restore_checkpoint(self, checkpoint_id, name, path=''):
         """Restore a file from one of its checkpoints"""
         raise NotImplementedError("must be implemented in a subclass")
 
     def delete_checkpoint(self, checkpoint_id, name, path=''):
         """delete a checkpoint for a file"""
         raise NotImplementedError("must be implemented in a subclass")
 
     # ContentsManager API part 2: methods that have useable default
     # implementations, but can be overridden in subclasses.
 
     def info_string(self):
         return "Serving contents"
 
     def get_kernel_path(self, name, path='', model=None):
         """ Return the path to start kernel in """
         return path
 
     def increment_filename(self, filename, path=''):
         """Increment a filename until it is unique.
 
         Parameters
         ----------
         filename : unicode
             The name of a file, including extension
         path : unicode
             The API path of the target's directory
 
         Returns
         -------
         name : unicode
             A filename that is unique, based on the input filename.
         """
         path = path.strip('/')
         basename, ext = os.path.splitext(filename)
         for i in itertools.count():
             name = u'{basename}{i}{ext}'.format(basename=basename, i=i,
                                                 ext=ext)
             if not self.file_exists(name, path):
                 break
         return name
 
     def validate_notebook_model(self, model):
         """Add failed-validation message to model"""
         try:
-            current.validate(model['content'])
-        except current.ValidationError as e:
+            validate(model['content'])
+        except ValidationError as e:
             model['message'] = 'Notebook Validation failed: {}:\n{}'.format(
                 e.message, json.dumps(e.instance, indent=1, default=lambda obj: '<UNKNOWN>'),
             )
         return model
 
     def create_file(self, model=None, path='', ext='.ipynb'):
         """Create a new file or directory and return its model with no content."""
         path = path.strip('/')
         if model is None:
             model = {}
         if 'content' not in model and model.get('type', None) != 'directory':
             if ext == '.ipynb':
-                model['content'] = current.new_notebook()
+                model['content'] = new_notebook()
                 model['type'] = 'notebook'
                 model['format'] = 'json'
             else:
                 model['content'] = ''
                 model['type'] = 'file'
                 model['format'] = 'text'
         if 'name' not in model:
             if model['type'] == 'directory':
                 untitled = self.untitled_directory
             elif model['type'] == 'notebook':
                 untitled = self.untitled_notebook
             elif model['type'] == 'file':
                 untitled = self.untitled_file
             else:
                 raise HTTPError(400, "Unexpected model type: %r" % model['type'])
             model['name'] = self.increment_filename(untitled + ext, path)
 
         model['path'] = path
         model = self.save(model, model['name'], model['path'])
         return model
 
     def copy(self, from_name, to_name=None, path=''):
         """Copy an existing file and return its new model.
 
         If to_name not specified, increment `from_name-Copy#.ext`.
 
         copy_from can be a full path to a file,
         or just a base name. If a base name, `path` is used.
         """
         path = path.strip('/')
         if '/' in from_name:
             from_path, from_name = from_name.rsplit('/', 1)
         else:
             from_path = path
         model = self.get_model(from_name, from_path)
         if model['type'] == 'directory':
             raise HTTPError(400, "Can't copy directories")
         if not to_name:
             base, ext = os.path.splitext(from_name)
             copy_name = u'{0}-Copy{1}'.format(base, ext)
             to_name = self.increment_filename(copy_name, path)
         model['name'] = to_name
         model['path'] = path
         model = self.save(model, to_name, path)
         return model
 
     def log_info(self):
         self.log.info(self.info_string())
 
     def trust_notebook(self, name, path=''):
         """Explicitly trust a notebook
 
         Parameters
         ----------
         name : string
             The filename of the notebook
         path : string
             The notebook's directory
         """
         model = self.get_model(name, path)
         nb = model['content']
         self.log.warn("Trusting notebook %s/%s", path, name)
         self.notary.mark_cells(nb, True)
         self.save(model, name, path)
 
     def check_and_sign(self, nb, name='', path=''):
         """Check for trusted cells, and sign the notebook.
 
         Called as a part of saving notebooks.
 
         Parameters
         ----------
         nb : dict
-            The notebook object (in nbformat.current format)
+            The notebook object (in current nbformat)
         name : string
             The filename of the notebook (for logging)
         path : string
             The notebook's directory (for logging)
         """
-        if nb['nbformat'] != current.nbformat:
+        # can't sign old notebooks
+        if nb['nbformat'] != 4:
             return
         if self.notary.check_cells(nb):
             self.notary.sign(nb)
         else:
             self.log.warn("Saving untrusted notebook %s/%s", path, name)
 
     def mark_trusted_cells(self, nb, name='', path=''):
         """Mark cells as trusted if the notebook signature matches.
 
         Called as a part of loading notebooks.
 
         Parameters
         ----------
         nb : dict
-            The notebook object (in nbformat.current format)
+            The notebook object (in current nbformat)
         name : string
             The filename of the notebook (for logging)
         path : string
             The notebook's directory (for logging)
         """
         trusted = self.notary.check_signature(nb)
         if not trusted:
             self.log.warn("Notebook %s/%s is not trusted", path, name)
         self.notary.mark_cells(nb, trusted)
 
     def should_list(self, name):
         """Should this file/directory name be displayed in a listing?"""
         return not any(fnmatch(name, glob) for glob in self.hide_globs)

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

@@ -1,480 +1,481
 # coding: utf-8
 """Test the contents webservice API."""
 
 import base64
 import io
 import json
 import os
 import shutil
 from unicodedata import normalize
 
 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 import current
-from IPython.nbformat.current import (new_notebook, write, read,
-                                      new_markdown_cell, from_dict)
+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
 
 
 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):
         response = requests.request(verb,
                 url_path_join(self.base_url, 'api/contents', path),
                 data=body,
         )
         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='/', ext=None):
         body = None
         if ext:
             body = json.dumps({'ext': ext})
         return self._req('POST', path, body)
 
     def upload_untitled(self, body, path='/'):
         return self._req('POST', path, body)
 
     def copy_untitled(self, copy_from, path='/'):
         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 mkdir(self, name, path='/'):
         return self._req('PUT', url_path_join(path, name), json.dumps({'type': 'directory'}))
 
     def copy(self, copy_from, copy_to, path='/'):
         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 = json.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'),
                 ('ordering', 'A'),
                 ('ordering', 'b'),
                 ('ordering', 'C'),
                 (u'å b', u'ç d'),
                ]
     hidden_dirs = ['.hidden', '__pycache__']
 
     dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
     del dirs[0]  # remove ''
     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 setUp(self):
         nbdir = self.notebook_dir.name
         self.blob = os.urandom(100)
         self.b64_blob = base64.encodestring(self.blob).decode('ascii')
 
 
 
         for d in (self.dirs + self.hidden_dirs):
             d.replace('/', os.sep)
             if not os.path.isdir(pjoin(nbdir, d)):
                 os.mkdir(pjoin(nbdir, d))
 
         for d, name in self.dirs_nbs:
             d = d.replace('/', os.sep)
             # create a notebook
             with io.open(pjoin(nbdir, d, '%s.ipynb' % name), 'w',
                          encoding='utf-8') as f:
                 nb = new_notebook()
-                write(nb, f, format='ipynb')
+                write(f, nb, version=4)
 
             # create a text file
             with io.open(pjoin(nbdir, d, '%s.txt' % name), 'w',
                          encoding='utf-8') as f:
                 f.write(self._txt_for_name(name))
 
             # create a binary file
             with io.open(pjoin(nbdir, d, '%s.blob' % name), 'wb') as f:
                 f.write(self._blob_for_name(name))
 
         self.api = API(self.base_url())
 
     def tearDown(self):
         nbdir = self.notebook_dir.name
 
         for dname in (list(self.top_level_dirs) + self.hidden_dirs):
             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 = 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é')
 
         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')
 
         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_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:
             nb = self.api.read('%s.ipynb' % name, d+'/').json()
             self.assertEqual(nb['name'], u'%s.ipynb' % name)
             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_contents_no_such_file(self):
         # Name that doesn't exist - should be a 404
         with assert_http_error(404):
             self.api.read('q.ipynb', 'foo')
 
     def test_get_text_file_contents(self):
         for d, name in self.dirs_nbs:
             model = self.api.read(u'%s.txt' % name, d+'/').json()
             self.assertEqual(model['name'], u'%s.txt' % name)
             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('q.txt', 'foo')
 
     def test_get_binary_file_contents(self):
         for d, name in self.dirs_nbs:
             model = self.api.read(u'%s.blob' % name, d+'/').json()
             self.assertEqual(model['name'], u'%s.blob' % name)
             self.assertIn('content', model)
             self.assertEqual(model['format'], 'base64')
             self.assertEqual(model['type'], 'file')
             b64_data = base64.encodestring(self._blob_for_name(name)).decode('ascii')
             self.assertEqual(model['content'], b64_data)
 
         # Name that doesn't exist - should be a 404
         with assert_http_error(404):
             self.api.read('q.txt', 'foo')
 
     def _check_created(self, resp, name, 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, name)))
         rjson = resp.json()
         self.assertEqual(rjson['name'], name)
         self.assertEqual(rjson['path'], path)
         self.assertEqual(rjson['type'], type)
         isright = os.path.isdir if type == 'directory' else os.path.isfile
         assert isright(pjoin(
             self.notebook_dir.name,
             path.replace('/', os.sep),
             name,
         ))
 
     def test_create_untitled(self):
         resp = self.api.create_untitled(path=u'å b')
         self._check_created(resp, 'Untitled0.ipynb', u'å b')
 
         # Second time
         resp = self.api.create_untitled(path=u'å b')
         self._check_created(resp, 'Untitled1.ipynb', u'å b')
 
         # And two directories down
         resp = self.api.create_untitled(path='foo/bar')
         self._check_created(resp, 'Untitled0.ipynb', 'foo/bar')
 
     def test_create_untitled_txt(self):
         resp = self.api.create_untitled(path='foo/bar', ext='.txt')
         self._check_created(resp, 'untitled0.txt', 'foo/bar', type='file')
 
         resp = self.api.read(path='foo/bar', name='untitled0.txt')
         model = resp.json()
         self.assertEqual(model['type'], 'file')
         self.assertEqual(model['format'], 'text')
         self.assertEqual(model['content'], '')
 
     def test_upload_untitled(self):
         nb = new_notebook()
         nbmodel = {'content': nb, 'type': 'notebook'}
         resp = self.api.upload_untitled(path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_created(resp, 'Untitled0.ipynb', u'å b')
 
     def test_upload(self):
         nb = new_notebook()
         nbmodel = {'content': nb, 'type': 'notebook'}
         resp = self.api.upload(u'Upload tést.ipynb', path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_created(resp, u'Upload tést.ipynb', u'å b')
 
     def test_mkdir(self):
         resp = self.api.mkdir(u'New ∂ir', path=u'å b')
         self._check_created(resp, u'New ∂ir', u'å b', type='directory')
 
     def test_mkdir_hidden_400(self):
         with assert_http_error(400):
             resp = self.api.mkdir(u'.hidden', path=u'å b')
 
     def test_upload_txt(self):
         body = u'ünicode téxt'
         model = {
             'content' : body,
             'format'  : 'text',
             'type'    : 'file',
         }
         resp = self.api.upload(u'Upload tést.txt', path=u'å b',
                                               body=json.dumps(model))
 
         # check roundtrip
         resp = self.api.read(path=u'å b', name=u'Upload tést.txt')
         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',
         }
         resp = self.api.upload(u'Upload tést.blob', path=u'å b',
                                               body=json.dumps(model))
 
         # check roundtrip
         resp = self.api.read(path=u'å b', name=u'Upload tést.blob')
         model = resp.json()
         self.assertEqual(model['type'], 'file')
         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'}
         resp = self.api.upload(u'Upload tést.ipynb', path=u'å b',
                                               body=json.dumps(nbmodel))
         self._check_created(resp, u'Upload tést.ipynb', u'å b')
         resp = self.api.read(u'Upload tést.ipynb', u'å b')
         data = resp.json()
-        self.assertEqual(data['content']['nbformat'], current.nbformat)
+        self.assertEqual(data['content']['nbformat'], 4)
 
     def test_copy_untitled(self):
         resp = self.api.copy_untitled(u'ç d.ipynb', path=u'å b')
         self._check_created(resp, u'ç d-Copy0.ipynb', u'å b')
 
     def test_copy(self):
         resp = self.api.copy(u'ç d.ipynb', u'cøpy.ipynb', path=u'å b')
         self._check_created(resp, u'cøpy.ipynb', u'å b')
 
     def test_copy_path(self):
         resp = self.api.copy(u'foo/a.ipynb', u'cøpyfoo.ipynb', path=u'å b')
         self._check_created(resp, u'cøpyfoo.ipynb', u'å b')
 
     def test_copy_dir_400(self):
         # can't copy directories
         with assert_http_error(400):
             resp = self.api.copy(u'å b', u'å c')
 
     def test_delete(self):
         for d, name in self.dirs_nbs:
             resp = self.api.delete('%s.ipynb' % name, d)
             self.assertEqual(resp.status_code, 204)
 
         for d in self.dirs + ['/']:
             nbs = notebooks_only(self.api.list(d).json())
             self.assertEqual(len(nbs), 0)
 
     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['name'], 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('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 = 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('a.ipynb', 'foo', 'b.ipynb')
 
     def test_save(self):
         resp = self.api.read('a.ipynb', 'foo')
         nbcontent = json.loads(resp.text)['content']
         nb = from_dict(nbcontent)
         nb.cells.append(new_markdown_cell(u'Created by test ³'))
 
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb, 'type': 'notebook'}
         resp = self.api.save('a.ipynb', path='foo', body=json.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')
+            newnb = read(f, as_version=4)
         self.assertEqual(newnb.cells[0].source,
                          u'Created by test ³')
         nbcontent = self.api.read('a.ipynb', 'foo').json()['content']
         newnb = from_dict(nbcontent)
         self.assertEqual(newnb.cells[0].source,
                          u'Created by test ³')
 
         # Save and rename
         nbmodel= {'name': 'a2.ipynb', 'path':'foo/bar', 'content': nb, 'type': 'notebook'}
         resp = self.api.save('a.ipynb', path='foo', body=json.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.api.read('a.ipynb', 'foo')
 
     def test_checkpoints(self):
         resp = self.api.read('a.ipynb', 'foo')
         r = self.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 = json.loads(resp.text)['content']
         nb = from_dict(nbcontent)
         hcell = new_markdown_cell('Created by test')
         nb.cells.append(hcell)
         # Save
         nbmodel= {'name': 'a.ipynb', 'path':'foo', 'content': nb, 'type': 'notebook'}
         resp = self.api.save('a.ipynb', path='foo', body=json.dumps(nbmodel))
 
         # List checkpoints
         cps = self.api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [cp1])
 
         nbcontent = self.api.read('a.ipynb', 'foo').json()['content']
         nb = from_dict(nbcontent)
         self.assertEqual(nb.cells[0].source, 'Created by test')
 
         # Restore cp1
         r = self.api.restore_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         nbcontent = self.api.read('a.ipynb', 'foo').json()['content']
         nb = from_dict(nbcontent)
         self.assertEqual(nb.cells, [])
 
         # Delete cp1
         r = self.api.delete_checkpoint('a.ipynb', 'foo', cp1['id'])
         self.assertEqual(r.status_code, 204)
         cps = self.api.get_checkpoints('a.ipynb', 'foo').json()
         self.assertEqual(cps, [])

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

@@ -1,332 +1,332
 # coding: utf-8
 """Tests for the notebook manager."""
 from __future__ import print_function
 
 import logging
 import os
 
 from tornado.web import HTTPError
 from unittest import TestCase
 from tempfile import NamedTemporaryFile
 
-from IPython.nbformat import current
+from IPython.nbformat import v4 as nbformat
 
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.utils.traitlets import TraitError
 from IPython.html.utils import url_path_join
 from IPython.testing import decorators as dec
 
 from ..filemanager import FileContentsManager
 from ..manager import ContentsManager
 
 
 class TestFileContentsManager(TestCase):
 
     def test_root_dir(self):
         with TemporaryDirectory() as td:
             fm = FileContentsManager(root_dir=td)
             self.assertEqual(fm.root_dir, td)
 
     def test_missing_root_dir(self):
         with TemporaryDirectory() as td:
             root = os.path.join(td, 'notebook', 'dir', 'is', 'missing')
             self.assertRaises(TraitError, FileContentsManager, root_dir=root)
 
     def test_invalid_root_dir(self):
         with NamedTemporaryFile() as tf:
             self.assertRaises(TraitError, FileContentsManager, root_dir=tf.name)
 
     def test_get_os_path(self):
         # full filesystem path should be returned with correct operating system
         # separators.
         with TemporaryDirectory() as td:
             root = td
             fm = FileContentsManager(root_dir=root)
             path = fm._get_os_path('test.ipynb', '/path/to/notebook/')
             rel_path_list =  '/path/to/notebook/test.ipynb'.split('/')
             fs_path = os.path.join(fm.root_dir, *rel_path_list)
             self.assertEqual(path, fs_path)
 
             fm = FileContentsManager(root_dir=root)
             path = fm._get_os_path('test.ipynb')
             fs_path = os.path.join(fm.root_dir, 'test.ipynb')
             self.assertEqual(path, fs_path)
 
             fm = FileContentsManager(root_dir=root)
             path = fm._get_os_path('test.ipynb', '////')
             fs_path = os.path.join(fm.root_dir, 'test.ipynb')
             self.assertEqual(path, fs_path)
 
     def test_checkpoint_subdir(self):
         subd = u'sub ∂ir'
         cp_name = 'test-cp.ipynb'
         with TemporaryDirectory() as td:
             root = td
             os.mkdir(os.path.join(td, subd))
             fm = FileContentsManager(root_dir=root)
             cp_dir = fm.get_checkpoint_path('cp', 'test.ipynb', '/')
             cp_subdir = fm.get_checkpoint_path('cp', 'test.ipynb', '/%s/' % subd)
         self.assertNotEqual(cp_dir, cp_subdir)
         self.assertEqual(cp_dir, os.path.join(root, fm.checkpoint_dir, cp_name))
         self.assertEqual(cp_subdir, os.path.join(root, subd, fm.checkpoint_dir, cp_name))
 
 
 class TestContentsManager(TestCase):
 
     def setUp(self):
         self._temp_dir = TemporaryDirectory()
         self.td = self._temp_dir.name
         self.contents_manager = FileContentsManager(
             root_dir=self.td,
             log=logging.getLogger()
         )
 
     def tearDown(self):
         self._temp_dir.cleanup()
 
     def make_dir(self, abs_path, rel_path):
         """make subdirectory, rel_path is the relative path
         to that directory from the location where the server started"""
         os_path = os.path.join(abs_path, rel_path)
         try:
             os.makedirs(os_path)
         except OSError:
             print("Directory already exists: %r" % os_path)
         return os_path
 
     def add_code_cell(self, nb):
-        output = current.new_output("display_data", {'application/javascript': "alert('hi');"})
-        cell = current.new_code_cell("print('hi')", outputs=[output])
+        output = nbformat.new_output("display_data", {'application/javascript': "alert('hi');"})
+        cell = nbformat.new_code_cell("print('hi')", outputs=[output])
         nb.cells.append(cell)
 
     def new_notebook(self):
         cm = self.contents_manager
         model = cm.create_file()
         name = model['name']
         path = model['path']
 
         full_model = cm.get_model(name, path)
         nb = full_model['content']
         self.add_code_cell(nb)
 
         cm.save(full_model, name, path)
         return nb, name, path
 
     def test_create_file(self):
         cm = self.contents_manager
         # Test in root directory
         model = cm.create_file()
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], 'Untitled0.ipynb')
         self.assertEqual(model['path'], '')
 
         # Test in sub-directory
         sub_dir = '/foo/'
         self.make_dir(cm.root_dir, 'foo')
         model = cm.create_file(None, sub_dir)
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], 'Untitled0.ipynb')
         self.assertEqual(model['path'], sub_dir.strip('/'))
 
     def test_get(self):
         cm = self.contents_manager
         # Create a notebook
         model = cm.create_file()
         name = model['name']
         path = model['path']
 
         # Check that we 'get' on the notebook we just created
         model2 = cm.get_model(name, path)
         assert isinstance(model2, dict)
         self.assertIn('name', model2)
         self.assertIn('path', model2)
         self.assertEqual(model['name'], name)
         self.assertEqual(model['path'], path)
 
         # Test in sub-directory
         sub_dir = '/foo/'
         self.make_dir(cm.root_dir, 'foo')
         model = cm.create_file(None, sub_dir)
         model2 = cm.get_model(name, sub_dir)
         assert isinstance(model2, dict)
         self.assertIn('name', model2)
         self.assertIn('path', model2)
         self.assertIn('content', model2)
         self.assertEqual(model2['name'], 'Untitled0.ipynb')
         self.assertEqual(model2['path'], sub_dir.strip('/'))
     
     @dec.skip_win32
     def test_bad_symlink(self):
         cm = self.contents_manager
         path = 'test bad symlink'
         os_path = self.make_dir(cm.root_dir, path)
         
         file_model = cm.create_file(path=path, ext='.txt')
         
         # create a broken symlink
         os.symlink("target", os.path.join(os_path, "bad symlink"))
         model = cm.get_model(path)
         self.assertEqual(model['content'], [file_model])
     
     @dec.skip_win32
     def test_good_symlink(self):
         cm = self.contents_manager
         path = 'test good symlink'
         os_path = self.make_dir(cm.root_dir, path)
         
         file_model = cm.create_file(path=path, ext='.txt')
         
         # create a good symlink
         os.symlink(file_model['name'], os.path.join(os_path, "good symlink"))
         symlink_model = cm.get_model(name="good symlink", path=path, content=False)
         
         dir_model = cm.get_model(path)
         self.assertEqual(
             sorted(dir_model['content'], key=lambda x: x['name']),
             [symlink_model, file_model],
         )
     
     def test_update(self):
         cm = self.contents_manager
         # Create a notebook
         model = cm.create_file()
         name = model['name']
         path = model['path']
 
         # Change the name in the model for rename
         model['name'] = 'test.ipynb'
         model = cm.update(model, name, path)
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], 'test.ipynb')
 
         # Make sure the old name is gone
         self.assertRaises(HTTPError, cm.get_model, name, path)
 
         # Test in sub-directory
         # Create a directory and notebook in that directory
         sub_dir = '/foo/'
         self.make_dir(cm.root_dir, 'foo')
         model = cm.create_file(None, sub_dir)
         name = model['name']
         path = model['path']
 
         # Change the name in the model for rename
         model['name'] = 'test_in_sub.ipynb'
         model = cm.update(model, name, path)
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], 'test_in_sub.ipynb')
         self.assertEqual(model['path'], sub_dir.strip('/'))
 
         # Make sure the old name is gone
         self.assertRaises(HTTPError, cm.get_model, name, path)
 
     def test_save(self):
         cm = self.contents_manager
         # Create a notebook
         model = cm.create_file()
         name = model['name']
         path = model['path']
 
         # Get the model with 'content'
         full_model = cm.get_model(name, path)
 
         # Save the notebook
         model = cm.save(full_model, name, path)
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], name)
         self.assertEqual(model['path'], path)
 
         # Test in sub-directory
         # Create a directory and notebook in that directory
         sub_dir = '/foo/'
         self.make_dir(cm.root_dir, 'foo')
         model = cm.create_file(None, sub_dir)
         name = model['name']
         path = model['path']
         model = cm.get_model(name, path)
 
         # Change the name in the model for rename
         model = cm.save(model, name, path)
         assert isinstance(model, dict)
         self.assertIn('name', model)
         self.assertIn('path', model)
         self.assertEqual(model['name'], 'Untitled0.ipynb')
         self.assertEqual(model['path'], sub_dir.strip('/'))
 
     def test_delete(self):
         cm = self.contents_manager
         # Create a notebook
         nb, name, path = self.new_notebook()
 
         # Delete the notebook
         cm.delete(name, path)
 
         # Check that a 'get' on the deleted notebook raises and error
         self.assertRaises(HTTPError, cm.get_model, name, path)
 
     def test_copy(self):
         cm = self.contents_manager
         path = u'å b'
         name = u'nb √.ipynb'
         os.mkdir(os.path.join(cm.root_dir, path))
         orig = cm.create_file({'name' : name}, path=path)
 
         # copy with unspecified name
         copy = cm.copy(name, path=path)
         self.assertEqual(copy['name'], orig['name'].replace('.ipynb', '-Copy0.ipynb'))
 
         # copy with specified name
         copy2 = cm.copy(name, u'copy 2.ipynb', path=path)
         self.assertEqual(copy2['name'], u'copy 2.ipynb')
 
     def test_trust_notebook(self):
         cm = self.contents_manager
         nb, name, path = self.new_notebook()
 
         untrusted = cm.get_model(name, path)['content']
         assert not cm.notary.check_cells(untrusted)
 
         # print(untrusted)
         cm.trust_notebook(name, path)
         trusted = cm.get_model(name, path)['content']
         # print(trusted)
         assert cm.notary.check_cells(trusted)
 
     def test_mark_trusted_cells(self):
         cm = self.contents_manager
         nb, name, path = self.new_notebook()
 
         cm.mark_trusted_cells(nb, name, path)
         for cell in nb.cells:
             if cell.cell_type == 'code':
                 assert not cell.metadata.trusted
 
         cm.trust_notebook(name, path)
         nb = cm.get_model(name, path)['content']
         for cell in nb.cells:
             if cell.cell_type == 'code':
                 assert cell.metadata.trusted
 
     def test_check_and_sign(self):
         cm = self.contents_manager
         nb, name, path = self.new_notebook()
 
         cm.mark_trusted_cells(nb, name, path)
         cm.check_and_sign(nb, name, path)
         assert not cm.notary.check_signature(nb)
 
         cm.trust_notebook(name, path)
         nb = cm.get_model(name, path)['content']
         cm.mark_trusted_cells(nb, name, path)
         cm.check_and_sign(nb, name, path)
         assert cm.notary.check_signature(nb)

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

@@ -1,116 +1,117
 """Test the sessions web service API."""
 
 import errno
 import io
 import os
 import json
 import requests
 import shutil
 
 pjoin = os.path.join
 
 from IPython.html.utils import url_path_join
 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
-from IPython.nbformat.current import new_notebook, write
+from IPython.nbformat.v4 import new_notebook
+from IPython.nbformat import write
 
 class SessionAPI(object):
     """Wrapper for notebook API calls."""
     def __init__(self, base_url):
         self.base_url = base_url
 
     def _req(self, verb, path, body=None):
         response = requests.request(verb,
                 url_path_join(self.base_url, 'api/sessions', path), data=body)
 
         if 400 <= response.status_code < 600:
             try:
                 response.reason = response.json()['message']
             except:
                 pass
         response.raise_for_status()
 
         return response
 
     def list(self):
         return self._req('GET', '')
 
     def get(self, id):
         return self._req('GET', id)
 
     def create(self, name, path, kernel_name='python'):
         body = json.dumps({'notebook': {'name':name, 'path':path},
                            'kernel': {'name': kernel_name}})
         return self._req('POST', '', body)
 
     def modify(self, id, name, path):
         body = json.dumps({'notebook': {'name':name, 'path':path}})
         return self._req('PATCH', id, body)
 
     def delete(self, id):
         return self._req('DELETE', id)
 
 class SessionAPITest(NotebookTestBase):
     """Test the sessions web service API"""
     def setUp(self):
         nbdir = self.notebook_dir.name
         try:
             os.mkdir(pjoin(nbdir, 'foo'))
         except OSError as e:
             # Deleting the folder in an earlier test may have failed
             if e.errno != errno.EEXIST:
                 raise
 
         with io.open(pjoin(nbdir, 'foo', 'nb1.ipynb'), 'w',
                      encoding='utf-8') as f:
             nb = new_notebook()
-            write(nb, f, format='ipynb')
+            write(f, nb, version=4)
 
         self.sess_api = SessionAPI(self.base_url())
 
     def tearDown(self):
         for session in self.sess_api.list().json():
             self.sess_api.delete(session['id'])
         shutil.rmtree(pjoin(self.notebook_dir.name, 'foo'),
                       ignore_errors=True)
 
     def test_create(self):
         sessions = self.sess_api.list().json()
         self.assertEqual(len(sessions), 0)
 
         resp = self.sess_api.create('nb1.ipynb', 'foo')
         self.assertEqual(resp.status_code, 201)
         newsession = resp.json()
         self.assertIn('id', newsession)
         self.assertEqual(newsession['notebook']['name'], 'nb1.ipynb')
         self.assertEqual(newsession['notebook']['path'], 'foo')
         self.assertEqual(resp.headers['Location'], '/api/sessions/{0}'.format(newsession['id']))
 
         sessions = self.sess_api.list().json()
         self.assertEqual(sessions, [newsession])
 
         # Retrieve it
         sid = newsession['id']
         got = self.sess_api.get(sid).json()
         self.assertEqual(got, newsession)
 
     def test_delete(self):
         newsession = self.sess_api.create('nb1.ipynb', 'foo').json()
         sid = newsession['id']
 
         resp = self.sess_api.delete(sid)
         self.assertEqual(resp.status_code, 204)
 
         sessions = self.sess_api.list().json()
         self.assertEqual(sessions, [])
 
         with assert_http_error(404):
             self.sess_api.get(sid)
 
     def test_modify(self):
         newsession = self.sess_api.create('nb1.ipynb', 'foo').json()
         sid = newsession['id']
 
         changed = self.sess_api.modify(sid, 'nb2.ipynb', '').json()
         self.assertEqual(changed['id'], sid)
         self.assertEqual(changed['notebook']['name'], 'nb2.ipynb')
         self.assertEqual(changed['notebook']['path'], '')

IPython/html/tests/test_files.py

@@ -1,151 +1,152
 # coding: utf-8
 """Test the /files/ handler."""
 
 import io
 import os
 from unicodedata import normalize
 
 pjoin = os.path.join
 
 import requests
 import json
 
-from IPython.nbformat.current import (new_notebook, write,
+from IPython.nbformat import write
+from IPython.nbformat.v4 import (new_notebook,
                               new_markdown_cell, new_code_cell,
                               new_output)
 
 from IPython.html.utils import url_path_join
 from .launchnotebook import NotebookTestBase
 from IPython.utils import py3compat
 
 
 class FilesTest(NotebookTestBase):
     def test_hidden_files(self):
         not_hidden = [
             u'å b',
             u'å b/ç. d',
         ]
         hidden = [
             u'.å b',
             u'å b/.ç d',
         ]
         dirs = not_hidden + hidden
         
         nbdir = self.notebook_dir.name
         for d in dirs:
             path = pjoin(nbdir, d.replace('/', os.sep))
             if not os.path.exists(path):
                 os.mkdir(path)
             with open(pjoin(path, 'foo'), 'w') as f:
                 f.write('foo')
             with open(pjoin(path, '.foo'), 'w') as f:
                 f.write('.foo')
         url = self.base_url()
         
         for d in not_hidden:
             path = pjoin(nbdir, d.replace('/', os.sep))
             r = requests.get(url_path_join(url, 'files', d, 'foo'))
             r.raise_for_status()
             self.assertEqual(r.text, 'foo')
             r = requests.get(url_path_join(url, 'files', d, '.foo'))
             self.assertEqual(r.status_code, 404)
             
         for d in hidden:
             path = pjoin(nbdir, d.replace('/', os.sep))
             for foo in ('foo', '.foo'):
                 r = requests.get(url_path_join(url, 'files', d, foo))
                 self.assertEqual(r.status_code, 404)
     
     def test_contents_manager(self):
         "make sure ContentsManager returns right files (ipynb, bin, txt)."
 
         nbdir = self.notebook_dir.name
         base = self.base_url()
 
         nb = new_notebook(
             cells=[
                 new_markdown_cell(u'Created by test ³'),
                 new_code_cell("print(2*6)", outputs=[
                     new_output("stream", text="12"),
                 ])
             ]
         )
 
         with io.open(pjoin(nbdir, 'testnb.ipynb'), 'w', 
             encoding='utf-8') as f:
-            write(nb, f)
+            write(f, nb, version=4)
 
         with io.open(pjoin(nbdir, 'test.bin'), 'wb') as f:
             f.write(b'\xff' + os.urandom(5))
             f.close()
 
         with io.open(pjoin(nbdir, 'test.txt'), 'w') as f:
             f.write(u'foobar')
             f.close()
 
         r = requests.get(url_path_join(base, 'files', 'testnb.ipynb'))
         self.assertEqual(r.status_code, 200)
         self.assertIn('print(2*6)', r.text)
         json.loads(r.text)
 
         r = requests.get(url_path_join(base, 'files', 'test.bin'))
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.headers['content-type'], 'application/octet-stream')
         self.assertEqual(r.content[:1], b'\xff')
         self.assertEqual(len(r.content), 6)
 
         r = requests.get(url_path_join(base, 'files', 'test.txt'))
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.headers['content-type'], 'text/plain')
         self.assertEqual(r.text, 'foobar')
     
     def test_download(self):
         nbdir = self.notebook_dir.name
         base = self.base_url()
         
         text = 'hello'
         with open(pjoin(nbdir, 'test.txt'), 'w') as f:
             f.write(text)
         
         r = requests.get(url_path_join(base, 'files', 'test.txt'))
         disposition = r.headers.get('Content-Disposition', '')
         self.assertNotIn('attachment', disposition)
 
         r = requests.get(url_path_join(base, 'files', 'test.txt') + '?download=1')
         disposition = r.headers.get('Content-Disposition', '')
         self.assertIn('attachment', disposition)
         self.assertIn('filename="test.txt"', disposition)
 
     def test_old_files_redirect(self):
         """pre-2.0 'files/' prefixed links are properly redirected"""
         nbdir = self.notebook_dir.name
         base = self.base_url()
         
         os.mkdir(pjoin(nbdir, 'files'))
         os.makedirs(pjoin(nbdir, 'sub', 'files'))
         
         for prefix in ('', 'sub'):
             with open(pjoin(nbdir, prefix, 'files', 'f1.txt'), 'w') as f:
                 f.write(prefix + '/files/f1')
             with open(pjoin(nbdir, prefix, 'files', 'f2.txt'), 'w') as f:
                 f.write(prefix + '/files/f2')
             with open(pjoin(nbdir, prefix, 'f2.txt'), 'w') as f:
                 f.write(prefix + '/f2')
             with open(pjoin(nbdir, prefix, 'f3.txt'), 'w') as f:
                 f.write(prefix + '/f3')
             
             url = url_path_join(base, 'notebooks', prefix, 'files', 'f1.txt')
             r = requests.get(url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/files/f1')
 
             url = url_path_join(base, 'notebooks', prefix, 'files', 'f2.txt')
             r = requests.get(url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/files/f2')
 
             url = url_path_join(base, 'notebooks', prefix, 'files', 'f3.txt')
             r = requests.get(url)
             self.assertEqual(r.status_code, 200)
             self.assertEqual(r.text, prefix + '/f3')