# Copyright 2015 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 io
import json
import mock
import pytest
import re
import requests
import unittest
from six.moves import http_client
from six.moves.urllib import parse as urlparse

from google.api_core import exceptions

from google.oauth2.service_account import Credentials
from . import _read_local_json

from google.cloud.storage.retry import DEFAULT_RETRY
from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED


_SERVICE_ACCOUNT_JSON = _read_local_json("url_signer_v4_test_account.json")
_CONFORMANCE_TESTS = _read_local_json("url_signer_v4_test_data.json")[
    "postPolicyV4Tests"
]
_POST_POLICY_TESTS = [test for test in _CONFORMANCE_TESTS if "policyInput" in test]
_FAKE_CREDENTIALS = Credentials.from_service_account_info(_SERVICE_ACCOUNT_JSON)


def _make_credentials(project=None):
    import google.auth.credentials

    if project is not None:
        return mock.Mock(spec=google.auth.credentials.Credentials, project_id=project)

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


def _create_signing_credentials():
    import google.auth.credentials

    class _SigningCredentials(
        google.auth.credentials.Credentials, google.auth.credentials.Signing
    ):
        pass

    credentials = mock.Mock(spec=_SigningCredentials)
    credentials.sign_bytes = mock.Mock(return_value=b"Signature_bytes")
    credentials.signer_email = "test@mail.com"
    return credentials


def _make_connection(*responses):
    import google.cloud.storage._http
    from google.cloud.exceptions import NotFound

    mock_conn = mock.create_autospec(google.cloud.storage._http.Connection)
    mock_conn.user_agent = "testing 1.2.3"
    mock_conn.api_request.side_effect = list(responses) + [NotFound("miss")]
    return mock_conn


def _make_response(status=http_client.OK, content=b"", headers={}):
    response = requests.Response()
    response.status_code = status
    response._content = content
    response.headers = headers
    response.request = requests.Request()
    return response


def _make_json_response(data, status=http_client.OK, headers=None):
    headers = headers or {}
    headers["Content-Type"] = "application/json"
    return _make_response(
        status=status, content=json.dumps(data).encode("utf-8"), headers=headers
    )


def _make_requests_session(responses):
    session = mock.create_autospec(requests.Session, instance=True)
    session.request.side_effect = responses
    session.is_mtls = False
    return session


class TestClient(unittest.TestCase):
    @staticmethod
    def _get_target_class():
        from google.cloud.storage.client import Client

        return Client

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

        return _DEFAULT_TIMEOUT

    def _make_one(self, *args, **kw):
        return self._get_target_class()(*args, **kw)

    def test_ctor_connection_type(self):
        from google.cloud._http import ClientInfo
        from google.cloud.storage._http import Connection

        PROJECT = "PROJECT"
        credentials = _make_credentials()

        client = self._make_one(project=PROJECT, credentials=credentials)

        self.assertEqual(client.project, PROJECT)
        self.assertIsInstance(client._connection, Connection)
        self.assertIs(client._connection.credentials, credentials)
        self.assertIsNone(client.current_batch)
        self.assertEqual(list(client._batch_stack), [])
        self.assertIsInstance(client._connection._client_info, ClientInfo)
        self.assertEqual(
            client._connection.API_BASE_URL, Connection.DEFAULT_API_ENDPOINT
        )

    def test_ctor_w_empty_client_options(self):
        from google.api_core.client_options import ClientOptions

        PROJECT = "PROJECT"
        credentials = _make_credentials()
        client_options = ClientOptions()

        client = self._make_one(
            project=PROJECT, credentials=credentials, client_options=client_options
        )

        self.assertEqual(
            client._connection.API_BASE_URL, client._connection.DEFAULT_API_ENDPOINT
        )

    def test_ctor_w_client_options_dict(self):
        PROJECT = "PROJECT"
        credentials = _make_credentials()
        api_endpoint = "https://www.foo-googleapis.com"
        client_options = {"api_endpoint": api_endpoint}

        client = self._make_one(
            project=PROJECT, credentials=credentials, client_options=client_options
        )

        self.assertEqual(client._connection.API_BASE_URL, api_endpoint)

    def test_ctor_w_client_options_object(self):
        from google.api_core.client_options import ClientOptions

        PROJECT = "PROJECT"
        credentials = _make_credentials()
        client_options = ClientOptions(api_endpoint="https://www.foo-googleapis.com")

        client = self._make_one(
            project=PROJECT, credentials=credentials, client_options=client_options
        )

        self.assertEqual(
            client._connection.API_BASE_URL, "https://www.foo-googleapis.com"
        )

    def test_ctor_wo_project(self):
        from google.cloud.storage._http import Connection

        PROJECT = "PROJECT"
        credentials = _make_credentials(project=PROJECT)

        client = self._make_one(credentials=credentials)

        self.assertEqual(client.project, PROJECT)
        self.assertIsInstance(client._connection, Connection)
        self.assertIs(client._connection.credentials, credentials)
        self.assertIsNone(client.current_batch)
        self.assertEqual(list(client._batch_stack), [])

    def test_ctor_w_project_explicit_none(self):
        from google.cloud.storage._http import Connection

        credentials = _make_credentials()

        client = self._make_one(project=None, credentials=credentials)

        self.assertIsNone(client.project)
        self.assertIsInstance(client._connection, Connection)
        self.assertIs(client._connection.credentials, credentials)
        self.assertIsNone(client.current_batch)
        self.assertEqual(list(client._batch_stack), [])

    def test_ctor_w_client_info(self):
        from google.cloud._http import ClientInfo
        from google.cloud.storage._http import Connection

        credentials = _make_credentials()
        client_info = ClientInfo()

        client = self._make_one(
            project=None, credentials=credentials, client_info=client_info
        )

        self.assertIsNone(client.project)
        self.assertIsInstance(client._connection, Connection)
        self.assertIs(client._connection.credentials, credentials)
        self.assertIsNone(client.current_batch)
        self.assertEqual(list(client._batch_stack), [])
        self.assertIs(client._connection._client_info, client_info)

    def test_ctor_mtls(self):
        PROJECT = "PROJECT"
        credentials = _make_credentials(project=PROJECT)

        client = self._make_one(credentials=credentials)
        self.assertEqual(client._connection.ALLOW_AUTO_SWITCH_TO_MTLS_URL, True)
        self.assertEqual(
            client._connection.API_BASE_URL, "https://storage.googleapis.com"
        )

        client = self._make_one(
            credentials=credentials, client_options={"api_endpoint": "http://foo"}
        )
        self.assertEqual(client._connection.ALLOW_AUTO_SWITCH_TO_MTLS_URL, False)
        self.assertEqual(client._connection.API_BASE_URL, "http://foo")

    def test_create_anonymous_client(self):
        from google.auth.credentials import AnonymousCredentials
        from google.cloud.storage._http import Connection

        klass = self._get_target_class()
        client = klass.create_anonymous_client()

        self.assertIsNone(client.project)
        self.assertIsInstance(client._connection, Connection)
        self.assertIsInstance(client._connection.credentials, AnonymousCredentials)

    def test__push_batch_and__pop_batch(self):
        from google.cloud.storage.batch import Batch

        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        batch1 = Batch(client)
        batch2 = Batch(client)
        client._push_batch(batch1)
        self.assertEqual(list(client._batch_stack), [batch1])
        self.assertIs(client.current_batch, batch1)
        client._push_batch(batch2)
        self.assertIs(client.current_batch, batch2)
        # list(_LocalStack) returns in reverse order.
        self.assertEqual(list(client._batch_stack), [batch2, batch1])
        self.assertIs(client._pop_batch(), batch2)
        self.assertEqual(list(client._batch_stack), [batch1])
        self.assertIs(client._pop_batch(), batch1)
        self.assertEqual(list(client._batch_stack), [])

    def test__connection_setter(self):
        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        client._base_connection = None  # Unset the value from the constructor
        client._connection = connection = object()
        self.assertIs(client._base_connection, connection)

    def test__connection_setter_when_set(self):
        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        self.assertRaises(ValueError, setattr, client, "_connection", None)

    def test__connection_getter_no_batch(self):
        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        self.assertIs(client._connection, client._base_connection)
        self.assertIsNone(client.current_batch)

    def test__connection_getter_with_batch(self):
        from google.cloud.storage.batch import Batch

        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        batch = Batch(client)
        client._push_batch(batch)
        self.assertIsNot(client._connection, client._base_connection)
        self.assertIs(client._connection, batch)
        self.assertIs(client.current_batch, batch)

    def test_get_service_account_email_wo_project(self):
        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        EMAIL = "storage-user-123@example.com"
        RESOURCE = {"kind": "storage#serviceAccount", "email_address": EMAIL}

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        http = _make_requests_session([_make_json_response(RESOURCE)])
        client._http_internal = http

        service_account_email = client.get_service_account_email(timeout=42)

        self.assertEqual(service_account_email, EMAIL)
        http.request.assert_called_once_with(
            method="GET", url=mock.ANY, data=None, headers=mock.ANY, timeout=42
        )
        _, kwargs = http.request.call_args
        scheme, netloc, path, qs, _ = urlparse.urlsplit(kwargs.get("url"))
        self.assertEqual("%s://%s" % (scheme, netloc), client._connection.API_BASE_URL)
        self.assertEqual(
            path,
            "/".join(
                [
                    "",
                    "storage",
                    client._connection.API_VERSION,
                    "projects",
                    PROJECT,
                    "serviceAccount",
                ]
            ),
        )

    def test_get_service_account_email_w_project(self):
        PROJECT = "PROJECT"
        OTHER_PROJECT = "OTHER_PROJECT"
        CREDENTIALS = _make_credentials()
        EMAIL = "storage-user-123@example.com"
        RESOURCE = {"kind": "storage#serviceAccount", "email_address": EMAIL}

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        http = _make_requests_session([_make_json_response(RESOURCE)])
        client._http_internal = http

        service_account_email = client.get_service_account_email(project=OTHER_PROJECT)

        self.assertEqual(service_account_email, EMAIL)
        http.request.assert_called_once_with(
            method="GET",
            url=mock.ANY,
            data=None,
            headers=mock.ANY,
            timeout=self._get_default_timeout(),
        )
        _, kwargs = http.request.call_args
        scheme, netloc, path, qs, _ = urlparse.urlsplit(kwargs.get("url"))
        self.assertEqual("%s://%s" % (scheme, netloc), client._connection.API_BASE_URL)
        self.assertEqual(
            path,
            "/".join(
                [
                    "",
                    "storage",
                    client._connection.API_VERSION,
                    "projects",
                    OTHER_PROJECT,
                    "serviceAccount",
                ]
            ),
        )

    def test_bucket(self):
        from google.cloud.storage.bucket import Bucket

        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()
        BUCKET_NAME = "BUCKET_NAME"

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        bucket = client.bucket(BUCKET_NAME)
        self.assertIsInstance(bucket, Bucket)
        self.assertIs(bucket.client, client)
        self.assertEqual(bucket.name, BUCKET_NAME)
        self.assertIsNone(bucket.user_project)

    def test_bucket_w_user_project(self):
        from google.cloud.storage.bucket import Bucket

        PROJECT = "PROJECT"
        USER_PROJECT = "USER_PROJECT"
        CREDENTIALS = _make_credentials()
        BUCKET_NAME = "BUCKET_NAME"

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        bucket = client.bucket(BUCKET_NAME, user_project=USER_PROJECT)
        self.assertIsInstance(bucket, Bucket)
        self.assertIs(bucket.client, client)
        self.assertEqual(bucket.name, BUCKET_NAME)
        self.assertEqual(bucket.user_project, USER_PROJECT)

    def test_batch(self):
        from google.cloud.storage.batch import Batch

        PROJECT = "PROJECT"
        CREDENTIALS = _make_credentials()

        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
        batch = client.batch()
        self.assertIsInstance(batch, Batch)
        self.assertIs(batch._client, client)

    def test__get_resource_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound

        project = "PROJECT"
        path = "/path/to/something"
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        with self.assertRaises(NotFound):
            client._get_resource(path)

        connection.api_request.assert_called_once_with(
            method="GET",
            path=path,
            query_params=None,
            headers=None,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=None,
        )

    def test__get_resource_hit_w_explicit(self):
        project = "PROJECT"
        path = "/path/to/something"
        query_params = {"foo": "Foo"}
        headers = {"bar": "Bar"}
        timeout = 100
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        expected = mock.Mock(spec={})
        connection = client._base_connection = _make_connection(expected)
        target = mock.Mock(spec={})

        found = client._get_resource(
            path,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

        self.assertIs(found, expected)

        connection.api_request.assert_called_once_with(
            method="GET",
            path=path,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

    def test__list_resource_w_defaults(self):
        import functools
        from google.api_core.page_iterator import HTTPIterator
        from google.api_core.page_iterator import _do_nothing_page_start

        project = "PROJECT"
        path = "/path/to/list/resource"
        item_to_value = mock.Mock(spec=[])
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        iterator = client._list_resource(path=path, item_to_value=item_to_value,)

        self.assertIsInstance(iterator, HTTPIterator)
        self.assertIs(iterator.client, client)
        self.assertIsInstance(iterator.api_request, functools.partial)
        self.assertIs(iterator.api_request.func, connection.api_request)
        self.assertEqual(iterator.api_request.args, ())
        expected_keywords = {
            "timeout": self._get_default_timeout(),
            "retry": DEFAULT_RETRY,
        }
        self.assertEqual(iterator.api_request.keywords, expected_keywords)
        self.assertEqual(iterator.path, path)
        self.assertEqual(iterator.next_page_token, None)
        self.assertEqual(iterator.max_results, None)
        self.assertIs(iterator._page_start, _do_nothing_page_start)

    def test__list_resource_w_explicit(self):
        import functools
        from google.api_core.page_iterator import HTTPIterator

        project = "PROJECT"
        path = "/path/to/list/resource"
        item_to_value = mock.Mock(spec=[])
        page_token = "PAGE-TOKEN"
        max_results = 47
        extra_params = {"foo": "Foo"}
        page_start = mock.Mock(spec=[])
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        iterator = client._list_resource(
            path=path,
            item_to_value=item_to_value,
            page_token=page_token,
            max_results=max_results,
            extra_params=extra_params,
            page_start=page_start,
        )

        self.assertIsInstance(iterator, HTTPIterator)
        self.assertIs(iterator.client, client)
        self.assertIsInstance(iterator.api_request, functools.partial)
        self.assertIs(iterator.api_request.func, connection.api_request)
        self.assertEqual(iterator.api_request.args, ())
        expected_keywords = {
            "timeout": self._get_default_timeout(),
            "retry": DEFAULT_RETRY,
        }
        self.assertEqual(iterator.api_request.keywords, expected_keywords)
        self.assertEqual(iterator.path, path)
        self.assertEqual(iterator.next_page_token, page_token)
        self.assertEqual(iterator.max_results, max_results)
        self.assertIs(iterator._page_start, page_start)

    def test__patch_resource_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound

        project = "PROJECT"
        path = "/path/to/something"
        credentials = _make_credentials()
        data = {"baz": "Baz"}

        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        with self.assertRaises(NotFound):
            client._patch_resource(path, data)

        connection.api_request.assert_called_once_with(
            method="PATCH",
            path=path,
            data=data,
            query_params=None,
            headers=None,
            timeout=self._get_default_timeout(),
            retry=None,
            _target_object=None,
        )

    def test__patch_resource_hit_w_explicit(self):
        project = "PROJECT"
        path = "/path/to/something"
        data = {"baz": "Baz"}
        query_params = {"foo": "Foo"}
        headers = {"bar": "Bar"}
        timeout = 100
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        expected = mock.Mock(spec={})
        connection = client._base_connection = _make_connection(expected)
        target = mock.Mock(spec={})

        found = client._patch_resource(
            path,
            data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

        self.assertIs(found, expected)

        connection.api_request.assert_called_once_with(
            method="PATCH",
            path=path,
            data=data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

    def test__put_resource_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound

        project = "PROJECT"
        path = "/path/to/something"
        credentials = _make_credentials()
        data = {"baz": "Baz"}

        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        with self.assertRaises(NotFound):
            client._put_resource(path, data)

        connection.api_request.assert_called_once_with(
            method="PUT",
            path=path,
            data=data,
            query_params=None,
            headers=None,
            timeout=self._get_default_timeout(),
            retry=None,
            _target_object=None,
        )

    def test__put_resource_hit_w_explicit(self):
        project = "PROJECT"
        path = "/path/to/something"
        data = {"baz": "Baz"}
        query_params = {"foo": "Foo"}
        headers = {"bar": "Bar"}
        timeout = 100
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        expected = mock.Mock(spec={})
        connection = client._base_connection = _make_connection(expected)
        target = mock.Mock(spec={})

        found = client._put_resource(
            path,
            data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

        self.assertIs(found, expected)

        connection.api_request.assert_called_once_with(
            method="PUT",
            path=path,
            data=data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

    def test__post_resource_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound

        project = "PROJECT"
        path = "/path/to/something"
        credentials = _make_credentials()
        data = {"baz": "Baz"}

        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        with self.assertRaises(NotFound):
            client._post_resource(path, data)

        connection.api_request.assert_called_once_with(
            method="POST",
            path=path,
            data=data,
            query_params=None,
            headers=None,
            timeout=self._get_default_timeout(),
            retry=None,
            _target_object=None,
        )

    def test__post_resource_hit_w_explicit(self):
        project = "PROJECT"
        path = "/path/to/something"
        data = {"baz": "Baz"}
        query_params = {"foo": "Foo"}
        headers = {"bar": "Bar"}
        timeout = 100
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        expected = mock.Mock(spec={})
        connection = client._base_connection = _make_connection(expected)
        target = mock.Mock(spec={})

        found = client._post_resource(
            path,
            data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

        self.assertIs(found, expected)

        connection.api_request.assert_called_once_with(
            method="POST",
            path=path,
            data=data,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

    def test__delete_resource_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound

        project = "PROJECT"
        path = "/path/to/something"
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        connection = client._base_connection = _make_connection()

        with self.assertRaises(NotFound):
            client._delete_resource(path)

        connection.api_request.assert_called_once_with(
            method="DELETE",
            path=path,
            query_params=None,
            headers=None,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=None,
        )

    def test__delete_resource_hit_w_explicit(self):
        project = "PROJECT"
        path = "/path/to/something"
        query_params = {"foo": "Foo"}
        headers = {"bar": "Bar"}
        timeout = 100
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()

        client = self._make_one(project=project, credentials=credentials)
        expected = mock.Mock(spec={})
        connection = client._base_connection = _make_connection(expected)
        target = mock.Mock(spec={})

        found = client._delete_resource(
            path,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

        self.assertIs(found, expected)

        connection.api_request.assert_called_once_with(
            method="DELETE",
            path=path,
            query_params=query_params,
            headers=headers,
            timeout=timeout,
            retry=retry,
            _target_object=target,
        )

    def test_get_bucket_miss_w_string_w_defaults(self):
        from google.cloud.exceptions import NotFound
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock()
        client._get_resource.side_effect = NotFound("testing")
        bucket_name = "nonesuch"

        with self.assertRaises(NotFound):
            client.get_bucket(bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=mock.ANY,
        )

        target = client._get_resource.call_args[1]["_target_object"]
        self.assertIsInstance(target, Bucket)
        self.assertEqual(target.name, bucket_name)

    def test_get_bucket_hit_w_string_w_timeout(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        timeout = 42
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)

        bucket = client.get_bucket(bucket_name, timeout=timeout)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=timeout,
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_get_bucket_hit_w_string_w_metageneration_match(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        metageneration_number = 6
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)

        bucket = client.get_bucket(
            bucket_name, if_metageneration_match=metageneration_number
        )

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {
            "projection": "noAcl",
            "ifMetagenerationMatch": metageneration_number,
        }
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_get_bucket_miss_w_object_w_retry(self):
        from google.cloud.exceptions import NotFound
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "nonesuch"
        retry = mock.Mock(spec=[])
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(side_effect=NotFound("testing"))
        bucket_obj = Bucket(client, bucket_name)

        with self.assertRaises(NotFound):
            client.get_bucket(bucket_obj, retry=retry)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=retry,
            _target_object=mock.ANY,
        )

        target = client._get_resource.call_args[1]["_target_object"]
        self.assertIsInstance(target, Bucket)
        self.assertEqual(target.name, bucket_name)

    def test_get_bucket_hit_w_object_defaults(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)
        bucket_obj = Bucket(client, bucket_name)

        bucket = client.get_bucket(bucket_obj)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_get_bucket_hit_w_object_w_retry_none(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)
        bucket_obj = Bucket(client, bucket_name)

        bucket = client.get_bucket(bucket_obj, retry=None)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=None,
            _target_object=bucket,
        )

    def test_lookup_bucket_miss_w_defaults(self):
        from google.cloud.exceptions import NotFound
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "nonesuch"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(side_effect=NotFound("testing"))

        bucket = client.lookup_bucket(bucket_name)

        self.assertIsNone(bucket)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=mock.ANY,
        )

        target = client._get_resource.call_args[1]["_target_object"]
        self.assertIsInstance(target, Bucket)
        self.assertEqual(target.name, bucket_name)

    def test_lookup_bucket_hit_w_timeout(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        timeout = 42
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)

        bucket = client.lookup_bucket(bucket_name, timeout=timeout)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=timeout,
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_lookup_bucket_hit_w_metageneration_match(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        metageneration_number = 6
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)

        bucket = client.lookup_bucket(
            bucket_name, if_metageneration_match=metageneration_number
        )

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {
            "projection": "noAcl",
            "ifMetagenerationMatch": metageneration_number,
        }
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_lookup_bucket_hit_w_retry(self):
        from google.cloud.storage.bucket import Bucket

        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._get_resource = mock.Mock(return_value=api_response)
        bucket_obj = Bucket(client, bucket_name)

        bucket = client.lookup_bucket(bucket_obj, retry=None)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)

        expected_path = "/b/%s" % (bucket_name,)
        expected_query_params = {"projection": "noAcl"}
        expected_headers = {}
        client._get_resource.assert_called_once_with(
            expected_path,
            query_params=expected_query_params,
            headers=expected_headers,
            timeout=self._get_default_timeout(),
            retry=None,
            _target_object=bucket,
        )

    def test_create_bucket_w_missing_client_project(self):
        credentials = _make_credentials()
        client = self._make_one(project=None, credentials=credentials)

        with self.assertRaises(ValueError):
            client.create_bucket("bucket")

    def test_create_bucket_w_conflict_w_user_project(self):
        from google.cloud.exceptions import Conflict

        project = "PROJECT"
        user_project = "USER_PROJECT"
        other_project = "OTHER_PROJECT"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.side_effect = Conflict("testing")

        bucket_name = "bucket-name"

        with self.assertRaises(Conflict):
            client.create_bucket(
                bucket_name, project=other_project, user_project=user_project
            )

        expected_path = "/b"
        expected_data = {"name": bucket_name}
        expected_query_params = {
            "project": other_project,
            "userProject": user_project,
        }
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=mock.ANY,
        )

    @mock.patch("warnings.warn")
    def test_create_bucket_w_requester_pays_deprecated(self, mock_warn):
        from google.cloud.storage.bucket import Bucket

        bucket_name = "bucket-name"
        project = "PROJECT"
        credentials = _make_credentials()
        api_respone = {"name": bucket_name, "billing": {"requesterPays": True}}
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_respone

        bucket = client.create_bucket(bucket_name, requester_pays=True)

        self.assertIsInstance(bucket, Bucket)
        self.assertEqual(bucket.name, bucket_name)
        self.assertTrue(bucket.requester_pays)

        expected_path = "/b"
        expected_data = api_respone
        expected_query_params = {"project": project}
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=mock.ANY,
        )

        mock_warn.assert_called_with(
            "requester_pays arg is deprecated. Use Bucket().requester_pays instead.",
            PendingDeprecationWarning,
            stacklevel=1,
        )

    def test_create_bucket_w_predefined_acl_invalid(self):
        project = "PROJECT"
        bucket_name = "bucket-name"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()

        with self.assertRaises(ValueError):
            client.create_bucket(bucket_name, predefined_acl="bogus")

        client._post_resource.assert_not_called()

    def test_create_bucket_w_predefined_acl_valid_w_timeout(self):
        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response
        timeout = 42

        bucket = client.create_bucket(
            bucket_name, predefined_acl="publicRead", timeout=timeout,
        )

        expected_path = "/b"
        expected_data = api_response
        expected_query_params = {
            "project": project,
            "predefinedAcl": "publicRead",
        }
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=timeout,
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_create_bucket_w_predefined_default_object_acl_invalid(self):
        project = "PROJECT"
        bucket_name = "bucket-name"

        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()

        with self.assertRaises(ValueError):
            client.create_bucket(bucket_name, predefined_default_object_acl="bogus")

        client._post_resource.assert_not_called()

    def test_create_bucket_w_predefined_default_object_acl_valid_w_retry(self):
        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response
        retry = mock.Mock(spec=[])

        bucket = client.create_bucket(
            bucket_name, predefined_default_object_acl="publicRead", retry=retry,
        )

        expected_path = "/b"
        expected_data = api_response
        expected_query_params = {
            "project": project,
            "predefinedDefaultObjectAcl": "publicRead",
        }
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=retry,
            _target_object=bucket,
        )

    def test_create_bucket_w_explicit_location(self):
        project = "PROJECT"
        bucket_name = "bucket-name"
        location = "us-central1"
        api_response = {"location": location, "name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response

        bucket = client.create_bucket(bucket_name, location=location)

        self.assertEqual(bucket.location, location)

        expected_path = "/b"
        expected_data = {"location": location, "name": bucket_name}
        expected_query_params = {"project": project}
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_create_bucket_w_explicit_project(self):
        project = "PROJECT"
        other_project = "other-project-123"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response

        bucket = client.create_bucket(bucket_name, project=other_project)

        expected_path = "/b"
        expected_data = api_response
        expected_query_params = {"project": other_project}
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_create_bucket_w_extra_properties(self):
        from google.cloud.storage.bucket import Bucket

        bucket_name = "bucket-name"
        project = "PROJECT"
        cors = [
            {
                "maxAgeSeconds": 60,
                "methods": ["*"],
                "origin": ["https://example.com/frontend"],
                "responseHeader": ["X-Custom-Header"],
            }
        ]
        lifecycle_rules = [{"action": {"type": "Delete"}, "condition": {"age": 365}}]
        location = "eu"
        labels = {"color": "red", "flavor": "cherry"}
        storage_class = "NEARLINE"
        api_response = {
            "name": bucket_name,
            "cors": cors,
            "lifecycle": {"rule": lifecycle_rules},
            "location": location,
            "storageClass": storage_class,
            "versioning": {"enabled": True},
            "billing": {"requesterPays": True},
            "labels": labels,
        }
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response

        bucket = Bucket(client=client, name=bucket_name)
        bucket.cors = cors
        bucket.lifecycle_rules = lifecycle_rules
        bucket.storage_class = storage_class
        bucket.versioning_enabled = True
        bucket.requester_pays = True
        bucket.labels = labels

        client.create_bucket(bucket, location=location)

        expected_path = "/b"
        expected_data = api_response
        expected_query_params = {"project": project}
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_create_bucket_w_name_only(self):
        project = "PROJECT"
        bucket_name = "bucket-name"
        api_response = {"name": bucket_name}
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response

        bucket = client.create_bucket(bucket_name)

        expected_path = "/b"
        expected_data = api_response
        expected_query_params = {"project": project}
        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
            _target_object=bucket,
        )

    def test_download_blob_to_file_with_failure(self):
        from google.resumable_media import InvalidResponse
        from google.cloud.storage.blob import Blob
        from google.cloud.storage.constants import _DEFAULT_TIMEOUT

        project = "PROJECT"
        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)
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)
        blob = mock.create_autospec(Blob)
        blob._encryption_key = None
        blob._get_download_url = mock.Mock()
        blob._do_download = mock.Mock()
        blob._do_download.side_effect = grmp_response

        file_obj = io.BytesIO()
        with self.assertRaises(exceptions.NotFound):
            client.download_blob_to_file(blob, file_obj)

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

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http,
            file_obj,
            blob._get_download_url(),
            headers,
            None,
            None,
            False,
            checksum="md5",
            timeout=_DEFAULT_TIMEOUT,
            retry=DEFAULT_RETRY,
        )

    def test_download_blob_to_file_with_uri(self):
        from google.cloud.storage.constants import _DEFAULT_TIMEOUT

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(project=project, credentials=credentials)
        blob = mock.Mock()
        file_obj = io.BytesIO()
        blob._encryption_key = None
        blob._get_download_url = mock.Mock()
        blob._do_download = mock.Mock()

        with mock.patch(
            "google.cloud.storage.client.Blob.from_string", return_value=blob
        ):
            client.download_blob_to_file("gs://bucket_name/path/to/object", file_obj)

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http,
            file_obj,
            blob._get_download_url(),
            headers,
            None,
            None,
            False,
            checksum="md5",
            timeout=_DEFAULT_TIMEOUT,
            retry=DEFAULT_RETRY,
        )

    def test_download_blob_to_file_with_invalid_uri(self):
        project = "PROJECT"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        file_obj = io.BytesIO()

        with pytest.raises(ValueError, match="URI scheme must be gs"):
            client.download_blob_to_file("http://bucket_name/path/to/object", file_obj)

    def test_download_blob_to_file_w_no_retry(self):
        self._download_blob_to_file_helper(
            use_chunks=True, raw_download=True, retry=None
        )

    def test_download_blob_to_file_w_conditional_retry_pass(self):
        self._download_blob_to_file_helper(
            use_chunks=True,
            raw_download=True,
            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
            if_generation_match=1,
        )

    def test_download_blob_to_file_w_conditional_retry_fail(self):
        self._download_blob_to_file_helper(
            use_chunks=True,
            raw_download=True,
            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
            expect_condition_fail=True,
        )

    def _download_blob_to_file_helper(
        self, use_chunks, raw_download, expect_condition_fail=False, **extra_kwargs
    ):
        from google.cloud.storage.blob import Blob
        from google.cloud.storage.constants import _DEFAULT_TIMEOUT

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)
        blob = mock.create_autospec(Blob)
        blob._encryption_key = None
        blob._get_download_url = mock.Mock()
        if use_chunks:
            blob._CHUNK_SIZE_MULTIPLE = 1
            blob.chunk_size = 3
        blob._do_download = mock.Mock()

        file_obj = io.BytesIO()
        if raw_download:
            client.download_blob_to_file(
                blob, file_obj, raw_download=True, **extra_kwargs
            )
        else:
            client.download_blob_to_file(blob, file_obj, **extra_kwargs)

        expected_retry = extra_kwargs.get("retry", DEFAULT_RETRY)
        if (
            expected_retry is DEFAULT_RETRY_IF_GENERATION_SPECIFIED
            and not expect_condition_fail
        ):
            expected_retry = DEFAULT_RETRY
        elif expect_condition_fail:
            expected_retry = None

        headers = {"accept-encoding": "gzip"}
        blob._do_download.assert_called_once_with(
            client._http,
            file_obj,
            blob._get_download_url(),
            headers,
            None,
            None,
            raw_download,
            checksum="md5",
            timeout=_DEFAULT_TIMEOUT,
            retry=expected_retry,
        )

    def test_download_blob_to_file_wo_chunks_wo_raw(self):
        self._download_blob_to_file_helper(use_chunks=False, raw_download=False)

    def test_download_blob_to_file_w_chunks_wo_raw(self):
        self._download_blob_to_file_helper(use_chunks=True, raw_download=False)

    def test_download_blob_to_file_wo_chunks_w_raw(self):
        self._download_blob_to_file_helper(use_chunks=False, raw_download=True)

    def test_download_blob_to_file_w_chunks_w_raw(self):
        self._download_blob_to_file_helper(use_chunks=True, raw_download=True)

    def test_list_blobs_w_defaults_w_bucket_obj(self):
        from google.cloud.storage.bucket import Bucket
        from google.cloud.storage.bucket import _blobs_page_start
        from google.cloud.storage.bucket import _item_to_blob

        project = "PROJECT"
        bucket_name = "bucket-name"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])
        bucket = Bucket(client, bucket_name)

        iterator = client.list_blobs(bucket)

        self.assertIs(iterator, client._list_resource.return_value)
        self.assertIs(iterator.bucket, bucket)
        self.assertEqual(iterator.prefixes, set())

        expected_path = "/b/{}/o".format(bucket_name)
        expected_item_to_value = _item_to_blob
        expected_page_token = None
        expected_max_results = None
        expected_extra_params = {"projection": "noAcl"}
        expected_page_start = _blobs_page_start
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            page_token=expected_page_token,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            page_start=expected_page_start,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
        )

    def test_list_blobs_w_explicit_w_user_project(self):
        from google.cloud.storage.bucket import _blobs_page_start
        from google.cloud.storage.bucket import _item_to_blob

        project = "PROJECT"
        user_project = "user-project-123"
        bucket_name = "name"
        max_results = 10
        page_token = "ABCD"
        prefix = "subfolder"
        delimiter = "/"
        start_offset = "c"
        end_offset = "g"
        include_trailing_delimiter = True
        versions = True
        projection = "full"
        fields = "items/contentLanguage,nextPageToken"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])
        client._bucket_arg_to_bucket = mock.Mock(spec=[])
        bucket = client._bucket_arg_to_bucket.return_value = mock.Mock(
            spec=["path", "user_project"],
        )
        bucket.path = "/b/{}".format(bucket_name)
        bucket.user_project = user_project
        timeout = 42
        retry = mock.Mock(spec=[])

        iterator = client.list_blobs(
            bucket_or_name=bucket_name,
            max_results=max_results,
            page_token=page_token,
            prefix=prefix,
            delimiter=delimiter,
            start_offset=start_offset,
            end_offset=end_offset,
            include_trailing_delimiter=include_trailing_delimiter,
            versions=versions,
            projection=projection,
            fields=fields,
            timeout=timeout,
            retry=retry,
        )

        self.assertIs(iterator, client._list_resource.return_value)
        self.assertIs(iterator.bucket, bucket)
        self.assertEqual(iterator.prefixes, set())

        expected_path = "/b/{}/o".format(bucket_name)
        expected_item_to_value = _item_to_blob
        expected_page_token = page_token
        expected_max_results = max_results
        expected_extra_params = {
            "projection": projection,
            "prefix": prefix,
            "delimiter": delimiter,
            "startOffset": start_offset,
            "endOffset": end_offset,
            "includeTrailingDelimiter": include_trailing_delimiter,
            "versions": versions,
            "fields": fields,
            "userProject": user_project,
        }
        expected_page_start = _blobs_page_start
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            page_token=expected_page_token,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            page_start=expected_page_start,
            timeout=timeout,
            retry=retry,
        )

    def test_list_buckets_wo_project(self):
        credentials = _make_credentials()
        client = self._make_one(project=None, credentials=credentials)

        with self.assertRaises(ValueError):
            client.list_buckets()

    def test_list_buckets_w_defaults(self):
        from google.cloud.storage.client import _item_to_bucket

        project = "PROJECT"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])

        iterator = client.list_buckets()

        self.assertIs(iterator, client._list_resource.return_value)

        expected_path = "/b"
        expected_item_to_value = _item_to_bucket
        expected_page_token = None
        expected_max_results = None
        expected_extra_params = {
            "project": project,
            "projection": "noAcl",
        }
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            page_token=expected_page_token,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
        )

    def test_list_buckets_w_explicit(self):
        from google.cloud.storage.client import _item_to_bucket

        project = "foo-bar"
        other_project = "OTHER_PROJECT"
        max_results = 10
        page_token = "ABCD"
        prefix = "subfolder"
        projection = "full"
        fields = "items/id,nextPageToken"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])
        timeout = 42
        retry = mock.Mock(spec=[])

        iterator = client.list_buckets(
            project=other_project,
            max_results=max_results,
            page_token=page_token,
            prefix=prefix,
            projection=projection,
            fields=fields,
            timeout=timeout,
            retry=retry,
        )

        self.assertIs(iterator, client._list_resource.return_value)

        expected_path = "/b"
        expected_item_to_value = _item_to_bucket
        expected_page_token = page_token
        expected_max_results = max_results
        expected_extra_params = {
            "project": other_project,
            "prefix": prefix,
            "projection": projection,
            "fields": fields,
        }
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            page_token=expected_page_token,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            timeout=timeout,
            retry=retry,
        )

    def _create_hmac_key_helper(
        self, explicit_project=None, user_project=None, timeout=None, retry=None,
    ):
        import datetime
        from pytz import UTC
        from google.cloud.storage.hmac_key import HMACKeyMetadata

        project = "PROJECT"
        access_id = "ACCESS-ID"
        credentials = _make_credentials()
        email = "storage-user-123@example.com"
        secret = "a" * 40
        now = datetime.datetime.utcnow().replace(tzinfo=UTC)
        now_stamp = "{}Z".format(now.isoformat())

        if explicit_project is not None:
            expected_project = explicit_project
        else:
            expected_project = project

        api_response = {
            "kind": "storage#hmacKey",
            "metadata": {
                "accessId": access_id,
                "etag": "ETAG",
                "id": "projects/{}/hmacKeys/{}".format(project, access_id),
                "project": expected_project,
                "state": "ACTIVE",
                "serviceAccountEmail": email,
                "timeCreated": now_stamp,
                "updated": now_stamp,
            },
            "secret": secret,
        }

        client = self._make_one(project=project, credentials=credentials)
        client._post_resource = mock.Mock()
        client._post_resource.return_value = api_response

        kwargs = {}
        if explicit_project is not None:
            kwargs["project_id"] = explicit_project

        if user_project is not None:
            kwargs["user_project"] = user_project

        if timeout is None:
            expected_timeout = self._get_default_timeout()
        else:
            expected_timeout = kwargs["timeout"] = timeout

        if retry is None:
            expected_retry = None
        else:
            expected_retry = kwargs["retry"] = retry

        metadata, secret = client.create_hmac_key(service_account_email=email, **kwargs)

        self.assertIsInstance(metadata, HMACKeyMetadata)

        self.assertIs(metadata._client, client)
        self.assertEqual(metadata._properties, api_response["metadata"])
        self.assertEqual(secret, api_response["secret"])

        expected_path = "/projects/{}/hmacKeys".format(expected_project)
        expected_data = None
        expected_query_params = {"serviceAccountEmail": email}

        if user_project is not None:
            expected_query_params["userProject"] = user_project

        client._post_resource.assert_called_once_with(
            expected_path,
            expected_data,
            query_params=expected_query_params,
            timeout=expected_timeout,
            retry=expected_retry,
        )

    def test_create_hmac_key_defaults(self):
        self._create_hmac_key_helper()

    def test_create_hmac_key_explicit_project(self):
        self._create_hmac_key_helper(explicit_project="other-project-456")

    def test_create_hmac_key_w_user_project(self):
        self._create_hmac_key_helper(user_project="billed-project")

    def test_create_hmac_key_w_timeout(self):
        self._create_hmac_key_helper(timeout=42)

    def test_create_hmac_key_w_retry(self):
        self._create_hmac_key_helper(retry=mock.Mock(spec=[]))

    def test_list_hmac_keys_w_defaults(self):
        from google.cloud.storage.client import _item_to_hmac_key_metadata

        project = "PROJECT"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])

        iterator = client.list_hmac_keys()

        self.assertIs(iterator, client._list_resource.return_value)

        expected_path = "/projects/{}/hmacKeys".format(project)
        expected_item_to_value = _item_to_hmac_key_metadata
        expected_max_results = None
        expected_extra_params = {}
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            timeout=self._get_default_timeout(),
            retry=DEFAULT_RETRY,
        )

    def test_list_hmac_keys_w_explicit(self):
        from google.cloud.storage.client import _item_to_hmac_key_metadata

        project = "PROJECT"
        other_project = "other-project-456"
        max_results = 3
        show_deleted_keys = True
        service_account_email = "storage-user-123@example.com"
        user_project = "billed-project"
        credentials = _make_credentials()
        client = self._make_one(project=project, credentials=credentials)
        client._list_resource = mock.Mock(spec=[])
        timeout = 42
        retry = mock.Mock(spec=[])

        iterator = client.list_hmac_keys(
            max_results=max_results,
            service_account_email=service_account_email,
            show_deleted_keys=show_deleted_keys,
            project_id=other_project,
            user_project=user_project,
            timeout=timeout,
            retry=retry,
        )

        self.assertIs(iterator, client._list_resource.return_value)

        expected_path = "/projects/{}/hmacKeys".format(other_project)
        expected_item_to_value = _item_to_hmac_key_metadata
        expected_max_results = max_results
        expected_extra_params = {
            "serviceAccountEmail": service_account_email,
            "showDeletedKeys": show_deleted_keys,
            "userProject": user_project,
        }
        client._list_resource.assert_called_once_with(
            expected_path,
            expected_item_to_value,
            max_results=expected_max_results,
            extra_params=expected_extra_params,
            timeout=timeout,
            retry=retry,
        )

    def test_get_hmac_key_metadata_wo_project(self):
        from google.cloud.storage.hmac_key import HMACKeyMetadata

        PROJECT = "PROJECT"
        EMAIL = "storage-user-123@example.com"
        ACCESS_ID = "ACCESS-ID"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)

        resource = {
            "kind": "storage#hmacKeyMetadata",
            "accessId": ACCESS_ID,
            "projectId": PROJECT,
            "serviceAccountEmail": EMAIL,
        }

        http = _make_requests_session([_make_json_response(resource)])
        client._http_internal = http

        metadata = client.get_hmac_key_metadata(ACCESS_ID, timeout=42)

        self.assertIsInstance(metadata, HMACKeyMetadata)
        self.assertIs(metadata._client, client)
        self.assertEqual(metadata.access_id, ACCESS_ID)
        self.assertEqual(metadata.project, PROJECT)

        http.request.assert_called_once_with(
            method="GET", url=mock.ANY, data=None, headers=mock.ANY, timeout=42
        )
        _, kwargs = http.request.call_args
        scheme, netloc, path, qs, _ = urlparse.urlsplit(kwargs.get("url"))
        self.assertEqual("%s://%s" % (scheme, netloc), client._connection.API_BASE_URL)
        self.assertEqual(
            path,
            "/".join(
                [
                    "",
                    "storage",
                    client._connection.API_VERSION,
                    "projects",
                    PROJECT,
                    "hmacKeys",
                    ACCESS_ID,
                ]
            ),
        )

    def test_get_hmac_key_metadata_w_project(self):
        from google.cloud.storage.hmac_key import HMACKeyMetadata

        PROJECT = "PROJECT"
        OTHER_PROJECT = "other-project-456"
        EMAIL = "storage-user-123@example.com"
        ACCESS_ID = "ACCESS-ID"
        USER_PROJECT = "billed-project"
        CREDENTIALS = _make_credentials()
        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)

        resource = {
            "kind": "storage#hmacKeyMetadata",
            "accessId": ACCESS_ID,
            "projectId": OTHER_PROJECT,
            "serviceAccountEmail": EMAIL,
        }

        http = _make_requests_session([_make_json_response(resource)])
        client._http_internal = http

        metadata = client.get_hmac_key_metadata(
            ACCESS_ID, project_id=OTHER_PROJECT, user_project=USER_PROJECT
        )

        self.assertIsInstance(metadata, HMACKeyMetadata)
        self.assertIs(metadata._client, client)
        self.assertEqual(metadata.access_id, ACCESS_ID)
        self.assertEqual(metadata.project, OTHER_PROJECT)

        http.request.assert_called_once_with(
            method="GET",
            url=mock.ANY,
            data=None,
            headers=mock.ANY,
            timeout=self._get_default_timeout(),
        )
        _, kwargs = http.request.call_args
        scheme, netloc, path, qs, _ = urlparse.urlsplit(kwargs.get("url"))
        self.assertEqual("%s://%s" % (scheme, netloc), client._connection.API_BASE_URL)
        self.assertEqual(
            path,
            "/".join(
                [
                    "",
                    "storage",
                    client._connection.API_VERSION,
                    "projects",
                    OTHER_PROJECT,
                    "hmacKeys",
                    ACCESS_ID,
                ]
            ),
        )
        parms = dict(urlparse.parse_qsl(qs))
        self.assertEqual(parms["userProject"], USER_PROJECT)

    def test_get_signed_policy_v4(self):
        import datetime

        BUCKET_NAME = "bucket-name"
        BLOB_NAME = "object-name"
        EXPECTED_SIGN = "5369676e61747572655f6279746573"
        EXPECTED_POLICY = "eyJjb25kaXRpb25zIjpbeyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsiYWNsIjoicHJpdmF0ZSJ9LFsic3RhcnRzLXdpdGgiLCIkQ29udGVudC1UeXBlIiwidGV4dC9wbGFpbiJdLHsiYnVja2V0IjoiYnVja2V0LW5hbWUifSx7ImtleSI6Im9iamVjdC1uYW1lIn0seyJ4LWdvb2ctZGF0ZSI6IjIwMjAwMzEyVDExNDcxNloifSx7IngtZ29vZy1jcmVkZW50aWFsIjoidGVzdEBtYWlsLmNvbS8yMDIwMDMxMi9hdXRvL3N0b3JhZ2UvZ29vZzRfcmVxdWVzdCJ9LHsieC1nb29nLWFsZ29yaXRobSI6IkdPT0c0LVJTQS1TSEEyNTYifV0sImV4cGlyYXRpb24iOiIyMDIwLTAzLTI2VDAwOjAwOjEwWiJ9"

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, now_patch, expire_secs_patch = _time_functions_patches()
        with dtstamps_patch, now_patch, expire_secs_patch:
            policy = client.generate_signed_post_policy_v4(
                BUCKET_NAME,
                BLOB_NAME,
                expiration=datetime.datetime(2020, 3, 12),
                conditions=[
                    {"bucket": BUCKET_NAME},
                    {"acl": "private"},
                    ["starts-with", "$Content-Type", "text/plain"],
                ],
                credentials=_create_signing_credentials(),
            )
        self.assertEqual(
            policy["url"], "https://storage.googleapis.com/" + BUCKET_NAME + "/"
        )
        fields = policy["fields"]

        self.assertEqual(fields["key"], BLOB_NAME)
        self.assertEqual(fields["x-goog-algorithm"], "GOOG4-RSA-SHA256")
        self.assertEqual(fields["x-goog-date"], "20200312T114716Z")
        self.assertEqual(
            fields["x-goog-credential"],
            "test@mail.com/20200312/auto/storage/goog4_request",
        )
        self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
        self.assertEqual(fields["policy"], EXPECTED_POLICY)

    def test_get_signed_policy_v4_without_credentials(self):
        import datetime

        BUCKET_NAME = "bucket-name"
        BLOB_NAME = "object-name"
        EXPECTED_SIGN = "5369676e61747572655f6279746573"
        EXPECTED_POLICY = "eyJjb25kaXRpb25zIjpbeyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsiYWNsIjoicHJpdmF0ZSJ9LFsic3RhcnRzLXdpdGgiLCIkQ29udGVudC1UeXBlIiwidGV4dC9wbGFpbiJdLHsiYnVja2V0IjoiYnVja2V0LW5hbWUifSx7ImtleSI6Im9iamVjdC1uYW1lIn0seyJ4LWdvb2ctZGF0ZSI6IjIwMjAwMzEyVDExNDcxNloifSx7IngtZ29vZy1jcmVkZW50aWFsIjoidGVzdEBtYWlsLmNvbS8yMDIwMDMxMi9hdXRvL3N0b3JhZ2UvZ29vZzRfcmVxdWVzdCJ9LHsieC1nb29nLWFsZ29yaXRobSI6IkdPT0c0LVJTQS1TSEEyNTYifV0sImV4cGlyYXRpb24iOiIyMDIwLTAzLTI2VDAwOjAwOjEwWiJ9"

        client = self._make_one(
            project="PROJECT", credentials=_create_signing_credentials()
        )

        dtstamps_patch, now_patch, expire_secs_patch = _time_functions_patches()
        with dtstamps_patch, now_patch, expire_secs_patch:
            policy = client.generate_signed_post_policy_v4(
                BUCKET_NAME,
                BLOB_NAME,
                expiration=datetime.datetime(2020, 3, 12),
                conditions=[
                    {"bucket": BUCKET_NAME},
                    {"acl": "private"},
                    ["starts-with", "$Content-Type", "text/plain"],
                ],
            )
        self.assertEqual(
            policy["url"], "https://storage.googleapis.com/" + BUCKET_NAME + "/"
        )
        fields = policy["fields"]

        self.assertEqual(fields["key"], BLOB_NAME)
        self.assertEqual(fields["x-goog-algorithm"], "GOOG4-RSA-SHA256")
        self.assertEqual(fields["x-goog-date"], "20200312T114716Z")
        self.assertEqual(
            fields["x-goog-credential"],
            "test@mail.com/20200312/auto/storage/goog4_request",
        )
        self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
        self.assertEqual(fields["policy"], EXPECTED_POLICY)

    def test_get_signed_policy_v4_with_fields(self):
        import datetime

        BUCKET_NAME = "bucket-name"
        BLOB_NAME = "object-name"
        FIELD1_VALUE = "Value1"
        EXPECTED_SIGN = "5369676e61747572655f6279746573"
        EXPECTED_POLICY = "eyJjb25kaXRpb25zIjpbeyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsiYWNsIjoicHJpdmF0ZSJ9LFsic3RhcnRzLXdpdGgiLCIkQ29udGVudC1UeXBlIiwidGV4dC9wbGFpbiJdLHsiZmllbGQxIjoiVmFsdWUxIn0seyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsia2V5Ijoib2JqZWN0LW5hbWUifSx7IngtZ29vZy1kYXRlIjoiMjAyMDAzMTJUMTE0NzE2WiJ9LHsieC1nb29nLWNyZWRlbnRpYWwiOiJ0ZXN0QG1haWwuY29tLzIwMjAwMzEyL2F1dG8vc3RvcmFnZS9nb29nNF9yZXF1ZXN0In0seyJ4LWdvb2ctYWxnb3JpdGhtIjoiR09PRzQtUlNBLVNIQTI1NiJ9XSwiZXhwaXJhdGlvbiI6IjIwMjAtMDMtMjZUMDA6MDA6MTBaIn0="

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, now_patch, expire_secs_patch = _time_functions_patches()
        with dtstamps_patch, now_patch, expire_secs_patch:
            policy = client.generate_signed_post_policy_v4(
                BUCKET_NAME,
                BLOB_NAME,
                expiration=datetime.datetime(2020, 3, 12),
                conditions=[
                    {"bucket": BUCKET_NAME},
                    {"acl": "private"},
                    ["starts-with", "$Content-Type", "text/plain"],
                ],
                fields={"field1": FIELD1_VALUE, "x-ignore-field": "Ignored_value"},
                credentials=_create_signing_credentials(),
            )
        self.assertEqual(
            policy["url"], "https://storage.googleapis.com/" + BUCKET_NAME + "/"
        )
        fields = policy["fields"]

        self.assertEqual(fields["key"], BLOB_NAME)
        self.assertEqual(fields["x-goog-algorithm"], "GOOG4-RSA-SHA256")
        self.assertEqual(fields["x-goog-date"], "20200312T114716Z")
        self.assertEqual(fields["field1"], FIELD1_VALUE)
        self.assertNotIn("x-ignore-field", fields.keys())
        self.assertEqual(
            fields["x-goog-credential"],
            "test@mail.com/20200312/auto/storage/goog4_request",
        )
        self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
        self.assertEqual(fields["policy"], EXPECTED_POLICY)

    def test_get_signed_policy_v4_virtual_hosted_style(self):
        import datetime

        BUCKET_NAME = "bucket-name"

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, _, _ = _time_functions_patches()
        with dtstamps_patch:
            policy = client.generate_signed_post_policy_v4(
                BUCKET_NAME,
                "object-name",
                expiration=datetime.datetime(2020, 3, 12),
                virtual_hosted_style=True,
                credentials=_create_signing_credentials(),
            )
        self.assertEqual(
            policy["url"], "https://{}.storage.googleapis.com/".format(BUCKET_NAME)
        )

    def test_get_signed_policy_v4_bucket_bound_hostname(self):
        import datetime

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, _, _ = _time_functions_patches()
        with dtstamps_patch:
            policy = client.generate_signed_post_policy_v4(
                "bucket-name",
                "object-name",
                expiration=datetime.datetime(2020, 3, 12),
                bucket_bound_hostname="https://bucket.bound_hostname",
                credentials=_create_signing_credentials(),
            )
        self.assertEqual(policy["url"], "https://bucket.bound_hostname")

    def test_get_signed_policy_v4_bucket_bound_hostname_with_scheme(self):
        import datetime

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, _, _ = _time_functions_patches()
        with dtstamps_patch:
            policy = client.generate_signed_post_policy_v4(
                "bucket-name",
                "object-name",
                expiration=datetime.datetime(2020, 3, 12),
                bucket_bound_hostname="bucket.bound_hostname",
                scheme="http",
                credentials=_create_signing_credentials(),
            )
        self.assertEqual(policy["url"], "http://bucket.bound_hostname/")

    def test_get_signed_policy_v4_no_expiration(self):
        BUCKET_NAME = "bucket-name"
        EXPECTED_POLICY = "eyJjb25kaXRpb25zIjpbeyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsia2V5Ijoib2JqZWN0LW5hbWUifSx7IngtZ29vZy1kYXRlIjoiMjAyMDAzMTJUMTE0NzE2WiJ9LHsieC1nb29nLWNyZWRlbnRpYWwiOiJ0ZXN0QG1haWwuY29tLzIwMjAwMzEyL2F1dG8vc3RvcmFnZS9nb29nNF9yZXF1ZXN0In0seyJ4LWdvb2ctYWxnb3JpdGhtIjoiR09PRzQtUlNBLVNIQTI1NiJ9XSwiZXhwaXJhdGlvbiI6IjIwMjAtMDMtMjZUMDA6MDA6MTBaIn0="

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, now_patch, expire_secs_patch = _time_functions_patches()
        with dtstamps_patch, now_patch, expire_secs_patch:
            policy = client.generate_signed_post_policy_v4(
                BUCKET_NAME,
                "object-name",
                expiration=None,
                credentials=_create_signing_credentials(),
            )

        self.assertEqual(
            policy["url"], "https://storage.googleapis.com/" + BUCKET_NAME + "/"
        )
        self.assertEqual(policy["fields"]["policy"], EXPECTED_POLICY)

    def test_get_signed_policy_v4_with_access_token(self):
        import datetime

        BUCKET_NAME = "bucket-name"
        BLOB_NAME = "object-name"
        EXPECTED_SIGN = "0c4003044105"
        EXPECTED_POLICY = "eyJjb25kaXRpb25zIjpbeyJidWNrZXQiOiJidWNrZXQtbmFtZSJ9LHsiYWNsIjoicHJpdmF0ZSJ9LFsic3RhcnRzLXdpdGgiLCIkQ29udGVudC1UeXBlIiwidGV4dC9wbGFpbiJdLHsiYnVja2V0IjoiYnVja2V0LW5hbWUifSx7ImtleSI6Im9iamVjdC1uYW1lIn0seyJ4LWdvb2ctZGF0ZSI6IjIwMjAwMzEyVDExNDcxNloifSx7IngtZ29vZy1jcmVkZW50aWFsIjoidGVzdEBtYWlsLmNvbS8yMDIwMDMxMi9hdXRvL3N0b3JhZ2UvZ29vZzRfcmVxdWVzdCJ9LHsieC1nb29nLWFsZ29yaXRobSI6IkdPT0c0LVJTQS1TSEEyNTYifV0sImV4cGlyYXRpb24iOiIyMDIwLTAzLTI2VDAwOjAwOjEwWiJ9"

        project = "PROJECT"
        credentials = _make_credentials(project=project)
        client = self._make_one(credentials=credentials)

        dtstamps_patch, now_patch, expire_secs_patch = _time_functions_patches()
        with dtstamps_patch, now_patch, expire_secs_patch:
            with mock.patch(
                "google.cloud.storage.client._sign_message", return_value=b"DEADBEEF"
            ):
                policy = client.generate_signed_post_policy_v4(
                    BUCKET_NAME,
                    BLOB_NAME,
                    expiration=datetime.datetime(2020, 3, 12),
                    conditions=[
                        {"bucket": BUCKET_NAME},
                        {"acl": "private"},
                        ["starts-with", "$Content-Type", "text/plain"],
                    ],
                    credentials=_create_signing_credentials(),
                    service_account_email="test@mail.com",
                    access_token="token",
                )
        self.assertEqual(
            policy["url"], "https://storage.googleapis.com/" + BUCKET_NAME + "/"
        )
        fields = policy["fields"]

        self.assertEqual(fields["key"], BLOB_NAME)
        self.assertEqual(fields["x-goog-algorithm"], "GOOG4-RSA-SHA256")
        self.assertEqual(fields["x-goog-date"], "20200312T114716Z")
        self.assertEqual(
            fields["x-goog-credential"],
            "test@mail.com/20200312/auto/storage/goog4_request",
        )
        self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
        self.assertEqual(fields["policy"], EXPECTED_POLICY)


class Test__item_to_bucket(unittest.TestCase):
    def _call_fut(self, iterator, item):
        from google.cloud.storage.client import _item_to_bucket

        return _item_to_bucket(iterator, item)

    def test_w_empty_item(self):
        from google.cloud.storage.bucket import Bucket

        iterator = mock.Mock(spec=["client"])
        item = {}

        bucket = self._call_fut(iterator, item)

        self.assertIsInstance(bucket, Bucket)
        self.assertIs(bucket.client, iterator.client)
        self.assertIsNone(bucket.name)

    def test_w_name(self):
        from google.cloud.storage.bucket import Bucket

        name = "name"
        iterator = mock.Mock(spec=["client"])
        item = {"name": name}

        bucket = self._call_fut(iterator, item)

        self.assertIsInstance(bucket, Bucket)
        self.assertIs(bucket.client, iterator.client)
        self.assertEqual(bucket.name, name)


class Test__item_to_hmac_key_metadata(unittest.TestCase):
    def _call_fut(self, iterator, item):
        from google.cloud.storage.client import _item_to_hmac_key_metadata

        return _item_to_hmac_key_metadata(iterator, item)

    def test_it(self):
        from google.cloud.storage.hmac_key import HMACKeyMetadata

        access_id = "ABCDE"
        iterator = mock.Mock(spec=["client"])
        item = {"id": access_id}

        metadata = self._call_fut(iterator, item)

        self.assertIsInstance(metadata, HMACKeyMetadata)
        self.assertIs(metadata._client, iterator.client)
        self.assertEqual(metadata._properties, item)


@pytest.mark.parametrize("test_data", _POST_POLICY_TESTS)
def test_conformance_post_policy(test_data):
    import datetime
    from google.cloud.storage.client import Client

    in_data = test_data["policyInput"]
    timestamp = datetime.datetime.strptime(in_data["timestamp"], "%Y-%m-%dT%H:%M:%SZ")

    client = Client(credentials=_FAKE_CREDENTIALS, project="PROJECT")

    # mocking time functions
    with mock.patch("google.cloud.storage._signing.NOW", return_value=timestamp):
        with mock.patch(
            "google.cloud.storage.client.get_expiration_seconds_v4",
            return_value=in_data["expiration"],
        ):
            with mock.patch("google.cloud.storage.client._NOW", return_value=timestamp):

                policy = client.generate_signed_post_policy_v4(
                    bucket_name=in_data["bucket"],
                    blob_name=in_data["object"],
                    conditions=_prepare_conditions(in_data),
                    fields=in_data.get("fields"),
                    credentials=_FAKE_CREDENTIALS,
                    expiration=in_data["expiration"],
                    virtual_hosted_style=in_data.get("urlStyle")
                    == "VIRTUAL_HOSTED_STYLE",
                    bucket_bound_hostname=in_data.get("bucketBoundHostname"),
                    scheme=in_data.get("scheme"),
                )
    fields = policy["fields"]
    out_data = test_data["policyOutput"]

    decoded_policy = base64.b64decode(fields["policy"]).decode("unicode_escape")
    assert decoded_policy == out_data["expectedDecodedPolicy"]

    for field in (
        "x-goog-algorithm",
        "x-goog-credential",
        "x-goog-date",
        "x-goog-signature",
    ):
        assert fields[field] == test_data["policyOutput"]["fields"][field]

    assert policy["url"] == out_data["url"]


def _prepare_conditions(in_data):
    """Helper for V4 POST policy generation conformance tests.

    Convert conformance test data conditions dict into list.

    Args:
        in_data (dict): conditions arg from conformance test data.

    Returns:
        list: conditions arg to pass into generate_signed_post_policy_v4().
    """
    if "conditions" in in_data:
        conditions = []
        for key, value in in_data["conditions"].items():
            # camel case to snake case with "-" separator
            field = re.sub(r"(?<!^)(?=[A-Z])", "-", key).lower()
            conditions.append([field] + value)

        return conditions


def _time_functions_patches():
    """Helper for POST policy generation - returns all necessary time functions patches."""
    import datetime

    dtstamps_patch = mock.patch(
        "google.cloud.storage.client.get_v4_now_dtstamps",
        return_value=("20200312T114716Z", "20200312"),
    )
    now_patch = mock.patch(
        "google.cloud.storage.client._NOW", return_value=datetime.datetime(2020, 3, 26)
    )
    expire_secs_patch = mock.patch(
        "google.cloud.storage.client.get_expiration_seconds_v4", return_value=10
    )
    return dtstamps_patch, now_patch, expire_secs_patch
