From 75f93a3f7854a7383febd047391e2bfd4caceee0 Mon Sep 17 00:00:00 2001
From: Yuval Adam <yuval@segmanta.com>
Date: Thu, 20 Mar 2014 17:49:16 +0200
Subject: Initial git fork of django-storages

---
 storages/tests/__init__.py |   2 +
 storages/tests/hashpath.py |  35 ++++++
 storages/tests/s3boto.py   | 308 +++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 345 insertions(+)
 create mode 100644 storages/tests/__init__.py
 create mode 100644 storages/tests/hashpath.py
 create mode 100644 storages/tests/s3boto.py

(limited to 'storages/tests')

diff --git a/storages/tests/__init__.py b/storages/tests/__init__.py
new file mode 100644
index 0000000..e17e6d5
--- /dev/null
+++ b/storages/tests/__init__.py
@@ -0,0 +1,2 @@
+from storages.tests.hashpath import *
+from storages.tests.s3boto import *
diff --git a/storages/tests/hashpath.py b/storages/tests/hashpath.py
new file mode 100644
index 0000000..5cc4d65
--- /dev/null
+++ b/storages/tests/hashpath.py
@@ -0,0 +1,35 @@
+import os
+import shutil
+
+from django.test import TestCase
+from django.core.files.base import ContentFile
+from django.conf import settings
+
+from storages.backends.hashpath import HashPathStorage
+
+TEST_PATH_PREFIX = 'django-storages-test'
+
+
+class HashPathStorageTest(TestCase):
+
+    def setUp(self):
+        self.test_path = os.path.join(settings.MEDIA_ROOT, TEST_PATH_PREFIX)
+        self.storage = HashPathStorage(location=self.test_path)
+
+        # make sure the profile upload folder exists
+        if not os.path.exists(self.test_path):
+            os.makedirs(self.test_path)
+
+    def tearDown(self):
+        # remove uploaded profile picture
+        if os.path.exists(self.test_path):
+            shutil.rmtree(self.test_path)
+
+    def test_save_same_file(self):
+        """
+        saves a file twice, the file should only be stored once, because the
+        content/hash is the same
+        """
+        path_1 = self.storage.save('test', ContentFile('new content'))
+        path_2 = self.storage.save('test', ContentFile('new content'))
+        self.assertEqual(path_1, path_2)
diff --git a/storages/tests/s3boto.py b/storages/tests/s3boto.py
new file mode 100644
index 0000000..3f49503
--- /dev/null
+++ b/storages/tests/s3boto.py
@@ -0,0 +1,308 @@
+import mock
+import datetime
+import urlparse
+
+from django.test import TestCase
+from django.core.files.base import ContentFile
+
+from boto.s3.key import Key
+
+from storages.backends import s3boto
+
+__all__ = (
+    'ParseTsExtendedCase',
+    'SafeJoinTest',
+    'S3BotoStorageTests',
+    #'S3BotoStorageFileTests',
+)
+
+class ParseTsExtendedCase(TestCase):
+    def test_normal(self):
+        value = s3boto.parse_ts_extended("Wed, 13 Mar 2013 12:45:49 GMT")
+        self.assertEquals(value, datetime.datetime(2013, 3, 13, 12, 45, 49))
+
+class S3BotoTestCase(TestCase):
+    @mock.patch('storages.backends.s3boto.S3Connection')
+    def setUp(self, S3Connection):
+        self.storage = s3boto.S3BotoStorage()
+        self.storage._connection = mock.MagicMock()
+
+class SafeJoinTest(TestCase):
+    def test_normal(self):
+        path = s3boto.safe_join("", "path/to/somewhere", "other", "path/to/somewhere")
+        self.assertEquals(path, "path/to/somewhere/other/path/to/somewhere")
+
+    def test_with_dot(self):
+        path = s3boto.safe_join("", "path/./somewhere/../other", "..",
+                                ".", "to/./somewhere")
+        self.assertEquals(path, "path/to/somewhere")
+
+    def test_base_url(self):
+        path = s3boto.safe_join("base_url", "path/to/somewhere")
+        self.assertEquals(path, "base_url/path/to/somewhere")
+
+    def test_base_url_with_slash(self):
+        path = s3boto.safe_join("base_url/", "path/to/somewhere")
+        self.assertEquals(path, "base_url/path/to/somewhere")
+
+    def test_suspicious_operation(self):
+        self.assertRaises(ValueError,
+            s3boto.safe_join, "base", "../../../../../../../etc/passwd")
+
+    def test_trailing_slash(self):
+        """
+        Test safe_join with paths that end with a trailing slash.
+        """
+        path = s3boto.safe_join("base_url/", "path/to/somewhere/")
+        self.assertEquals(path, "base_url/path/to/somewhere/")
+
+    def test_trailing_slash_multi(self):
+        """
+        Test safe_join with multiple paths that end with a trailing slash.
+        """
+        path = s3boto.safe_join("base_url/", "path/to/" "somewhere/")
+        self.assertEquals(path, "base_url/path/to/somewhere/")
+
+
+class S3BotoStorageTests(S3BotoTestCase):
+
+    def test_clean_name(self):
+        """
+        Test the base case of _clean_name
+        """
+        path = self.storage._clean_name("path/to/somewhere")
+        self.assertEqual(path, "path/to/somewhere")
+
+    def test_clean_name_normalize(self):
+        """
+        Test the normalization of _clean_name
+        """
+        path = self.storage._clean_name("path/to/../somewhere")
+        self.assertEqual(path, "path/somewhere")
+
+    def test_clean_name_trailing_slash(self):
+        """
+        Test the _clean_name when the path has a trailing slash
+        """
+        path = self.storage._clean_name("path/to/somewhere/")
+        self.assertEqual(path, "path/to/somewhere/")
+
+    def test_clean_name_windows(self):
+        """
+        Test the _clean_name when the path has a trailing slash
+        """
+        path = self.storage._clean_name("path\\to\\somewhere")
+        self.assertEqual(path, "path/to/somewhere")
+
+    def test_storage_url(self):
+        """
+        Test URL generation.
+        """
+        self.storage.custom_domain = 'example.com'
+
+        # We expect no leading slashes in the path,
+        # and trailing slashes should be preserved.
+        self.assertEqual(self.storage.url(''), 'https://example.com/')
+        self.assertEqual(self.storage.url('path'), 'https://example.com/path')
+        self.assertEqual(self.storage.url('path/'), 'https://example.com/path/')
+        self.assertEqual(self.storage.url('path/1'), 'https://example.com/path/1')
+        self.assertEqual(self.storage.url('path/1/'), 'https://example.com/path/1/')
+
+    def test_storage_save(self):
+        """
+        Test saving a file
+        """
+        name = 'test_storage_save.txt'
+        content = ContentFile('new content')
+        self.storage.save(name, content)
+        self.storage.bucket.get_key.assert_called_once_with(name)
+
+        key = self.storage.bucket.get_key.return_value
+        key.set_metadata.assert_called_with('Content-Type', 'text/plain')
+        key.set_contents_from_file.assert_called_with(
+            content,
+            headers={'Content-Type': 'text/plain'},
+            policy=self.storage.default_acl,
+            reduced_redundancy=self.storage.reduced_redundancy,
+            rewind=True
+        )
+
+    def test_storage_save_gzip(self):
+        """
+        Test saving a file with gzip enabled.
+        """
+        if not s3boto.S3BotoStorage.gzip:  # Gzip not available.
+            return
+        name = 'test_storage_save.css'
+        content = ContentFile("I should be gzip'd")
+        self.storage.save(name, content)
+        key = self.storage.bucket.get_key.return_value
+        key.set_metadata.assert_called_with('Content-Type', 'text/css')
+        key.set_contents_from_file.assert_called_with(
+            content,
+            headers={'Content-Type': 'text/css', 'Content-Encoding': 'gzip'},
+            policy=self.storage.default_acl,
+            reduced_redundancy=self.storage.reduced_redundancy,
+            rewind=True,
+        )
+
+    def test_compress_content_len(self):
+        """
+        Test that file returned by _compress_content() is readable.
+        """
+        if not s3boto.S3BotoStorage.gzip:  # Gzip not available.
+            return
+        content = ContentFile("I should be gzip'd")
+        content = self.storage._compress_content(content)
+        self.assertTrue(len(content.read()) > 0)
+
+    def test_storage_open_write(self):
+        """
+        Test opening a file in write mode
+        """
+        name = 'test_open_for_writing.txt'
+        content = 'new content'
+
+        # Set the ACL header used when creating/writing data.
+        self.storage.bucket.connection.provider.acl_header = 'x-amz-acl'
+        # Set the mocked key's bucket
+        self.storage.bucket.get_key.return_value.bucket = self.storage.bucket
+        # Set the name of the mock object
+        self.storage.bucket.get_key.return_value.name = name
+
+        file = self.storage.open(name, 'w')
+        self.storage.bucket.get_key.assert_called_with(name)
+
+        file.write(content)
+        self.storage.bucket.initiate_multipart_upload.assert_called_with(
+            name,
+            headers={
+                'Content-Type': 'text/plain',
+                'x-amz-acl': 'public-read',
+            },
+            reduced_redundancy=self.storage.reduced_redundancy,
+        )
+
+        # Save the internal file before closing
+        _file = file.file
+        file.close()
+        file._multipart.upload_part_from_file.assert_called_with(
+            _file, 1, headers=self.storage.headers,
+        )
+        file._multipart.complete_upload.assert_called_once()
+    
+    def test_storage_exists(self):
+        key = self.storage.bucket.new_key.return_value
+        key.exists.return_value = True
+        self.assertTrue(self.storage.exists("file.txt"))
+
+    def test_storage_exists_false(self):
+        key = self.storage.bucket.new_key.return_value
+        key.exists.return_value = False 
+        self.assertFalse(self.storage.exists("file.txt"))
+
+    def test_storage_delete(self):
+        self.storage.delete("path/to/file.txt")
+        self.storage.bucket.delete_key.assert_called_with("path/to/file.txt")
+
+    def test_storage_listdir_base(self):
+        file_names = ["some/path/1.txt", "2.txt", "other/path/3.txt", "4.txt"]
+
+        self.storage.bucket.list.return_value = []
+        for p in file_names:
+            key = mock.MagicMock(spec=Key)
+            key.name = p
+            self.storage.bucket.list.return_value.append(key)
+
+        dirs, files = self.storage.listdir("")
+
+        self.assertEqual(len(dirs), 2)
+        for directory in ["some", "other"]:
+            self.assertTrue(directory in dirs,
+                            """ "%s" not in directory list "%s".""" % (
+                                directory, dirs))
+
+        self.assertEqual(len(files), 2)
+        for filename in ["2.txt", "4.txt"]:
+            self.assertTrue(filename in files,
+                            """ "%s" not in file list "%s".""" % (
+                                filename, files))
+
+    def test_storage_listdir_subdir(self):
+        file_names = ["some/path/1.txt", "some/2.txt"]
+
+        self.storage.bucket.list.return_value = []
+        for p in file_names:
+            key = mock.MagicMock(spec=Key)
+            key.name = p
+            self.storage.bucket.list.return_value.append(key)
+
+        dirs, files = self.storage.listdir("some/")
+        self.assertEqual(len(dirs), 1)
+        self.assertTrue('path' in dirs,
+                        """ "path" not in directory list "%s".""" % (dirs,))
+
+        self.assertEqual(len(files), 1)
+        self.assertTrue('2.txt' in files,
+                        """ "2.txt" not in files list "%s".""" % (files,))
+
+    def test_storage_size(self):
+        key = self.storage.bucket.get_key.return_value
+        key.size = 4098
+
+        name = 'file.txt'
+        self.assertEqual(self.storage.size(name), key.size)
+
+    def test_storage_url(self):
+        name = 'test_storage_size.txt'
+        url = 'http://aws.amazon.com/%s' % name
+        self.storage.connection.generate_url.return_value = url
+
+        self.assertEquals(self.storage.url(name), url)
+        self.storage.connection.generate_url.assert_called_with(
+            self.storage.querystring_expire,
+            method='GET',
+            bucket=self.storage.bucket.name,
+            key=name,
+            query_auth=self.storage.querystring_auth,
+            force_http=not self.storage.secure_urls,
+            headers=None,
+            response_headers=None,
+        )
+
+    def test_generated_url_is_encoded(self):
+        self.storage.custom_domain = "mock.cloudfront.net"
+        filename = "whacky & filename.mp4"
+        url = self.storage.url(filename)
+        parsed_url = urlparse.urlparse(url)
+        self.assertEqual(parsed_url.path,
+                         "/whacky%20%26%20filename.mp4")
+        
+#class S3BotoStorageFileTests(S3BotoTestCase):
+#    def test_multipart_upload(self):
+#        nparts = 2
+#        name = self.prefix_path("test_multipart_upload.txt")
+#        mode = 'w'
+#        f = s3boto.S3BotoStorageFile(name, mode, self.storage)
+#        content_length = 1024 * 1024# 1 MB
+#        content = 'a' * content_length
+#
+#        bytes = 0
+#        target = f._write_buffer_size * nparts
+#        while bytes < target:
+#            f.write(content)
+#            bytes += content_length
+#
+#        # make the buffer roll over so f._write_counter
+#        # is incremented
+#        f.write("finished")
+#
+#        # verify upload was multipart and correctly partitioned
+#        self.assertEqual(f._write_counter, nparts)
+#
+#        # complete the upload
+#        f.close()
+#
+#        # verify that the remaining buffered bytes were
+#        # uploaded when the file was closed.
+#        self.assertEqual(f._write_counter, nparts+1)
-- 
cgit v1.3.1