Scripts to download files from google storage based on sha1 sums

third_party/gsutil/boto/tests/unit/s3/test_uri.py

Index: third_party/gsutil/boto/tests/unit/s3/test_uri.py
diff --git a/third_party/gsutil/boto/tests/unit/s3/test_uri.py b/third_party/gsutil/boto/tests/unit/s3/test_uri.py
new file mode 100644
index 0000000000000000000000000000000000000000..ab6821912b27ebba7b2914396ef028c7de07c749
--- /dev/null
+++ b/third_party/gsutil/boto/tests/unit/s3/test_uri.py
@@ -0,0 +1,257 @@
+#!/usr/bin/env python
+# Copyright (c) 2013 Google, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import boto
+import tempfile
+import urllib
+from boto.exception import InvalidUriError
+from boto import storage_uri
+from boto.s3.keyfile import KeyFile
+from tests.integration.s3.mock_storage_service import MockBucket
+from tests.integration.s3.mock_storage_service import MockBucketStorageUri
+from tests.integration.s3.mock_storage_service import MockConnection
+from tests.unit import unittest
+
+"""Unit tests for StorageUri interface."""
+
+class UriTest(unittest.TestCase):
+
+    def test_provider_uri(self):
+        for prov in ('gs', 's3'):
+            uri_str = '%s://' % prov
+            uri = boto.storage_uri(uri_str, validate=False,
+                suppress_consec_slashes=False)
+            self.assertEqual(prov, uri.scheme)
+            self.assertEqual(uri_str, uri.uri)
+            self.assertFalse(hasattr(uri, 'versionless_uri'))
+            self.assertEqual('', uri.bucket_name)
+            self.assertEqual('', uri.object_name)
+            self.assertEqual(None, uri.version_id)
+            self.assertEqual(None, uri.generation)
+            self.assertEqual(uri.names_provider(), True)
+            self.assertEqual(uri.names_container(), True)
+            self.assertEqual(uri.names_bucket(), False)
+            self.assertEqual(uri.names_object(), False)
+            self.assertEqual(uri.names_directory(), False)
+            self.assertEqual(uri.names_file(), False)
+            self.assertEqual(uri.is_stream(), False)
+            self.assertEqual(uri.is_version_specific, False)
+
+    def test_bucket_uri_no_trailing_slash(self):
+        for prov in ('gs', 's3'):
+            uri_str = '%s://bucket' % prov
+            uri = boto.storage_uri(uri_str, validate=False,
+                suppress_consec_slashes=False)
+            self.assertEqual(prov, uri.scheme)
+            self.assertEqual('%s/' % uri_str, uri.uri)
+            self.assertFalse(hasattr(uri, 'versionless_uri'))
+            self.assertEqual('bucket', uri.bucket_name)
+            self.assertEqual('', uri.object_name)
+            self.assertEqual(None, uri.version_id)
+            self.assertEqual(None, uri.generation)
+            self.assertEqual(uri.names_provider(), False)
+            self.assertEqual(uri.names_container(), True)
+            self.assertEqual(uri.names_bucket(), True)
+            self.assertEqual(uri.names_object(), False)
+            self.assertEqual(uri.names_directory(), False)
+            self.assertEqual(uri.names_file(), False)
+            self.assertEqual(uri.is_stream(), False)
+            self.assertEqual(uri.is_version_specific, False)
+
+    def test_bucket_uri_with_trailing_slash(self):
+        for prov in ('gs', 's3'):
+            uri_str = '%s://bucket/' % prov
+            uri = boto.storage_uri(uri_str, validate=False,
+                suppress_consec_slashes=False)
+            self.assertEqual(prov, uri.scheme)
+            self.assertEqual(uri_str, uri.uri)
+            self.assertFalse(hasattr(uri, 'versionless_uri'))
+            self.assertEqual('bucket', uri.bucket_name)
+            self.assertEqual('', uri.object_name)
+            self.assertEqual(None, uri.version_id)
+            self.assertEqual(None, uri.generation)
+            self.assertEqual(uri.names_provider(), False)
+            self.assertEqual(uri.names_container(), True)
+            self.assertEqual(uri.names_bucket(), True)
+            self.assertEqual(uri.names_object(), False)
+            self.assertEqual(uri.names_directory(), False)
+            self.assertEqual(uri.names_file(), False)
+            self.assertEqual(uri.is_stream(), False)
+            self.assertEqual(uri.is_version_specific, False)
+
+    def test_non_versioned_object_uri(self):
+        for prov in ('gs', 's3'):
+            uri_str = '%s://bucket/obj/a/b' % prov
+            uri = boto.storage_uri(uri_str, validate=False,
+                suppress_consec_slashes=False)
+            self.assertEqual(prov, uri.scheme)
+            self.assertEqual(uri_str, uri.uri)
+            self.assertEqual(uri_str, uri.versionless_uri)
+            self.assertEqual('bucket', uri.bucket_name)
+            self.assertEqual('obj/a/b', uri.object_name)
+            self.assertEqual(None, uri.version_id)
+            self.assertEqual(None, uri.generation)
+            self.assertEqual(uri.names_provider(), False)
+            self.assertEqual(uri.names_container(), False)
+            self.assertEqual(uri.names_bucket(), False)
+            self.assertEqual(uri.names_object(), True)
+            self.assertEqual(uri.names_directory(), False)
+            self.assertEqual(uri.names_file(), False)
+            self.assertEqual(uri.is_stream(), False)
+            self.assertEqual(uri.is_version_specific, False)
+
+    def test_versioned_gs_object_uri(self):
+        uri_str = 'gs://bucket/obj/a/b#1359908801674000'
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('gs', uri.scheme)
+        self.assertEqual(uri_str, uri.uri)
+        self.assertEqual('gs://bucket/obj/a/b', uri.versionless_uri)
+        self.assertEqual('bucket', uri.bucket_name)
+        self.assertEqual('obj/a/b', uri.object_name)
+        self.assertEqual(None, uri.version_id)
+        self.assertEqual(1359908801674000, uri.generation)
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_container(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        self.assertEqual(uri.names_object(), True)
+        self.assertEqual(uri.names_directory(), False)
+        self.assertEqual(uri.names_file(), False)
+        self.assertEqual(uri.is_stream(), False)
+        self.assertEqual(uri.is_version_specific, True)
+
+    def test_versioned_gs_object_uri_with_legacy_generation_value(self):
+        uri_str = 'gs://bucket/obj/a/b#1'
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('gs', uri.scheme)
+        self.assertEqual(uri_str, uri.uri)
+        self.assertEqual('gs://bucket/obj/a/b', uri.versionless_uri)
+        self.assertEqual('bucket', uri.bucket_name)
+        self.assertEqual('obj/a/b', uri.object_name)
+        self.assertEqual(None, uri.version_id)
+        self.assertEqual(1, uri.generation)
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_container(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        self.assertEqual(uri.names_object(), True)
+        self.assertEqual(uri.names_directory(), False)
+        self.assertEqual(uri.names_file(), False)
+        self.assertEqual(uri.is_stream(), False)
+        self.assertEqual(uri.is_version_specific, True)
+
+    def test_roundtrip_versioned_gs_object_uri_parsed(self):
+        uri_str = 'gs://bucket/obj#1359908801674000'
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        roundtrip_uri = boto.storage_uri(uri.uri, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual(uri.uri, roundtrip_uri.uri)
+        self.assertEqual(uri.is_version_specific, True)
+
+    def test_versioned_s3_object_uri(self):
+        uri_str = 's3://bucket/obj/a/b#eMuM0J15HkJ9QHlktfNP5MfA.oYR2q6S'
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('s3', uri.scheme)
+        self.assertEqual(uri_str, uri.uri)
+        self.assertEqual('s3://bucket/obj/a/b', uri.versionless_uri)
+        self.assertEqual('bucket', uri.bucket_name)
+        self.assertEqual('obj/a/b', uri.object_name)
+        self.assertEqual('eMuM0J15HkJ9QHlktfNP5MfA.oYR2q6S', uri.version_id)
+        self.assertEqual(None, uri.generation)
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_container(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        self.assertEqual(uri.names_object(), True)
+        self.assertEqual(uri.names_directory(), False)
+        self.assertEqual(uri.names_file(), False)
+        self.assertEqual(uri.is_stream(), False)
+        self.assertEqual(uri.is_version_specific, True)
+
+    def test_explicit_file_uri(self):
+        tmp_dir = tempfile.tempdir
+        uri_str = 'file://%s' % urllib.pathname2url(tmp_dir)
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('file', uri.scheme)
+        self.assertEqual(uri_str, uri.uri)
+        self.assertFalse(hasattr(uri, 'versionless_uri'))
+        self.assertEqual('', uri.bucket_name)
+        self.assertEqual(tmp_dir, uri.object_name)
+        self.assertFalse(hasattr(uri, 'version_id'))
+        self.assertFalse(hasattr(uri, 'generation'))
+        self.assertFalse(hasattr(uri, 'is_version_specific'))
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        # Don't check uri.names_container(), uri.names_directory(),
+        # uri.names_file(), or uri.names_object(), because for file URIs these
+        # functions look at the file system and apparently unit tests run
+        # chroot'd.
+        self.assertEqual(uri.is_stream(), False)
+
+    def test_implicit_file_uri(self):
+        tmp_dir = tempfile.tempdir
+        uri_str = '%s' % urllib.pathname2url(tmp_dir)
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('file', uri.scheme)
+        self.assertEqual('file://%s' % tmp_dir, uri.uri)
+        self.assertFalse(hasattr(uri, 'versionless_uri'))
+        self.assertEqual('', uri.bucket_name)
+        self.assertEqual(tmp_dir, uri.object_name)
+        self.assertFalse(hasattr(uri, 'version_id'))
+        self.assertFalse(hasattr(uri, 'generation'))
+        self.assertFalse(hasattr(uri, 'is_version_specific'))
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        # Don't check uri.names_container(), uri.names_directory(),
+        # uri.names_file(), or uri.names_object(), because for file URIs these
+        # functions look at the file system and apparently unit tests run
+        # chroot'd.
+        self.assertEqual(uri.is_stream(), False)
+
+    def test_gs_object_uri_contains_sharp_not_matching_version_syntax(self):
+        uri_str = 'gs://bucket/obj#13a990880167400'
+        uri = boto.storage_uri(uri_str, validate=False,
+            suppress_consec_slashes=False)
+        self.assertEqual('gs', uri.scheme)
+        self.assertEqual(uri_str, uri.uri)
+        self.assertEqual('gs://bucket/obj#13a990880167400',
+                         uri.versionless_uri)
+        self.assertEqual('bucket', uri.bucket_name)
+        self.assertEqual('obj#13a990880167400', uri.object_name)
+        self.assertEqual(None, uri.version_id)
+        self.assertEqual(None, uri.generation)
+        self.assertEqual(uri.names_provider(), False)
+        self.assertEqual(uri.names_container(), False)
+        self.assertEqual(uri.names_bucket(), False)
+        self.assertEqual(uri.names_object(), True)
+        self.assertEqual(uri.names_directory(), False)
+        self.assertEqual(uri.names_file(), False)
+        self.assertEqual(uri.is_stream(), False)
+        self.assertEqual(uri.is_version_specific, False)
+
+    def test_invalid_scheme(self):
+        uri_str = 'mars://bucket/object'
+        try:
+            boto.storage_uri(uri_str, validate=False,
+                suppress_consec_slashes=False)
+        except InvalidUriError as e:
+            self.assertIn('Unrecognized scheme', e.message)
+
+
+if __name__ == '__main__':
+    unittest.main()