Index: tools/telemetry/third_party/gsutil/third_party/boto/tests/unit/s3/test_bucket.py |
diff --git a/tools/telemetry/third_party/gsutil/third_party/boto/tests/unit/s3/test_bucket.py b/tools/telemetry/third_party/gsutil/third_party/boto/tests/unit/s3/test_bucket.py |
deleted file mode 100644 |
index 72e10ed2f925369483852105ad0c354361f19564..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/third_party/boto/tests/unit/s3/test_bucket.py |
+++ /dev/null |
@@ -1,231 +0,0 @@ |
-# -*- coding: utf-8 -*- |
-from mock import patch |
-import xml.dom.minidom |
- |
-from tests.unit import unittest |
-from tests.unit import AWSMockServiceTestCase |
- |
-from boto.exception import BotoClientError |
-from boto.s3.connection import S3Connection |
-from boto.s3.bucket import Bucket |
-from boto.s3.deletemarker import DeleteMarker |
-from boto.s3.key import Key |
-from boto.s3.multipart import MultiPartUpload |
-from boto.s3.prefix import Prefix |
- |
- |
-class TestS3Bucket(AWSMockServiceTestCase): |
- connection_class = S3Connection |
- |
- def setUp(self): |
- super(TestS3Bucket, self).setUp() |
- |
- def test_bucket_create_bucket(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket_create') |
- self.assertEqual(bucket.name, 'mybucket_create') |
- |
- def test_bucket_constructor(self): |
- self.set_http_response(status_code=200) |
- bucket = Bucket(self.service_connection, 'mybucket_constructor') |
- self.assertEqual(bucket.name, 'mybucket_constructor') |
- |
- def test_bucket_basics(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- self.assertEqual(bucket.__repr__(), '<Bucket: mybucket>') |
- |
- def test_bucket_new_key(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- key = bucket.new_key('mykey') |
- |
- self.assertEqual(key.bucket, bucket) |
- self.assertEqual(key.key, 'mykey') |
- |
- def test_bucket_new_key_missing_name(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- |
- with self.assertRaises(ValueError): |
- key = bucket.new_key('') |
- |
- def test_bucket_delete_key_missing_name(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- |
- with self.assertRaises(ValueError): |
- key = bucket.delete_key('') |
- |
- def test_bucket_kwargs_misspelling(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- |
- with self.assertRaises(TypeError): |
- bucket.get_all_keys(delimeter='foo') |
- |
- def test__get_all_query_args(self): |
- bukket = Bucket() |
- |
- # Default. |
- qa = bukket._get_all_query_args({}) |
- self.assertEqual(qa, '') |
- |
- # Default with initial. |
- qa = bukket._get_all_query_args({}, 'initial=1') |
- self.assertEqual(qa, 'initial=1') |
- |
- # Single param. |
- qa = bukket._get_all_query_args({ |
- 'foo': 'true' |
- }) |
- self.assertEqual(qa, 'foo=true') |
- |
- # Single param with initial. |
- qa = bukket._get_all_query_args({ |
- 'foo': 'true' |
- }, 'initial=1') |
- self.assertEqual(qa, 'initial=1&foo=true') |
- |
- # Multiple params with all the weird cases. |
- multiple_params = { |
- 'foo': 'true', |
- # Ensure Unicode chars get encoded. |
- 'bar': '☃', |
- # Ensure unicode strings with non-ascii characters get encoded |
- 'baz': u'χ', |
- # Underscores are bad, m'kay? |
- 'some_other': 'thing', |
- # Change the variant of ``max-keys``. |
- 'maxkeys': 0, |
- # ``None`` values get excluded. |
- 'notthere': None, |
- # Empty values also get excluded. |
- 'notpresenteither': '', |
- } |
- qa = bukket._get_all_query_args(multiple_params) |
- self.assertEqual( |
- qa, |
- 'bar=%E2%98%83&baz=%CF%87&foo=true&max-keys=0&some-other=thing' |
- ) |
- |
- # Multiple params with initial. |
- qa = bukket._get_all_query_args(multiple_params, 'initial=1') |
- self.assertEqual( |
- qa, |
- 'initial=1&bar=%E2%98%83&baz=%CF%87&foo=true&max-keys=0&some-other=thing' |
- ) |
- |
- @patch.object(S3Connection, 'head_bucket') |
- def test_bucket_copy_key_no_validate(self, mock_head_bucket): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.create_bucket('mybucket') |
- |
- self.assertFalse(mock_head_bucket.called) |
- self.service_connection.get_bucket('mybucket', validate=True) |
- self.assertTrue(mock_head_bucket.called) |
- |
- mock_head_bucket.reset_mock() |
- self.assertFalse(mock_head_bucket.called) |
- try: |
- bucket.copy_key('newkey', 'srcbucket', 'srckey', preserve_acl=True) |
- except: |
- # Will throw because of empty response. |
- pass |
- self.assertFalse(mock_head_bucket.called) |
- |
- @patch.object(Bucket, '_get_all') |
- def test_bucket_encoding(self, mock_get_all): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.get_bucket('mybucket') |
- |
- # First, without the encoding. |
- mock_get_all.reset_mock() |
- bucket.get_all_keys() |
- mock_get_all.assert_called_with( |
- [ |
- ('Contents', Key), |
- ('CommonPrefixes', Prefix) |
- ], '', None |
- ) |
- |
- # Now the variants with the encoding. |
- mock_get_all.reset_mock() |
- bucket.get_all_keys(encoding_type='url') |
- mock_get_all.assert_called_with( |
- [ |
- ('Contents', Key), |
- ('CommonPrefixes', Prefix) |
- ], '', None, |
- encoding_type='url' |
- ) |
- |
- mock_get_all.reset_mock() |
- bucket.get_all_versions(encoding_type='url') |
- mock_get_all.assert_called_with( |
- [ |
- ('Version', Key), |
- ('CommonPrefixes', Prefix), |
- ('DeleteMarker', DeleteMarker), |
- ], 'versions', None, |
- encoding_type='url' |
- ) |
- |
- mock_get_all.reset_mock() |
- bucket.get_all_multipart_uploads(encoding_type='url') |
- mock_get_all.assert_called_with( |
- [ |
- ('Upload', MultiPartUpload), |
- ('CommonPrefixes', Prefix) |
- ], 'uploads', None, |
- encoding_type='url' |
- ) |
- |
- @patch.object(Bucket, 'get_all_keys') |
- @patch.object(Bucket, '_get_key_internal') |
- def test_bucket_get_key_no_validate(self, mock_gki, mock_gak): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.get_bucket('mybucket') |
- key = bucket.get_key('mykey', validate=False) |
- |
- self.assertEqual(len(mock_gki.mock_calls), 0) |
- self.assertTrue(isinstance(key, Key)) |
- self.assertEqual(key.name, 'mykey') |
- |
- with self.assertRaises(BotoClientError): |
- bucket.get_key( |
- 'mykey', |
- version_id='something', |
- validate=False |
- ) |
- |
- def acl_policy(self): |
- return """<?xml version="1.0" encoding="UTF-8"?> |
- <AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"> |
- <Owner> |
- <ID>owner_id</ID> |
- <DisplayName>owner_display_name</DisplayName> |
- </Owner> |
- <AccessControlList> |
- <Grant> |
- <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" |
- xsi:type="CanonicalUser"> |
- <ID>grantee_id</ID> |
- <DisplayName>grantee_display_name</DisplayName> |
- </Grantee> |
- <Permission>FULL_CONTROL</Permission> |
- </Grant> |
- </AccessControlList> |
- </AccessControlPolicy>""" |
- |
- def test_bucket_acl_policy_namespace(self): |
- self.set_http_response(status_code=200) |
- bucket = self.service_connection.get_bucket('mybucket') |
- |
- self.set_http_response(status_code=200, body=self.acl_policy()) |
- policy = bucket.get_acl() |
- |
- xml_policy = policy.to_xml() |
- document = xml.dom.minidom.parseString(xml_policy) |
- namespace = document.documentElement.namespaceURI |
- self.assertEqual(namespace, 'http://s3.amazonaws.com/doc/2006-03-01/') |