Merge pull request #7191 from minrk/contents-test-reuse

abstract some methods in contents service tests

IPython/html/services/contents/filemanager.py

@@ -357,6 +357,7 class FileContentsManager(ContentsManager):
             if model['content'] is None:
                 model['content'] = base64.encodestring(bcontent).decode('ascii')
                 model['format'] = 'base64'
+            if model['format'] == 'base64':
                 default_mime = 'application/octet-stream'
 
             model['mimetype'] = mimetypes.guess_type(os_path)[0] or default_mime

IPython/html/services/contents/manager.py

@@ -347,6 +347,9 class ContentsManager(LoggingConfigurable):
         from_path must be a full path to a file.
         """
         path = from_path.strip('/')
+        if to_path is not None:
+            to_path = to_path.strip('/')
+
         if '/' in path:
             from_dir, from_name = path.rsplit('/', 1)
         else:
@@ -359,7 +362,7 class ContentsManager(LoggingConfigurable):
         if model['type'] == 'directory':
             raise HTTPError(400, "Can't copy directories")
         
-        if not to_path:
+        if to_path is None:
             to_path = from_dir
         if self.dir_exists(to_path):
             name = copy_pat.sub(u'.', from_name)

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

@@ -12,7 +12,7 pjoin = os.path.join
 
 import requests
 
-from IPython.html.utils import url_path_join, url_escape
+from IPython.html.utils import url_path_join, url_escape, to_os_path
 from IPython.html.tests.launchnotebook import NotebookTestBase, assert_http_error
 from IPython.nbformat import read, write, from_dict
 from IPython.nbformat.v4 import (
@@ -73,9 +73,6 class API(object):
     def upload(self, path, body):
         return self._req('PUT', path, body)
 
-    def mkdir_untitled(self, path='/'):
-        return self._req('POST', path, json.dumps({'type': 'directory'}))
-
     def mkdir(self, path='/'):
         return self._req('PUT', path, json.dumps({'type': 'directory'}))
 
@@ -122,8 +119,8 class APITest(NotebookTestBase):
                ]
     hidden_dirs = ['.hidden', '__pycache__']
 
-    dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs])
-    del dirs[0]  # remove ''
+    # Don't include root dir.
+    dirs = uniq_stable([py3compat.cast_unicode(d) for (d,n) in dirs_nbs[1:]])
     top_level_dirs = {normalize('NFC', d.split('/')[0]) for d in dirs}
 
     @staticmethod
@@ -133,44 +130,77 class APITest(NotebookTestBase):
     @staticmethod
     def _txt_for_name(name):
         return u'%s text file' % name
-
+    
+    def to_os_path(self, api_path):
+        return to_os_path(api_path, root=self.notebook_dir.name)
+    
+    def make_dir(self, api_path):
+        """Create a directory at api_path"""
+        os_path = self.to_os_path(api_path)
+        try:
+            os.makedirs(os_path)
+        except OSError:
+            print("Directory already exists: %r" % os_path)
+
+    def make_txt(self, api_path, txt):
+        """Make a text file at a given api_path"""
+        os_path = self.to_os_path(api_path)
+        with io.open(os_path, 'w', encoding='utf-8') as f:
+            f.write(txt)
+    
+    def make_blob(self, api_path, blob):
+        """Make a binary file at a given api_path"""
+        os_path = self.to_os_path(api_path)
+        with io.open(os_path, 'wb') as f:
+            f.write(blob)
+    
+    def make_nb(self, api_path, nb):
+        """Make a notebook file at a given api_path"""
+        os_path = self.to_os_path(api_path)
+        
+        with io.open(os_path, 'w', encoding='utf-8') as f:
+            write(nb, f, version=4)
+
+    def delete_dir(self, api_path):
+        """Delete a directory at api_path, removing any contents."""
+        os_path = self.to_os_path(api_path)
+        shutil.rmtree(os_path, ignore_errors=True)
+
+    def delete_file(self, api_path):
+        """Delete a file at the given path if it exists."""
+        if self.isfile(api_path):
+            os.unlink(self.to_os_path(api_path))
+    
+    def isfile(self, api_path):
+        return os.path.isfile(self.to_os_path(api_path))
+    
+    def isdir(self, api_path):
+        return os.path.isdir(self.to_os_path(api_path))
+    
     def setUp(self):
-        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))
+            self.make_dir(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, version=4)
-
+            nb = new_notebook()
+            self.make_nb(u'{}/{}.ipynb'.format(d, name), nb)
+            
             # 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))
-
+            txt = self._txt_for_name(name)
+            self.make_txt(u'{}/{}.txt'.format(d, name), txt)
+            
             # create a binary file
-            with io.open(pjoin(nbdir, d, '%s.blob' % name), 'wb') as f:
-                f.write(self._blob_for_name(name))
+            blob = self._blob_for_name(name)
+            self.make_blob(u'{}/{}.blob'.format(d, name), blob)
 
         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'))
+            self.delete_dir(dname)
+        self.delete_file('inroot.ipynb')
 
     def test_list_notebooks(self):
         nbs = notebooks_only(self.api.list().json())
@@ -204,11 +234,8 class APITest(NotebookTestBase):
         self.assertEqual(nbnames, expected)
 
     def test_list_dirs(self):
-        print(self.api.list().json())
         dirs = dirs_only(self.api.list().json())
         dir_names = {normalize('NFC', d['name']) for d in dirs}
-        print(dir_names)
-        print(self.top_level_dirs)
         self.assertEqual(dir_names, self.top_level_dirs)  # Excluding hidden dirs
 
     def test_list_nonexistant_dir(self):
@@ -261,8 +288,10 class APITest(NotebookTestBase):
             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)
+            self.assertEqual(
+                base64.decodestring(model['content'].encode('ascii')),
+                self._blob_for_name(name),
+            )
 
         # Name that doesn't exist - should be a 404
         with assert_http_error(404):
@@ -283,11 +312,8 class APITest(NotebookTestBase):
         self.assertEqual(rjson['name'], path.rsplit('/', 1)[-1])
         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),
-        ))
+        isright = self.isdir if type == 'directory' else self.isfile
+        assert isright(path)
 
     def test_create_untitled(self):
         resp = self.api.create_untitled(path=u'å b')
@@ -451,7 +477,7 class APITest(NotebookTestBase):
         self.assertEqual(resp.headers['Location'].split('/')[-1], 'z.ipynb')
         self.assertEqual(resp.json()['name'], 'z.ipynb')
         self.assertEqual(resp.json()['path'], 'foo/z.ipynb')
-        assert os.path.isfile(pjoin(self.notebook_dir.name, 'foo', 'z.ipynb'))
+        assert self.isfile('foo/z.ipynb')
 
         nbs = notebooks_only(self.api.list('foo').json())
         nbnames = set(n['name'] for n in nbs)
@@ -471,11 +497,6 class APITest(NotebookTestBase):
         nbmodel= {'content': nb, 'type': 'notebook'}
         resp = self.api.save('foo/a.ipynb', 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, as_version=4)
-        self.assertEqual(newnb.cells[0].source,
-                         u'Created by test ³')
         nbcontent = self.api.read('foo/a.ipynb').json()['content']
         newnb = from_dict(nbcontent)
         self.assertEqual(newnb.cells[0].source,

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

@@ -2,7 +2,6
 """Tests for the notebook manager."""
 from __future__ import print_function
 
-import logging
 import os
 
 from tornado.web import HTTPError
@@ -17,11 +16,31 from IPython.html.utils import url_path_join
 from IPython.testing import decorators as dec
 
 from ..filemanager import FileContentsManager
-from ..manager import ContentsManager
+
+
+def _make_dir(contents_manager, api_path):
+    """
+    Make a directory.
+    """
+    os_path = contents_manager._get_os_path(api_path)
+    try:
+        os.makedirs(os_path)
+    except OSError:
+        print("Directory already exists: %r" % os_path)
 
 
 class TestFileContentsManager(TestCase):
 
+    def symlink(self, contents_manager, src, dst):
+        """Make a symlink to src from dst
+        
+        src and dst are api_paths
+        """
+        src_os_path = contents_manager._get_os_path(src)
+        dst_os_path = contents_manager._get_os_path(dst)
+        print(src_os_path, dst_os_path, os.path.isfile(src_os_path))
+        os.symlink(src_os_path, dst_os_path)
+
     def test_root_dir(self):
         with TemporaryDirectory() as td:
             fm = FileContentsManager(root_dir=td)
@@ -69,10 +88,44 class TestFileContentsManager(TestCase):
         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))
+    
+    @dec.skip_win32
+    def test_bad_symlink(self):
+        with TemporaryDirectory() as td:
+            cm = FileContentsManager(root_dir=td)
+            path = 'test bad symlink'
+            _make_dir(cm, path)
 
+            file_model = cm.new_untitled(path=path, ext='.txt')
+
+            # create a broken symlink
+            self.symlink(cm, "target", '%s/%s' % (path, 'bad symlink'))
+            model = cm.get(path)
+            self.assertEqual(model['content'], [file_model])
+    
+    @dec.skip_win32
+    def test_good_symlink(self):
+        with TemporaryDirectory() as td:
+            cm = FileContentsManager(root_dir=td)
+            parent = 'test good symlink'
+            name = 'good symlink'
+            path = '{0}/{1}'.format(parent, name)
+            _make_dir(cm, parent)
+
+            file_model = cm.new(path=parent + '/zfoo.txt')
+
+            # create a good symlink
+            self.symlink(cm, file_model['path'], path)
+            symlink_model = cm.get(path, content=False)
+            dir_model = cm.get(parent)
+            self.assertEqual(
+                sorted(dir_model['content'], key=lambda x: x['name']),
+                [symlink_model, file_model],
+            )
 
-class TestContentsManager(TestCase):
 
+class TestContentsManager(TestCase):
+    
     def setUp(self):
         self._temp_dir = TemporaryDirectory()
         self.td = self._temp_dir.name
@@ -83,15 +136,12 class TestContentsManager(TestCase):
     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 make_dir(self, api_path):
+        """make a subdirectory at api_path
+        
+        override in subclasses if contents are not on the filesystem.
+        """
+        _make_dir(self.contents_manager, api_path)
 
     def add_code_cell(self, nb):
         output = nbformat.new_output("display_data", {'application/javascript': "alert('hi');"})
@@ -169,7 +219,7 class TestContentsManager(TestCase):
 
         # Test in sub-directory
         sub_dir = '/foo/'
-        self.make_dir(cm.root_dir, 'foo')
+        self.make_dir('foo')
         model = cm.new_untitled(path=sub_dir, ext='.ipynb')
         model2 = cm.get(sub_dir + name)
         assert isinstance(model2, dict)
@@ -179,46 +229,59 class TestContentsManager(TestCase):
         self.assertEqual(model2['name'], 'Untitled.ipynb')
         self.assertEqual(model2['path'], '{0}/{1}'.format(sub_dir.strip('/'), name))
 
+        # Test with a regular file.
+        file_model_path = cm.new_untitled(path=sub_dir, ext='.txt')['path']
+        file_model = cm.get(file_model_path)
+        self.assertDictContainsSubset(
+            {
+                'content': u'',
+                'format': u'text',
+                'mimetype': u'text/plain',
+                'name': u'untitled.txt',
+                'path': u'foo/untitled.txt',
+                'type': u'file',
+                'writable': True,
+            },
+            file_model,
+        )
+        self.assertIn('created', file_model)
+        self.assertIn('last_modified', file_model)
+
         # Test getting directory model
+
+        # Create a sub-sub directory to test getting directory contents with a
+        # subdir.
+        self.make_dir('foo/bar')
         dirmodel = cm.get('foo')
         self.assertEqual(dirmodel['type'], 'directory')
+        self.assertIsInstance(dirmodel['content'], list)
+        self.assertEqual(len(dirmodel['content']), 3)
+        self.assertEqual(dirmodel['path'], 'foo')
+        self.assertEqual(dirmodel['name'], 'foo')
+
+        # Directory contents should match the contents of each individual entry
+        # when requested with content=False.
+        model2_no_content = cm.get(sub_dir + name, content=False)
+        file_model_no_content = cm.get(u'foo/untitled.txt', content=False)
+        sub_sub_dir_no_content = cm.get('foo/bar', content=False)
+        self.assertEqual(sub_sub_dir_no_content['path'], 'foo/bar')
+        self.assertEqual(sub_sub_dir_no_content['name'], 'bar')
+
+        for entry in dirmodel['content']:
+            # Order isn't guaranteed by the spec, so this is a hacky way of
+            # verifying that all entries are matched.
+            if entry['path'] == sub_sub_dir_no_content['path']:
+                self.assertEqual(entry, sub_sub_dir_no_content)
+            elif entry['path'] == model2_no_content['path']:
+                self.assertEqual(entry, model2_no_content)
+            elif entry['path'] == file_model_no_content['path']:
+                self.assertEqual(entry, file_model_no_content)
+            else:
+                self.fail("Unexpected directory entry: %s" % entry())
 
         with self.assertRaises(HTTPError):
             cm.get('foo', type='file')
 
-    
-    @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.new_untitled(path=path, ext='.txt')
-        
-        # create a broken symlink
-        os.symlink("target", os.path.join(os_path, "bad symlink"))
-        model = cm.get(path)
-        self.assertEqual(model['content'], [file_model])
-    
-    @dec.skip_win32
-    def test_good_symlink(self):
-        cm = self.contents_manager
-        parent = 'test good symlink'
-        name = 'good symlink'
-        path = '{0}/{1}'.format(parent, name)
-        os_path = self.make_dir(cm.root_dir, parent)
-        
-        file_model = cm.new(path=parent + '/zfoo.txt')
-        
-        # create a good symlink
-        os.symlink(file_model['name'], os.path.join(os_path, name))
-        symlink_model = cm.get(path, content=False)
-        dir_model = cm.get(parent)
-        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
@@ -240,9 +303,8 class TestContentsManager(TestCase):
         # Test in sub-directory
         # Create a directory and notebook in that directory
         sub_dir = '/foo/'
-        self.make_dir(cm.root_dir, 'foo')
+        self.make_dir('foo')
         model = cm.new_untitled(path=sub_dir, type='notebook')
-        name = model['name']
         path = model['path']
 
         # Change the name in the model for rename
@@ -279,7 +341,7 class TestContentsManager(TestCase):
         # Test in sub-directory
         # Create a directory and notebook in that directory
         sub_dir = '/foo/'
-        self.make_dir(cm.root_dir, 'foo')
+        self.make_dir('foo')
         model = cm.new_untitled(path=sub_dir, type='notebook')
         name = model['name']
         path = model['path']
@@ -301,6 +363,9 class TestContentsManager(TestCase):
         # Delete the notebook
         cm.delete(path)
 
+        # Check that deleting a non-existent path raises an error.
+        self.assertRaises(HTTPError, cm.delete, path)
+
         # Check that a 'get' on the deleted notebook raises and error
         self.assertRaises(HTTPError, cm.get, path)
 
@@ -309,9 +374,9 class TestContentsManager(TestCase):
         parent = u'å b'
         name = u'nb √.ipynb'
         path = u'{0}/{1}'.format(parent, name)
-        os.mkdir(os.path.join(cm.root_dir, parent))
-        orig = cm.new(path=path)
+        self.make_dir(parent)
 
+        orig = cm.new(path=path)
         # copy with unspecified name
         copy = cm.copy(path)
         self.assertEqual(copy['name'], orig['name'].replace('.ipynb', '-Copy1.ipynb'))
@@ -366,3 +431,4 class TestContentsManager(TestCase):
         cm.mark_trusted_cells(nb, path)
         cm.check_and_sign(nb, path)
         assert cm.notary.check_signature(nb)
+

IPython/html/tests/launchnotebook.py

@@ -30,6 +30,7 class NotebookTestBase(TestCase):
     """
 
     port = 12341
+    config = None
 
     @classmethod
     def wait_until_alive(cls):
@@ -65,6 +66,7 class NotebookTestBase(TestCase):
             open_browser=False,
             ipython_dir=cls.ipython_dir.name,
             notebook_dir=cls.notebook_dir.name,
+            config=cls.config,
         )
         
         # clear log handlers and propagate to root for nose to capture it