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 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

    @staticmethod
    def _get_default_timeout():
        from google.cloud.storage.constants import _DEFAULT_TIMEOUT

        return _DEFAULT_TIMEOUT

    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,
        access_token=None,
        service_account_email=None,
        virtual_hosted_style=False,
    ):
        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,
                access_token=access_token,
                service_account_email=service_account_email,
                virtual_hosted_style=virtual_hosted_style,
            )

        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")
        quoted_name = parse.quote(encoded_name, safe=b"/~")

        if virtual_hosted_style:
            expected_api_access_endpoint = "https://{}.storage.googleapis.com".format(
                bucket.name
            )
            expected_resource = "/{}".format(quoted_name)
        else:
            expected_api_access_endpoint = api_access_endpoint
            expected_resource = "/{}/{}".format(bucket.name, quoted_name)

        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": expected_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,
            "access_token": access_token,
            "service_account_email": service_account_email,
        }
        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_virtual_hostname(self):
        self._generate_signed_url_v4_helper(virtual_hosted_style=True)

    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(timeout=42))
        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,
                "timeout": 42,
            },
        )

    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,
                "timeout": self._get_default_timeout(),
            },
        )

    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,
                "timeout": self._get_default_timeout(),
            },
        )

    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, self._get_default_timeout())]
        )

    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(timeout=42)
        self.assertFalse(blob.exists())
        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, GENERATION, 42)])

    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://storage.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://storage.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://storage.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://storage.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""):
        import requests

        response = requests.Response()
        response.status_code = status_code
        response.headers.update(headers)
        response.raw = None
        response._content = content

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

    def _do_download_helper_wo_chunks(self, w_range, raw_download):
        blob_name = "blob-name"
        client = mock.Mock()
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        self.assertIsNone(blob.chunk_size)

        transport = object()
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}

        if raw_download:
            patch = mock.patch("google.cloud.storage.blob.RawDownload")
        else:
            patch = mock.patch("google.cloud.storage.blob.Download")

        with patch as patched:
            if w_range:
                blob._do_download(
                    transport,
                    file_obj,
                    download_url,
                    headers,
                    start=1,
                    end=3,
                    raw_download=raw_download,
                )
            else:
                blob._do_download(
                    transport,
                    file_obj,
                    download_url,
                    headers,
                    raw_download=raw_download,
                )

        if w_range:
            patched.assert_called_once_with(
                download_url, stream=file_obj, headers=headers, start=1, end=3
            )
        else:
            patched.assert_called_once_with(
                download_url, stream=file_obj, headers=headers, start=None, end=None
            )
        patched.return_value.consume.assert_called_once_with(transport)

    def test__do_download_wo_chunks_wo_range_wo_raw(self):
        self._do_download_helper_wo_chunks(w_range=False, raw_download=False)

    def test__do_download_wo_chunks_w_range_wo_raw(self):
        self._do_download_helper_wo_chunks(w_range=True, raw_download=False)

    def test__do_download_wo_chunks_wo_range_w_raw(self):
        self._do_download_helper_wo_chunks(w_range=False, raw_download=True)

    def test__do_download_wo_chunks_w_range_w_raw(self):
        self._do_download_helper_wo_chunks(w_range=True, raw_download=True)

    def _do_download_helper_w_chunks(self, w_range, raw_download):
        blob_name = "blob-name"
        client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        blob._CHUNK_SIZE_MULTIPLE = 1
        chunk_size = blob.chunk_size = 3

        transport = object()
        file_obj = io.BytesIO()
        download_url = "http://test.invalid"
        headers = {}

        download = mock.Mock(finished=False, spec=["finished", "consume_next_chunk"])

        def side_effect(_):
            download.finished = True

        download.consume_next_chunk.side_effect = side_effect

        if raw_download:
            patch = mock.patch("google.cloud.storage.blob.RawChunkedDownload")
        else:
            patch = mock.patch("google.cloud.storage.blob.ChunkedDownload")

        with patch as patched:
            patched.return_value = download
            if w_range:
                blob._do_download(
                    transport,
                    file_obj,
                    download_url,
                    headers,
                    start=1,
                    end=3,
                    raw_download=raw_download,
                )
            else:
                blob._do_download(
                    transport,
                    file_obj,
                    download_url,
                    headers,
                    raw_download=raw_download,
                )

        if w_range:
            patched.assert_called_once_with(
                download_url, chunk_size, file_obj, headers=headers, start=1, end=3
            )
        else:
            patched.assert_called_once_with(
                download_url, chunk_size, file_obj, headers=headers, start=0, end=None
            )
        download.consume_next_chunk.assert_called_once_with(transport)

    def test__do_download_w_chunks_wo_range_wo_raw(self):
        self._do_download_helper_w_chunks(w_range=False, raw_download=False)

    def test__do_download_w_chunks_w_range_wo_raw(self):
        self._do_download_helper_w_chunks(w_range=True, raw_download=False)

    def test__do_download_w_chunks_wo_range_w_raw(self):
        self._do_download_helper_w_chunks(w_range=False, raw_download=True)

    def test__do_download_w_chunks_w_range_w_raw(self):
        self._do_download_helper_w_chunks(w_range=True, raw_download=True)

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

        raw_response = requests.Response()
        raw_response.status_code = http_client.NOT_FOUND
        raw_request = requests.Request("GET", "http://example.com")
        raw_response.request = raw_request.prepare()
        grmp_response = InvalidResponse(raw_response)

        blob_name = "blob-name"
        media_link = "http://test.invalid"
        client = mock.Mock(spec=[u"_http"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        blob._properties["mediaLink"] = media_link
        blob._do_download = mock.Mock()
        blob._do_download.side_effect = grmp_response

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

        self.assertEqual(file_obj.tell(), 0)

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, file_obj, media_link, headers, None, None, False
        )

    def test_download_to_file_wo_media_link(self):
        blob_name = "blob-name"
        client = mock.Mock(spec=[u"_http"])
        bucket = _Bucket(client)
        blob = self._make_one(blob_name, bucket=bucket)
        blob._do_download = mock.Mock()
        file_obj = io.BytesIO()

        blob.download_to_file(file_obj)

        # Make sure the media link is still unknown.
        self.assertIsNone(blob.media_link)

        expected_url = (
            "https://storage.googleapis.com/download/storage/v1/b/"
            "name/o/blob-name?alt=media"
        )
        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, file_obj, expected_url, headers, None, None, False
        )

    def _download_to_file_helper(self, use_chunks, raw_download):
        blob_name = "blob-name"
        client = mock.Mock(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:
            blob._CHUNK_SIZE_MULTIPLE = 1
            blob.chunk_size = 3
        blob._do_download = mock.Mock()

        file_obj = io.BytesIO()
        if raw_download:
            blob.download_to_file(file_obj, raw_download=True)
        else:
            blob.download_to_file(file_obj)

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, file_obj, media_link, headers, None, None, raw_download
        )

    def test_download_to_file_wo_chunks_wo_raw(self):
        self._download_to_file_helper(use_chunks=False, raw_download=False)

    def test_download_to_file_w_chunks_wo_raw(self):
        self._download_to_file_helper(use_chunks=True, raw_download=False)

    def test_download_to_file_wo_chunks_w_raw(self):
        self._download_to_file_helper(use_chunks=False, raw_download=True)

    def test_download_to_file_w_chunks_w_raw(self):
        self._download_to_file_helper(use_chunks=True, raw_download=True)

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

        blob_name = "blob-name"
        client = mock.Mock(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)
        blob._do_download = mock.Mock()

        with _NamedTemporaryFile() as temp:
            blob.download_to_filename(temp.name, raw_download=raw_download)
            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)

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, mock.ANY, media_link, headers, None, None, raw_download
        )
        stream = blob._do_download.mock_calls[0].args[1]
        self.assertEqual(stream.name, temp.name)

    def test_download_to_filename_w_updated_wo_raw(self):
        updated = "2014-12-06T13:13:50.690Z"
        self._download_to_filename_helper(updated=updated, raw_download=False)

    def test_download_to_filename_wo_updated_wo_raw(self):
        self._download_to_filename_helper(updated=None, raw_download=False)

    def test_download_to_filename_w_updated_w_raw(self):
        updated = "2014-12-06T13:13:50.690Z"
        self._download_to_filename_helper(updated=updated, raw_download=True)

    def test_download_to_filename_wo_updated_w_raw(self):
        self._download_to_filename_helper(updated=None, raw_download=True)

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

        blob_name = "blob-name"
        client = mock.Mock(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)
        blob._do_download = mock.Mock()
        blob._do_download.side_effect = DataCorruption("testing")

        # 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)
        self.assertTrue(os.path.exists(filename))

        with self.assertRaises(DataCorruption):
            blob.download_to_filename(filename)

        # Make sure the file was cleaned up.
        self.assertFalse(os.path.exists(filename))

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, mock.ANY, media_link, headers, None, None, False
        )
        stream = blob._do_download.mock_calls[0].args[1]
        self.assertEqual(stream.name, filename)

    def test_download_to_filename_w_key(self):
        from google.cloud._testing import _NamedTemporaryFile
        from google.cloud.storage.blob import _get_encryption_headers

        blob_name = "blob-name"
        # Create a fake client/bucket and use them in the Blob() constructor.
        client = mock.Mock(spec=["_http"])
        bucket = _Bucket(client)
        media_link = "http://example.com/media/"
        properties = {"mediaLink": media_link}
        key = b"aa426195405adee2c8081bb9e7e74b19"
        blob = self._make_one(
            blob_name, bucket=bucket, properties=properties, encryption_key=key
        )
        blob._do_download = mock.Mock()

        with _NamedTemporaryFile() as temp:
            blob.download_to_filename(temp.name)

        headers = {"accept-encoding": "gzip"}
        headers.update(_get_encryption_headers(key))
        blob._do_download.assert_called_once_with(
            client._http, mock.ANY, media_link, headers, None, None, False
        )
        stream = blob._do_download.mock_calls[0].args[1]
        self.assertEqual(stream.name, temp.name)

    def _download_as_string_helper(self, raw_download):
        blob_name = "blob-name"
        client = mock.Mock(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)
        blob._do_download = mock.Mock()

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

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http, mock.ANY, media_link, headers, None, None, raw_download
        )
        stream = blob._do_download.mock_calls[0].args[1]
        self.assertIsInstance(stream, io.BytesIO)

    def test_download_as_string_wo_raw(self):
        self._download_as_string_helper(raw_download=False)

    def test_download_as_string_w_raw(self):
        self._download_as_string_helper(raw_download=True)

    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__set_metadata_to_none(self):
        name = u"blob-name"
        blob = self._make_one(name, bucket=None)
        blob.storage_class = "NEARLINE"
        blob.cache_control = "max-age=3600"

        with mock.patch("google.cloud.storage.blob.Blob._patch_property") as patch_prop:
            blob.metadata = None
            patch_prop.assert_called_once_with("metadata", None)

    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://storage.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
        from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE

        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://storage.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, _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://storage.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
    ):
        from google.cloud.storage.blob import _MAX_MULTIPART_SIZE

        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 <= _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):
        from google.cloud.storage.blob import _MAX_MULTIPART_SIZE

        self._do_upload_helper(size=_MAX_MULTIPART_SIZE)

    def test__do_upload_uses_resumable(self):
        from google.cloud.storage.blob import _MAX_MULTIPART_SIZE

        chunk_size = 256 * 1024  # 256KB
        self._do_upload_helper(chunk_size=chunk_size, size=_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://storage.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 = 1
        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(timeout=42)

        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,
                "timeout": 42,
            },
        )

    def test_get_iam_policy_w_requested_policy_version(self):
        from google.cloud.storage.iam import STORAGE_OWNER_ROLE

        BLOB_NAME = "blob-name"
        PATH = "/b/name/o/%s" % (BLOB_NAME,)
        ETAG = "DEADBEEF"
        VERSION = 1
        OWNER1 = "user:phred@example.com"
        OWNER2 = "group:cloud-logs@google.com"
        RETURNED = {
            "resourceId": PATH,
            "etag": ETAG,
            "version": VERSION,
            "bindings": [{"role": STORAGE_OWNER_ROLE, "members": [OWNER1, OWNER2]}],
        }
        after = ({"status": http_client.OK}, RETURNED)
        connection = _Connection(after)
        client = _Client(connection)
        bucket = _Bucket(client=client)
        blob = self._make_one(BLOB_NAME, bucket=bucket)

        blob.get_iam_policy(requested_policy_version=3)

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

    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 = 1
        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,
                "timeout": self._get_default_timeout(),
            },
        )

    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 = 1
        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, timeout=42)

        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"], {})
        self.assertEqual(kw[0]["timeout"], 42)
        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 = 1
        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, timeout=42)

        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})
        self.assertEqual(kw[0]["timeout"], 42)

    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},
        )
        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())

    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,
                "timeout": self._get_default_timeout(),
            },
        )

    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], timeout=42)

        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,
                "timeout": 42,
            },
        )

    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,
                "timeout": self._get_default_timeout(),
            },
        )

    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, timeout=42)

        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})
        self.assertEqual(kw["timeout"], 42)

    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)
        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())

        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):
        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)
        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())

        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):
        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)
        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())

        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):
        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}
        )
        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
        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):
        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_metadata_setter_w_nan(self):
        BLOB_NAME = "blob-name"
        METADATA = {"foo": float("nan")}
        bucket = _Bucket()
        blob = self._make_one(BLOB_NAME, bucket=bucket)
        self.assertIsNone(blob.metadata)
        blob.metadata = METADATA
        value = blob.metadata["foo"]
        self.assertIsInstance(value, str)

    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, timeout=None):
        del self._blobs[blob_name]
        self._deleted.append((blob_name, client, generation, timeout))


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