Added gsutil to depot tools

third_party/gsutil/boto/tests/devpay/test_s3.py

+#!/usr/bin/env python
+
+# Copyright (c) 2006,2007 Mitch Garnaat http://garnaat.org/
+#
+# 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 S3Connection
+"""
+
+import time
+import os
+import urllib
+
+from boto.s3.connection import S3Connection
+from boto.exception import S3PermissionsError
+
+# this test requires a devpay product and user token to run:
+
+AMAZON_USER_TOKEN = '{UserToken}...your token here...'
+DEVPAY_HEADERS = { 'x-amz-security-token': AMAZON_USER_TOKEN }
+
+def test():
+    print '--- running S3Connection tests (DevPay) ---'
+    c = S3Connection()
+    # create a new, empty bucket
+    bucket_name = 'test-%d' % int(time.time())
+    bucket = c.create_bucket(bucket_name, headers=DEVPAY_HEADERS)
+    # now try a get_bucket call and see if it's really there
+    bucket = c.get_bucket(bucket_name, headers=DEVPAY_HEADERS)
+    # test logging
+    logging_bucket = c.create_bucket(bucket_name + '-log', headers=DEVPAY_HEADERS)
+    logging_bucket.set_as_logging_target(headers=DEVPAY_HEADERS)
+    bucket.enable_logging(target_bucket=logging_bucket, target_prefix=bucket.name, headers=DEVPAY_HEADERS)
+    bucket.disable_logging(headers=DEVPAY_HEADERS)
+    c.delete_bucket(logging_bucket, headers=DEVPAY_HEADERS)
+    # create a new key and store it's content from a string
+    k = bucket.new_key()
+    k.name = 'foobar'
+    s1 = 'This is a test of file upload and download'
+    s2 = 'This is a second string to test file upload and download'
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    fp = open('foobar', 'wb')
+    # now get the contents from s3 to a local file
+    k.get_contents_to_file(fp, headers=DEVPAY_HEADERS)
+    fp.close()
+    fp = open('foobar')
+    # check to make sure content read from s3 is identical to original
+    assert s1 == fp.read(), 'corrupted file'
+    fp.close()
+    # test generated URLs
+    url = k.generate_url(3600, headers=DEVPAY_HEADERS)
+    file = urllib.urlopen(url)
+    assert s1 == file.read(), 'invalid URL %s' % url
+    url = k.generate_url(3600, force_http=True, headers=DEVPAY_HEADERS)
+    file = urllib.urlopen(url)
+    assert s1 == file.read(), 'invalid URL %s' % url
+    bucket.delete_key(k, headers=DEVPAY_HEADERS)
+    # test a few variations on get_all_keys - first load some data
+    # for the first one, let's override the content type
+    phony_mimetype = 'application/x-boto-test'
+    headers = {'Content-Type': phony_mimetype}
+    headers.update(DEVPAY_HEADERS)
+    k.name = 'foo/bar'
+    k.set_contents_from_string(s1, headers)
+    k.name = 'foo/bas'
+    k.set_contents_from_filename('foobar', headers=DEVPAY_HEADERS)
+    k.name = 'foo/bat'
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    k.name = 'fie/bar'
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    k.name = 'fie/bas'
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    k.name = 'fie/bat'
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    # try resetting the contents to another value
+    md5 = k.md5
+    k.set_contents_from_string(s2, headers=DEVPAY_HEADERS)
+    assert k.md5 != md5
+    os.unlink('foobar')
+    all = bucket.get_all_keys(headers=DEVPAY_HEADERS)
+    assert len(all) == 6
+    rs = bucket.get_all_keys(prefix='foo', headers=DEVPAY_HEADERS)
+    assert len(rs) == 3
+    rs = bucket.get_all_keys(prefix='', delimiter='/', headers=DEVPAY_HEADERS)
+    assert len(rs) == 2
+    rs = bucket.get_all_keys(maxkeys=5, headers=DEVPAY_HEADERS)
+    assert len(rs) == 5
+    # test the lookup method
+    k = bucket.lookup('foo/bar', headers=DEVPAY_HEADERS)
+    assert isinstance(k, bucket.key_class)
+    assert k.content_type == phony_mimetype
+    k = bucket.lookup('notthere', headers=DEVPAY_HEADERS)
+    assert k == None
+    # try some metadata stuff
+    k = bucket.new_key()
+    k.name = 'has_metadata'
+    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)
+    k.set_contents_from_string(s1, headers=DEVPAY_HEADERS)
+    k = bucket.lookup('has_metadata', headers=DEVPAY_HEADERS)
+    assert k.get_metadata(mdkey1) == mdval1
+    assert k.get_metadata(mdkey2) == mdval2
+    k = bucket.new_key()
+    k.name = 'has_metadata'
+    k.get_contents_as_string(headers=DEVPAY_HEADERS)
+    assert k.get_metadata(mdkey1) == mdval1
+    assert k.get_metadata(mdkey2) == mdval2
+    bucket.delete_key(k, headers=DEVPAY_HEADERS)
+    # test list and iterator
+    rs1 = bucket.list(headers=DEVPAY_HEADERS)
+    num_iter = 0
+    for r in rs1:
+        num_iter = num_iter + 1
+    rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
+    num_keys = len(rs)
+    assert num_iter == num_keys
+    # try a key with a funny character
+    k = bucket.new_key()
+    k.name = 'testnewline\n'
+    k.set_contents_from_string('This is a test', headers=DEVPAY_HEADERS)
+    rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
+    assert len(rs) == num_keys + 1
+    bucket.delete_key(k, headers=DEVPAY_HEADERS)
+    rs = bucket.get_all_keys(headers=DEVPAY_HEADERS)
+    assert len(rs) == num_keys
+    # try some acl stuff
+    bucket.set_acl('public-read', headers=DEVPAY_HEADERS)
+    policy = bucket.get_acl(headers=DEVPAY_HEADERS)
+    assert len(policy.acl.grants) == 2
+    bucket.set_acl('private', headers=DEVPAY_HEADERS)
+    policy = bucket.get_acl(headers=DEVPAY_HEADERS)
+    assert len(policy.acl.grants) == 1
+    k = bucket.lookup('foo/bar', headers=DEVPAY_HEADERS)
+    k.set_acl('public-read', headers=DEVPAY_HEADERS)
+    policy = k.get_acl(headers=DEVPAY_HEADERS)
+    assert len(policy.acl.grants) == 2
+    k.set_acl('private', headers=DEVPAY_HEADERS)
+    policy = k.get_acl(headers=DEVPAY_HEADERS)
+    assert len(policy.acl.grants) == 1
+    # try the convenience methods for grants
+    # this doesn't work with devpay
+    #bucket.add_user_grant('FULL_CONTROL',
+    #                      'c1e724fbfa0979a4448393c59a8c055011f739b6d102fb37a65f26414653cd67',
+    #                      headers=DEVPAY_HEADERS)
+    try:
+        bucket.add_email_grant('foobar', 'foo@bar.com', headers=DEVPAY_HEADERS)
+    except S3PermissionsError:
+        pass
+    # now delete all keys in bucket
+    for k in all:
+        bucket.delete_key(k, headers=DEVPAY_HEADERS)
+    # now delete bucket
+
+    c.delete_bucket(bucket, headers=DEVPAY_HEADERS)
+
+    print '--- tests completed ---'
+
+if __name__ == '__main__':
+    test()