test_blob.py

# Copyright 2014 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import base64
import datetime
import hashlib
import io
import json
import os
import tempfile
import unittest

import google.cloud.storage.blob
import mock
import pytest
import six
from six.moves import http_client


def _make_credentials():
    import google.auth.credentials

    return mock.Mock(spec=google.auth.credentials.Credentials)


class Test_Blob(unittest.TestCase):
    @staticmethod
    def _make_one(*args, **kw):
        from google.cloud.storage.blob import Blob

        properties = kw.pop("properties", {})
        blob = Blob(*args, **kw)
        blob._properties.update(properties)
        return blob

    def test_ctor_wo_encryption_key(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {"key": "value"}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertIs(blob.bucket, bucket)
        self.assertEqual(blob.name, BLOB_NAME)
        self.assertEqual(blob._properties, properties)
        self.assertFalse(blob._acl.loaded)
        self.assertIs(blob._acl.blob, blob)
        self.assertEqual(blob._encryption_key, None)
        self.assertEqual(blob.kms_key_name, None)

    def test_ctor_with_encoded_unicode(self):
        blob_name = b"wet \xe2\x9b\xb5"
        blob = self._make_one(blob_name, bucket=None)
        unicode_name = u"wet \N{sailboat}"
        self.assertNotIsInstance(blob.name, bytes)
        self.assertIsInstance(blob.name, six.text_type)
        self.assertEqual(blob.name, unicode_name)

    def test_ctor_w_encryption_key(self):
        KEY = b"01234567890123456789012345678901"  # 32 bytes
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=KEY)
        self.assertEqual(blob._encryption_key, KEY)
        self.assertEqual(blob.kms_key_name, None)

    def test_ctor_w_kms_key_name_and_encryption_key(self):
        KEY = b"01234567890123456789012345678901"  # 32 bytes
        KMS_RESOURCE = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        BLOB_NAME = "blob-name"
        bucket = _Bucket()

        with self.assertRaises(ValueError):
            self._make_one(
                BLOB_NAME, bucket=bucket, encryption_key=KEY, kms_key_name=KMS_RESOURCE
            )

    def test_ctor_w_kms_key_name(self):
        KMS_RESOURCE = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket, kms_key_name=KMS_RESOURCE)
        self.assertEqual(blob._encryption_key, None)
        self.assertEqual(blob.kms_key_name, KMS_RESOURCE)

    def test_ctor_with_generation(self):
        BLOB_NAME = "blob-name"
        GENERATION = 12345
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket, generation=GENERATION)
        self.assertEqual(blob.generation, GENERATION)

    def _set_properties_helper(self, kms_key_name=None):
        import datetime
        from google.cloud._helpers import UTC
        from google.cloud._helpers import _RFC3339_MICROS

        now = datetime.datetime.utcnow().replace(tzinfo=UTC)
        NOW = now.strftime(_RFC3339_MICROS)
        BLOB_NAME = "blob-name"
        GENERATION = 12345
        BLOB_ID = "name/{}/{}".format(BLOB_NAME, GENERATION)
        SELF_LINK = "http://example.com/self/"
        METAGENERATION = 23456
        SIZE = 12345
        MD5_HASH = "DEADBEEF"
        MEDIA_LINK = "http://example.com/media/"
        ENTITY = "project-owner-12345"
        ENTITY_ID = "23456"
        CRC32C = "FACE0DAC"
        COMPONENT_COUNT = 2
        ETAG = "ETAG"
        resource = {
            "id": BLOB_ID,
            "selfLink": SELF_LINK,
            "generation": GENERATION,
            "metageneration": METAGENERATION,
            "contentType": "text/plain",
            "timeCreated": NOW,
            "updated": NOW,
            "timeDeleted": NOW,
            "storageClass": "NEARLINE",
            "timeStorageClassUpdated": NOW,
            "size": SIZE,
            "md5Hash": MD5_HASH,
            "mediaLink": MEDIA_LINK,
            "contentEncoding": "gzip",
            "contentDisposition": "inline",
            "contentLanguage": "en-US",
            "cacheControl": "private",
            "metadata": {"foo": "Foo"},
            "owner": {"entity": ENTITY, "entityId": ENTITY_ID},
            "crc32c": CRC32C,
            "componentCount": COMPONENT_COUNT,
            "etag": ETAG,
        }

        if kms_key_name is not None:
            resource["kmsKeyName"] = kms_key_name

        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        blob._set_properties(resource)

        self.assertEqual(blob.id, BLOB_ID)
        self.assertEqual(blob.self_link, SELF_LINK)
        self.assertEqual(blob.generation, GENERATION)
        self.assertEqual(blob.metageneration, METAGENERATION)
        self.assertEqual(blob.content_type, "text/plain")
        self.assertEqual(blob.time_created, now)
        self.assertEqual(blob.updated, now)
        self.assertEqual(blob.time_deleted, now)
        self.assertEqual(blob.storage_class, "NEARLINE")
        self.assertEqual(blob.size, SIZE)
        self.assertEqual(blob.md5_hash, MD5_HASH)
        self.assertEqual(blob.media_link, MEDIA_LINK)
        self.assertEqual(blob.content_encoding, "gzip")
        self.assertEqual(blob.content_disposition, "inline")
        self.assertEqual(blob.content_language, "en-US")
        self.assertEqual(blob.cache_control, "private")
        self.assertEqual(blob.metadata, {"foo": "Foo"})
        self.assertEqual(blob.owner, {"entity": ENTITY, "entityId": ENTITY_ID})
        self.assertEqual(blob.crc32c, CRC32C)
        self.assertEqual(blob.component_count, COMPONENT_COUNT)
        self.assertEqual(blob.etag, ETAG)

        if kms_key_name is not None:
            self.assertEqual(blob.kms_key_name, kms_key_name)
        else:
            self.assertIsNone(blob.kms_key_name)

    def test__set_properties_wo_kms_key_name(self):
        self._set_properties_helper()

    def test__set_properties_w_kms_key_name(self):
        kms_resource = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        self._set_properties_helper(kms_key_name=kms_resource)

    def test_chunk_size_ctor(self):
        from google.cloud.storage.blob import Blob

        BLOB_NAME = "blob-name"
        BUCKET = object()
        chunk_size = 10 * Blob._CHUNK_SIZE_MULTIPLE
        blob = self._make_one(BLOB_NAME, bucket=BUCKET, chunk_size=chunk_size)
        self.assertEqual(blob._chunk_size, chunk_size)

    def test_chunk_size_getter(self):
        BLOB_NAME = "blob-name"
        BUCKET = object()
        blob = self._make_one(BLOB_NAME, bucket=BUCKET)
        self.assertIsNone(blob.chunk_size)
        VALUE = object()
        blob._chunk_size = VALUE
        self.assertIs(blob.chunk_size, VALUE)

    def test_chunk_size_setter(self):
        BLOB_NAME = "blob-name"
        BUCKET = object()
        blob = self._make_one(BLOB_NAME, bucket=BUCKET)
        self.assertIsNone(blob._chunk_size)
        blob._CHUNK_SIZE_MULTIPLE = 10
        blob.chunk_size = 20
        self.assertEqual(blob._chunk_size, 20)

    def test_chunk_size_setter_bad_value(self):
        BLOB_NAME = "blob-name"
        BUCKET = object()
        blob = self._make_one(BLOB_NAME, bucket=BUCKET)
        self.assertIsNone(blob._chunk_size)
        blob._CHUNK_SIZE_MULTIPLE = 10
        with self.assertRaises(ValueError):
            blob.chunk_size = 11

    def test_acl_property(self):
        from google.cloud.storage.acl import ObjectACL

        fake_bucket = _Bucket()
        blob = self._make_one(u"name", bucket=fake_bucket)
        acl = blob.acl
        self.assertIsInstance(acl, ObjectACL)
        self.assertIs(acl, blob._acl)

    def test_path_bad_bucket(self):
        fake_bucket = object()
        name = u"blob-name"
        blob = self._make_one(name, bucket=fake_bucket)
        self.assertRaises(AttributeError, getattr, blob, "path")

    def test_path_no_name(self):
        bucket = _Bucket()
        blob = self._make_one(u"", bucket=bucket)
        self.assertRaises(ValueError, getattr, blob, "path")

    def test_path_normal(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(blob.path, "/b/name/o/%s" % BLOB_NAME)

    def test_path_w_slash_in_name(self):
        BLOB_NAME = "parent/child"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(blob.path, "/b/name/o/parent%2Fchild")

    def test_path_with_non_ascii(self):
        blob_name = u"Caf\xe9"
        bucket = _Bucket()
        blob = self._make_one(blob_name, bucket=bucket)
        self.assertEqual(blob.path, "/b/name/o/Caf%C3%A9")

    def test_bucket_readonly_property(self):
        blob_name = "BLOB"
        bucket = _Bucket()
        other = _Bucket()
        blob = self._make_one(blob_name, bucket=bucket)
        with self.assertRaises(AttributeError):
            blob.bucket = other

    def test_client(self):
        blob_name = "BLOB"
        bucket = _Bucket()
        blob = self._make_one(blob_name, bucket=bucket)
        self.assertIs(blob.client, bucket.client)

    def test_user_project(self):
        user_project = "user-project-123"
        blob_name = "BLOB"
        bucket = _Bucket(user_project=user_project)
        blob = self._make_one(blob_name, bucket=bucket)
        self.assertEqual(blob.user_project, user_project)

    def test__encryption_headers_wo_encryption_key(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        expected = {}
        self.assertEqual(blob._encryption_headers(), expected)

    def test__encryption_headers_w_encryption_key(self):
        key = b"aa426195405adee2c8081bb9e7e74b19"
        header_key_value = "YWE0MjYxOTU0MDVhZGVlMmM4MDgxYmI5ZTdlNzRiMTk="
        header_key_hash_value = "V3Kwe46nKc3xLv96+iJ707YfZfFvlObta8TQcx2gpm0="
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=key)
        expected = {
            "X-Goog-Encryption-Algorithm": "AES256",
            "X-Goog-Encryption-Key": header_key_value,
            "X-Goog-Encryption-Key-Sha256": header_key_hash_value,
        }
        self.assertEqual(blob._encryption_headers(), expected)

    def test__query_params_default(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(blob._query_params, {})

    def test__query_params_w_user_project(self):
        user_project = "user-project-123"
        BLOB_NAME = "BLOB"
        bucket = _Bucket(user_project=user_project)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(blob._query_params, {"userProject": user_project})

    def test__query_params_w_generation(self):
        generation = 123456
        BLOB_NAME = "BLOB"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket, generation=generation)
        self.assertEqual(blob._query_params, {"generation": generation})

    def test_public_url(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(
            blob.public_url, "https://storage.googleapis.com/name/%s" % BLOB_NAME
        )

    def test_public_url_w_slash_in_name(self):
        BLOB_NAME = "parent/child"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(
            blob.public_url, "https://storage.googleapis.com/name/parent/child"
        )

    def test_public_url_w_tilde_in_name(self):
        BLOB_NAME = "foo~bar"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertEqual(blob.public_url, "https://storage.googleapis.com/name/foo~bar")

    def test_public_url_with_non_ascii(self):
        blob_name = u"winter \N{snowman}"
        bucket = _Bucket()
        blob = self._make_one(blob_name, bucket=bucket)
        expected_url = "https://storage.googleapis.com/name/winter%20%E2%98%83"
        self.assertEqual(blob.public_url, expected_url)

    def test_generate_signed_url_w_invalid_version(self):
        BLOB_NAME = "blob-name"
        EXPIRATION = "2014-10-16T20:34:37.000Z"
        connection = _Connection()
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        with self.assertRaises(ValueError):
            blob.generate_signed_url(EXPIRATION, version="nonesuch")

    def _generate_signed_url_helper(
        self,
        version=None,
        blob_name="blob-name",
        api_access_endpoint=None,
        method="GET",
        content_md5=None,
        content_type=None,
        response_type=None,
        response_disposition=None,
        generation=None,
        headers=None,
        query_parameters=None,
        credentials=None,
        expiration=None,
        encryption_key=None,
    ):
        from six.moves.urllib import parse
        from google.cloud._helpers import UTC
        from google.cloud.storage.blob import _API_ACCESS_ENDPOINT
        from google.cloud.storage.blob import _get_encryption_headers

        api_access_endpoint = api_access_endpoint or _API_ACCESS_ENDPOINT

        delta = datetime.timedelta(hours=1)

        if expiration is None:
            expiration = datetime.datetime.utcnow().replace(tzinfo=UTC) + delta

        connection = _Connection()
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket, encryption_key=encryption_key)

        if version is None:
            effective_version = "v2"
        else:
            effective_version = version

        to_patch = "google.cloud.storage.blob.generate_signed_url_{}".format(
            effective_version
        )

        with mock.patch(to_patch) as signer:
            signed_uri = blob.generate_signed_url(
                expiration=expiration,
                api_access_endpoint=api_access_endpoint,
                method=method,
                credentials=credentials,
                content_md5=content_md5,
                content_type=content_type,
                response_type=response_type,
                response_disposition=response_disposition,
                generation=generation,
                headers=headers,
                query_parameters=query_parameters,
                version=version,
            )

        self.assertEqual(signed_uri, signer.return_value)

        if credentials is None:
            expected_creds = _Connection.credentials
        else:
            expected_creds = credentials

        encoded_name = blob_name.encode("utf-8")
        expected_resource = "/name/{}".format(parse.quote(encoded_name, safe=b"/~"))
        if encryption_key is not None:
            expected_headers = headers or {}
            if effective_version == "v2":
                expected_headers["X-Goog-Encryption-Algorithm"] = "AES256"
            else:
                expected_headers.update(_get_encryption_headers(encryption_key))
        else:
            expected_headers = headers

        expected_kwargs = {
            "resource": expected_resource,
            "expiration": expiration,
            "api_access_endpoint": api_access_endpoint,
            "method": method.upper(),
            "content_md5": content_md5,
            "content_type": content_type,
            "response_type": response_type,
            "response_disposition": response_disposition,
            "generation": generation,
            "headers": expected_headers,
            "query_parameters": query_parameters,
        }
        signer.assert_called_once_with(expected_creds, **expected_kwargs)

    def test_generate_signed_url_no_version_passed_warning(self):
        self._generate_signed_url_helper()

    def _generate_signed_url_v2_helper(self, **kw):
        version = "v2"
        self._generate_signed_url_helper(version, **kw)

    def test_generate_signed_url_v2_w_defaults(self):
        self._generate_signed_url_v2_helper()

    def test_generate_signed_url_v2_w_expiration(self):
        from google.cloud._helpers import UTC

        expiration = datetime.datetime.utcnow().replace(tzinfo=UTC)
        self._generate_signed_url_v2_helper(expiration=expiration)

    def test_generate_signed_url_v2_w_non_ascii_name(self):
        BLOB_NAME = u"\u0410\u043a\u043a\u043e\u0440\u0434\u044b.txt"
        self._generate_signed_url_v2_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v2_w_slash_in_name(self):
        BLOB_NAME = "parent/child"
        self._generate_signed_url_v2_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v2_w_tilde_in_name(self):
        BLOB_NAME = "foo~bar"
        self._generate_signed_url_v2_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v2_w_endpoint(self):
        self._generate_signed_url_v2_helper(
            api_access_endpoint="https://api.example.com/v1"
        )

    def test_generate_signed_url_v2_w_method(self):
        self._generate_signed_url_v2_helper(method="POST")

    def test_generate_signed_url_v2_w_lowercase_method(self):
        self._generate_signed_url_v2_helper(method="get")

    def test_generate_signed_url_v2_w_content_md5(self):
        self._generate_signed_url_v2_helper(content_md5="FACEDACE")

    def test_generate_signed_url_v2_w_content_type(self):
        self._generate_signed_url_v2_helper(content_type="text.html")

    def test_generate_signed_url_v2_w_response_type(self):
        self._generate_signed_url_v2_helper(response_type="text.html")

    def test_generate_signed_url_v2_w_response_disposition(self):
        self._generate_signed_url_v2_helper(response_disposition="inline")

    def test_generate_signed_url_v2_w_generation(self):
        self._generate_signed_url_v2_helper(generation=12345)

    def test_generate_signed_url_v2_w_headers(self):
        self._generate_signed_url_v2_helper(headers={"x-goog-foo": "bar"})

    def test_generate_signed_url_v2_w_csek(self):
        self._generate_signed_url_v2_helper(encryption_key=os.urandom(32))

    def test_generate_signed_url_v2_w_csek_and_headers(self):
        self._generate_signed_url_v2_helper(
            encryption_key=os.urandom(32), headers={"x-goog-foo": "bar"}
        )

    def test_generate_signed_url_v2_w_credentials(self):
        credentials = object()
        self._generate_signed_url_v2_helper(credentials=credentials)

    def _generate_signed_url_v4_helper(self, **kw):
        version = "v4"
        self._generate_signed_url_helper(version, **kw)

    def test_generate_signed_url_v4_w_defaults(self):
        self._generate_signed_url_v4_helper()

    def test_generate_signed_url_v4_w_non_ascii_name(self):
        BLOB_NAME = u"\u0410\u043a\u043a\u043e\u0440\u0434\u044b.txt"
        self._generate_signed_url_v4_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v4_w_slash_in_name(self):
        BLOB_NAME = "parent/child"
        self._generate_signed_url_v4_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v4_w_tilde_in_name(self):
        BLOB_NAME = "foo~bar"
        self._generate_signed_url_v4_helper(blob_name=BLOB_NAME)

    def test_generate_signed_url_v4_w_endpoint(self):
        self._generate_signed_url_v4_helper(
            api_access_endpoint="https://api.example.com/v1"
        )

    def test_generate_signed_url_v4_w_method(self):
        self._generate_signed_url_v4_helper(method="POST")

    def test_generate_signed_url_v4_w_lowercase_method(self):
        self._generate_signed_url_v4_helper(method="get")

    def test_generate_signed_url_v4_w_content_md5(self):
        self._generate_signed_url_v4_helper(content_md5="FACEDACE")

    def test_generate_signed_url_v4_w_content_type(self):
        self._generate_signed_url_v4_helper(content_type="text.html")

    def test_generate_signed_url_v4_w_response_type(self):
        self._generate_signed_url_v4_helper(response_type="text.html")

    def test_generate_signed_url_v4_w_response_disposition(self):
        self._generate_signed_url_v4_helper(response_disposition="inline")

    def test_generate_signed_url_v4_w_generation(self):
        self._generate_signed_url_v4_helper(generation=12345)

    def test_generate_signed_url_v4_w_headers(self):
        self._generate_signed_url_v4_helper(headers={"x-goog-foo": "bar"})

    def test_generate_signed_url_v4_w_csek(self):
        self._generate_signed_url_v4_helper(encryption_key=os.urandom(32))

    def test_generate_signed_url_v4_w_csek_and_headers(self):
        self._generate_signed_url_v4_helper(
            encryption_key=os.urandom(32), headers={"x-goog-foo": "bar"}
        )

    def test_generate_signed_url_v4_w_credentials(self):
        credentials = object()
        self._generate_signed_url_v4_helper(credentials=credentials)

    def test_exists_miss(self):
        NONESUCH = "nonesuch"
        not_found_response = ({"status": http_client.NOT_FOUND}, b"")
        connection = _Connection(not_found_response)
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(NONESUCH, bucket=bucket)
        self.assertFalse(blob.exists())
        self.assertEqual(len(connection._requested), 1)
        self.assertEqual(
            connection._requested[0],
            {
                "method": "GET",
                "path": "/b/name/o/{}".format(NONESUCH),
                "query_params": {"fields": "name"},
                "_target_object": None,
            },
        )

    def test_exists_hit_w_user_project(self):
        BLOB_NAME = "blob-name"
        USER_PROJECT = "user-project-123"
        found_response = ({"status": http_client.OK}, b"")
        connection = _Connection(found_response)
        client = _Client(connection)
        bucket = _Bucket(client, user_project=USER_PROJECT)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        bucket._blobs[BLOB_NAME] = 1
        self.assertTrue(blob.exists())
        self.assertEqual(len(connection._requested), 1)
        self.assertEqual(
            connection._requested[0],
            {
                "method": "GET",
                "path": "/b/name/o/{}".format(BLOB_NAME),
                "query_params": {"fields": "name", "userProject": USER_PROJECT},
                "_target_object": None,
            },
        )

    def test_exists_hit_w_generation(self):
        BLOB_NAME = "blob-name"
        GENERATION = 123456
        found_response = ({"status": http_client.OK}, b"")
        connection = _Connection(found_response)
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(BLOB_NAME, bucket=bucket, generation=GENERATION)
        bucket._blobs[BLOB_NAME] = 1
        self.assertTrue(blob.exists())
        self.assertEqual(len(connection._requested), 1)
        self.assertEqual(
            connection._requested[0],
            {
                "method": "GET",
                "path": "/b/name/o/{}".format(BLOB_NAME),
                "query_params": {"fields": "name", "generation": GENERATION},
                "_target_object": None,
            },
        )

    def test_delete_wo_generation(self):
        BLOB_NAME = "blob-name"
        not_found_response = ({"status": http_client.NOT_FOUND}, b"")
        connection = _Connection(not_found_response)
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        bucket._blobs[BLOB_NAME] = 1
        blob.delete()
        self.assertFalse(blob.exists())
        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, None)])

    def test_delete_w_generation(self):
        BLOB_NAME = "blob-name"
        GENERATION = 123456
        not_found_response = ({"status": http_client.NOT_FOUND}, b"")
        connection = _Connection(not_found_response)
        client = _Client(connection)
        bucket = _Bucket(client)
        blob = self._make_one(BLOB_NAME, bucket=bucket, generation=GENERATION)
        bucket._blobs[BLOB_NAME] = 1
        blob.delete()
        self.assertFalse(blob.exists())
        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, GENERATION)])

    def test__get_transport(self):
        client = mock.Mock(spec=[u"_credentials", "_http"])
        client._http = mock.sentinel.transport
        blob = self._make_one(u"blob-name", bucket=None)

        transport = blob._get_transport(client)

        self.assertIs(transport, mock.sentinel.transport)

    def test__get_download_url_with_media_link(self):
        blob_name = "something.txt"
        bucket = _Bucket(name="IRRELEVANT")
        blob = self._make_one(blob_name, bucket=bucket)
        media_link = "http://test.invalid"
        # Set the media link on the blob
        blob._properties["mediaLink"] = media_link

        download_url = blob._get_download_url()
        self.assertEqual(download_url, media_link)

    def test__get_download_url_with_media_link_w_user_project(self):
        blob_name = "something.txt"
        user_project = "user-project-123"
        bucket = _Bucket(name="IRRELEVANT", user_project=user_project)
        blob = self._make_one(blob_name, bucket=bucket)
        media_link = "http://test.invalid"
        # Set the media link on the blob
        blob._properties["mediaLink"] = media_link

        download_url = blob._get_download_url()
        self.assertEqual(
            download_url, "{}?userProject={}".format(media_link, user_project)
        )

    def test__get_download_url_on_the_fly(self):
        blob_name = "bzzz-fly.txt"
        bucket = _Bucket(name="buhkit")
        blob = self._make_one(blob_name, bucket=bucket)

        self.assertIsNone(blob.media_link)
        download_url = blob._get_download_url()
        expected_url = (
            "https://www.googleapis.com/download/storage/v1/b/"
            "buhkit/o/bzzz-fly.txt?alt=media"
        )
        self.assertEqual(download_url, expected_url)

    def test__get_download_url_on_the_fly_with_generation(self):
        blob_name = "pretend.txt"
        bucket = _Bucket(name="fictional")
        blob = self._make_one(blob_name, bucket=bucket)
        generation = 1493058489532987
        # Set the media link on the blob
        blob._properties["generation"] = str(generation)

        self.assertIsNone(blob.media_link)
        download_url = blob._get_download_url()
        expected_url = (
            "https://www.googleapis.com/download/storage/v1/b/"
            "fictional/o/pretend.txt?alt=media&generation=1493058489532987"
        )
        self.assertEqual(download_url, expected_url)

    def test__get_download_url_on_the_fly_with_user_project(self):
        blob_name = "pretend.txt"
        user_project = "user-project-123"
        bucket = _Bucket(name="fictional", user_project=user_project)
        blob = self._make_one(blob_name, bucket=bucket)

        self.assertIsNone(blob.media_link)
        download_url = blob._get_download_url()
        expected_url = (
            "https://www.googleapis.com/download/storage/v1/b/"
            "fictional/o/pretend.txt?alt=media&userProject={}".format(user_project)
        )
        self.assertEqual(download_url, expected_url)

    def test__get_download_url_on_the_fly_with_kms_key_name(self):
        kms_resource = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        blob_name = "bzzz-fly.txt"
        bucket = _Bucket(name="buhkit")
        blob = self._make_one(blob_name, bucket=bucket, kms_key_name=kms_resource)

        self.assertIsNone(blob.media_link)
        download_url = blob._get_download_url()
        expected_url = (
            "https://www.googleapis.com/download/storage/v1/b/"
            "buhkit/o/bzzz-fly.txt?alt=media"
        )
        self.assertEqual(download_url, expected_url)

    @staticmethod
    def _mock_requests_response(status_code, headers, content=b"", stream=False):
        import requests

        response = requests.Response()
        response.status_code = status_code
        response.headers.update(headers)
        if stream:
            raw = io.BytesIO(content)
            raw.headers = headers
            response.raw = raw
            response._content = False
        else:
            response.raw = None
            response._content = content

        response.request = requests.Request("POST", "http://example.com").prepare()
        return response

    def _mock_download_transport(self):
        fake_transport = mock.Mock(spec=["request"])
        # Give the transport two fake responses.
        chunk1_response = self._mock_requests_response(
            http_client.PARTIAL_CONTENT,
            {"content-length": "3", "content-range": "bytes 0-2/6"},
            content=b"abc",
        )
        chunk2_response = self._mock_requests_response(
            http_client.PARTIAL_CONTENT,
            {"content-length": "3", "content-range": "bytes 3-5/6"},
            content=b"def",
        )
        fake_transport.request.side_effect = [chunk1_response, chunk2_response]
        return fake_transport

    def _mock_download_transport_range(self):
        fake_transport = mock.Mock(spec=["request"])
        # Give the transport two fake responses.
        chunk1_response = self._mock_requests_response(
            http_client.PARTIAL_CONTENT,
            {"content-length": "2", "content-range": "bytes 1-2/6"},
            content=b"bc",
        )
        chunk2_response = self._mock_requests_response(
            http_client.PARTIAL_CONTENT,
            {"content-length": "2", "content-range": "bytes 3-4/6"},
            content=b"de",
        )
        fake_transport.request.side_effect = [chunk1_response, chunk2_response]
        return fake_transport

    def _check_session_mocks(self, client, transport, expected_url, headers=None):
        # Check that the transport was called exactly twice.
        self.assertEqual(transport.request.call_count, 2)
        if headers is None:
            headers = {}
        # NOTE: bytes=0-2 never shows up because the mock was called with
        #       **MUTABLE** headers and it was mutated before the
        #       second request.
        headers["range"] = "bytes=3-5"
        headers["accept-encoding"] = "gzip"
        call = mock.call(
            "GET", expected_url, data=None, headers=headers, timeout=mock.ANY
        )
        self.assertEqual(transport.request.mock_calls, [call, call])

    def test__do_download_simple(self):
        blob_name = "blob-name"
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)

        # Make sure this will not be chunked.
        self.assertIsNone(blob.chunk_size)

        transport = mock.Mock(spec=["request"])
        transport.request.return_value = self._mock_requests_response(
            http_client.OK,
            {"content-length": "6", "content-range": "bytes 0-5/6"},
            content=b"abcdef",
            stream=True,
        )
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}
        blob._do_download(transport, file_obj, download_url, headers)
        # Make sure the download was as expected.
        self.assertEqual(file_obj.getvalue(), b"abcdef")

        transport.request.assert_called_once_with(
            "GET",
            download_url,
            data=None,
            headers=headers,
            stream=True,
            timeout=mock.ANY,
        )

    def test__do_download_simple_with_range(self):
        blob_name = "blob-name"
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)

        # Make sure this will not be chunked.
        self.assertIsNone(blob.chunk_size)

        transport = mock.Mock(spec=["request"])
        transport.request.return_value = self._mock_requests_response(
            http_client.OK,
            {"content-length": "3", "content-range": "bytes 1-3"},
            content=b"bcd",
            stream=True,
        )
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}
        blob._do_download(transport, file_obj, download_url, headers, start=1, end=3)
        # Make sure the download was as expected.
        self.assertEqual(file_obj.getvalue(), b"bcd")
        self.assertEqual(headers["range"], "bytes=1-3")

        transport.request.assert_called_once_with(
            "GET",
            download_url,
            data=None,
            headers=headers,
            stream=True,
            timeout=mock.ANY,
        )

    def test__do_download_chunked(self):
        blob_name = "blob-name"
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)

        # Modify the blob so there there will be 2 chunks of size 3.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 3

        transport = self._mock_download_transport()
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}
        blob._do_download(transport, file_obj, download_url, headers)
        # Make sure the download was as expected.
        self.assertEqual(file_obj.getvalue(), b"abcdef")

        # Check that the transport was called exactly twice.
        self.assertEqual(transport.request.call_count, 2)
        # ``headers`` was modified (in place) once for each API call.
        self.assertEqual(headers, {"range": "bytes=3-5"})
        call = mock.call(
            "GET", download_url, data=None, headers=headers, timeout=mock.ANY
        )
        self.assertEqual(transport.request.mock_calls, [call, call])

    def test__do_download_chunked_with_range(self):
        blob_name = "blob-name"
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)

        # Modify the blob so there there will be 2 chunks of size 2.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 2

        transport = self._mock_download_transport_range()
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}
        blob._do_download(transport, file_obj, download_url, headers, start=1, end=4)
        # Make sure the download was as expected.
        self.assertEqual(file_obj.getvalue(), b"bcde")

        # Check that the transport was called exactly twice.
        self.assertEqual(transport.request.call_count, 2)
        # ``headers`` was modified (in place) once for each API call.
        self.assertEqual(headers, {"range": "bytes=3-4"})
        call = mock.call(
            "GET", download_url, data=None, headers=headers, timeout=mock.ANY
        )
        self.assertEqual(transport.request.mock_calls, [call, call])

    def test_download_to_file_with_failure(self):
        from google.cloud import exceptions

        blob_name = "blob-name"
        transport = mock.Mock(spec=["request"])
        bad_response_headers = {
            "Content-Length": "9",
            "Content-Type": "text/html; charset=UTF-8",
        }
        transport.request.return_value = self._mock_requests_response(
            http_client.NOT_FOUND, bad_response_headers, content=b"Not found"
        )
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=[u"_http"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        # Set the media link on the blob
        blob._properties["mediaLink"] = "http://test.invalid"

        file_obj = io.BytesIO()
        with self.assertRaises(exceptions.NotFound):
            blob.download_to_file(file_obj)

        self.assertEqual(file_obj.tell(), 0)
        # Check that the transport was called once.
        transport.request.assert_called_once_with(
            "GET",
            blob.media_link,
            data=None,
            headers={"accept-encoding": "gzip"},
            stream=True,
            timeout=mock.ANY,
        )

    def test_download_to_file_wo_media_link(self):
        blob_name = "blob-name"
        transport = self._mock_download_transport()
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=[u"_http"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        # Modify the blob so there there will be 2 chunks of size 3.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 3

        file_obj = io.BytesIO()
        blob.download_to_file(file_obj)
        self.assertEqual(file_obj.getvalue(), b"abcdef")
        # Make sure the media link is still unknown.
        self.assertIsNone(blob.media_link)

        expected_url = (
            "https://www.googleapis.com/download/storage/v1/b/"
            "name/o/blob-name?alt=media"
        )
        self._check_session_mocks(client, transport, expected_url)

    def _download_to_file_helper(self, use_chunks=False):
        blob_name = "blob-name"
        transport = self._mock_download_transport()
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=[u"_http"])
        bucket = _Bucket(client)
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link}
        blob = self._make_one(blob_name, bucket=bucket, properties=properties)
        if use_chunks:
            # Modify the blob so there there will be 2 chunks of size 3.
            blob._CHUNK_SIZE_MULTIPLE = 1
            blob.chunk_size = 3
        else:
            # Modify the response.
            single_chunk_response = self._mock_requests_response(
                http_client.OK,
                {"content-length": "6", "content-range": "bytes 0-5/6"},
                content=b"abcdef",
                stream=True,
            )
            transport.request.side_effect = [single_chunk_response]

        file_obj = io.BytesIO()
        blob.download_to_file(file_obj)
        self.assertEqual(file_obj.getvalue(), b"abcdef")

        if use_chunks:
            self._check_session_mocks(client, transport, media_link)
        else:
            transport.request.assert_called_once_with(
                "GET",
                media_link,
                data=None,
                headers={"accept-encoding": "gzip"},
                stream=True,
                timeout=mock.ANY,
            )

    def test_download_to_file_default(self):
        self._download_to_file_helper()

    def test_download_to_file_with_chunk_size(self):
        self._download_to_file_helper(use_chunks=True)

    def _download_to_filename_helper(self, updated=None):
        import os
        import time
        from google.cloud._testing import _NamedTemporaryFile

        blob_name = "blob-name"
        transport = self._mock_download_transport()
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=["_http"])
        bucket = _Bucket(client)
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link}
        if updated is not None:
            properties["updated"] = updated

        blob = self._make_one(blob_name, bucket=bucket, properties=properties)
        # Modify the blob so there there will be 2 chunks of size 3.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 3

        with _NamedTemporaryFile() as temp:
            blob.download_to_filename(temp.name)
            with open(temp.name, "rb") as file_obj:
                wrote = file_obj.read()
                if updated is None:
                    self.assertIsNone(blob.updated)
                else:
                    mtime = os.path.getmtime(temp.name)
                    updated_time = time.mktime(blob.updated.timetuple())
                    self.assertEqual(mtime, updated_time)

        self.assertEqual(wrote, b"abcdef")

        self._check_session_mocks(client, transport, media_link)

    def test_download_to_filename(self):
        updated = "2014-12-06T13:13:50.690Z"
        self._download_to_filename_helper(updated=updated)

    def test_download_to_filename_wo_updated(self):
        self._download_to_filename_helper()

    def test_download_to_filename_corrupted(self):
        from google.resumable_media import DataCorruption
        from google.resumable_media.requests.download import _CHECKSUM_MISMATCH

        blob_name = "blob-name"
        transport = mock.Mock(spec=["request"])
        empty_hash = base64.b64encode(hashlib.md5(b"").digest()).decode(u"utf-8")
        headers = {"x-goog-hash": "md5=" + empty_hash}
        mock_raw = mock.Mock(headers=headers, spec=["headers"])
        response = mock.MagicMock(
            headers=headers,
            status_code=http_client.OK,
            raw=mock_raw,
            spec=[
                "__enter__",
                "__exit__",
                "headers",
                "iter_content",
                "status_code",
                "raw",
            ],
        )
        # i.e. context manager returns ``self``.
        response.__enter__.return_value = response
        response.__exit__.return_value = None
        chunks = (b"noms1", b"coooookies2")
        response.iter_content.return_value = iter(chunks)

        transport.request.return_value = response
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=["_http"])
        bucket = mock.Mock(
            client=client, user_project=None, spec=["client", "user_project"]
        )
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link}
        blob = self._make_one(blob_name, bucket=bucket, properties=properties)
        # Make sure the download is **not** chunked.
        self.assertIsNone(blob.chunk_size)

        # Make sure the hash will be wrong.
        content = b"".join(chunks)
        expected_hash = base64.b64encode(hashlib.md5(content).digest()).decode(u"utf-8")
        self.assertNotEqual(empty_hash, expected_hash)

        # Try to download into a temporary file (don't use
        # `_NamedTemporaryFile` it will try to remove after the file is
        # already removed)
        filehandle, filename = tempfile.mkstemp()
        os.close(filehandle)
        with self.assertRaises(DataCorruption) as exc_info:
            blob.download_to_filename(filename)

        msg = _CHECKSUM_MISMATCH.format(media_link, empty_hash, expected_hash)
        self.assertEqual(exc_info.exception.args, (msg,))
        # Make sure the file was cleaned up.
        self.assertFalse(os.path.exists(filename))

        # Check the mocks.
        response.__enter__.assert_called_once_with()
        response.__exit__.assert_called_once_with(None, None, None)
        response.iter_content.assert_called_once_with(
            chunk_size=8192, decode_unicode=False
        )
        transport.request.assert_called_once_with(
            "GET",
            media_link,
            data=None,
            headers={"accept-encoding": "gzip"},
            stream=True,
            timeout=mock.ANY,
        )

    def test_download_to_filename_w_key(self):
        import os
        import time
        from google.cloud._testing import _NamedTemporaryFile

        blob_name = "blob-name"
        transport = self._mock_download_transport()
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=["_http"])
        bucket = _Bucket(client)
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link, "updated": "2014-12-06T13:13:50.690Z"}
        key = b"aa426195405adee2c8081bb9e7e74b19"
        blob = self._make_one(
            blob_name, bucket=bucket, properties=properties, encryption_key=key
        )
        # Modify the blob so there there will be 2 chunks of size 3.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 3

        with _NamedTemporaryFile() as temp:
            blob.download_to_filename(temp.name)
            with open(temp.name, "rb") as file_obj:
                wrote = file_obj.read()
                mtime = os.path.getmtime(temp.name)
                updated_time = time.mktime(blob.updated.timetuple())

        self.assertEqual(wrote, b"abcdef")
        self.assertEqual(mtime, updated_time)

        header_key_value = "YWE0MjYxOTU0MDVhZGVlMmM4MDgxYmI5ZTdlNzRiMTk="
        header_key_hash_value = "V3Kwe46nKc3xLv96+iJ707YfZfFvlObta8TQcx2gpm0="
        key_headers = {
            "X-Goog-Encryption-Key-Sha256": header_key_hash_value,
            "X-Goog-Encryption-Algorithm": "AES256",
            "X-Goog-Encryption-Key": header_key_value,
            "accept-encoding": "gzip",
        }
        self._check_session_mocks(client, transport, media_link, headers=key_headers)

    def test_download_as_string(self):
        blob_name = "blob-name"
        transport = self._mock_download_transport()
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(_http=transport, spec=["_http"])
        bucket = _Bucket(client)
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link}
        blob = self._make_one(blob_name, bucket=bucket, properties=properties)
        # Modify the blob so there there will be 2 chunks of size 3.
        blob._CHUNK_SIZE_MULTIPLE = 1
        blob.chunk_size = 3

        fetched = blob.download_as_string()
        self.assertEqual(fetched, b"abcdef")

        self._check_session_mocks(client, transport, media_link)

    def test__get_content_type_explicit(self):
        blob = self._make_one(u"blob-name", bucket=None)

        content_type = u"text/plain"
        return_value = blob._get_content_type(content_type)
        self.assertEqual(return_value, content_type)

    def test__get_content_type_from_blob(self):
        blob = self._make_one(u"blob-name", bucket=None)
        blob.content_type = u"video/mp4"

        return_value = blob._get_content_type(None)
        self.assertEqual(return_value, blob.content_type)

    def test__get_content_type_from_filename(self):
        blob = self._make_one(u"blob-name", bucket=None)

        return_value = blob._get_content_type(None, filename="archive.tar")
        self.assertEqual(return_value, "application/x-tar")

    def test__get_content_type_default(self):
        blob = self._make_one(u"blob-name", bucket=None)

        return_value = blob._get_content_type(None)
        self.assertEqual(return_value, u"application/octet-stream")

    def test__get_writable_metadata_no_changes(self):
        name = u"blob-name"
        blob = self._make_one(name, bucket=None)

        object_metadata = blob._get_writable_metadata()
        expected = {"name": name}
        self.assertEqual(object_metadata, expected)

    def test__get_writable_metadata_with_changes(self):
        name = u"blob-name"
        blob = self._make_one(name, bucket=None)
        blob.storage_class = "NEARLINE"
        blob.cache_control = "max-age=3600"
        blob.metadata = {"color": "red"}

        object_metadata = blob._get_writable_metadata()
        expected = {
            "cacheControl": blob.cache_control,
            "metadata": blob.metadata,
            "name": name,
            "storageClass": blob.storage_class,
        }
        self.assertEqual(object_metadata, expected)

    def test__get_writable_metadata_unwritable_field(self):
        name = u"blob-name"
        properties = {"updated": "2016-10-16T18:18:18.181Z"}
        blob = self._make_one(name, bucket=None, properties=properties)
        # Fake that `updated` is in changes.
        blob._changes.add("updated")

        object_metadata = blob._get_writable_metadata()
        expected = {"name": name}
        self.assertEqual(object_metadata, expected)

    def test__get_upload_arguments(self):
        name = u"blob-name"
        key = b"[pXw@,p@@AfBfrR3x-2b2SCHR,.?YwRO"
        blob = self._make_one(name, bucket=None, encryption_key=key)
        blob.content_disposition = "inline"

        content_type = u"image/jpeg"
        info = blob._get_upload_arguments(content_type)

        headers, object_metadata, new_content_type = info
        header_key_value = "W3BYd0AscEBAQWZCZnJSM3gtMmIyU0NIUiwuP1l3Uk8="
        header_key_hash_value = "G0++dxF4q5rG4o9kE8gvEKn15RH6wLm0wXV1MgAlXOg="
        expected_headers = {
            "X-Goog-Encryption-Algorithm": "AES256",
            "X-Goog-Encryption-Key": header_key_value,
            "X-Goog-Encryption-Key-Sha256": header_key_hash_value,
        }
        self.assertEqual(headers, expected_headers)
        expected_metadata = {
            "contentDisposition": blob.content_disposition,
            "name": name,
        }
        self.assertEqual(object_metadata, expected_metadata)
        self.assertEqual(new_content_type, content_type)

    def _mock_transport(self, status_code, headers, content=b""):
        fake_transport = mock.Mock(spec=["request"])
        fake_response = self._mock_requests_response(
            status_code, headers, content=content
        )
        fake_transport.request.return_value = fake_response
        return fake_transport

    def _do_multipart_success(
        self,
        mock_get_boundary,
        size=None,
        num_retries=None,
        user_project=None,
        predefined_acl=None,
        kms_key_name=None,
    ):
        from six.moves.urllib.parse import urlencode

        bucket = _Bucket(name="w00t", user_project=user_project)
        blob = self._make_one(u"blob-name", bucket=bucket, kms_key_name=kms_key_name)
        self.assertIsNone(blob.chunk_size)

        # Create mocks to be checked for doing transport.
        transport = self._mock_transport(http_client.OK, {})

        # Create some mock arguments.
        client = mock.Mock(_http=transport, spec=["_http"])
        data = b"data here hear hier"
        stream = io.BytesIO(data)
        content_type = u"application/xml"
        response = blob._do_multipart_upload(
            client, stream, content_type, size, num_retries, predefined_acl
        )

        # Check the mocks and the returned value.
        self.assertIs(response, transport.request.return_value)
        if size is None:
            data_read = data
            self.assertEqual(stream.tell(), len(data))
        else:
            data_read = data[:size]
            self.assertEqual(stream.tell(), size)

        mock_get_boundary.assert_called_once_with()

        upload_url = "https://www.googleapis.com/upload/storage/v1" + bucket.path + "/o"

        qs_params = [("uploadType", "multipart")]

        if user_project is not None:
            qs_params.append(("userProject", user_project))

        if predefined_acl is not None:
            qs_params.append(("predefinedAcl", predefined_acl))

        if kms_key_name is not None:
            qs_params.append(("kmsKeyName", kms_key_name))

        upload_url += "?" + urlencode(qs_params)

        payload = (
            b"--==0==\r\n"
            + b"content-type: application/json; charset=UTF-8\r\n\r\n"
            + b'{"name": "blob-name"}\r\n'
            + b"--==0==\r\n"
            + b"content-type: application/xml\r\n\r\n"
            + data_read
            + b"\r\n--==0==--"
        )
        headers = {"content-type": b'multipart/related; boundary="==0=="'}
        transport.request.assert_called_once_with(
            "POST", upload_url, data=payload, headers=headers, timeout=mock.ANY
        )

    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
    def test__do_multipart_upload_no_size(self, mock_get_boundary):
        self._do_multipart_success(mock_get_boundary, predefined_acl="private")

    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
    def test__do_multipart_upload_with_size(self, mock_get_boundary):
        self._do_multipart_success(mock_get_boundary, size=10)

    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
    def test__do_multipart_upload_with_user_project(self, mock_get_boundary):
        user_project = "user-project-123"
        self._do_multipart_success(mock_get_boundary, user_project=user_project)

    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
    def test__do_multipart_upload_with_kms(self, mock_get_boundary):
        kms_resource = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        self._do_multipart_success(mock_get_boundary, kms_key_name=kms_resource)

    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
    def test__do_multipart_upload_with_retry(self, mock_get_boundary):
        self._do_multipart_success(mock_get_boundary, num_retries=8)

    def test__do_multipart_upload_bad_size(self):
        blob = self._make_one(u"blob-name", bucket=None)

        data = b"data here hear hier"
        stream = io.BytesIO(data)
        size = 50
        self.assertGreater(size, len(data))

        with self.assertRaises(ValueError) as exc_info:
            blob._do_multipart_upload(None, stream, None, size, None, None)

        exc_contents = str(exc_info.exception)
        self.assertIn("was specified but the file-like object only had", exc_contents)
        self.assertEqual(stream.tell(), len(data))

    def _initiate_resumable_helper(
        self,
        size=None,
        extra_headers=None,
        chunk_size=None,
        num_retries=None,
        user_project=None,
        predefined_acl=None,
        blob_chunk_size=786432,
        kms_key_name=None,
    ):
        from six.moves.urllib.parse import urlencode
        from google.resumable_media.requests import ResumableUpload

        bucket = _Bucket(name="whammy", user_project=user_project)
        blob = self._make_one(u"blob-name", bucket=bucket, kms_key_name=kms_key_name)
        blob.metadata = {"rook": "takes knight"}
        blob.chunk_size = blob_chunk_size
        if blob_chunk_size is not None:
            self.assertIsNotNone(blob.chunk_size)
        else:
            self.assertIsNone(blob.chunk_size)

        # Need to make sure **same** dict is used because ``json.dumps()``
        # will depend on the hash order.
        object_metadata = blob._get_writable_metadata()
        blob._get_writable_metadata = mock.Mock(return_value=object_metadata, spec=[])

        # Create mocks to be checked for doing transport.
        resumable_url = "http://test.invalid?upload_id=hey-you"
        response_headers = {"location": resumable_url}
        transport = self._mock_transport(http_client.OK, response_headers)

        # Create some mock arguments and call the method under test.
        client = mock.Mock(_http=transport, spec=[u"_http"])
        data = b"hello hallo halo hi-low"
        stream = io.BytesIO(data)
        content_type = u"text/plain"
        upload, transport = blob._initiate_resumable_upload(
            client,
            stream,
            content_type,
            size,
            num_retries,
            extra_headers=extra_headers,
            chunk_size=chunk_size,
            predefined_acl=predefined_acl,
        )

        # Check the returned values.
        self.assertIsInstance(upload, ResumableUpload)

        upload_url = "https://www.googleapis.com/upload/storage/v1" + bucket.path + "/o"
        qs_params = [("uploadType", "resumable")]

        if user_project is not None:
            qs_params.append(("userProject", user_project))

        if predefined_acl is not None:
            qs_params.append(("predefinedAcl", predefined_acl))

        if kms_key_name is not None:
            qs_params.append(("kmsKeyName", kms_key_name))

        upload_url += "?" + urlencode(qs_params)

        self.assertEqual(upload.upload_url, upload_url)
        if extra_headers is None:
            self.assertEqual(upload._headers, {})
        else:
            self.assertEqual(upload._headers, extra_headers)
            self.assertIsNot(upload._headers, extra_headers)
        self.assertFalse(upload.finished)
        if chunk_size is None:
            if blob_chunk_size is None:
                self.assertEqual(
                    upload._chunk_size, google.cloud.storage.blob._DEFAULT_CHUNKSIZE
                )
            else:
                self.assertEqual(upload._chunk_size, blob.chunk_size)
        else:
            self.assertNotEqual(blob.chunk_size, chunk_size)
            self.assertEqual(upload._chunk_size, chunk_size)
        self.assertIs(upload._stream, stream)
        if size is None:
            self.assertIsNone(upload._total_bytes)
        else:
            self.assertEqual(upload._total_bytes, size)
        self.assertEqual(upload._content_type, content_type)
        self.assertEqual(upload.resumable_url, resumable_url)
        retry_strategy = upload._retry_strategy
        self.assertEqual(retry_strategy.max_sleep, 64.0)
        if num_retries is None:
            self.assertEqual(retry_strategy.max_cumulative_retry, 600.0)
            self.assertIsNone(retry_strategy.max_retries)
        else:
            self.assertIsNone(retry_strategy.max_cumulative_retry)
            self.assertEqual(retry_strategy.max_retries, num_retries)
        self.assertIs(transport, transport)
        # Make sure we never read from the stream.
        self.assertEqual(stream.tell(), 0)

        # Check the mocks.
        blob._get_writable_metadata.assert_called_once_with()
        payload = json.dumps(object_metadata).encode("utf-8")
        expected_headers = {
            "content-type": "application/json; charset=UTF-8",
            "x-upload-content-type": content_type,
        }
        if size is not None:
            expected_headers["x-upload-content-length"] = str(size)
        if extra_headers is not None:
            expected_headers.update(extra_headers)
        transport.request.assert_called_once_with(
            "POST", upload_url, data=payload, headers=expected_headers, timeout=mock.ANY
        )

    def test__initiate_resumable_upload_no_size(self):
        self._initiate_resumable_helper()

    def test__initiate_resumable_upload_with_size(self):
        self._initiate_resumable_helper(size=10000)

    def test__initiate_resumable_upload_with_user_project(self):
        user_project = "user-project-123"
        self._initiate_resumable_helper(user_project=user_project)

    def test__initiate_resumable_upload_with_kms(self):
        kms_resource = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        self._initiate_resumable_helper(kms_key_name=kms_resource)

    def test__initiate_resumable_upload_without_chunk_size(self):
        self._initiate_resumable_helper(blob_chunk_size=None)

    def test__initiate_resumable_upload_with_chunk_size(self):
        one_mb = 1048576
        self._initiate_resumable_helper(chunk_size=one_mb)

    def test__initiate_resumable_upload_with_extra_headers(self):
        extra_headers = {"origin": "http://not-in-kansas-anymore.invalid"}
        self._initiate_resumable_helper(extra_headers=extra_headers)

    def test__initiate_resumable_upload_with_retry(self):
        self._initiate_resumable_helper(num_retries=11)

    def test__initiate_resumable_upload_with_predefined_acl(self):
        self._initiate_resumable_helper(predefined_acl="private")

    def _make_resumable_transport(self, headers1, headers2, headers3, total_bytes):
        from google import resumable_media

        fake_transport = mock.Mock(spec=["request"])

        fake_response1 = self._mock_requests_response(http_client.OK, headers1)
        fake_response2 = self._mock_requests_response(
            resumable_media.PERMANENT_REDIRECT, headers2
        )
        json_body = '{{"size": "{:d}"}}'.format(total_bytes)
        fake_response3 = self._mock_requests_response(
            http_client.OK, headers3, content=json_body.encode("utf-8")
        )

        responses = [fake_response1, fake_response2, fake_response3]
        fake_transport.request.side_effect = responses
        return fake_transport, responses

    @staticmethod
    def _do_resumable_upload_call0(blob, content_type, size=None, predefined_acl=None):
        # First mock transport.request() does initiates upload.
        upload_url = (
            "https://www.googleapis.com/upload/storage/v1"
            + blob.bucket.path
            + "/o?uploadType=resumable"
        )
        if predefined_acl is not None:
            upload_url += "&predefinedAcl={}".format(predefined_acl)
        expected_headers = {
            "content-type": "application/json; charset=UTF-8",
            "x-upload-content-type": content_type,
        }
        if size is not None:
            expected_headers["x-upload-content-length"] = str(size)
        payload = json.dumps({"name": blob.name}).encode("utf-8")
        return mock.call(
            "POST", upload_url, data=payload, headers=expected_headers, timeout=mock.ANY
        )

    @staticmethod
    def _do_resumable_upload_call1(
        blob, content_type, data, resumable_url, size=None, predefined_acl=None
    ):
        # Second mock transport.request() does sends first chunk.
        if size is None:
            content_range = "bytes 0-{:d}/*".format(blob.chunk_size - 1)
        else:
            content_range = "bytes 0-{:d}/{:d}".format(blob.chunk_size - 1, size)

        expected_headers = {
            "content-type": content_type,
            "content-range": content_range,
        }
        payload = data[: blob.chunk_size]
        return mock.call(
            "PUT",
            resumable_url,
            data=payload,
            headers=expected_headers,
            timeout=mock.ANY,
        )

    @staticmethod
    def _do_resumable_upload_call2(
        blob, content_type, data, resumable_url, total_bytes, predefined_acl=None
    ):
        # Third mock transport.request() does sends last chunk.
        content_range = "bytes {:d}-{:d}/{:d}".format(
            blob.chunk_size, total_bytes - 1, total_bytes
        )
        expected_headers = {
            "content-type": content_type,
            "content-range": content_range,
        }
        payload = data[blob.chunk_size :]
        return mock.call(
            "PUT",
            resumable_url,
            data=payload,
            headers=expected_headers,
            timeout=mock.ANY,
        )

    def _do_resumable_helper(
        self, use_size=False, num_retries=None, predefined_acl=None
    ):
        bucket = _Bucket(name="yesterday")
        blob = self._make_one(u"blob-name", bucket=bucket)
        blob.chunk_size = blob._CHUNK_SIZE_MULTIPLE
        self.assertIsNotNone(blob.chunk_size)

        # Data to be uploaded.
        data = b"<html>" + (b"A" * blob.chunk_size) + b"</html>"
        total_bytes = len(data)
        if use_size:
            size = total_bytes
        else:
            size = None

        # Create mocks to be checked for doing transport.
        resumable_url = "http://test.invalid?upload_id=and-then-there-was-1"
        headers1 = {"location": resumable_url}
        headers2 = {"range": "bytes=0-{:d}".format(blob.chunk_size - 1)}
        transport, responses = self._make_resumable_transport(
            headers1, headers2, {}, total_bytes
        )

        # Create some mock arguments and call the method under test.
        client = mock.Mock(_http=transport, spec=["_http"])
        stream = io.BytesIO(data)
        content_type = u"text/html"
        response = blob._do_resumable_upload(
            client, stream, content_type, size, num_retries, predefined_acl
        )

        # Check the returned values.
        self.assertIs(response, responses[2])
        self.assertEqual(stream.tell(), total_bytes)

        # Check the mocks.
        call0 = self._do_resumable_upload_call0(
            blob, content_type, size=size, predefined_acl=predefined_acl
        )
        call1 = self._do_resumable_upload_call1(
            blob,
            content_type,
            data,
            resumable_url,
            size=size,
            predefined_acl=predefined_acl,
        )
        call2 = self._do_resumable_upload_call2(
            blob,
            content_type,
            data,
            resumable_url,
            total_bytes,
            predefined_acl=predefined_acl,
        )
        self.assertEqual(transport.request.mock_calls, [call0, call1, call2])

    def test__do_resumable_upload_no_size(self):
        self._do_resumable_helper()

    def test__do_resumable_upload_with_size(self):
        self._do_resumable_helper(use_size=True)

    def test__do_resumable_upload_with_retry(self):
        self._do_resumable_helper(num_retries=6)

    def test__do_resumable_upload_with_predefined_acl(self):
        self._do_resumable_helper(predefined_acl="private")

    def _do_upload_helper(
        self, chunk_size=None, num_retries=None, predefined_acl=None, size=None
    ):
        blob = self._make_one(u"blob-name", bucket=None)

        # Create a fake response.
        response = mock.Mock(spec=[u"json"])
        response.json.return_value = mock.sentinel.json
        # Mock **both** helpers.
        blob._do_multipart_upload = mock.Mock(return_value=response, spec=[])
        blob._do_resumable_upload = mock.Mock(return_value=response, spec=[])

        if chunk_size is None:
            self.assertIsNone(blob.chunk_size)
        else:
            blob.chunk_size = chunk_size
            self.assertIsNotNone(blob.chunk_size)

        client = mock.sentinel.client
        stream = mock.sentinel.stream
        content_type = u"video/mp4"
        if size is None:
            size = 12345654321
        # Make the request and check the mocks.
        created_json = blob._do_upload(
            client, stream, content_type, size, num_retries, predefined_acl
        )
        self.assertIs(created_json, mock.sentinel.json)
        response.json.assert_called_once_with()
        if size is not None and size <= google.cloud.storage.blob._MAX_MULTIPART_SIZE:
            blob._do_multipart_upload.assert_called_once_with(
                client, stream, content_type, size, num_retries, predefined_acl
            )
            blob._do_resumable_upload.assert_not_called()
        else:
            blob._do_multipart_upload.assert_not_called()
            blob._do_resumable_upload.assert_called_once_with(
                client, stream, content_type, size, num_retries, predefined_acl
            )

    def test__do_upload_uses_multipart(self):
        self._do_upload_helper(size=google.cloud.storage.blob._MAX_MULTIPART_SIZE)

    def test__do_upload_uses_resumable(self):
        self._do_upload_helper(
            chunk_size=256 * 1024,  # 256KB
            size=google.cloud.storage.blob._MAX_MULTIPART_SIZE + 1,
        )

    def test__do_upload_with_retry(self):
        self._do_upload_helper(num_retries=20)

    def _upload_from_file_helper(self, side_effect=None, **kwargs):
        from google.cloud._helpers import UTC

        blob = self._make_one("blob-name", bucket=None)
        # Mock low-level upload helper on blob (it is tested elsewhere).
        created_json = {"updated": "2017-01-01T09:09:09.081Z"}
        blob._do_upload = mock.Mock(return_value=created_json, spec=[])
        if side_effect is not None:
            blob._do_upload.side_effect = side_effect
        # Make sure `updated` is empty before the request.
        self.assertIsNone(blob.updated)

        data = b"data is here"
        stream = io.BytesIO(data)
        stream.seek(2)  # Not at zero.
        content_type = u"font/woff"
        client = mock.sentinel.client
        predefined_acl = kwargs.get("predefined_acl", None)
        ret_val = blob.upload_from_file(
            stream, size=len(data), content_type=content_type, client=client, **kwargs
        )

        # Check the response and side-effects.
        self.assertIsNone(ret_val)
        new_updated = datetime.datetime(2017, 1, 1, 9, 9, 9, 81000, tzinfo=UTC)
        self.assertEqual(blob.updated, new_updated)

        # Check the mock.
        num_retries = kwargs.get("num_retries")
        blob._do_upload.assert_called_once_with(
            client, stream, content_type, len(data), num_retries, predefined_acl
        )
        return stream

    def test_upload_from_file_success(self):
        stream = self._upload_from_file_helper(predefined_acl="private")
        assert stream.tell() == 2

    @mock.patch("warnings.warn")
    def test_upload_from_file_with_retries(self, mock_warn):
        from google.cloud.storage import blob as blob_module

        self._upload_from_file_helper(num_retries=20)
        mock_warn.assert_called_once_with(
            blob_module._NUM_RETRIES_MESSAGE, DeprecationWarning, stacklevel=2
        )

    def test_upload_from_file_with_rewind(self):
        stream = self._upload_from_file_helper(rewind=True)
        assert stream.tell() == 0

    def test_upload_from_file_failure(self):
        import requests

        from google.resumable_media import InvalidResponse
        from google.cloud import exceptions

        message = "Someone is already in this spot."
        response = requests.Response()
        response.status_code = http_client.CONFLICT
        response.request = requests.Request("POST", "http://example.com").prepare()
        side_effect = InvalidResponse(response, message)

        with self.assertRaises(exceptions.Conflict) as exc_info:
            self._upload_from_file_helper(side_effect=side_effect)

        self.assertIn(message, exc_info.exception.message)
        self.assertEqual(exc_info.exception.errors, [])

    def _do_upload_mock_call_helper(self, blob, client, content_type, size):
        self.assertEqual(blob._do_upload.call_count, 1)
        mock_call = blob._do_upload.mock_calls[0]
        call_name, pos_args, kwargs = mock_call
        self.assertEqual(call_name, "")
        self.assertEqual(len(pos_args), 6)
        self.assertEqual(pos_args[0], client)
        self.assertEqual(pos_args[2], content_type)
        self.assertEqual(pos_args[3], size)
        self.assertIsNone(pos_args[4])  # num_retries
        self.assertIsNone(pos_args[5])  # predefined_acl
        self.assertEqual(kwargs, {})

        return pos_args[1]

    def test_upload_from_filename(self):
        from google.cloud._testing import _NamedTemporaryFile

        blob = self._make_one("blob-name", bucket=None)
        # Mock low-level upload helper on blob (it is tested elsewhere).
        created_json = {"metadata": {"mint": "ice-cream"}}
        blob._do_upload = mock.Mock(return_value=created_json, spec=[])
        # Make sure `metadata` is empty before the request.
        self.assertIsNone(blob.metadata)

        data = b"soooo much data"
        content_type = u"image/svg+xml"
        client = mock.sentinel.client
        with _NamedTemporaryFile() as temp:
            with open(temp.name, "wb") as file_obj:
                file_obj.write(data)

            ret_val = blob.upload_from_filename(
                temp.name, content_type=content_type, client=client
            )

        # Check the response and side-effects.
        self.assertIsNone(ret_val)
        self.assertEqual(blob.metadata, created_json["metadata"])

        # Check the mock.
        stream = self._do_upload_mock_call_helper(blob, client, content_type, len(data))
        self.assertTrue(stream.closed)
        self.assertEqual(stream.mode, "rb")
        self.assertEqual(stream.name, temp.name)

    def _upload_from_string_helper(self, data, **kwargs):
        from google.cloud._helpers import _to_bytes

        blob = self._make_one("blob-name", bucket=None)

        # Mock low-level upload helper on blob (it is tested elsewhere).
        created_json = {"componentCount": "5"}
        blob._do_upload = mock.Mock(return_value=created_json, spec=[])
        # Make sure `metadata` is empty before the request.
        self.assertIsNone(blob.component_count)

        client = mock.sentinel.client
        ret_val = blob.upload_from_string(data, client=client, **kwargs)

        # Check the response and side-effects.
        self.assertIsNone(ret_val)
        self.assertEqual(blob.component_count, 5)

        # Check the mock.
        payload = _to_bytes(data, encoding="utf-8")
        stream = self._do_upload_mock_call_helper(
            blob, client, "text/plain", len(payload)
        )
        self.assertIsInstance(stream, io.BytesIO)
        self.assertEqual(stream.getvalue(), payload)

    def test_upload_from_string_w_bytes(self):
        data = b"XB]jb\xb8tad\xe0"
        self._upload_from_string_helper(data)

    def test_upload_from_string_w_text(self):
        data = u"\N{snowman} \N{sailboat}"
        self._upload_from_string_helper(data)

    def _create_resumable_upload_session_helper(self, origin=None, side_effect=None):
        bucket = _Bucket(name="alex-trebek")
        blob = self._make_one("blob-name", bucket=bucket)
        chunk_size = 99 * blob._CHUNK_SIZE_MULTIPLE
        blob.chunk_size = chunk_size

        # Create mocks to be checked for doing transport.
        resumable_url = "http://test.invalid?upload_id=clean-up-everybody"
        response_headers = {"location": resumable_url}
        transport = self._mock_transport(http_client.OK, response_headers)
        if side_effect is not None:
            transport.request.side_effect = side_effect

        # Create some mock arguments and call the method under test.
        content_type = u"text/plain"
        size = 10000
        client = mock.Mock(_http=transport, spec=[u"_http"])
        new_url = blob.create_resumable_upload_session(
            content_type=content_type, size=size, origin=origin, client=client
        )

        # Check the returned value and (lack of) side-effect.
        self.assertEqual(new_url, resumable_url)
        self.assertEqual(blob.chunk_size, chunk_size)

        # Check the mocks.
        upload_url = (
            "https://www.googleapis.com/upload/storage/v1"
            + bucket.path
            + "/o?uploadType=resumable"
        )
        payload = b'{"name": "blob-name"}'
        expected_headers = {
            "content-type": "application/json; charset=UTF-8",
            "x-upload-content-length": str(size),
            "x-upload-content-type": content_type,
        }
        if origin is not None:
            expected_headers["Origin"] = origin
        transport.request.assert_called_once_with(
            "POST", upload_url, data=payload, headers=expected_headers, timeout=mock.ANY
        )

    def test_create_resumable_upload_session(self):
        self._create_resumable_upload_session_helper()

    def test_create_resumable_upload_session_with_origin(self):
        self._create_resumable_upload_session_helper(origin="http://google.com")

    def test_create_resumable_upload_session_with_failure(self):
        from google.resumable_media import InvalidResponse
        from google.cloud import exceptions

        message = "5-oh-3 woe is me."
        response = self._mock_requests_response(
            status_code=http_client.SERVICE_UNAVAILABLE, headers={}
        )
        side_effect = InvalidResponse(response, message)

        with self.assertRaises(exceptions.ServiceUnavailable) as exc_info:
            self._create_resumable_upload_session_helper(side_effect=side_effect)

        self.assertIn(message, exc_info.exception.message)
        self.assertEqual(exc_info.exception.errors, [])

    def test_get_iam_policy(self):
        from google.cloud.storage.iam import STORAGE_OWNER_ROLE
        from google.cloud.storage.iam import STORAGE_EDITOR_ROLE
        from google.cloud.storage.iam import STORAGE_VIEWER_ROLE
        from google.api_core.iam import Policy

        BLOB_NAME = "blob-name"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        ETAG = "DEADBEEF"
        VERSION = 17
        OWNER1 = "user:phred@example.com"
        OWNER2 = "group:cloud-logs@google.com"
        EDITOR1 = "domain:google.com"
        EDITOR2 = "user:phred@example.com"
        VIEWER1 = "serviceAccount:1234-abcdef@service.example.com"
        VIEWER2 = "user:phred@example.com"
        RETURNED = {
            "resourceId": PATH,
            "etag": ETAG,
            "version": VERSION,
            "bindings": [
                {"role": STORAGE_OWNER_ROLE, "members": [OWNER1, OWNER2]},
                {"role": STORAGE_EDITOR_ROLE, "members": [EDITOR1, EDITOR2]},
                {"role": STORAGE_VIEWER_ROLE, "members": [VIEWER1, VIEWER2]},
            ],
        }
        after = ({"status": http_client.OK}, RETURNED)
        EXPECTED = {
            binding["role"]: set(binding["members"]) for binding in RETURNED["bindings"]
        }
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        policy = blob.get_iam_policy()

        self.assertIsInstance(policy, Policy)
        self.assertEqual(policy.etag, RETURNED["etag"])
        self.assertEqual(policy.version, RETURNED["version"])
        self.assertEqual(dict(policy), EXPECTED)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(
            kw[0],
            {
                "method": "GET",
                "path": "%s/iam" % (PATH,),
                "query_params": {},
                "_target_object": None,
            },
        )

    def test_get_iam_policy_w_user_project(self):
        from google.api_core.iam import Policy

        BLOB_NAME = "blob-name"
        USER_PROJECT = "user-project-123"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        ETAG = "DEADBEEF"
        VERSION = 17
        RETURNED = {
            "resourceId": PATH,
            "etag": ETAG,
            "version": VERSION,
            "bindings": [],
        }
        after = ({"status": http_client.OK}, RETURNED)
        EXPECTED = {}
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        policy = blob.get_iam_policy()

        self.assertIsInstance(policy, Policy)
        self.assertEqual(policy.etag, RETURNED["etag"])
        self.assertEqual(policy.version, RETURNED["version"])
        self.assertEqual(dict(policy), EXPECTED)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(
            kw[0],
            {
                "method": "GET",
                "path": "%s/iam" % (PATH,),
                "query_params": {"userProject": USER_PROJECT},
                "_target_object": None,
            },
        )

    def test_set_iam_policy(self):
        import operator
        from google.cloud.storage.iam import STORAGE_OWNER_ROLE
        from google.cloud.storage.iam import STORAGE_EDITOR_ROLE
        from google.cloud.storage.iam import STORAGE_VIEWER_ROLE
        from google.api_core.iam import Policy

        BLOB_NAME = "blob-name"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        ETAG = "DEADBEEF"
        VERSION = 17
        OWNER1 = "user:phred@example.com"
        OWNER2 = "group:cloud-logs@google.com"
        EDITOR1 = "domain:google.com"
        EDITOR2 = "user:phred@example.com"
        VIEWER1 = "serviceAccount:1234-abcdef@service.example.com"
        VIEWER2 = "user:phred@example.com"
        BINDINGS = [
            {"role": STORAGE_OWNER_ROLE, "members": [OWNER1, OWNER2]},
            {"role": STORAGE_EDITOR_ROLE, "members": [EDITOR1, EDITOR2]},
            {"role": STORAGE_VIEWER_ROLE, "members": [VIEWER1, VIEWER2]},
        ]
        RETURNED = {"etag": ETAG, "version": VERSION, "bindings": BINDINGS}
        after = ({"status": http_client.OK}, RETURNED)
        policy = Policy()
        for binding in BINDINGS:
            policy[binding["role"]] = binding["members"]

        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        returned = blob.set_iam_policy(policy)

        self.assertEqual(returned.etag, ETAG)
        self.assertEqual(returned.version, VERSION)
        self.assertEqual(dict(returned), dict(policy))

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "PUT")
        self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
        self.assertEqual(kw[0]["query_params"], {})
        sent = kw[0]["data"]
        self.assertEqual(sent["resourceId"], PATH)
        self.assertEqual(len(sent["bindings"]), len(BINDINGS))
        key = operator.itemgetter("role")
        for found, expected in zip(
            sorted(sent["bindings"], key=key), sorted(BINDINGS, key=key)
        ):
            self.assertEqual(found["role"], expected["role"])
            self.assertEqual(sorted(found["members"]), sorted(expected["members"]))

    def test_set_iam_policy_w_user_project(self):
        from google.api_core.iam import Policy

        BLOB_NAME = "blob-name"
        USER_PROJECT = "user-project-123"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        ETAG = "DEADBEEF"
        VERSION = 17
        BINDINGS = []
        RETURNED = {"etag": ETAG, "version": VERSION, "bindings": BINDINGS}
        after = ({"status": http_client.OK}, RETURNED)
        policy = Policy()

        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        returned = blob.set_iam_policy(policy)

        self.assertEqual(returned.etag, ETAG)
        self.assertEqual(returned.version, VERSION)
        self.assertEqual(dict(returned), dict(policy))

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "PUT")
        self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
        self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
        self.assertEqual(kw[0]["data"], {"resourceId": PATH})

    def test_test_iam_permissions(self):
        from google.cloud.storage.iam import STORAGE_OBJECTS_LIST
        from google.cloud.storage.iam import STORAGE_BUCKETS_GET
        from google.cloud.storage.iam import STORAGE_BUCKETS_UPDATE

        BLOB_NAME = "blob-name"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        PERMISSIONS = [
            STORAGE_OBJECTS_LIST,
            STORAGE_BUCKETS_GET,
            STORAGE_BUCKETS_UPDATE,
        ]
        ALLOWED = PERMISSIONS[1:]
        RETURNED = {"permissions": ALLOWED}
        after = ({"status": http_client.OK}, RETURNED)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        allowed = blob.test_iam_permissions(PERMISSIONS)

        self.assertEqual(allowed, ALLOWED)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "GET")
        self.assertEqual(kw[0]["path"], "%s/iam/testPermissions" % (PATH,))
        self.assertEqual(kw[0]["query_params"], {"permissions": PERMISSIONS})

    def test_test_iam_permissions_w_user_project(self):
        from google.cloud.storage.iam import STORAGE_OBJECTS_LIST
        from google.cloud.storage.iam import STORAGE_BUCKETS_GET
        from google.cloud.storage.iam import STORAGE_BUCKETS_UPDATE

        BLOB_NAME = "blob-name"
        USER_PROJECT = "user-project-123"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        PERMISSIONS = [
            STORAGE_OBJECTS_LIST,
            STORAGE_BUCKETS_GET,
            STORAGE_BUCKETS_UPDATE,
        ]
        ALLOWED = PERMISSIONS[1:]
        RETURNED = {"permissions": ALLOWED}
        after = ({"status": http_client.OK}, RETURNED)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        allowed = blob.test_iam_permissions(PERMISSIONS)

        self.assertEqual(allowed, ALLOWED)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "GET")
        self.assertEqual(kw[0]["path"], "%s/iam/testPermissions" % (PATH,))
        self.assertEqual(
            kw[0]["query_params"],
            {"permissions": PERMISSIONS, "userProject": USER_PROJECT},
        )

    def test_make_public(self):
        from google.cloud.storage.acl import _ACLEntity

        BLOB_NAME = "blob-name"
        permissive = [{"entity": "allUsers", "role": _ACLEntity.READER_ROLE}]
        after = ({"status": http_client.OK}, {"acl": permissive})
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        blob.acl.loaded = True
        blob.make_public()
        self.assertEqual(list(blob.acl), permissive)
        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "PATCH")
        self.assertEqual(kw[0]["path"], "/b/name/o/%s" % BLOB_NAME)
        self.assertEqual(kw[0]["data"], {"acl": permissive})
        self.assertEqual(kw[0]["query_params"], {"projection": "full"})

    def test_make_private(self):
        BLOB_NAME = "blob-name"
        no_permissions = []
        after = ({"status": http_client.OK}, {"acl": no_permissions})
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        blob.acl.loaded = True
        blob.make_private()
        self.assertEqual(list(blob.acl), no_permissions)
        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "PATCH")
        self.assertEqual(kw[0]["path"], "/b/name/o/%s" % BLOB_NAME)
        self.assertEqual(kw[0]["data"], {"acl": no_permissions})
        self.assertEqual(kw[0]["query_params"], {"projection": "full"})

    def test_compose_wo_content_type_set(self):
        SOURCE_1 = "source-1"
        SOURCE_2 = "source-2"
        DESTINATION = "destinaton"
        RESOURCE = {}
        after = ({"status": http_client.OK}, RESOURCE)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        source_1 = self._make_one(SOURCE_1, bucket=bucket)
        source_2 = self._make_one(SOURCE_2, bucket=bucket)
        destination = self._make_one(DESTINATION, bucket=bucket)
        # no destination.content_type set

        destination.compose(sources=[source_1, source_2])

        self.assertIsNone(destination.content_type)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(
            kw[0],
            {
                "method": "POST",
                "path": "/b/name/o/%s/compose" % DESTINATION,
                "query_params": {},
                "data": {
                    "sourceObjects": [{"name": source_1.name}, {"name": source_2.name}],
                    "destination": {},
                },
                "_target_object": destination,
            },
        )

    def test_compose_minimal_w_user_project(self):
        SOURCE_1 = "source-1"
        SOURCE_2 = "source-2"
        DESTINATION = "destinaton"
        RESOURCE = {"etag": "DEADBEEF"}
        USER_PROJECT = "user-project-123"
        after = ({"status": http_client.OK}, RESOURCE)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        source_1 = self._make_one(SOURCE_1, bucket=bucket)
        source_2 = self._make_one(SOURCE_2, bucket=bucket)
        destination = self._make_one(DESTINATION, bucket=bucket)
        destination.content_type = "text/plain"

        destination.compose(sources=[source_1, source_2])

        self.assertEqual(destination.etag, "DEADBEEF")

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(
            kw[0],
            {
                "method": "POST",
                "path": "/b/name/o/%s/compose" % DESTINATION,
                "query_params": {"userProject": USER_PROJECT},
                "data": {
                    "sourceObjects": [{"name": source_1.name}, {"name": source_2.name}],
                    "destination": {"contentType": "text/plain"},
                },
                "_target_object": destination,
            },
        )

    def test_compose_w_additional_property_changes(self):
        SOURCE_1 = "source-1"
        SOURCE_2 = "source-2"
        DESTINATION = "destinaton"
        RESOURCE = {"etag": "DEADBEEF"}
        after = ({"status": http_client.OK}, RESOURCE)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        source_1 = self._make_one(SOURCE_1, bucket=bucket)
        source_2 = self._make_one(SOURCE_2, bucket=bucket)
        destination = self._make_one(DESTINATION, bucket=bucket)
        destination.content_type = "text/plain"
        destination.content_language = "en-US"
        destination.metadata = {"my-key": "my-value"}

        destination.compose(sources=[source_1, source_2])

        self.assertEqual(destination.etag, "DEADBEEF")

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(
            kw[0],
            {
                "method": "POST",
                "path": "/b/name/o/%s/compose" % DESTINATION,
                "query_params": {},
                "data": {
                    "sourceObjects": [{"name": source_1.name}, {"name": source_2.name}],
                    "destination": {
                        "contentType": "text/plain",
                        "contentLanguage": "en-US",
                        "metadata": {"my-key": "my-value"},
                    },
                },
                "_target_object": destination,
            },
        )

    def test_rewrite_response_without_resource(self):
        SOURCE_BLOB = "source"
        DEST_BLOB = "dest"
        DEST_BUCKET = "other-bucket"
        TOKEN = "TOKEN"
        RESPONSE = {
            "totalBytesRewritten": 33,
            "objectSize": 42,
            "done": False,
            "rewriteToken": TOKEN,
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        source_bucket = _Bucket(client=client)
        source_blob = self._make_one(SOURCE_BLOB, bucket=source_bucket)
        dest_bucket = _Bucket(client=client, name=DEST_BUCKET)
        dest_blob = self._make_one(DEST_BLOB, bucket=dest_bucket)

        token, rewritten, size = dest_blob.rewrite(source_blob)

        self.assertEqual(token, TOKEN)
        self.assertEqual(rewritten, 33)
        self.assertEqual(size, 42)

    def test_rewrite_w_generations(self):
        SOURCE_BLOB = "source"
        SOURCE_GENERATION = 42
        DEST_BLOB = "dest"
        DEST_BUCKET = "other-bucket"
        DEST_GENERATION = 43
        TOKEN = "TOKEN"
        RESPONSE = {
            "totalBytesRewritten": 33,
            "objectSize": 42,
            "done": False,
            "rewriteToken": TOKEN,
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        source_bucket = _Bucket(client=client)
        source_blob = self._make_one(
            SOURCE_BLOB, bucket=source_bucket, generation=SOURCE_GENERATION
        )
        dest_bucket = _Bucket(client=client, name=DEST_BUCKET)
        dest_blob = self._make_one(
            DEST_BLOB, bucket=dest_bucket, generation=DEST_GENERATION
        )

        token, rewritten, size = dest_blob.rewrite(source_blob)

        self.assertEqual(token, TOKEN)
        self.assertEqual(rewritten, 33)
        self.assertEqual(size, 42)

        kw, = connection._requested
        self.assertEqual(kw["method"], "POST")
        self.assertEqual(
            kw["path"],
            "/b/%s/o/%s/rewriteTo/b/%s/o/%s"
            % (
                (source_bucket.name, source_blob.name, dest_bucket.name, dest_blob.name)
            ),
        )
        self.assertEqual(kw["query_params"], {"sourceGeneration": SOURCE_GENERATION})

    def test_rewrite_other_bucket_other_name_no_encryption_partial(self):
        SOURCE_BLOB = "source"
        DEST_BLOB = "dest"
        DEST_BUCKET = "other-bucket"
        TOKEN = "TOKEN"
        RESPONSE = {
            "totalBytesRewritten": 33,
            "objectSize": 42,
            "done": False,
            "rewriteToken": TOKEN,
            "resource": {"etag": "DEADBEEF"},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        source_bucket = _Bucket(client=client)
        source_blob = self._make_one(SOURCE_BLOB, bucket=source_bucket)
        dest_bucket = _Bucket(client=client, name=DEST_BUCKET)
        dest_blob = self._make_one(DEST_BLOB, bucket=dest_bucket)

        token, rewritten, size = dest_blob.rewrite(source_blob)

        self.assertEqual(token, TOKEN)
        self.assertEqual(rewritten, 33)
        self.assertEqual(size, 42)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/%s/o/%s" % (
            SOURCE_BLOB,
            DEST_BUCKET,
            DEST_BLOB,
        )
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(kw[0]["query_params"], {})
        SENT = {}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Algorithm", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key-Sha256", headers)
        self.assertNotIn("X-Goog-Encryption-Algorithm", headers)
        self.assertNotIn("X-Goog-Encryption-Key", headers)
        self.assertNotIn("X-Goog-Encryption-Key-Sha256", headers)

    def test_rewrite_same_name_no_old_key_new_key_done_w_user_project(self):
        import base64
        import hashlib

        KEY = b"01234567890123456789012345678901"  # 32 bytes
        KEY_B64 = base64.b64encode(KEY).rstrip().decode("ascii")
        KEY_HASH = hashlib.sha256(KEY).digest()
        KEY_HASH_B64 = base64.b64encode(KEY_HASH).rstrip().decode("ascii")
        BLOB_NAME = "blob"
        USER_PROJECT = "user-project-123"
        RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 42,
            "done": True,
            "resource": {"etag": "DEADBEEF"},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        plain = self._make_one(BLOB_NAME, bucket=bucket)
        encrypted = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=KEY)

        token, rewritten, size = encrypted.rewrite(plain)

        self.assertIsNone(token)
        self.assertEqual(rewritten, 42)
        self.assertEqual(size, 42)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/name/o/%s" % (BLOB_NAME, BLOB_NAME)
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
        SENT = {}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Algorithm", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key-Sha256", headers)
        self.assertEqual(headers["X-Goog-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Encryption-Key"], KEY_B64)
        self.assertEqual(headers["X-Goog-Encryption-Key-Sha256"], KEY_HASH_B64)

    def test_rewrite_same_name_no_key_new_key_w_token(self):
        import base64
        import hashlib

        SOURCE_KEY = b"01234567890123456789012345678901"  # 32 bytes
        SOURCE_KEY_B64 = base64.b64encode(SOURCE_KEY).rstrip().decode("ascii")
        SOURCE_KEY_HASH = hashlib.sha256(SOURCE_KEY).digest()
        SOURCE_KEY_HASH_B64 = base64.b64encode(SOURCE_KEY_HASH).rstrip().decode("ascii")
        DEST_KEY = b"90123456789012345678901234567890"  # 32 bytes
        DEST_KEY_B64 = base64.b64encode(DEST_KEY).rstrip().decode("ascii")
        DEST_KEY_HASH = hashlib.sha256(DEST_KEY).digest()
        DEST_KEY_HASH_B64 = base64.b64encode(DEST_KEY_HASH).rstrip().decode("ascii")
        BLOB_NAME = "blob"
        TOKEN = "TOKEN"
        RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 42,
            "done": True,
            "resource": {"etag": "DEADBEEF"},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        source = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=SOURCE_KEY)
        dest = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=DEST_KEY)

        token, rewritten, size = dest.rewrite(source, token=TOKEN)

        self.assertIsNone(token)
        self.assertEqual(rewritten, 42)
        self.assertEqual(size, 42)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/name/o/%s" % (BLOB_NAME, BLOB_NAME)
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(kw[0]["query_params"], {"rewriteToken": TOKEN})
        SENT = {}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Key"], SOURCE_KEY_B64)
        self.assertEqual(
            headers["X-Goog-Copy-Source-Encryption-Key-Sha256"], SOURCE_KEY_HASH_B64
        )
        self.assertEqual(headers["X-Goog-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Encryption-Key"], DEST_KEY_B64)
        self.assertEqual(headers["X-Goog-Encryption-Key-Sha256"], DEST_KEY_HASH_B64)

    def test_rewrite_same_name_w_old_key_new_kms_key(self):
        import base64
        import hashlib

        SOURCE_KEY = b"01234567890123456789012345678901"  # 32 bytes
        SOURCE_KEY_B64 = base64.b64encode(SOURCE_KEY).rstrip().decode("ascii")
        SOURCE_KEY_HASH = hashlib.sha256(SOURCE_KEY).digest()
        SOURCE_KEY_HASH_B64 = base64.b64encode(SOURCE_KEY_HASH).rstrip().decode("ascii")
        DEST_KMS_RESOURCE = (
            "projects/test-project-123/"
            "locations/us/"
            "keyRings/test-ring/"
            "cryptoKeys/test-key"
        )
        BLOB_NAME = "blob"
        RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 42,
            "done": True,
            "resource": {"etag": "DEADBEEF"},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        source = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=SOURCE_KEY)
        dest = self._make_one(BLOB_NAME, bucket=bucket, kms_key_name=DEST_KMS_RESOURCE)

        token, rewritten, size = dest.rewrite(source)

        self.assertIsNone(token)
        self.assertEqual(rewritten, 42)
        self.assertEqual(size, 42)

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/name/o/%s" % (BLOB_NAME, BLOB_NAME)
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(
            kw[0]["query_params"], {"destinationKmsKeyName": DEST_KMS_RESOURCE}
        )
        SENT = {"kmsKeyName": DEST_KMS_RESOURCE}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Key"], SOURCE_KEY_B64)
        self.assertEqual(
            headers["X-Goog-Copy-Source-Encryption-Key-Sha256"], SOURCE_KEY_HASH_B64
        )

    def test_update_storage_class_invalid(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        with self.assertRaises(ValueError):
            blob.update_storage_class(u"BOGUS")

    def test_update_storage_class_large_file(self):
        BLOB_NAME = "blob-name"
        STORAGE_CLASS = u"NEARLINE"
        TOKEN = "TOKEN"
        INCOMPLETE_RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 84,
            "done": False,
            "rewriteToken": TOKEN,
            "resource": {"storageClass": STORAGE_CLASS},
        }
        COMPLETE_RESPONSE = {
            "totalBytesRewritten": 84,
            "objectSize": 84,
            "done": True,
            "resource": {"storageClass": STORAGE_CLASS},
        }
        response_1 = ({"status": http_client.OK}, INCOMPLETE_RESPONSE)
        response_2 = ({"status": http_client.OK}, COMPLETE_RESPONSE)
        connection = _Connection(response_1, response_2)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        blob.update_storage_class("NEARLINE")

        self.assertEqual(blob.storage_class, "NEARLINE")

    def test_update_storage_class_wo_encryption_key(self):
        BLOB_NAME = "blob-name"
        STORAGE_CLASS = u"NEARLINE"
        RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 42,
            "done": True,
            "resource": {"storageClass": STORAGE_CLASS},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        blob.update_storage_class("NEARLINE")

        self.assertEqual(blob.storage_class, "NEARLINE")

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/name/o/%s" % (BLOB_NAME, BLOB_NAME)
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(kw[0]["query_params"], {})
        SENT = {"storageClass": STORAGE_CLASS}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        # Blob has no key, and therefore the relevant headers are not sent.
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Algorithm", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key", headers)
        self.assertNotIn("X-Goog-Copy-Source-Encryption-Key-Sha256", headers)
        self.assertNotIn("X-Goog-Encryption-Algorithm", headers)
        self.assertNotIn("X-Goog-Encryption-Key", headers)
        self.assertNotIn("X-Goog-Encryption-Key-Sha256", headers)

    def test_update_storage_class_w_encryption_key_w_user_project(self):
        import base64
        import hashlib

        BLOB_NAME = "blob-name"
        BLOB_KEY = b"01234567890123456789012345678901"  # 32 bytes
        BLOB_KEY_B64 = base64.b64encode(BLOB_KEY).rstrip().decode("ascii")
        BLOB_KEY_HASH = hashlib.sha256(BLOB_KEY).digest()
        BLOB_KEY_HASH_B64 = base64.b64encode(BLOB_KEY_HASH).rstrip().decode("ascii")
        STORAGE_CLASS = u"NEARLINE"
        USER_PROJECT = "user-project-123"
        RESPONSE = {
            "totalBytesRewritten": 42,
            "objectSize": 42,
            "done": True,
            "resource": {"storageClass": STORAGE_CLASS},
        }
        response = ({"status": http_client.OK}, RESPONSE)
        connection = _Connection(response)
        client = _Client(connection)
        bucket = _Bucket(client=client, user_project=USER_PROJECT)
        blob = self._make_one(BLOB_NAME, bucket=bucket, encryption_key=BLOB_KEY)

        blob.update_storage_class("NEARLINE")

        self.assertEqual(blob.storage_class, "NEARLINE")

        kw = connection._requested
        self.assertEqual(len(kw), 1)
        self.assertEqual(kw[0]["method"], "POST")
        PATH = "/b/name/o/%s/rewriteTo/b/name/o/%s" % (BLOB_NAME, BLOB_NAME)
        self.assertEqual(kw[0]["path"], PATH)
        self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
        SENT = {"storageClass": STORAGE_CLASS}
        self.assertEqual(kw[0]["data"], SENT)

        headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
        # Blob has key, and therefore the relevant headers are sent.
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Key"], BLOB_KEY_B64)
        self.assertEqual(
            headers["X-Goog-Copy-Source-Encryption-Key-Sha256"], BLOB_KEY_HASH_B64
        )
        self.assertEqual(headers["X-Goog-Encryption-Algorithm"], "AES256")
        self.assertEqual(headers["X-Goog-Encryption-Key"], BLOB_KEY_B64)
        self.assertEqual(headers["X-Goog-Encryption-Key-Sha256"], BLOB_KEY_HASH_B64)

    def test_cache_control_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CACHE_CONTROL = "no-cache"
        properties = {"cacheControl": CACHE_CONTROL}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.cache_control, CACHE_CONTROL)

    def test_cache_control_setter(self):
        BLOB_NAME = "blob-name"
        CACHE_CONTROL = "no-cache"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.cache_control)
        blob.cache_control = CACHE_CONTROL
        self.assertEqual(blob.cache_control, CACHE_CONTROL)

    def test_component_count(self):
        BUCKET = object()
        COMPONENT_COUNT = 42
        blob = self._make_one(
            "blob-name", bucket=BUCKET, properties={"componentCount": COMPONENT_COUNT}
        )
        self.assertEqual(blob.component_count, COMPONENT_COUNT)

    def test_component_count_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.component_count)

    def test_component_count_string_val(self):
        BUCKET = object()
        COMPONENT_COUNT = 42
        blob = self._make_one(
            "blob-name",
            bucket=BUCKET,
            properties={"componentCount": str(COMPONENT_COUNT)},
        )
        self.assertEqual(blob.component_count, COMPONENT_COUNT)

    def test_content_disposition_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CONTENT_DISPOSITION = "Attachment; filename=example.jpg"
        properties = {"contentDisposition": CONTENT_DISPOSITION}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.content_disposition, CONTENT_DISPOSITION)

    def test_content_disposition_setter(self):
        BLOB_NAME = "blob-name"
        CONTENT_DISPOSITION = "Attachment; filename=example.jpg"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.content_disposition)
        blob.content_disposition = CONTENT_DISPOSITION
        self.assertEqual(blob.content_disposition, CONTENT_DISPOSITION)

    def test_content_encoding_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CONTENT_ENCODING = "gzip"
        properties = {"contentEncoding": CONTENT_ENCODING}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.content_encoding, CONTENT_ENCODING)

    def test_content_encoding_setter(self):
        BLOB_NAME = "blob-name"
        CONTENT_ENCODING = "gzip"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.content_encoding)
        blob.content_encoding = CONTENT_ENCODING
        self.assertEqual(blob.content_encoding, CONTENT_ENCODING)

    def test_content_language_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CONTENT_LANGUAGE = "pt-BR"
        properties = {"contentLanguage": CONTENT_LANGUAGE}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.content_language, CONTENT_LANGUAGE)

    def test_content_language_setter(self):
        BLOB_NAME = "blob-name"
        CONTENT_LANGUAGE = "pt-BR"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.content_language)
        blob.content_language = CONTENT_LANGUAGE
        self.assertEqual(blob.content_language, CONTENT_LANGUAGE)

    def test_content_type_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CONTENT_TYPE = "image/jpeg"
        properties = {"contentType": CONTENT_TYPE}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.content_type, CONTENT_TYPE)

    def test_content_type_setter(self):
        BLOB_NAME = "blob-name"
        CONTENT_TYPE = "image/jpeg"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.content_type)
        blob.content_type = CONTENT_TYPE
        self.assertEqual(blob.content_type, CONTENT_TYPE)

    def test_crc32c_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        CRC32C = "DEADBEEF"
        properties = {"crc32c": CRC32C}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.crc32c, CRC32C)

    def test_crc32c_setter(self):
        BLOB_NAME = "blob-name"
        CRC32C = "DEADBEEF"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.crc32c)
        blob.crc32c = CRC32C
        self.assertEqual(blob.crc32c, CRC32C)

    def test_etag(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        ETAG = "ETAG"
        properties = {"etag": ETAG}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.etag, ETAG)

    def test_event_based_hold_getter_missing(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertIsNone(blob.event_based_hold)

    def test_event_based_hold_getter_false(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {"eventBasedHold": False}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertFalse(blob.event_based_hold)

    def test_event_based_hold_getter_true(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {"eventBasedHold": True}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertTrue(blob.event_based_hold)

    def test_event_based_hold_setter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.event_based_hold)
        blob.event_based_hold = True
        self.assertEqual(blob.event_based_hold, True)

    def test_generation(self):
        BUCKET = object()
        GENERATION = 42
        blob = self._make_one(
            "blob-name", bucket=BUCKET, properties={"generation": GENERATION}
        )
        self.assertEqual(blob.generation, GENERATION)

    def test_generation_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.generation)

    def test_generation_string_val(self):
        BUCKET = object()
        GENERATION = 42
        blob = self._make_one(
            "blob-name", bucket=BUCKET, properties={"generation": str(GENERATION)}
        )
        self.assertEqual(blob.generation, GENERATION)

    def test_id(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        ID = "ID"
        properties = {"id": ID}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.id, ID)

    def test_md5_hash_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        MD5_HASH = "DEADBEEF"
        properties = {"md5Hash": MD5_HASH}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.md5_hash, MD5_HASH)

    def test_md5_hash_setter(self):
        BLOB_NAME = "blob-name"
        MD5_HASH = "DEADBEEF"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.md5_hash)
        blob.md5_hash = MD5_HASH
        self.assertEqual(blob.md5_hash, MD5_HASH)

    def test_media_link(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        MEDIA_LINK = "http://example.com/media/"
        properties = {"mediaLink": MEDIA_LINK}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.media_link, MEDIA_LINK)

    def test_metadata_getter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        METADATA = {"foo": "Foo"}
        properties = {"metadata": METADATA}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.metadata, METADATA)

    def test_metadata_setter(self):
        BLOB_NAME = "blob-name"
        METADATA = {"foo": "Foo"}
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.metadata)
        blob.metadata = METADATA
        self.assertEqual(blob.metadata, METADATA)

    def test_metageneration(self):
        BUCKET = object()
        METAGENERATION = 42
        blob = self._make_one(
            "blob-name", bucket=BUCKET, properties={"metageneration": METAGENERATION}
        )
        self.assertEqual(blob.metageneration, METAGENERATION)

    def test_metageneration_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.metageneration)

    def test_metageneration_string_val(self):
        BUCKET = object()
        METAGENERATION = 42
        blob = self._make_one(
            "blob-name",
            bucket=BUCKET,
            properties={"metageneration": str(METAGENERATION)},
        )
        self.assertEqual(blob.metageneration, METAGENERATION)

    def test_owner(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        OWNER = {"entity": "project-owner-12345", "entityId": "23456"}
        properties = {"owner": OWNER}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        owner = blob.owner
        self.assertEqual(owner["entity"], "project-owner-12345")
        self.assertEqual(owner["entityId"], "23456")

    def test_retention_expiration_time(self):
        from google.cloud._helpers import _RFC3339_MICROS
        from google.cloud._helpers import UTC

        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        TIMESTAMP = datetime.datetime(2014, 11, 5, 20, 34, 37, tzinfo=UTC)
        TIME_CREATED = TIMESTAMP.strftime(_RFC3339_MICROS)
        properties = {"retentionExpirationTime": TIME_CREATED}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.retention_expiration_time, TIMESTAMP)

    def test_retention_expiration_time_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.retention_expiration_time)

    def test_self_link(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        SELF_LINK = "http://example.com/self/"
        properties = {"selfLink": SELF_LINK}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.self_link, SELF_LINK)

    def test_size(self):
        BUCKET = object()
        SIZE = 42
        blob = self._make_one("blob-name", bucket=BUCKET, properties={"size": SIZE})
        self.assertEqual(blob.size, SIZE)

    def test_size_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.size)

    def test_size_string_val(self):
        BUCKET = object()
        SIZE = 42
        blob = self._make_one(
            "blob-name", bucket=BUCKET, properties={"size": str(SIZE)}
        )
        self.assertEqual(blob.size, SIZE)

    def test_storage_class_getter(self):
        blob_name = "blob-name"
        bucket = _Bucket()
        storage_class = "COLDLINE"
        properties = {"storageClass": storage_class}
        blob = self._make_one(blob_name, bucket=bucket, properties=properties)
        self.assertEqual(blob.storage_class, storage_class)

    def test_storage_class_setter(self):
        blob_name = "blob-name"
        bucket = _Bucket()
        storage_class = "COLDLINE"
        blob = self._make_one(blob_name, bucket=bucket)
        self.assertIsNone(blob.storage_class)
        blob.storage_class = storage_class
        self.assertEqual(blob.storage_class, storage_class)
        self.assertEqual(blob._properties, {"storageClass": storage_class})

    def test_temporary_hold_getter_missing(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertIsNone(blob.temporary_hold)

    def test_temporary_hold_getter_false(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {"temporaryHold": False}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertFalse(blob.temporary_hold)

    def test_temporary_hold_getter_true(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        properties = {"temporaryHold": True}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertTrue(blob.temporary_hold)

    def test_temporary_hold_setter(self):
        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.temporary_hold)
        blob.temporary_hold = True
        self.assertEqual(blob.temporary_hold, True)

    def test_time_deleted(self):
        from google.cloud._helpers import _RFC3339_MICROS
        from google.cloud._helpers import UTC

        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        TIMESTAMP = datetime.datetime(2014, 11, 5, 20, 34, 37, tzinfo=UTC)
        TIME_DELETED = TIMESTAMP.strftime(_RFC3339_MICROS)
        properties = {"timeDeleted": TIME_DELETED}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.time_deleted, TIMESTAMP)

    def test_time_deleted_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.time_deleted)

    def test_time_created(self):
        from google.cloud._helpers import _RFC3339_MICROS
        from google.cloud._helpers import UTC

        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        TIMESTAMP = datetime.datetime(2014, 11, 5, 20, 34, 37, tzinfo=UTC)
        TIME_CREATED = TIMESTAMP.strftime(_RFC3339_MICROS)
        properties = {"timeCreated": TIME_CREATED}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.time_created, TIMESTAMP)

    def test_time_created_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.time_created)

    def test_updated(self):
        from google.cloud._helpers import _RFC3339_MICROS
        from google.cloud._helpers import UTC

        BLOB_NAME = "blob-name"
        bucket = _Bucket()
        TIMESTAMP = datetime.datetime(2014, 11, 5, 20, 34, 37, tzinfo=UTC)
        UPDATED = TIMESTAMP.strftime(_RFC3339_MICROS)
        properties = {"updated": UPDATED}
        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
        self.assertEqual(blob.updated, TIMESTAMP)

    def test_updated_unset(self):
        BUCKET = object()
        blob = self._make_one("blob-name", bucket=BUCKET)
        self.assertIsNone(blob.updated)

    def test_from_string_w_valid_uri(self):
        from google.cloud.storage.blob import Blob

        connection = _Connection()
        client = _Client(connection)
        uri = "gs://BUCKET_NAME/b"
        blob = Blob.from_string(uri, client)

        self.assertIsInstance(blob, Blob)
        self.assertIs(blob.client, client)
        self.assertEqual(blob.name, "b")
        self.assertEqual(blob.bucket.name, "BUCKET_NAME")

    def test_from_string_w_invalid_uri(self):
        from google.cloud.storage.blob import Blob

        connection = _Connection()
        client = _Client(connection)

        with pytest.raises(ValueError, match="URI scheme must be gs"):
            Blob.from_string("http://bucket_name/b", client)

    def test_from_string_w_domain_name_bucket(self):
        from google.cloud.storage.blob import Blob

        connection = _Connection()
        client = _Client(connection)
        uri = "gs://buckets.example.com/b"
        blob = Blob.from_string(uri, client)

        self.assertIsInstance(blob, Blob)
        self.assertIs(blob.client, client)
        self.assertEqual(blob.name, "b")
        self.assertEqual(blob.bucket.name, "buckets.example.com")


class Test__quote(unittest.TestCase):
    @staticmethod
    def _call_fut(*args, **kw):
        from google.cloud.storage.blob import _quote

        return _quote(*args, **kw)

    def test_bytes(self):
        quoted = self._call_fut(b"\xDE\xAD\xBE\xEF")
        self.assertEqual(quoted, "%DE%AD%BE%EF")

    def test_unicode(self):
        helicopter = u"\U0001f681"
        quoted = self._call_fut(helicopter)
        self.assertEqual(quoted, "%F0%9F%9A%81")

    def test_bad_type(self):
        with self.assertRaises(TypeError):
            self._call_fut(None)

    def test_w_slash_default(self):
        with_slash = "foo/bar/baz"
        quoted = self._call_fut(with_slash)
        self.assertEqual(quoted, "foo%2Fbar%2Fbaz")

    def test_w_slash_w_safe(self):
        with_slash = "foo/bar/baz"
        quoted_safe = self._call_fut(with_slash, safe=b"/")
        self.assertEqual(quoted_safe, with_slash)

    def test_w_tilde(self):
        with_tilde = "bam~qux"
        quoted = self._call_fut(with_tilde, safe=b"~")
        self.assertEqual(quoted, with_tilde)


class Test__maybe_rewind(unittest.TestCase):
    @staticmethod
    def _call_fut(*args, **kwargs):
        from google.cloud.storage.blob import _maybe_rewind

        return _maybe_rewind(*args, **kwargs)

    def test_default(self):
        stream = mock.Mock(spec=[u"seek"])
        ret_val = self._call_fut(stream)
        self.assertIsNone(ret_val)

        stream.seek.assert_not_called()

    def test_do_not_rewind(self):
        stream = mock.Mock(spec=[u"seek"])
        ret_val = self._call_fut(stream, rewind=False)
        self.assertIsNone(ret_val)

        stream.seek.assert_not_called()

    def test_do_rewind(self):
        stream = mock.Mock(spec=[u"seek"])
        ret_val = self._call_fut(stream, rewind=True)
        self.assertIsNone(ret_val)

        stream.seek.assert_called_once_with(0, os.SEEK_SET)


class Test__raise_from_invalid_response(unittest.TestCase):
    @staticmethod
    def _call_fut(error):
        from google.cloud.storage.blob import _raise_from_invalid_response

        return _raise_from_invalid_response(error)

    def _helper(self, message, code=http_client.BAD_REQUEST, args=()):
        import requests

        from google.resumable_media import InvalidResponse
        from google.api_core import exceptions

        response = requests.Response()
        response.request = requests.Request("GET", "http://example.com").prepare()
        response.status_code = code
        error = InvalidResponse(response, message, *args)

        with self.assertRaises(exceptions.GoogleAPICallError) as exc_info:
            self._call_fut(error)

        return exc_info

    def test_default(self):
        message = "Failure"
        exc_info = self._helper(message)
        expected = "GET http://example.com/: {}".format(message)
        self.assertEqual(exc_info.exception.message, expected)
        self.assertEqual(exc_info.exception.errors, [])

    def test_w_206_and_args(self):
        message = "Failure"
        args = ("one", "two")
        exc_info = self._helper(message, code=http_client.PARTIAL_CONTENT, args=args)
        expected = "GET http://example.com/: {}".format((message,) + args)
        self.assertEqual(exc_info.exception.message, expected)
        self.assertEqual(exc_info.exception.errors, [])


class Test__add_query_parameters(unittest.TestCase):
    @staticmethod
    def _call_fut(*args, **kwargs):
        from google.cloud.storage.blob import _add_query_parameters

        return _add_query_parameters(*args, **kwargs)

    def test_w_empty_list(self):
        BASE_URL = "https://test.example.com/base"
        self.assertEqual(self._call_fut(BASE_URL, []), BASE_URL)

    def test_wo_existing_qs(self):
        BASE_URL = "https://test.example.com/base"
        NV_LIST = [("one", "One"), ("two", "Two")]
        expected = "&".join(["{}={}".format(name, value) for name, value in NV_LIST])
        self.assertEqual(
            self._call_fut(BASE_URL, NV_LIST), "{}?{}".format(BASE_URL, expected)
        )

    def test_w_existing_qs(self):
        BASE_URL = "https://test.example.com/base?one=Three"
        NV_LIST = [("one", "One"), ("two", "Two")]
        expected = "&".join(["{}={}".format(name, value) for name, value in NV_LIST])
        self.assertEqual(
            self._call_fut(BASE_URL, NV_LIST), "{}&{}".format(BASE_URL, expected)
        )


class _Connection(object):

    API_BASE_URL = "http://example.com"
    USER_AGENT = "testing 1.2.3"
    credentials = object()

    def __init__(self, *responses):
        self._responses = responses[:]
        self._requested = []
        self._signed = []

    def _respond(self, **kw):
        self._requested.append(kw)
        response, self._responses = self._responses[0], self._responses[1:]
        return response

    def api_request(self, **kw):
        from google.cloud.exceptions import NotFound

        info, content = self._respond(**kw)
        if info.get("status") == http_client.NOT_FOUND:
            raise NotFound(info)
        return content


class _Bucket(object):
    def __init__(self, client=None, name="name", user_project=None):
        if client is None:
            connection = _Connection()
            client = _Client(connection)
        self.client = client
        self._blobs = {}
        self._copied = []
        self._deleted = []
        self.name = name
        self.path = "/b/" + name
        self.user_project = user_project

    def delete_blob(self, blob_name, client=None, generation=None):
        del self._blobs[blob_name]
        self._deleted.append((blob_name, client, generation))


class _Client(object):
    def __init__(self, connection):
        self._base_connection = connection

    @property
    def _connection(self):
        return self._base_connection

    @property
    def _credentials(self):
        return self._base_connection.credentials