test_http.py

#!/usr/bin/env python
#
# Copyright 2014 Google Inc. All Rights Reserved.
#
# 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.

"""Http tests

Unit tests for the googleapiclient.http.
"""
from __future__ import absolute_import

__author__ = "jcgregorio@google.com (Joe Gregorio)"

import io
from io import FileIO

# Do not remove the httplib2 import
import json
import logging
import os
import random
import socket
import ssl
import time
import unittest
from unittest import mock
import urllib

import httplib2

from googleapiclient.discovery import build
from googleapiclient.errors import BatchError, HttpError, InvalidChunkSizeError
from googleapiclient.http import (
    MAX_URI_LENGTH,
    BatchHttpRequest,
    HttpMock,
    HttpMockSequence,
    HttpRequest,
    MediaFileUpload,
    MediaInMemoryUpload,
    MediaIoBaseDownload,
    MediaIoBaseUpload,
    MediaUpload,
    _StreamSlice,
    build_http,
    set_user_agent,
)
from googleapiclient.model import JsonModel


class MockCredentials:
    """Mock class for all Credentials objects."""

    def __init__(self, bearer_token, expired=False):
        super(MockCredentials, self).__init__()
        self._authorized = 0
        self._refreshed = 0
        self._applied = 0
        self._bearer_token = bearer_token
        self._access_token_expired = expired

    @property
    def access_token(self):
        return self._bearer_token

    @property
    def access_token_expired(self):
        return self._access_token_expired

    def authorize(self, http):
        self._authorized += 1

        request_orig = http.request

        # The closure that will replace 'httplib2.Http.request'.
        def new_request(
            uri,
            method="GET",
            body=None,
            headers=None,
            redirections=httplib2.DEFAULT_MAX_REDIRECTS,
            connection_type=None,
        ):
            # Modify the request headers to add the appropriate
            # Authorization header.
            if headers is None:
                headers = {}
            self.apply(headers)

            resp, content = request_orig(
                uri, method, body, headers, redirections, connection_type
            )

            return resp, content

        # Replace the request method with our own closure.
        http.request = new_request

        # Set credentials as a property of the request method.
        setattr(http.request, "credentials", self)

        return http

    def refresh(self, http):
        self._refreshed += 1

    def apply(self, headers):
        self._applied += 1
        headers["authorization"] = self._bearer_token + " " + str(self._refreshed)


class HttpMockWithErrors(object):
    def __init__(self, num_errors, success_json, success_data):
        self.num_errors = num_errors
        self.success_json = success_json
        self.success_data = success_data

    def request(self, *args, **kwargs):
        if not self.num_errors:
            return httplib2.Response(self.success_json), self.success_data
        elif self.num_errors == 5:
            ex = ConnectionResetError  # noqa: F821
        elif self.num_errors == 4:
            ex = httplib2.ServerNotFoundError()
        elif self.num_errors == 3:
            ex = OSError()
            ex.errno = socket.errno.EPIPE
        elif self.num_errors == 2:
            ex = ssl.SSLError()
        else:
            # Initialize the timeout error code to the platform's error code.
            try:
                # For Windows:
                ex = OSError()
                ex.errno = socket.errno.WSAETIMEDOUT
            except AttributeError:
                # For Linux/Mac:
                ex = socket.timeout()

        self.num_errors -= 1
        raise ex


class HttpMockWithNonRetriableErrors(object):
    def __init__(self, num_errors, success_json, success_data):
        self.num_errors = num_errors
        self.success_json = success_json
        self.success_data = success_data

    def request(self, *args, **kwargs):
        if not self.num_errors:
            return httplib2.Response(self.success_json), self.success_data
        else:
            self.num_errors -= 1
            ex = OSError()
            # set errno to a non-retriable value
            try:
                # For Windows:
                ex.errno = socket.errno.WSAEHOSTUNREACH
            except AttributeError:
                # For Linux/Mac:
                ex.errno = socket.errno.EHOSTUNREACH
            # Now raise the correct timeout error.
            raise ex


DATA_DIR = os.path.join(os.path.dirname(__file__), "data")


def datafile(filename):
    return os.path.join(DATA_DIR, filename)


def _postproc_none(*kwargs):
    pass


class TestUserAgent(unittest.TestCase):
    def test_set_user_agent(self):
        http = HttpMockSequence([({"status": "200"}, "echo_request_headers")])

        http = set_user_agent(http, "my_app/5.5")
        resp, content = http.request("http://example.com")
        self.assertEqual("my_app/5.5", content["user-agent"])

    def test_set_user_agent_nested(self):
        http = HttpMockSequence([({"status": "200"}, "echo_request_headers")])

        http = set_user_agent(http, "my_app/5.5")
        http = set_user_agent(http, "my_library/0.1")
        resp, content = http.request("http://example.com")
        self.assertEqual("my_app/5.5 my_library/0.1", content["user-agent"])


class TestMediaUpload(unittest.TestCase):
    def test_media_file_upload_closes_fd_in___del__(self):
        file_desc = mock.Mock(spec=io.TextIOWrapper)
        opener = mock.mock_open(file_desc)
        with mock.patch("builtins.open", return_value=opener):
            upload = MediaFileUpload(datafile("test_close"), mimetype="text/plain")
        self.assertIs(upload.stream(), file_desc)
        del upload
        file_desc.close.assert_called_once_with()

    def test_media_file_upload_mimetype_detection(self):
        upload = MediaFileUpload(datafile("small.png"))
        self.assertEqual("image/png", upload.mimetype())

        upload = MediaFileUpload(datafile("empty"))
        self.assertEqual("application/octet-stream", upload.mimetype())

    def test_media_file_upload_to_from_json(self):
        upload = MediaFileUpload(datafile("small.png"), chunksize=500, resumable=True)
        self.assertEqual("image/png", upload.mimetype())
        self.assertEqual(190, upload.size())
        self.assertEqual(True, upload.resumable())
        self.assertEqual(500, upload.chunksize())
        self.assertEqual(b"PNG", upload.getbytes(1, 3))

        json = upload.to_json()
        new_upload = MediaUpload.new_from_json(json)

        self.assertEqual("image/png", new_upload.mimetype())
        self.assertEqual(190, new_upload.size())
        self.assertEqual(True, new_upload.resumable())
        self.assertEqual(500, new_upload.chunksize())
        self.assertEqual(b"PNG", new_upload.getbytes(1, 3))

    def test_media_file_upload_raises_on_file_not_found(self):
        with self.assertRaises(FileNotFoundError):
            MediaFileUpload(datafile("missing.png"))

    def test_media_file_upload_raises_on_invalid_chunksize(self):
        self.assertRaises(
            InvalidChunkSizeError,
            MediaFileUpload,
            datafile("small.png"),
            mimetype="image/png",
            chunksize=-2,
            resumable=True,
        )

    def test_media_inmemory_upload(self):
        media = MediaInMemoryUpload(
            b"abcdef", mimetype="text/plain", chunksize=10, resumable=True
        )
        self.assertEqual("text/plain", media.mimetype())
        self.assertEqual(10, media.chunksize())
        self.assertTrue(media.resumable())
        self.assertEqual(b"bc", media.getbytes(1, 2))
        self.assertEqual(6, media.size())

    def test_http_request_to_from_json(self):
        http = build_http()
        media_upload = MediaFileUpload(
            datafile("small.png"), chunksize=500, resumable=True
        )
        req = HttpRequest(
            http,
            _postproc_none,
            "http://example.com",
            method="POST",
            body="{}",
            headers={"content-type": 'multipart/related; boundary="---flubber"'},
            methodId="foo",
            resumable=media_upload,
        )

        json = req.to_json()
        new_req = HttpRequest.from_json(json, http, _postproc_none)

        self.assertEqual(
            {"content-type": 'multipart/related; boundary="---flubber"'},
            new_req.headers,
        )
        self.assertEqual("http://example.com", new_req.uri)
        self.assertEqual("{}", new_req.body)
        self.assertEqual(http, new_req.http)
        self.assertEqual(media_upload.to_json(), new_req.resumable.to_json())

        self.assertEqual(random.random, new_req._rand)
        self.assertEqual(time.sleep, new_req._sleep)


class TestMediaIoBaseUpload(unittest.TestCase):
    def test_media_io_base_upload_from_file_io(self):
        fd = FileIO(datafile("small.png"), "r")
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )
        self.assertEqual("image/png", upload.mimetype())
        self.assertEqual(190, upload.size())
        self.assertEqual(True, upload.resumable())
        self.assertEqual(500, upload.chunksize())
        self.assertEqual(b"PNG", upload.getbytes(1, 3))

    def test_media_io_base_upload_from_file_object(self):
        f = open(datafile("small.png"), "rb")
        upload = MediaIoBaseUpload(
            fd=f, mimetype="image/png", chunksize=500, resumable=True
        )
        self.assertEqual("image/png", upload.mimetype())
        self.assertEqual(190, upload.size())
        self.assertEqual(True, upload.resumable())
        self.assertEqual(500, upload.chunksize())
        self.assertEqual(b"PNG", upload.getbytes(1, 3))
        f.close()

    def test_media_io_base_upload_serializable(self):
        f = open(datafile("small.png"), "rb")
        upload = MediaIoBaseUpload(fd=f, mimetype="image/png")

        try:
            json = upload.to_json()
            self.fail("MediaIoBaseUpload should not be serializable.")
        except NotImplementedError:
            pass

    def test_media_io_base_upload_from_bytes(self):
        f = open(datafile("small.png"), "rb")
        fd = io.BytesIO(f.read())
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )
        self.assertEqual("image/png", upload.mimetype())
        self.assertEqual(190, upload.size())
        self.assertEqual(True, upload.resumable())
        self.assertEqual(500, upload.chunksize())
        self.assertEqual(b"PNG", upload.getbytes(1, 3))

    def test_media_io_base_upload_raises_on_invalid_chunksize(self):
        f = open(datafile("small.png"), "rb")
        fd = io.BytesIO(f.read())
        self.assertRaises(
            InvalidChunkSizeError,
            MediaIoBaseUpload,
            fd,
            "image/png",
            chunksize=-2,
            resumable=True,
        )

    def test_media_io_base_upload_streamable(self):
        fd = io.BytesIO(b"stuff")
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )
        self.assertEqual(True, upload.has_stream())
        self.assertEqual(fd, upload.stream())

    def test_media_io_base_next_chunk_retries(self):
        f = open(datafile("small.png"), "rb")
        fd = io.BytesIO(f.read())
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )

        # Simulate errors for both the request that creates the resumable upload
        # and the upload itself.
        http = HttpMockSequence(
            [
                ({"status": "500"}, ""),
                ({"status": "500"}, ""),
                ({"status": "503"}, ""),
                ({"status": "200", "location": "location"}, ""),
                ({"status": "403"}, USER_RATE_LIMIT_EXCEEDED_RESPONSE_NO_STATUS),
                ({"status": "403"}, RATE_LIMIT_EXCEEDED_RESPONSE),
                ({"status": "429"}, ""),
                ({"status": "200"}, "{}"),
            ]
        )

        model = JsonModel()
        uri = "https://www.googleapis.com/someapi/v1/upload/?foo=bar"
        method = "POST"
        request = HttpRequest(
            http, model.response, uri, method=method, headers={}, resumable=upload
        )

        sleeptimes = []
        request._sleep = lambda x: sleeptimes.append(x)
        request._rand = lambda: 10

        request.execute(num_retries=3)
        self.assertEqual([20, 40, 80, 20, 40, 80], sleeptimes)

    def test_media_io_base_next_chunk_no_retry_403_not_configured(self):
        fd = io.BytesIO(b"i am png")
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )

        http = HttpMockSequence(
            [({"status": "403"}, NOT_CONFIGURED_RESPONSE), ({"status": "200"}, "{}")]
        )

        model = JsonModel()
        uri = "https://www.googleapis.com/someapi/v1/upload/?foo=bar"
        method = "POST"
        request = HttpRequest(
            http, model.response, uri, method=method, headers={}, resumable=upload
        )

        request._rand = lambda: 1.0
        request._sleep = mock.MagicMock()

        with self.assertRaises(HttpError):
            request.execute(num_retries=3)
        request._sleep.assert_not_called()

    def test_media_io_base_empty_file(self):
        fd = io.BytesIO()
        upload = MediaIoBaseUpload(
            fd=fd, mimetype="image/png", chunksize=500, resumable=True
        )

        http = HttpMockSequence(
            [
                (
                    {
                        "status": "200",
                        "location": "https://www.googleapis.com/someapi/v1/upload?foo=bar",
                    },
                    "{}",
                ),
                (
                    {
                        "status": "200",
                        "location": "https://www.googleapis.com/someapi/v1/upload?foo=bar",
                    },
                    "{}",
                ),
            ]
        )

        model = JsonModel()
        uri = "https://www.googleapis.com/someapi/v1/upload/?foo=bar"
        method = "POST"
        request = HttpRequest(
            http, model.response, uri, method=method, headers={}, resumable=upload
        )

        request.execute()

        # Check that "Content-Range" header is not set in the PUT request
        self.assertTrue("Content-Range" not in http.request_sequence[-1][-1])
        self.assertEqual("0", http.request_sequence[-1][-1]["Content-Length"])


class TestMediaIoBaseDownload(unittest.TestCase):
    def setUp(self):
        http = HttpMock(datafile("zoo.json"), {"status": "200"})
        zoo = build("zoo", "v1", http=http, static_discovery=False)
        self.request = zoo.animals().get_media(name="Lion")
        self.fd = io.BytesIO()

    def test_media_io_base_download(self):
        self.request.http = HttpMockSequence(
            [
                ({"status": "200", "content-range": "0-2/5"}, b"123"),
                ({"status": "200", "content-range": "3-4/5"}, b"45"),
            ]
        )
        self.assertEqual(True, self.request.http.follow_redirects)

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(self.fd, download._fd)
        self.assertEqual(3, download._chunksize)
        self.assertEqual(0, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(False, download._done)
        self.assertEqual(self.request.uri, download._uri)

        status, done = download.next_chunk()

        self.assertEqual(self.fd.getvalue(), b"123")
        self.assertEqual(False, done)
        self.assertEqual(3, download._progress)
        self.assertEqual(5, download._total_size)
        self.assertEqual(3, status.resumable_progress)

        status, done = download.next_chunk()

        self.assertEqual(self.fd.getvalue(), b"12345")
        self.assertEqual(True, done)
        self.assertEqual(5, download._progress)
        self.assertEqual(5, download._total_size)

    def test_media_io_base_download_range_request_header(self):
        self.request.http = HttpMockSequence(
            [
                (
                    {"status": "200", "content-range": "0-2/5"},
                    "echo_request_headers_as_json",
                ),
            ]
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        status, done = download.next_chunk()
        result = json.loads(self.fd.getvalue().decode("utf-8"))

        self.assertEqual(result.get("range"), "bytes=0-2")

    def test_media_io_base_download_custom_request_headers(self):
        self.request.http = HttpMockSequence(
            [
                (
                    {"status": "200", "content-range": "0-2/5"},
                    "echo_request_headers_as_json",
                ),
                (
                    {"status": "200", "content-range": "3-4/5"},
                    "echo_request_headers_as_json",
                ),
            ]
        )
        self.assertEqual(True, self.request.http.follow_redirects)

        self.request.headers["Cache-Control"] = "no-store"

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(download._headers.get("Cache-Control"), "no-store")

        status, done = download.next_chunk()

        result = json.loads(self.fd.getvalue().decode("utf-8"))

        # assert that that the header we added to the original request is
        # sent up to the server on each call to next_chunk

        self.assertEqual(result.get("Cache-Control"), "no-store")

        download._fd = self.fd = io.BytesIO()
        status, done = download.next_chunk()

        result = json.loads(self.fd.getvalue().decode("utf-8"))
        self.assertEqual(result.get("Cache-Control"), "no-store")

    def test_media_io_base_download_handle_redirects(self):
        self.request.http = HttpMockSequence(
            [
                (
                    {
                        "status": "200",
                        "content-location": "https://secure.example.net/lion",
                    },
                    b"",
                ),
                ({"status": "200", "content-range": "0-2/5"}, b"abc"),
            ]
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        status, done = download.next_chunk()

        self.assertEqual("https://secure.example.net/lion", download._uri)

    def test_media_io_base_download_handle_4xx(self):
        self.request.http = HttpMockSequence([({"status": "400"}, "")])

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        try:
            status, done = download.next_chunk()
            self.fail("Should raise an exception")
        except HttpError:
            pass

        # Even after raising an exception we can pick up where we left off.
        self.request.http = HttpMockSequence(
            [({"status": "200", "content-range": "0-2/5"}, b"123")]
        )

        status, done = download.next_chunk()

        self.assertEqual(self.fd.getvalue(), b"123")

    def test_media_io_base_download_retries_connection_errors(self):
        self.request.http = HttpMockWithErrors(
            5, {"status": "200", "content-range": "0-2/3"}, b"123"
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)
        download._sleep = lambda _x: 0  # do nothing
        download._rand = lambda: 10

        status, done = download.next_chunk(num_retries=5)

        self.assertEqual(self.fd.getvalue(), b"123")
        self.assertEqual(True, done)

    def test_media_io_base_download_retries_5xx(self):
        self.request.http = HttpMockSequence(
            [
                ({"status": "500"}, ""),
                ({"status": "500"}, ""),
                ({"status": "500"}, ""),
                ({"status": "200", "content-range": "0-2/5"}, b"123"),
                ({"status": "503"}, ""),
                ({"status": "503"}, ""),
                ({"status": "503"}, ""),
                ({"status": "200", "content-range": "3-4/5"}, b"45"),
            ]
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(self.fd, download._fd)
        self.assertEqual(3, download._chunksize)
        self.assertEqual(0, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(False, download._done)
        self.assertEqual(self.request.uri, download._uri)

        # Set time.sleep and random.random stubs.
        sleeptimes = []
        download._sleep = lambda x: sleeptimes.append(x)
        download._rand = lambda: 10

        status, done = download.next_chunk(num_retries=3)

        # Check for exponential backoff using the rand function above.
        self.assertEqual([20, 40, 80], sleeptimes)

        self.assertEqual(self.fd.getvalue(), b"123")
        self.assertEqual(False, done)
        self.assertEqual(3, download._progress)
        self.assertEqual(5, download._total_size)
        self.assertEqual(3, status.resumable_progress)

        # Reset time.sleep stub.
        del sleeptimes[0 : len(sleeptimes)]

        status, done = download.next_chunk(num_retries=3)

        # Check for exponential backoff using the rand function above.
        self.assertEqual([20, 40, 80], sleeptimes)

        self.assertEqual(self.fd.getvalue(), b"12345")
        self.assertEqual(True, done)
        self.assertEqual(5, download._progress)
        self.assertEqual(5, download._total_size)

    def test_media_io_base_download_empty_file(self):
        self.request.http = HttpMockSequence(
            [({"status": "200", "content-range": "0-0/0"}, b"")]
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(self.fd, download._fd)
        self.assertEqual(0, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(False, download._done)
        self.assertEqual(self.request.uri, download._uri)

        status, done = download.next_chunk()

        self.assertEqual(True, done)
        self.assertEqual(0, download._progress)
        self.assertEqual(0, download._total_size)
        self.assertEqual(0, status.progress())

    def test_media_io_base_download_empty_file_416_response(self):
        self.request.http = HttpMockSequence(
            [({"status": "416", "content-range": "0-0/0"}, b"")]
        )

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(self.fd, download._fd)
        self.assertEqual(0, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(False, download._done)
        self.assertEqual(self.request.uri, download._uri)

        status, done = download.next_chunk()

        self.assertEqual(True, done)
        self.assertEqual(0, download._progress)
        self.assertEqual(0, download._total_size)
        self.assertEqual(0, status.progress())

    def test_media_io_base_download_unknown_media_size(self):
        self.request.http = HttpMockSequence([({"status": "200"}, b"123")])

        download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)

        self.assertEqual(self.fd, download._fd)
        self.assertEqual(0, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(False, download._done)
        self.assertEqual(self.request.uri, download._uri)

        status, done = download.next_chunk()

        self.assertEqual(self.fd.getvalue(), b"123")
        self.assertEqual(True, done)
        self.assertEqual(3, download._progress)
        self.assertEqual(None, download._total_size)
        self.assertEqual(0, status.progress())


EXPECTED = """POST /someapi/v1/collection/?foo=bar HTTP/1.1
Content-Type: application/json
MIME-Version: 1.0
Host: www.googleapis.com
content-length: 2\r\n\r\n{}"""


NO_BODY_EXPECTED = """POST /someapi/v1/collection/?foo=bar HTTP/1.1
Content-Type: application/json
MIME-Version: 1.0
Host: www.googleapis.com
content-length: 0\r\n\r\n"""

NO_BODY_EXPECTED_GET = """GET /someapi/v1/collection/?foo=bar HTTP/1.1
Content-Type: application/json
MIME-Version: 1.0
Host: www.googleapis.com\r\n\r\n"""


RESPONSE = """HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/pony"\r\n\r\n{"answer": 42}"""


BATCH_RESPONSE = b"""--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 1>

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/pony"\r\n\r\n{"foo": 42}

--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 2>

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/sheep"\r\n\r\n{"baz": "qux"}
--batch_foobarbaz--"""


BATCH_ERROR_RESPONSE = b"""--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 1>

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/pony"\r\n\r\n{"foo": 42}

--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 2>

HTTP/1.1 403 Access Not Configured
Content-Type: application/json
Content-Length: 245
ETag: "etag/sheep"\r\n\r\n{
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "accessNotConfigured",
    "message": "Access Not Configured",
    "debugInfo": "QuotaState: BLOCKED"
   }
  ],
  "code": 403,
  "message": "Access Not Configured"
 }
}

--batch_foobarbaz--"""


BATCH_RESPONSE_WITH_401 = b"""--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 1>

HTTP/1.1 401 Authorization Required
Content-Type: application/json
Content-Length: 14
ETag: "etag/pony"\r\n\r\n{"error": {"message":
  "Authorizaton failed."}}

--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 2>

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/sheep"\r\n\r\n{"baz": "qux"}
--batch_foobarbaz--"""


BATCH_SINGLE_RESPONSE = b"""--batch_foobarbaz
Content-Type: application/http
Content-Transfer-Encoding: binary
Content-ID: <randomness + 1>

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 14
ETag: "etag/pony"\r\n\r\n{"foo": 42}
--batch_foobarbaz--"""


USER_RATE_LIMIT_EXCEEDED_RESPONSE_NO_STATUS = """{
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "userRateLimitExceeded",
    "message": "User Rate Limit Exceeded"
   }
  ],
  "code": 403,
  "message": "User Rate Limit Exceeded"
 }
}"""

USER_RATE_LIMIT_EXCEEDED_RESPONSE_WITH_STATUS = """{
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "userRateLimitExceeded",
    "message": "User Rate Limit Exceeded"
   }
  ],
  "code": 403,
  "message": "User Rate Limit Exceeded",
  "status": "PERMISSION_DENIED"
 }
}"""

RATE_LIMIT_EXCEEDED_RESPONSE = """{
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "rateLimitExceeded",
    "message": "Rate Limit Exceeded"
   }
  ],
  "code": 403,
  "message": "Rate Limit Exceeded"
 }
}"""

FAILED_PRECONDITION_RESPONSE = """{
 "error": {
  "errors": [
   {
    "domain": "global",
    "reason": "failedPrecondition",
    "message": "Precondition Failed"
   }
  ],
  "code": 400,
  "message": "Precondition Failed"
 }
}"""


NOT_CONFIGURED_RESPONSE = """{
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "accessNotConfigured",
    "message": "Access Not Configured"
   }
  ],
  "code": 403,
  "message": "Access Not Configured"
 }
}"""

LIST_NOT_CONFIGURED_RESPONSE = """[
 "error": {
  "errors": [
   {
    "domain": "usageLimits",
    "reason": "accessNotConfigured",
    "message": "Access Not Configured"
   }
  ],
  "code": 403,
  "message": "Access Not Configured"
 }
]"""


class Callbacks(object):
    def __init__(self):
        self.responses = {}
        self.exceptions = {}

    def f(self, request_id, response, exception):
        self.responses[request_id] = response
        self.exceptions[request_id] = exception


class TestHttpRequest(unittest.TestCase):
    def test_unicode(self):
        http = HttpMock(datafile("zoo.json"), headers={"status": "200"})
        model = JsonModel()
        uri = "https://www.googleapis.com/someapi/v1/collection/?foo=bar"
        method = "POST"
        request = HttpRequest(
            http,
            model.response,
            uri,
            method=method,
            body="{}",
            headers={"content-type": "application/json"},
        )
        request.execute()
        self.assertEqual(uri, http.uri)
        self.assertEqual(str, type(http.uri))
        self.assertEqual(method, http.method)
        self.assertEqual(str, type(http.method))

    def test_empty_content_type(self):
        """Test for #284"""
        http = HttpMock(None, headers={"status": 200})
        uri = "https://www.googleapis.com/someapi/v1/upload/?foo=bar"
        method = "POST"
        request = HttpRequest(
            http, _postproc_none, uri, method=method, headers={"content-type": ""}
        )
        request.execute()
        self.assertEqual("", http.headers.get("content-type"))

    def test_no_retry_connection_errors(self):
        model = JsonModel()
        request = HttpRequest(
            HttpMockWithNonRetriableErrors(1, {"status": "200"}, '{"foo": "bar"}'),
            model.response,
            "https://www.example.com/json_api_endpoint",
        )
        request._sleep = lambda _x: 0  # do nothing
        request._rand = lambda: 10
        with self.assertRaises(OSError):
            response = request.execute(num_retries=3)

    def test_retry_connection_errors_non_resumable(self):
        model = JsonModel()
        request = HttpRequest(
            HttpMockWithErrors(5, {"status": "200"}, '{"foo": "bar"}'),
            model.response,
            "https://www.example.com/json_api_endpoint",
        )
        request._sleep = lambda _x: 0  # do nothing
        request._rand = lambda: 10
        response = request.execute(num_retries=5)
        self.assertEqual({"foo": "bar"}, response)

    def test_retry_connection_errors_resumable(self):
        with open(datafile("small.png"), "rb") as small_png_file:
            small_png_fd = io.BytesIO(small_png_file.read())
        upload = MediaIoBaseUpload(
            fd=small_png_fd, mimetype="image/png", chunksize=500, resumable=True
        )
        model = JsonModel()

        request = HttpRequest(
            HttpMockWithErrors(
                5, {"status": "200", "location": "location"}, '{"foo": "bar"}'