Update checked in version of gsutil to version 4.6

third_party/boto/tests/unit/s3/test_bucket.py

 # -*- 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
@@ skipping 88 matching lines @@
             'bar=%E2%98%83&max-keys=0&foo=true&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&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):
         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(
@@ skipping 27 matching lines @@
 
         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/')