Revert "Add gsutil 4.13 to telemetry/third_party"

tools/telemetry/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py

Index: tools/telemetry/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py
diff --git a/tools/telemetry/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py b/tools/telemetry/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py
deleted file mode 100644
index d7b848bb2e91b8ea3902a160104ddb2f51434341..0000000000000000000000000000000000000000
--- a/tools/telemetry/third_party/gsutil/third_party/boto/tests/integration/s3/test_connection.py
+++ /dev/null
@@ -1,245 +0,0 @@
-# -*- 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.")