9 files changed, 656 insertions, 0 deletions
diff --git a/testing/mozbase/mozfile/tests/files/missing_file_attributes.zip b/testing/mozbase/mozfile/tests/files/missing_file_attributes.zip
new file mode 100644
index 0000000000..2b5409e89c
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/files/missing_file_attributes.zip
diff --git a/testing/mozbase/mozfile/tests/manifest.ini b/testing/mozbase/mozfile/tests/manifest.ini
new file mode 100644
index 0000000000..c7889beca6
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/manifest.ini
@@ -0,0 +1,6 @@
+[test_extract.py]
+[test_load.py]
+[test_move_remove.py]
+[test_tempdir.py]
+[test_tempfile.py]
+[test_url.py]
diff --git a/testing/mozbase/mozfile/tests/stubs.py b/testing/mozbase/mozfile/tests/stubs.py
new file mode 100644
index 0000000000..06d79e7af1
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/stubs.py
@@ -0,0 +1,37 @@
+import os
+import shutil
+import tempfile
+
+
+# stub file paths
+files = [('foo.txt',),
+         ('foo', 'bar.txt',),
+         ('foo', 'bar', 'fleem.txt',),
+         ('foobar', 'fleem.txt',),
+         ('bar.txt',),
+         ('nested_tree', 'bar', 'fleem.txt',),
+         ('readonly.txt',),
+         ]
+
+
+def create_stub():
+    """create a stub directory"""
+
+    tempdir = tempfile.mkdtemp()
+    try:
+        for path in files:
+            fullpath = os.path.join(tempdir, *path)
+            dirname = os.path.dirname(fullpath)
+            if not os.path.exists(dirname):
+                os.makedirs(dirname)
+            contents = path[-1]
+            f = file(fullpath, 'w')
+            f.write(contents)
+            f.close()
+        return tempdir
+    except Exception:
+        try:
+            shutil.rmtree(tempdir)
+        except:
+            pass
+        raise
diff --git a/testing/mozbase/mozfile/tests/test_extract.py b/testing/mozbase/mozfile/tests/test_extract.py
new file mode 100644
index 0000000000..e91f52349b
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_extract.py
@@ -0,0 +1,154 @@
+#!/usr/bin/env python
+
+import os
+import shutil
+import tarfile
+import tempfile
+import unittest
+import zipfile
+
+import mozfile
+
+import stubs
+
+
+class TestExtract(unittest.TestCase):
+    """test extracting archives"""
+
+    def ensure_directory_contents(self, directory):
+        """ensure the directory contents match"""
+        for f in stubs.files:
+            path = os.path.join(directory, *f)
+            exists = os.path.exists(path)
+            if not exists:
+                print "%s does not exist" % (os.path.join(f))
+            self.assertTrue(exists)
+            if exists:
+                contents = file(path).read().strip()
+                self.assertTrue(contents == f[-1])
+
+    def test_extract_zipfile(self):
+        """test extracting a zipfile"""
+        _zipfile = self.create_zip()
+        self.assertTrue(os.path.exists(_zipfile))
+        try:
+            dest = tempfile.mkdtemp()
+            try:
+                mozfile.extract_zip(_zipfile, dest)
+                self.ensure_directory_contents(dest)
+            finally:
+                shutil.rmtree(dest)
+        finally:
+            os.remove(_zipfile)
+
+    def test_extract_zipfile_missing_file_attributes(self):
+        """if files do not have attributes set the default permissions have to be inherited."""
+        _zipfile = os.path.join(os.path.dirname(__file__), 'files', 'missing_file_attributes.zip')
+        self.assertTrue(os.path.exists(_zipfile))
+        dest = tempfile.mkdtemp()
+        try:
+            # Get the default file permissions for the user
+            fname = os.path.join(dest, 'foo')
+            with open(fname, 'w'):
+                pass
+            default_stmode = os.stat(fname).st_mode
+
+            files = mozfile.extract_zip(_zipfile, dest)
+            for filename in files:
+                self.assertEqual(os.stat(os.path.join(dest, filename)).st_mode,
+                                 default_stmode)
+        finally:
+            shutil.rmtree(dest)
+
+    def test_extract_tarball(self):
+        """test extracting a tarball"""
+        tarball = self.create_tarball()
+        self.assertTrue(os.path.exists(tarball))
+        try:
+            dest = tempfile.mkdtemp()
+            try:
+                mozfile.extract_tarball(tarball, dest)
+                self.ensure_directory_contents(dest)
+            finally:
+                shutil.rmtree(dest)
+        finally:
+            os.remove(tarball)
+
+    def test_extract(self):
+        """test the generalized extract function"""
+
+        # test extracting a tarball
+        tarball = self.create_tarball()
+        self.assertTrue(os.path.exists(tarball))
+        try:
+            dest = tempfile.mkdtemp()
+            try:
+                mozfile.extract(tarball, dest)
+                self.ensure_directory_contents(dest)
+            finally:
+                shutil.rmtree(dest)
+        finally:
+            os.remove(tarball)
+
+        # test extracting a zipfile
+        _zipfile = self.create_zip()
+        self.assertTrue(os.path.exists(_zipfile))
+        try:
+            dest = tempfile.mkdtemp()
+            try:
+                mozfile.extract_zip(_zipfile, dest)
+                self.ensure_directory_contents(dest)
+            finally:
+                shutil.rmtree(dest)
+        finally:
+            os.remove(_zipfile)
+
+        # test extracting some non-archive; this should fail
+        fd, filename = tempfile.mkstemp()
+        os.write(fd, 'This is not a zipfile or tarball')
+        os.close(fd)
+        exception = None
+        try:
+            dest = tempfile.mkdtemp()
+            mozfile.extract(filename, dest)
+        except Exception as exception:
+            pass
+        finally:
+            os.remove(filename)
+            os.rmdir(dest)
+        self.assertTrue(isinstance(exception, Exception))
+
+    # utility functions
+
+    def create_tarball(self):
+        """create a stub tarball for testing"""
+        tempdir = stubs.create_stub()
+        filename = tempfile.mktemp(suffix='.tar')
+        archive = tarfile.TarFile(filename, mode='w')
+        try:
+            for path in stubs.files:
+                archive.add(os.path.join(tempdir, *path), arcname=os.path.join(*path))
+        except:
+            os.remove(archive)
+            raise
+        finally:
+            shutil.rmtree(tempdir)
+        archive.close()
+        return filename
+
+    def create_zip(self):
+        """create a stub zipfile for testing"""
+
+        tempdir = stubs.create_stub()
+        filename = tempfile.mktemp(suffix='.zip')
+        archive = zipfile.ZipFile(filename, mode='w')
+        try:
+            for path in stubs.files:
+                archive.write(os.path.join(tempdir, *path), arcname=os.path.join(*path))
+        except:
+            os.remove(filename)
+            raise
+        finally:
+            shutil.rmtree(tempdir)
+        archive.close()
+        return filename
diff --git a/testing/mozbase/mozfile/tests/test_load.py b/testing/mozbase/mozfile/tests/test_load.py
new file mode 100755
index 0000000000..13a5b519c1
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_load.py
@@ -0,0 +1,62 @@
+#!/usr/bin/env python
+
+"""
+tests for mozfile.load
+"""
+
+import mozhttpd
+import os
+import tempfile
+import unittest
+from mozfile import load
+
+
+class TestLoad(unittest.TestCase):
+    """test the load function"""
+
+    def test_http(self):
+        """test with mozhttpd and a http:// URL"""
+
+        def example(request):
+            """example request handler"""
+            body = 'example'
+            return (200, {'Content-type': 'text/plain',
+                          'Content-length': len(body)
+                          }, body)
+
+        host = '127.0.0.1'
+        httpd = mozhttpd.MozHttpd(host=host,
+                                  urlhandlers=[{'method': 'GET',
+                                                'path': '.*',
+                                                'function': example}])
+        try:
+            httpd.start(block=False)
+            content = load(httpd.get_url()).read()
+            self.assertEqual(content, 'example')
+        finally:
+            httpd.stop()
+
+    def test_file_path(self):
+        """test loading from file path"""
+        try:
+            # create a temporary file
+            tmp = tempfile.NamedTemporaryFile(delete=False)
+            tmp.write('foo bar')
+            tmp.close()
+
+            # read the file
+            contents = file(tmp.name).read()
+            self.assertEqual(contents, 'foo bar')
+
+            # read the file with load and a file path
+            self.assertEqual(load(tmp.name).read(), contents)
+
+            # read the file with load and a file URL
+            self.assertEqual(load('file://%s' % tmp.name).read(), contents)
+        finally:
+            # remove the tempfile
+            if os.path.exists(tmp.name):
+                os.remove(tmp.name)
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/testing/mozbase/mozfile/tests/test_move_remove.py b/testing/mozbase/mozfile/tests/test_move_remove.py
new file mode 100644
index 0000000000..e9d0cd4347
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_move_remove.py
@@ -0,0 +1,232 @@
+#!/usr/bin/env python
+
+import os
+import stat
+import shutil
+import threading
+import time
+import unittest
+import errno
+from contextlib import contextmanager
+
+import mozfile
+import mozinfo
+
+import stubs
+
+
+def mark_readonly(path):
+    """Removes all write permissions from given file/directory.
+
+    :param path: path of directory/file of which modes must be changed
+    """
+    mode = os.stat(path)[stat.ST_MODE]
+    os.chmod(path, mode & ~stat.S_IWUSR & ~stat.S_IWGRP & ~stat.S_IWOTH)
+
+
+class FileOpenCloseThread(threading.Thread):
+    """Helper thread for asynchronous file handling"""
+
+    def __init__(self, path, delay, delete=False):
+        threading.Thread.__init__(self)
+        self.file_opened = threading.Event()
+        self.delay = delay
+        self.path = path
+        self.delete = delete
+
+    def run(self):
+        with open(self.path):
+            self.file_opened.set()
+            time.sleep(self.delay)
+        if self.delete:
+            try:
+                os.remove(self.path)
+            except:
+                pass
+
+
+@contextmanager
+def wait_file_opened_in_thread(*args, **kwargs):
+    thread = FileOpenCloseThread(*args, **kwargs)
+    thread.start()
+    thread.file_opened.wait()
+    try:
+        yield thread
+    finally:
+        thread.join()
+
+
+class MozfileRemoveTestCase(unittest.TestCase):
+    """Test our ability to remove directories and files"""
+
+    def setUp(self):
+        # Generate a stub
+        self.tempdir = stubs.create_stub()
+
+    def tearDown(self):
+        if os.path.isdir(self.tempdir):
+            shutil.rmtree(self.tempdir)
+
+    def test_remove_directory(self):
+        """Test the removal of a directory"""
+        self.assertTrue(os.path.isdir(self.tempdir))
+        mozfile.remove(self.tempdir)
+        self.assertFalse(os.path.exists(self.tempdir))
+
+    def test_remove_directory_with_open_file(self):
+        """Test removing a directory with an open file"""
+        # Open a file in the generated stub
+        filepath = os.path.join(self.tempdir, *stubs.files[1])
+        f = file(filepath, 'w')
+        f.write('foo-bar')
+
+        # keep file open and then try removing the dir-tree
+        if mozinfo.isWin:
+            # On the Windows family WindowsError should be raised.
+            self.assertRaises(OSError, mozfile.remove, self.tempdir)
+            self.assertTrue(os.path.exists(self.tempdir))
+        else:
+            # Folder should be deleted on all other platforms
+            mozfile.remove(self.tempdir)
+            self.assertFalse(os.path.exists(self.tempdir))
+
+    def test_remove_closed_file(self):
+        """Test removing a closed file"""
+        # Open a file in the generated stub
+        filepath = os.path.join(self.tempdir, *stubs.files[1])
+        with open(filepath, 'w') as f:
+            f.write('foo-bar')
+
+        # Folder should be deleted on all platforms
+        mozfile.remove(self.tempdir)
+        self.assertFalse(os.path.exists(self.tempdir))
+
+    def test_removing_open_file_with_retry(self):
+        """Test removing a file in use with retry"""
+        filepath = os.path.join(self.tempdir, *stubs.files[1])
+
+        with wait_file_opened_in_thread(filepath, 0.2):
+            # on windows first attempt will fail,
+            # and it will be retried until the thread leave the handle
+            mozfile.remove(filepath)
+
+        # Check deletion was successful
+        self.assertFalse(os.path.exists(filepath))
+
+    def test_removing_already_deleted_file_with_retry(self):
+        """Test removing a meanwhile removed file with retry"""
+        filepath = os.path.join(self.tempdir, *stubs.files[1])
+
+        with wait_file_opened_in_thread(filepath, 0.2, True):
+            # on windows first attempt will fail, and before
+            # the retry the opened file will be deleted in the thread
+            mozfile.remove(filepath)
+
+        # Check deletion was successful
+        self.assertFalse(os.path.exists(filepath))
+
+    def test_remove_readonly_tree(self):
+        """Test removing a read-only directory"""
+
+        dirpath = os.path.join(self.tempdir, "nested_tree")
+        mark_readonly(dirpath)
+
+        # However, mozfile should change write permissions and remove dir.
+        mozfile.remove(dirpath)
+
+        self.assertFalse(os.path.exists(dirpath))
+
+    def test_remove_readonly_file(self):
+        """Test removing read-only files"""
+        filepath = os.path.join(self.tempdir, *stubs.files[1])
+        mark_readonly(filepath)
+
+        # However, mozfile should change write permission and then remove file.
+        mozfile.remove(filepath)
+
+        self.assertFalse(os.path.exists(filepath))
+
+    @unittest.skipIf(mozinfo.isWin, "Symlinks are not supported on Windows")
+    def test_remove_symlink(self):
+        """Test removing a symlink"""
+        file_path = os.path.join(self.tempdir, *stubs.files[1])
+        symlink_path = os.path.join(self.tempdir, 'symlink')
+
+        os.symlink(file_path, symlink_path)
+        self.assertTrue(os.path.islink(symlink_path))
+
+        # The linked folder and files should not be deleted
+        mozfile.remove(symlink_path)
+        self.assertFalse(os.path.exists(symlink_path))
+        self.assertTrue(os.path.exists(file_path))
+
+    @unittest.skipIf(mozinfo.isWin, "Symlinks are not supported on Windows")
+    def test_remove_symlink_in_subfolder(self):
+        """Test removing a folder with an contained symlink"""
+        file_path = os.path.join(self.tempdir, *stubs.files[0])
+        dir_path = os.path.dirname(os.path.join(self.tempdir, *stubs.files[1]))
+        symlink_path = os.path.join(dir_path, 'symlink')
+
+        os.symlink(file_path, symlink_path)
+        self.assertTrue(os.path.islink(symlink_path))
+
+        # The folder with the contained symlink will be deleted but not the
+        # original linked file
+        mozfile.remove(dir_path)
+        self.assertFalse(os.path.exists(dir_path))
+        self.assertFalse(os.path.exists(symlink_path))
+        self.assertTrue(os.path.exists(file_path))
+
+    @unittest.skipIf(mozinfo.isWin or not os.geteuid(),
+                     "Symlinks are not supported on Windows and cannot run test as root")
+    def test_remove_symlink_for_system_path(self):
+        """Test removing a symlink which points to a system folder"""
+        symlink_path = os.path.join(self.tempdir, 'symlink')
+
+        os.symlink(os.path.dirname(self.tempdir), symlink_path)
+        self.assertTrue(os.path.islink(symlink_path))
+
+        # The folder with the contained symlink will be deleted but not the
+        # original linked file
+        mozfile.remove(symlink_path)
+        self.assertFalse(os.path.exists(symlink_path))
+
+    def test_remove_path_that_does_not_exists(self):
+        not_existing_path = os.path.join(self.tempdir, 'I_do_not_not_exists')
+        try:
+            mozfile.remove(not_existing_path)
+        except OSError as exc:
+            if exc.errno == errno.ENOENT:
+                self.fail("removing non existing path must not raise error")
+            raise
+
+
+class MozFileMoveTestCase(unittest.TestCase):
+
+    def setUp(self):
+        # Generate a stub
+        self.tempdir = stubs.create_stub()
+        self.addCleanup(mozfile.rmtree, self.tempdir)
+
+    def test_move_file(self):
+        file_path = os.path.join(self.tempdir, *stubs.files[1])
+        moved_path = file_path + '.moved'
+        self.assertTrue(os.path.isfile(file_path))
+        self.assertFalse(os.path.exists(moved_path))
+        mozfile.move(file_path, moved_path)
+        self.assertFalse(os.path.exists(file_path))
+        self.assertTrue(os.path.isfile(moved_path))
+
+    def test_move_file_with_retry(self):
+        file_path = os.path.join(self.tempdir, *stubs.files[1])
+        moved_path = file_path + '.moved'
+
+        with wait_file_opened_in_thread(file_path, 0.2):
+            # first move attempt should fail on windows and be retried
+            mozfile.move(file_path, moved_path)
+            self.assertFalse(os.path.exists(file_path))
+            self.assertTrue(os.path.isfile(moved_path))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/testing/mozbase/mozfile/tests/test_tempdir.py b/testing/mozbase/mozfile/tests/test_tempdir.py
new file mode 100644
index 0000000000..81f03d095b
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_tempdir.py
@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+"""
+tests for mozfile.TemporaryDirectory
+"""
+
+from mozfile import TemporaryDirectory
+import os
+import unittest
+
+
+class TestTemporaryDirectory(unittest.TestCase):
+
+    def test_removed(self):
+        """ensure that a TemporaryDirectory gets removed"""
+        path = None
+        with TemporaryDirectory() as tmp:
+            path = tmp
+            self.assertTrue(os.path.isdir(tmp))
+            tmpfile = os.path.join(tmp, "a_temp_file")
+            open(tmpfile, "w").write("data")
+            self.assertTrue(os.path.isfile(tmpfile))
+        self.assertFalse(os.path.isdir(path))
+        self.assertFalse(os.path.exists(path))
+
+    def test_exception(self):
+        """ensure that TemporaryDirectory handles exceptions"""
+        path = None
+        with self.assertRaises(Exception):
+            with TemporaryDirectory() as tmp:
+                path = tmp
+                self.assertTrue(os.path.isdir(tmp))
+                raise Exception("oops")
+        self.assertFalse(os.path.isdir(path))
+        self.assertFalse(os.path.exists(path))
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/testing/mozbase/mozfile/tests/test_tempfile.py b/testing/mozbase/mozfile/tests/test_tempfile.py
new file mode 100644
index 0000000000..3c3d26d5d6
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_tempfile.py
@@ -0,0 +1,102 @@
+#!/usr/bin/env python
+
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+"""
+tests for mozfile.NamedTemporaryFile
+"""
+
+import mozfile
+import os
+import unittest
+
+
+class TestNamedTemporaryFile(unittest.TestCase):
+    """test our fix for NamedTemporaryFile"""
+
+    def test_named_temporary_file(self):
+        """ Ensure the fix for re-opening a NamedTemporaryFile works
+
+            Refer to https://bugzilla.mozilla.org/show_bug.cgi?id=818777
+            and https://bugzilla.mozilla.org/show_bug.cgi?id=821362
+        """
+
+        test_string = "A simple test"
+        with mozfile.NamedTemporaryFile() as temp:
+            # Test we can write to file
+            temp.write(test_string)
+            # Forced flush, so that we can read later
+            temp.flush()
+
+            # Test we can open the file again on all platforms
+            self.assertEqual(open(temp.name).read(), test_string)
+
+    def test_iteration(self):
+        """ensure the line iterator works"""
+
+        # make a file and write to it
+        tf = mozfile.NamedTemporaryFile()
+        notes = ['doe', 'rae', 'mi']
+        for note in notes:
+            tf.write('%s\n' % note)
+        tf.flush()
+
+        # now read from it
+        tf.seek(0)
+        lines = [line.rstrip('\n') for line in tf.readlines()]
+        self.assertEqual(lines, notes)
+
+        # now read from it iteratively
+        lines = []
+        for line in tf:
+            lines.append(line.strip())
+        self.assertEqual(lines, [])  # because we did not seek(0)
+        tf.seek(0)
+        lines = []
+        for line in tf:
+            lines.append(line.strip())
+        self.assertEqual(lines, notes)
+
+    def test_delete(self):
+        """ensure ``delete=True/False`` works as expected"""
+
+        # make a deleteable file; ensure it gets cleaned up
+        path = None
+        with mozfile.NamedTemporaryFile(delete=True) as tf:
+            path = tf.name
+        self.assertTrue(isinstance(path, basestring))
+        self.assertFalse(os.path.exists(path))
+
+        # it is also deleted when __del__ is called
+        # here we will do so explicitly
+        tf = mozfile.NamedTemporaryFile(delete=True)
+        path = tf.name
+        self.assertTrue(os.path.exists(path))
+        del tf
+        self.assertFalse(os.path.exists(path))
+
+        # Now the same thing but we won't delete the file
+        path = None
+        try:
+            with mozfile.NamedTemporaryFile(delete=False) as tf:
+                path = tf.name
+            self.assertTrue(os.path.exists(path))
+        finally:
+            if path and os.path.exists(path):
+                os.remove(path)
+
+        path = None
+        try:
+            tf = mozfile.NamedTemporaryFile(delete=False)
+            path = tf.name
+            self.assertTrue(os.path.exists(path))
+            del tf
+            self.assertTrue(os.path.exists(path))
+        finally:
+            if path and os.path.exists(path):
+                os.remove(path)
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/testing/mozbase/mozfile/tests/test_url.py b/testing/mozbase/mozfile/tests/test_url.py
new file mode 100755
index 0000000000..7d2b12b399
--- /dev/null
+++ b/testing/mozbase/mozfile/tests/test_url.py
@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+
+"""
+tests for is_url
+"""
+
+import unittest
+from mozfile import is_url
+
+
+class TestIsUrl(unittest.TestCase):
+    """test the is_url function"""
+
+    def test_is_url(self):
+        self.assertTrue(is_url('http://mozilla.org'))
+        self.assertFalse(is_url('/usr/bin/mozilla.org'))
+        self.assertTrue(is_url('file:///usr/bin/mozilla.org'))
+        self.assertFalse(is_url('c:\foo\bar'))
+
+if __name__ == '__main__':
+    unittest.main()