| Index: third_party/boto/tests/unit/s3/test_bucket.py
|
| ===================================================================
|
| --- third_party/boto/tests/unit/s3/test_bucket.py (revision 33376)
|
| +++ third_party/boto/tests/unit/s3/test_bucket.py (working copy)
|
| @@ -1,9 +1,11 @@
|
| # -*- 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
|
| @@ -112,23 +114,23 @@
|
| 'initial=1&bar=%E2%98%83&max-keys=0&foo=true&some-other=thing'
|
| )
|
|
|
| - @patch.object(Bucket, 'get_all_keys')
|
| - def test_bucket_copy_key_no_validate(self, mock_get_all_keys):
|
| + @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_get_all_keys.called)
|
| + self.assertFalse(mock_head_bucket.called)
|
| self.service_connection.get_bucket('mybucket', validate=True)
|
| - self.assertTrue(mock_get_all_keys.called)
|
| + self.assertTrue(mock_head_bucket.called)
|
|
|
| - mock_get_all_keys.reset_mock()
|
| - self.assertFalse(mock_get_all_keys.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_get_all_keys.called)
|
| + self.assertFalse(mock_head_bucket.called)
|
|
|
| @patch.object(Bucket, '_get_all')
|
| def test_bucket_encoding(self, mock_get_all):
|
| @@ -176,3 +178,52 @@
|
| ], '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/')
|
|
|