Index: third_party/gsutil/boto/tests/integration/gs/test_basic.py |
diff --git a/third_party/gsutil/boto/tests/integration/gs/test_basic.py b/third_party/gsutil/boto/tests/integration/gs/test_basic.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..d425f306bed9ed4adb3f45c3d18d7ef5af4e18b2 |
--- /dev/null |
+++ b/third_party/gsutil/boto/tests/integration/gs/test_basic.py |
@@ -0,0 +1,379 @@ |
+# -*- coding: utf-8 -*- |
+# Copyright (c) 2006-2011 Mitch Garnaat http://garnaat.org/ |
+# Copyright (c) 2010, Eucalyptus Systems, Inc. |
+# Copyright (c) 2011, Nexenta Systems, Inc. |
+# Copyright (c) 2012, Google, Inc. |
+# All rights reserved. |
+# |
+# Permission is hereby granted, free of charge, to any person obtaining a |
+# copy of this software and associated documentation files (the |
+# "Software"), to deal in the Software without restriction, including |
+# without limitation the rights to use, copy, modify, merge, publish, dis- |
+# tribute, sublicense, and/or sell copies of the Software, and to permit |
+# persons to whom the Software is furnished to do so, subject to the fol- |
+# lowing conditions: |
+# |
+# The above copyright notice and this permission notice shall be included |
+# in all copies or substantial portions of the Software. |
+# |
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
+# IN THE SOFTWARE. |
+ |
+""" |
+Some unit tests for the GSConnection |
+""" |
+ |
+import os |
+import re |
+import StringIO |
+import xml.sax |
+ |
+from boto import handler |
+from boto import storage_uri |
+from boto.gs.acl import ACL |
+from boto.gs.cors import Cors |
+from tests.integration.gs.testcase import GSTestCase |
+ |
+ |
+CORS_EMPTY = '<CorsConfig></CorsConfig>' |
+CORS_DOC = ('<CorsConfig><Cors><Origins><Origin>origin1.example.com' |
+ '</Origin><Origin>origin2.example.com</Origin></Origins>' |
+ '<Methods><Method>GET</Method><Method>PUT</Method>' |
+ '<Method>POST</Method></Methods><ResponseHeaders>' |
+ '<ResponseHeader>foo</ResponseHeader>' |
+ '<ResponseHeader>bar</ResponseHeader></ResponseHeaders>' |
+ '</Cors></CorsConfig>') |
+ |
+# Regexp for matching project-private default object ACL. |
+PROJECT_PRIVATE_RE = ('\s*<AccessControlList>\s*<Entries>\s*<Entry>' |
+ '\s*<Scope type="GroupById"><ID>[0-9a-fA-F]+</ID></Scope>' |
+ '\s*<Permission>FULL_CONTROL</Permission>\s*</Entry>\s*<Entry>' |
+ '\s*<Scope type="GroupById"><ID>[0-9a-fA-F]+</ID></Scope>' |
+ '\s*<Permission>FULL_CONTROL</Permission>\s*</Entry>\s*<Entry>' |
+ '\s*<Scope type="GroupById"><ID>[0-9a-fA-F]+</ID></Scope>' |
+ '\s*<Permission>READ</Permission></Entry>\s*</Entries>' |
+ '\s*</AccessControlList>\s*') |
+ |
+ |
+class GSBasicTest(GSTestCase): |
+ """Tests some basic GCS functionality.""" |
+ |
+ def test_read_write(self): |
+ """Tests basic read/write to keys.""" |
+ bucket = self._MakeBucket() |
+ bucket_name = bucket.name |
+ # now try a get_bucket call and see if it's really there |
+ bucket = self._GetConnection().get_bucket(bucket_name) |
+ key_name = 'foobar' |
+ k = bucket.new_key(key_name) |
+ s1 = 'This is a test of file upload and download' |
+ k.set_contents_from_string(s1) |
+ tmpdir = self._MakeTempDir() |
+ fpath = os.path.join(tmpdir, key_name) |
+ fp = open(fpath, 'wb') |
+ # now get the contents from gcs to a local file |
+ k.get_contents_to_file(fp) |
+ fp.close() |
+ fp = open(fpath) |
+ # check to make sure content read from gcs is identical to original |
+ self.assertEqual(s1, fp.read()) |
+ fp.close() |
+ # check to make sure set_contents_from_file is working |
+ sfp = StringIO.StringIO('foo') |
+ k.set_contents_from_file(sfp) |
+ self.assertEqual(k.get_contents_as_string(), 'foo') |
+ sfp2 = StringIO.StringIO('foo2') |
+ k.set_contents_from_file(sfp2) |
+ self.assertEqual(k.get_contents_as_string(), 'foo2') |
+ |
+ def test_get_all_keys(self): |
+ """Tests get_all_keys.""" |
+ phony_mimetype = 'application/x-boto-test' |
+ headers = {'Content-Type': phony_mimetype} |
+ tmpdir = self._MakeTempDir() |
+ fpath = os.path.join(tmpdir, 'foobar1') |
+ fpath2 = os.path.join(tmpdir, 'foobar') |
+ with open(fpath2, 'w') as f: |
+ f.write('test-data') |
+ bucket = self._MakeBucket() |
+ |
+ # First load some data for the first one, overriding content type. |
+ k = bucket.new_key('foobar') |
+ s1 = 'test-contents' |
+ s2 = 'test-contents2' |
+ k.name = 'foo/bar' |
+ k.set_contents_from_string(s1, headers) |
+ k.name = 'foo/bas' |
+ k.set_contents_from_filename(fpath2) |
+ k.name = 'foo/bat' |
+ k.set_contents_from_string(s1) |
+ k.name = 'fie/bar' |
+ k.set_contents_from_string(s1) |
+ k.name = 'fie/bas' |
+ k.set_contents_from_string(s1) |
+ k.name = 'fie/bat' |
+ k.set_contents_from_string(s1) |
+ # try resetting the contents to another value |
+ md5 = k.md5 |
+ k.set_contents_from_string(s2) |
+ self.assertNotEqual(k.md5, md5) |
+ |
+ fp2 = open(fpath2, 'rb') |
+ k.md5 = None |
+ k.base64md5 = None |
+ k.set_contents_from_stream(fp2) |
+ fp = open(fpath, 'wb') |
+ k.get_contents_to_file(fp) |
+ fp.close() |
+ fp2.seek(0, 0) |
+ fp = open(fpath, 'rb') |
+ self.assertEqual(fp2.read(), fp.read()) |
+ fp.close() |
+ fp2.close() |
+ all = bucket.get_all_keys() |
+ self.assertEqual(len(all), 6) |
+ rs = bucket.get_all_keys(prefix='foo') |
+ self.assertEqual(len(rs), 3) |
+ rs = bucket.get_all_keys(prefix='', delimiter='/') |
+ self.assertEqual(len(rs), 2) |
+ rs = bucket.get_all_keys(maxkeys=5) |
+ self.assertEqual(len(rs), 5) |
+ |
+ def test_bucket_lookup(self): |
+ """Test the bucket lookup method.""" |
+ bucket = self._MakeBucket() |
+ k = bucket.new_key('foo/bar') |
+ phony_mimetype = 'application/x-boto-test' |
+ headers = {'Content-Type': phony_mimetype} |
+ k.set_contents_from_string('testdata', headers) |
+ |
+ k = bucket.lookup('foo/bar') |
+ self.assertIsInstance(k, bucket.key_class) |
+ self.assertEqual(k.content_type, phony_mimetype) |
+ k = bucket.lookup('notthere') |
+ self.assertIsNone(k) |
+ |
+ def test_metadata(self): |
+ """Test key metadata operations.""" |
+ bucket = self._MakeBucket() |
+ k = self._MakeKey(bucket=bucket) |
+ key_name = k.name |
+ s1 = 'This is a test of file upload and download' |
+ |
+ mdkey1 = 'meta1' |
+ mdval1 = 'This is the first metadata value' |
+ k.set_metadata(mdkey1, mdval1) |
+ mdkey2 = 'meta2' |
+ mdval2 = 'This is the second metadata value' |
+ k.set_metadata(mdkey2, mdval2) |
+ |
+ # Test unicode character. |
+ mdval3 = u'föö' |
+ mdkey3 = 'meta3' |
+ k.set_metadata(mdkey3, mdval3) |
+ k.set_contents_from_string(s1) |
+ |
+ k = bucket.lookup(key_name) |
+ self.assertEqual(k.get_metadata(mdkey1), mdval1) |
+ self.assertEqual(k.get_metadata(mdkey2), mdval2) |
+ self.assertEqual(k.get_metadata(mdkey3), mdval3) |
+ k = bucket.new_key(key_name) |
+ k.get_contents_as_string() |
+ self.assertEqual(k.get_metadata(mdkey1), mdval1) |
+ self.assertEqual(k.get_metadata(mdkey2), mdval2) |
+ self.assertEqual(k.get_metadata(mdkey3), mdval3) |
+ |
+ def test_list_iterator(self): |
+ """Test list and iterator.""" |
+ bucket = self._MakeBucket() |
+ num_iter = len([k for k in bucket.list()]) |
+ rs = bucket.get_all_keys() |
+ num_keys = len(rs) |
+ self.assertEqual(num_iter, num_keys) |
+ |
+ def test_acl(self): |
+ """Test bucket and key ACLs.""" |
+ bucket = self._MakeBucket() |
+ |
+ # try some acl stuff |
+ bucket.set_acl('public-read') |
+ acl = bucket.get_acl() |
+ self.assertEqual(len(acl.entries.entry_list), 2) |
+ bucket.set_acl('private') |
+ acl = bucket.get_acl() |
+ self.assertEqual(len(acl.entries.entry_list), 1) |
+ k = self._MakeKey(bucket=bucket) |
+ k.set_acl('public-read') |
+ acl = k.get_acl() |
+ self.assertEqual(len(acl.entries.entry_list), 2) |
+ k.set_acl('private') |
+ acl = k.get_acl() |
+ self.assertEqual(len(acl.entries.entry_list), 1) |
+ |
+ # Test case-insensitivity of XML ACL parsing. |
+ acl_xml = ( |
+ '<ACCESSControlList><EntrIes><Entry>' + |
+ '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' + |
+ '</Entry></EntrIes></ACCESSControlList>') |
+ acl = ACL() |
+ h = handler.XmlHandler(acl, bucket) |
+ xml.sax.parseString(acl_xml, h) |
+ bucket.set_acl(acl) |
+ self.assertEqual(len(acl.entries.entry_list), 1) |
+ aclstr = k.get_xml_acl() |
+ self.assertGreater(aclstr.count('/Entry', 1), 0) |
+ |
+ def test_logging(self): |
+ """Test set/get raw logging subresource.""" |
+ bucket = self._MakeBucket() |
+ empty_logging_str="<?xml version='1.0' encoding='UTF-8'?><Logging/>" |
+ logging_str = ( |
+ "<?xml version='1.0' encoding='UTF-8'?><Logging>" |
+ "<LogBucket>log-bucket</LogBucket>" + |
+ "<LogObjectPrefix>example</LogObjectPrefix>" + |
+ "</Logging>") |
+ bucket.set_subresource('logging', logging_str) |
+ self.assertEqual(bucket.get_subresource('logging'), logging_str) |
+ # try disable/enable logging |
+ bucket.disable_logging() |
+ self.assertEqual(bucket.get_subresource('logging'), empty_logging_str) |
+ bucket.enable_logging('log-bucket', 'example') |
+ self.assertEqual(bucket.get_subresource('logging'), logging_str) |
+ |
+ def test_copy_key(self): |
+ """Test copying a key from one bucket to another.""" |
+ # create two new, empty buckets |
+ bucket1 = self._MakeBucket() |
+ bucket2 = self._MakeBucket() |
+ bucket_name_1 = bucket1.name |
+ bucket_name_2 = bucket2.name |
+ # verify buckets got created |
+ bucket1 = self._GetConnection().get_bucket(bucket_name_1) |
+ bucket2 = self._GetConnection().get_bucket(bucket_name_2) |
+ # create a key in bucket1 and give it some content |
+ key_name = 'foobar' |
+ k1 = bucket1.new_key(key_name) |
+ self.assertIsInstance(k1, bucket1.key_class) |
+ k1.name = key_name |
+ s = 'This is a test.' |
+ k1.set_contents_from_string(s) |
+ # copy the new key from bucket1 to bucket2 |
+ k1.copy(bucket_name_2, key_name) |
+ # now copy the contents from bucket2 to a local file |
+ k2 = bucket2.lookup(key_name) |
+ self.assertIsInstance(k2, bucket2.key_class) |
+ tmpdir = self._MakeTempDir() |
+ fpath = os.path.join(tmpdir, 'foobar') |
+ fp = open(fpath, 'wb') |
+ k2.get_contents_to_file(fp) |
+ fp.close() |
+ fp = open(fpath) |
+ # check to make sure content read is identical to original |
+ self.assertEqual(s, fp.read()) |
+ fp.close() |
+ # delete keys |
+ bucket1.delete_key(k1) |
+ bucket2.delete_key(k2) |
+ |
+ def test_default_object_acls(self): |
+ """Test default object acls.""" |
+ # create a new bucket |
+ bucket = self._MakeBucket() |
+ # get default acl and make sure it's project-private |
+ acl = bucket.get_def_acl() |
+ self.assertIsNotNone(re.search(PROJECT_PRIVATE_RE, acl.to_xml())) |
+ # set default acl to a canned acl and verify it gets set |
+ bucket.set_def_acl('public-read') |
+ acl = bucket.get_def_acl() |
+ # save public-read acl for later test |
+ public_read_acl = acl |
+ self.assertEqual(acl.to_xml(), ('<AccessControlList><Entries><Entry>' |
+ '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' |
+ '</Entry></Entries></AccessControlList>')) |
+ # back to private acl |
+ bucket.set_def_acl('private') |
+ acl = bucket.get_def_acl() |
+ self.assertEqual(acl.to_xml(), |
+ '<AccessControlList></AccessControlList>') |
+ # set default acl to an xml acl and verify it gets set |
+ bucket.set_def_acl(public_read_acl) |
+ acl = bucket.get_def_acl() |
+ self.assertEqual(acl.to_xml(), ('<AccessControlList><Entries><Entry>' |
+ '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' |
+ '</Entry></Entries></AccessControlList>')) |
+ # back to private acl |
+ bucket.set_def_acl('private') |
+ acl = bucket.get_def_acl() |
+ self.assertEqual(acl.to_xml(), |
+ '<AccessControlList></AccessControlList>') |
+ |
+ def test_default_object_acls_storage_uri(self): |
+ """Test default object acls using storage_uri.""" |
+ # create a new bucket |
+ bucket = self._MakeBucket() |
+ bucket_name = bucket.name |
+ uri = storage_uri('gs://' + bucket_name) |
+ # get default acl and make sure it's project-private |
+ acl = uri.get_def_acl() |
+ self.assertIsNotNone(re.search(PROJECT_PRIVATE_RE, acl.to_xml())) |
+ # set default acl to a canned acl and verify it gets set |
+ uri.set_def_acl('public-read') |
+ acl = uri.get_def_acl() |
+ # save public-read acl for later test |
+ public_read_acl = acl |
+ self.assertEqual(acl.to_xml(), ('<AccessControlList><Entries><Entry>' |
+ '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' |
+ '</Entry></Entries></AccessControlList>')) |
+ # back to private acl |
+ uri.set_def_acl('private') |
+ acl = uri.get_def_acl() |
+ self.assertEqual(acl.to_xml(), |
+ '<AccessControlList></AccessControlList>') |
+ # set default acl to an xml acl and verify it gets set |
+ uri.set_def_acl(public_read_acl) |
+ acl = uri.get_def_acl() |
+ self.assertEqual(acl.to_xml(), ('<AccessControlList><Entries><Entry>' |
+ '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' |
+ '</Entry></Entries></AccessControlList>')) |
+ # back to private acl |
+ uri.set_def_acl('private') |
+ acl = uri.get_def_acl() |
+ self.assertEqual(acl.to_xml(), |
+ '<AccessControlList></AccessControlList>') |
+ |
+ def test_cors_xml_bucket(self): |
+ """Test setting and getting of CORS XML documents on Bucket.""" |
+ # create a new bucket |
+ bucket = self._MakeBucket() |
+ bucket_name = bucket.name |
+ # now call get_bucket to see if it's really there |
+ bucket = self._GetConnection().get_bucket(bucket_name) |
+ # get new bucket cors and make sure it's empty |
+ cors = re.sub(r'\s', '', bucket.get_cors().to_xml()) |
+ self.assertEqual(cors, CORS_EMPTY) |
+ # set cors document on new bucket |
+ bucket.set_cors(CORS_DOC) |
+ cors = re.sub(r'\s', '', bucket.get_cors().to_xml()) |
+ self.assertEqual(cors, CORS_DOC) |
+ |
+ def test_cors_xml_storage_uri(self): |
+ """Test setting and getting of CORS XML documents with storage_uri.""" |
+ # create a new bucket |
+ bucket = self._MakeBucket() |
+ bucket_name = bucket.name |
+ uri = storage_uri('gs://' + bucket_name) |
+ # get new bucket cors and make sure it's empty |
+ cors = re.sub(r'\s', '', uri.get_cors().to_xml()) |
+ self.assertEqual(cors, CORS_EMPTY) |
+ # set cors document on new bucket |
+ cors_obj = Cors() |
+ h = handler.XmlHandler(cors_obj, None) |
+ xml.sax.parseString(CORS_DOC, h) |
+ uri.set_cors(cors_obj) |
+ cors = re.sub(r'\s', '', uri.get_cors().to_xml()) |
+ self.assertEqual(cors, CORS_DOC) |