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