Project import generated by Copybara.
GitOrigin-RevId: d9e9e3fb4e31372ec1fb43b178994ca78fa8fe70
diff --git a/framework/test/gcs_helpers_test.py b/framework/test/gcs_helpers_test.py
new file mode 100644
index 0000000..3500e40
--- /dev/null
+++ b/framework/test/gcs_helpers_test.py
@@ -0,0 +1,185 @@
+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""Unit tests for the framework_helpers module."""
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+
+import mock
+import unittest
+import uuid
+
+import mox
+
+from google.appengine.api import app_identity
+from google.appengine.api import images
+from google.appengine.api import urlfetch
+from google.appengine.ext import testbed
+from third_party import cloudstorage
+
+from framework import filecontent
+from framework import gcs_helpers
+from testing import fake
+from testing import testing_helpers
+
+
+class GcsHelpersTest(unittest.TestCase):
+
+ def setUp(self):
+ self.mox = mox.Mox()
+ self.testbed = testbed.Testbed()
+ self.testbed.activate()
+ self.testbed.init_memcache_stub()
+
+ def tearDown(self):
+ self.mox.UnsetStubs()
+ self.mox.ResetAll()
+ self.testbed.deactivate()
+
+ def testDeleteObjectFromGCS(self):
+ object_id = 'aaaaa'
+ bucket_name = 'test_bucket'
+ object_path = '/' + bucket_name + object_id
+
+ self.mox.StubOutWithMock(app_identity, 'get_default_gcs_bucket_name')
+ app_identity.get_default_gcs_bucket_name().AndReturn(bucket_name)
+
+ self.mox.StubOutWithMock(cloudstorage, 'delete')
+ cloudstorage.delete(object_path)
+
+ self.mox.ReplayAll()
+
+ gcs_helpers.DeleteObjectFromGCS(object_id)
+ self.mox.VerifyAll()
+
+ def testStoreObjectInGCS_ResizableMimeType(self):
+ guid = 'aaaaa'
+ project_id = 100
+ object_id = '/%s/attachments/%s' % (project_id, guid)
+ bucket_name = 'test_bucket'
+ object_path = '/' + bucket_name + object_id
+ mime_type = 'image/png'
+ content = 'content'
+ thumb_content = 'thumb_content'
+
+ self.mox.StubOutWithMock(app_identity, 'get_default_gcs_bucket_name')
+ app_identity.get_default_gcs_bucket_name().AndReturn(bucket_name)
+
+ self.mox.StubOutWithMock(uuid, 'uuid4')
+ uuid.uuid4().AndReturn(guid)
+
+ self.mox.StubOutWithMock(cloudstorage, 'open')
+ cloudstorage.open(
+ object_path, 'w', mime_type, options={}
+ ).AndReturn(fake.FakeFile())
+ cloudstorage.open(object_path + '-thumbnail', 'w', mime_type).AndReturn(
+ fake.FakeFile())
+
+ self.mox.StubOutWithMock(images, 'resize')
+ images.resize(content, gcs_helpers.DEFAULT_THUMB_WIDTH,
+ gcs_helpers.DEFAULT_THUMB_HEIGHT).AndReturn(thumb_content)
+
+ self.mox.ReplayAll()
+
+ ret_id = gcs_helpers.StoreObjectInGCS(
+ content, mime_type, project_id, gcs_helpers.DEFAULT_THUMB_WIDTH,
+ gcs_helpers.DEFAULT_THUMB_HEIGHT)
+ self.mox.VerifyAll()
+ self.assertEqual(object_id, ret_id)
+
+ def testStoreObjectInGCS_NotResizableMimeType(self):
+ guid = 'aaaaa'
+ project_id = 100
+ object_id = '/%s/attachments/%s' % (project_id, guid)
+ bucket_name = 'test_bucket'
+ object_path = '/' + bucket_name + object_id
+ mime_type = 'not_resizable_mime_type'
+ content = 'content'
+
+ self.mox.StubOutWithMock(app_identity, 'get_default_gcs_bucket_name')
+ app_identity.get_default_gcs_bucket_name().AndReturn(bucket_name)
+
+ self.mox.StubOutWithMock(uuid, 'uuid4')
+ uuid.uuid4().AndReturn(guid)
+
+ self.mox.StubOutWithMock(cloudstorage, 'open')
+ options = {'Content-Disposition': 'inline; filename="file.ext"'}
+ cloudstorage.open(
+ object_path, 'w', mime_type, options=options
+ ).AndReturn(fake.FakeFile())
+
+ self.mox.ReplayAll()
+
+ ret_id = gcs_helpers.StoreObjectInGCS(
+ content, mime_type, project_id, gcs_helpers.DEFAULT_THUMB_WIDTH,
+ gcs_helpers.DEFAULT_THUMB_HEIGHT, filename='file.ext')
+ self.mox.VerifyAll()
+ self.assertEqual(object_id, ret_id)
+
+ def testCheckMemeTypeResizable(self):
+ for resizable_mime_type in gcs_helpers.RESIZABLE_MIME_TYPES:
+ gcs_helpers.CheckMimeTypeResizable(resizable_mime_type)
+
+ with self.assertRaises(gcs_helpers.UnsupportedMimeType):
+ gcs_helpers.CheckMimeTypeResizable('not_resizable_mime_type')
+
+ def testStoreLogoInGCS(self):
+ file_name = 'test_file.png'
+ mime_type = 'image/png'
+ content = 'test content'
+ project_id = 100
+ object_id = 123
+
+ self.mox.StubOutWithMock(filecontent, 'GuessContentTypeFromFilename')
+ filecontent.GuessContentTypeFromFilename(file_name).AndReturn(mime_type)
+
+ self.mox.StubOutWithMock(gcs_helpers, 'StoreObjectInGCS')
+ gcs_helpers.StoreObjectInGCS(
+ content, mime_type, project_id,
+ thumb_width=gcs_helpers.LOGO_THUMB_WIDTH,
+ thumb_height=gcs_helpers.LOGO_THUMB_HEIGHT).AndReturn(object_id)
+
+ self.mox.ReplayAll()
+
+ ret_id = gcs_helpers.StoreLogoInGCS(file_name, content, project_id)
+ self.mox.VerifyAll()
+ self.assertEqual(object_id, ret_id)
+
+ @mock.patch('google.appengine.api.urlfetch.fetch')
+ def testFetchSignedURL_Success(self, mock_fetch):
+ mock_fetch.return_value = testing_helpers.Blank(
+ headers={'Location': 'signed url'})
+ actual = gcs_helpers._FetchSignedURL('signing req url')
+ mock_fetch.assert_called_with('signing req url', follow_redirects=False)
+ self.assertEqual('signed url', actual)
+
+ @mock.patch('google.appengine.api.urlfetch.fetch')
+ def testFetchSignedURL_UnderpopulatedResult(self, mock_fetch):
+ mock_fetch.return_value = testing_helpers.Blank(headers={})
+ self.assertRaises(
+ KeyError, gcs_helpers._FetchSignedURL, 'signing req url')
+
+ @mock.patch('google.appengine.api.urlfetch.fetch')
+ def testFetchSignedURL_DownloadError(self, mock_fetch):
+ mock_fetch.side_effect = urlfetch.DownloadError
+ self.assertRaises(
+ urlfetch.DownloadError,
+ gcs_helpers._FetchSignedURL, 'signing req url')
+
+ @mock.patch('framework.gcs_helpers._FetchSignedURL')
+ def testSignUrl_Success(self, mock_FetchSignedURL):
+ with mock.patch(
+ 'google.appengine.api.app_identity.get_access_token') as gat:
+ gat.return_value = ['token']
+ mock_FetchSignedURL.return_value = 'signed url'
+ signed_url = gcs_helpers.SignUrl('bucket', '/object')
+ self.assertEqual('signed url', signed_url)
+
+ @mock.patch('framework.gcs_helpers._FetchSignedURL')
+ def testSignUrl_DownloadError(self, mock_FetchSignedURL):
+ mock_FetchSignedURL.side_effect = urlfetch.DownloadError
+ self.assertEqual(
+ '/missing-gcs-url', gcs_helpers.SignUrl('bucket', '/object'))