xref: /external/chromium-trace/catapult/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py

# -*- coding: utf-8 -*-
# Copyright (c) 2006-2011 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 unittest
import time
import os

from boto.s3.connection import S3Connection
from boto.s3.bucket import Bucket
from boto.exception import S3PermissionsError, S3ResponseError
from boto.compat import http_client, six, urlopen, urlsplit


class S3ConnectionTest (unittest.TestCase):
    s3 = True

    def test_1_basic(self):
        print('--- running S3Connection tests ---')
        c = S3Connection()
        # create a new, empty bucket
        bucket_name = 'test-%d' % int(time.time())
        bucket = c.create_bucket(bucket_name)
        # now try a get_bucket call and see if it's really there
        bucket = c.get_bucket(bucket_name)
        # test logging
        logging_bucket = c.create_bucket(bucket_name + '-log')
        logging_bucket.set_as_logging_target()
        bucket.enable_logging(target_bucket=logging_bucket, target_prefix=bucket.name)
        bucket.disable_logging()
        c.delete_bucket(logging_bucket)
        k = bucket.new_key('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)
        fp = open('foobar', 'wb')
        # now get the contents from s3 to a local file
        k.get_contents_to_file(fp)
        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)
        file = urlopen(url)
        assert s1 == file.read().decode('utf-8'), 'invalid URL %s' % url
        url = k.generate_url(3600, force_http=True)
        file = urlopen(url)
        assert s1 == file.read().decode('utf-8'), 'invalid URL %s' % url
        url = k.generate_url(3600, force_http=True, headers={'x-amz-x-token' : 'XYZ'})
        file = urlopen(url)
        assert s1 == file.read().decode('utf-8'), 'invalid URL %s' % url
        rh = {'response-content-disposition': 'attachment; filename="foo.txt"'}
        url = k.generate_url(60, response_headers=rh)
        file = urlopen(url)
        assert s1 == file.read().decode('utf-8'), 'invalid URL %s' % url
        #test whether amperands and to-be-escaped characters work in header filename
        rh = {'response-content-disposition': 'attachment; filename="foo&z%20ar&ar&zar&bar.txt"'}
        url = k.generate_url(60, response_headers=rh, force_http=True)
        file = urlopen(url)
        assert s1 == file.read().decode('utf-8'), 'invalid URL %s' % url
        # overwrite foobar contents with a PUT
        url = k.generate_url(3600, 'PUT', force_http=True, policy='private', reduced_redundancy=True)
        up = urlsplit(url)
        con = http_client.HTTPConnection(up.hostname, up.port)
        con.request("PUT", up.path + '?' + up.query, body="hello there")
        resp = con.getresponse()
        assert 200 == resp.status
        assert b"hello there" == k.get_contents_as_string()
        bucket.delete_key(k)
        # 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}
        k.name = 'foo/bar'
        k.set_contents_from_string(s1, headers)
        k.name = 'foo/bas'
        size = k.set_contents_from_filename('foobar')
        assert size == 42
        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)
        assert k.md5 != md5
        os.unlink('foobar')
        all = bucket.get_all_keys()
        assert len(all) == 6
        rs = bucket.get_all_keys(prefix='foo')
        assert len(rs) == 3
        rs = bucket.get_all_keys(prefix='', delimiter='/')
        assert len(rs) == 2
        rs = bucket.get_all_keys(maxkeys=5)
        assert len(rs) == 5
        # test the lookup method
        k = bucket.lookup('foo/bar')
        assert isinstance(k, bucket.key_class)
        assert k.content_type == phony_mimetype
        k = bucket.lookup('notthere')
        assert k == None
        # try some metadata stuff
        k = bucket.new_key('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)
        # try a unicode metadata value
        mdval3 = u'föö'
        mdkey3 = 'meta3'
        k.set_metadata(mdkey3, mdval3)
        k.set_contents_from_string(s1)
        k = bucket.lookup('has_metadata')
        assert k.get_metadata(mdkey1) == mdval1
        assert k.get_metadata(mdkey2) == mdval2
        assert k.get_metadata(mdkey3) == mdval3
        k = bucket.new_key('has_metadata')
        k.get_contents_as_string()
        assert k.get_metadata(mdkey1) == mdval1
        assert k.get_metadata(mdkey2) == mdval2
        assert k.get_metadata(mdkey3) == mdval3
        bucket.delete_key(k)
        # test list and iterator
        rs1 = bucket.list()
        num_iter = 0
        for r in rs1:
            num_iter = num_iter + 1
        rs = bucket.get_all_keys()
        num_keys = len(rs)
        assert num_iter == num_keys
        # try a key with a funny character
        k = bucket.new_key('testnewline\n')
        k.set_contents_from_string('This is a test')
        rs = bucket.get_all_keys()
        assert len(rs) == num_keys + 1
        bucket.delete_key(k)
        rs = bucket.get_all_keys()
        assert len(rs) == num_keys
        # try some acl stuff
        bucket.set_acl('public-read')
        policy = bucket.get_acl()
        assert len(policy.acl.grants) == 2
        bucket.set_acl('private')
        policy = bucket.get_acl()
        assert len(policy.acl.grants) == 1
        k = bucket.lookup('foo/bar')
        k.set_acl('public-read')
        policy = k.get_acl()
        assert len(policy.acl.grants) == 2
        k.set_acl('private')
        policy = k.get_acl()
        assert len(policy.acl.grants) == 1
        # try the convenience methods for grants
        bucket.add_user_grant('FULL_CONTROL',
                              'c1e724fbfa0979a4448393c59a8c055011f739b6d102fb37a65f26414653cd67')
        try:
            bucket.add_email_grant('foobar', 'foo@bar.com')
        except S3PermissionsError:
            pass
        # now try to create an RRS key
        k = bucket.new_key('reduced_redundancy')
        k.set_contents_from_string('This key has reduced redundancy',
                                   reduced_redundancy=True)

        # now try to inject a response header
        data = k.get_contents_as_string(response_headers={'response-content-type' : 'foo/bar'})
        assert k.content_type == 'foo/bar'

        # now delete all keys in bucket
        for k in bucket:
            if k.name == 'reduced_redundancy':
                assert k.storage_class == 'REDUCED_REDUNDANCY'
            bucket.delete_key(k)
        # now delete bucket
        time.sleep(5)
        c.delete_bucket(bucket)
        print('--- tests completed ---')

    def test_basic_anon(self):
        auth_con = S3Connection()
        # create a new, empty bucket
        bucket_name = 'test-%d' % int(time.time())
        auth_bucket = auth_con.create_bucket(bucket_name)

        # try read the bucket anonymously
        anon_con = S3Connection(anon=True)
        anon_bucket = Bucket(anon_con, bucket_name)
        try:
            next(iter(anon_bucket.list()))
            self.fail("anon bucket list should fail")
        except S3ResponseError:
            pass

        # give bucket anon user access and anon read again
        auth_bucket.set_acl('public-read')
        time.sleep(5)
        try:
            next(iter(anon_bucket.list()))
            self.fail("not expecting contents")
        except S3ResponseError as e:
            self.fail("We should have public-read access, but received "
                      "an error: %s" % e)
        except StopIteration:
            pass

        # cleanup
        auth_con.delete_bucket(auth_bucket)

    def test_error_code_populated(self):
        c = S3Connection()
        try:
            c.create_bucket('bad$bucket$name')
        except S3ResponseError as e:
            self.assertEqual(e.error_code, 'InvalidBucketName')
        else:
            self.fail("S3ResponseError not raised.")