1# -*- coding: utf-8 -*- 2from mock import patch 3import xml.dom.minidom 4 5from tests.unit import unittest 6from tests.unit import AWSMockServiceTestCase 7 8from boto.exception import BotoClientError 9from boto.s3.connection import Location, S3Connection 10from boto.s3.bucket import Bucket 11from boto.s3.deletemarker import DeleteMarker 12from boto.s3.key import Key 13from boto.s3.multipart import MultiPartUpload 14from boto.s3.prefix import Prefix 15 16 17class TestS3Bucket(AWSMockServiceTestCase): 18 connection_class = S3Connection 19 20 def setUp(self): 21 super(TestS3Bucket, self).setUp() 22 23 def test_bucket_create_bucket(self): 24 self.set_http_response(status_code=200) 25 bucket = self.service_connection.create_bucket('mybucket_create') 26 self.assertEqual(bucket.name, 'mybucket_create') 27 28 def test_bucket_create_eu_central_1_location(self): 29 self.set_http_response(status_code=200) 30 bucket = self.service_connection.create_bucket( 31 'eu_central_1_bucket', 32 location=Location.EUCentral1 33 ) 34 self.assertEqual(bucket.name, 'eu_central_1_bucket') 35 36 def test_bucket_constructor(self): 37 self.set_http_response(status_code=200) 38 bucket = Bucket(self.service_connection, 'mybucket_constructor') 39 self.assertEqual(bucket.name, 'mybucket_constructor') 40 41 def test_bucket_basics(self): 42 self.set_http_response(status_code=200) 43 bucket = self.service_connection.create_bucket('mybucket') 44 self.assertEqual(bucket.__repr__(), '<Bucket: mybucket>') 45 46 def test_bucket_new_key(self): 47 self.set_http_response(status_code=200) 48 bucket = self.service_connection.create_bucket('mybucket') 49 key = bucket.new_key('mykey') 50 51 self.assertEqual(key.bucket, bucket) 52 self.assertEqual(key.key, 'mykey') 53 54 def test_bucket_new_key_missing_name(self): 55 self.set_http_response(status_code=200) 56 bucket = self.service_connection.create_bucket('mybucket') 57 58 with self.assertRaises(ValueError): 59 key = bucket.new_key('') 60 61 def test_bucket_delete_key_missing_name(self): 62 self.set_http_response(status_code=200) 63 bucket = self.service_connection.create_bucket('mybucket') 64 65 with self.assertRaises(ValueError): 66 key = bucket.delete_key('') 67 68 def test_bucket_kwargs_misspelling(self): 69 self.set_http_response(status_code=200) 70 bucket = self.service_connection.create_bucket('mybucket') 71 72 with self.assertRaises(TypeError): 73 bucket.get_all_keys(delimeter='foo') 74 75 def test__get_all_query_args(self): 76 bukket = Bucket() 77 78 # Default. 79 qa = bukket._get_all_query_args({}) 80 self.assertEqual(qa, '') 81 82 # Default with initial. 83 qa = bukket._get_all_query_args({}, 'initial=1') 84 self.assertEqual(qa, 'initial=1') 85 86 # Single param. 87 qa = bukket._get_all_query_args({ 88 'foo': 'true' 89 }) 90 self.assertEqual(qa, 'foo=true') 91 92 # Single param with initial. 93 qa = bukket._get_all_query_args({ 94 'foo': 'true' 95 }, 'initial=1') 96 self.assertEqual(qa, 'initial=1&foo=true') 97 98 # Multiple params with all the weird cases. 99 multiple_params = { 100 'foo': 'true', 101 # Ensure Unicode chars get encoded. 102 'bar': '☃', 103 # Ensure unicode strings with non-ascii characters get encoded 104 'baz': u'χ', 105 # Underscores are bad, m'kay? 106 'some_other': 'thing', 107 # Change the variant of ``max-keys``. 108 'maxkeys': 0, 109 # ``None`` values get excluded. 110 'notthere': None, 111 # Empty values also get excluded. 112 'notpresenteither': '', 113 } 114 qa = bukket._get_all_query_args(multiple_params) 115 self.assertEqual( 116 qa, 117 'bar=%E2%98%83&baz=%CF%87&foo=true&max-keys=0&some-other=thing' 118 ) 119 120 # Multiple params with initial. 121 qa = bukket._get_all_query_args(multiple_params, 'initial=1') 122 self.assertEqual( 123 qa, 124 'initial=1&bar=%E2%98%83&baz=%CF%87&foo=true&max-keys=0&some-other=thing' 125 ) 126 127 @patch.object(S3Connection, 'head_bucket') 128 def test_bucket_copy_key_no_validate(self, mock_head_bucket): 129 self.set_http_response(status_code=200) 130 bucket = self.service_connection.create_bucket('mybucket') 131 132 self.assertFalse(mock_head_bucket.called) 133 self.service_connection.get_bucket('mybucket', validate=True) 134 self.assertTrue(mock_head_bucket.called) 135 136 mock_head_bucket.reset_mock() 137 self.assertFalse(mock_head_bucket.called) 138 try: 139 bucket.copy_key('newkey', 'srcbucket', 'srckey', preserve_acl=True) 140 except: 141 # Will throw because of empty response. 142 pass 143 self.assertFalse(mock_head_bucket.called) 144 145 @patch.object(Bucket, '_get_all') 146 def test_bucket_encoding(self, mock_get_all): 147 self.set_http_response(status_code=200) 148 bucket = self.service_connection.get_bucket('mybucket') 149 150 # First, without the encoding. 151 mock_get_all.reset_mock() 152 bucket.get_all_keys() 153 mock_get_all.assert_called_with( 154 [ 155 ('Contents', Key), 156 ('CommonPrefixes', Prefix) 157 ], '', None 158 ) 159 160 # Now the variants with the encoding. 161 mock_get_all.reset_mock() 162 bucket.get_all_keys(encoding_type='url') 163 mock_get_all.assert_called_with( 164 [ 165 ('Contents', Key), 166 ('CommonPrefixes', Prefix) 167 ], '', None, 168 encoding_type='url' 169 ) 170 171 mock_get_all.reset_mock() 172 bucket.get_all_versions(encoding_type='url') 173 mock_get_all.assert_called_with( 174 [ 175 ('Version', Key), 176 ('CommonPrefixes', Prefix), 177 ('DeleteMarker', DeleteMarker), 178 ], 'versions', None, 179 encoding_type='url' 180 ) 181 182 mock_get_all.reset_mock() 183 bucket.get_all_multipart_uploads(encoding_type='url') 184 mock_get_all.assert_called_with( 185 [ 186 ('Upload', MultiPartUpload), 187 ('CommonPrefixes', Prefix) 188 ], 'uploads', None, 189 encoding_type='url' 190 ) 191 192 @patch.object(Bucket, 'get_all_keys') 193 @patch.object(Bucket, '_get_key_internal') 194 def test_bucket_get_key_no_validate(self, mock_gki, mock_gak): 195 self.set_http_response(status_code=200) 196 bucket = self.service_connection.get_bucket('mybucket') 197 key = bucket.get_key('mykey', validate=False) 198 199 self.assertEqual(len(mock_gki.mock_calls), 0) 200 self.assertTrue(isinstance(key, Key)) 201 self.assertEqual(key.name, 'mykey') 202 203 with self.assertRaises(BotoClientError): 204 bucket.get_key( 205 'mykey', 206 version_id='something', 207 validate=False 208 ) 209 210 def acl_policy(self): 211 return """<?xml version="1.0" encoding="UTF-8"?> 212 <AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"> 213 <Owner> 214 <ID>owner_id</ID> 215 <DisplayName>owner_display_name</DisplayName> 216 </Owner> 217 <AccessControlList> 218 <Grant> 219 <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 220 xsi:type="CanonicalUser"> 221 <ID>grantee_id</ID> 222 <DisplayName>grantee_display_name</DisplayName> 223 </Grantee> 224 <Permission>FULL_CONTROL</Permission> 225 </Grant> 226 </AccessControlList> 227 </AccessControlPolicy>""" 228 229 def test_bucket_acl_policy_namespace(self): 230 self.set_http_response(status_code=200) 231 bucket = self.service_connection.get_bucket('mybucket') 232 233 self.set_http_response(status_code=200, body=self.acl_policy()) 234 policy = bucket.get_acl() 235 236 xml_policy = policy.to_xml() 237 document = xml.dom.minidom.parseString(xml_policy) 238 namespace = document.documentElement.namespaceURI 239 self.assertEqual(namespace, 'http://s3.amazonaws.com/doc/2006-03-01/') 240