#!/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

from six.moves import range

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

from six import PY3

from six import BytesIO, StringIO

from io import FileIO

from six.moves.urllib.parse import urlencode

# Do not remove the httplib2 import

import httplib2

import logging

import mock

import os

import unittest2 as unittest

import random

import socket

import ssl

import time

from googleapiclient.discovery import build

from googleapiclient.errors import BatchError

from googleapiclient.errors import HttpError

from googleapiclient.errors import InvalidChunkSizeError

from googleapiclient.http import BatchHttpRequest

from googleapiclient.http import HttpMock

from googleapiclient.http import HttpMockSequence

from googleapiclient.http import HttpRequest

from googleapiclient.http import MAX_URI_LENGTH

from googleapiclient.http import MediaFileUpload

from googleapiclient.http import MediaInMemoryUpload

from googleapiclient.http import MediaIoBaseDownload

from googleapiclient.http import MediaIoBaseUpload

from googleapiclient.http import MediaUpload

from googleapiclient.http import _StreamSlice

from googleapiclient.http import set_user_agent

from googleapiclient.model import JsonModel

from oauth2client.client import Credentials

class MockCredentials(Credentials):

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

def __init__(self, bearer_token):

super(MockCredentials, self).__init__()

self._authorized = 0

self._refreshed = 0

self._applied = 0

self._bearer_token = bearer_token

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

else:

self.num_errors -= 1

if self.num_errors == 1:

raise ssl.SSLError()

else:

if PY3:

ex = TimeoutError()

else:

ex = socket.error()

if self.num_errors == 2:

#first try a broken pipe error (#218)

ex.errno = socket.errno.EPIPE

else:

# Initialize the timeout error code to the platform's error code.

try:

# For Windows:

ex.errno = socket.errno.WSAETIMEDOUT

except AttributeError:

# For Linux/Mac:

ex.errno = socket.errno.ETIMEDOUT

# Now raise the correct error.

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 = socket.error()

# set errno to a non-retriable value

try:

# For Windows:

ex.errno = socket.errno.WSAECONNREFUSED

except AttributeError:

# For Linux/Mac:

ex.errno = socket.errno.ECONNREFUSED

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

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_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_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):

def _postproc(*kwargs):

pass

http = httplib2.Http()

media_upload = MediaFileUpload(

datafile('small.png'), chunksize=500, resumable=True)

req = HttpRequest(

http,

_postproc,

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

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

@unittest.skipIf(PY3, 'Strings and Bytes are different types')

def test_media_io_base_upload_from_string_io(self):

f = open(datafile('small.png'), 'rb')

fd = StringIO(f.read())

f.close()

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

f.close()

def test_media_io_base_upload_from_bytes(self):

f = open(datafile('small.png'), 'rb')

fd = 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 = BytesIO(f.read())

self.assertRaises(InvalidChunkSizeError, MediaIoBaseUpload,

fd, 'image/png', chunksize=-2, resumable=True)

def test_media_io_base_upload_streamable(self):

fd = 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 = BytesIO(f.read())

upload = MediaIoBaseUpload(

fd=fd, mimetype='image/png', chunksize=500, resumable=True)

# Simulate 5XXs 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': '500'}, ''),

({'status': '500'}, ''),

({'status': '503'}, ''),

({'status': '200'}, '{}'),

])

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/upload/?foo=bar'

method = u'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)

class TestMediaIoBaseDownload(unittest.TestCase):

def setUp(self):

http = HttpMock(datafile('zoo.json'), {'status': '200'})

zoo = build('zoo', 'v1', http=http)

self.request = zoo.animals().get_media(name='Lion')

self.fd = 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_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(

3, {'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=3)

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)

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 = """{

"error": {

"errors": [

{

"domain": "usageLimits",

"reason": "userRateLimitExceeded",

"message": "User Rate Limit Exceeded"

}

],

"code": 403,

"message": "User Rate Limit Exceeded"

}

}"""

RATE_LIMIT_EXCEEDED_RESPONSE = """{

"error": {

"errors": [

{

"domain": "usageLimits",

"reason": "rateLimitExceeded",

"message": "Rate Limit Exceeded"

}

],

"code": 403,

"message": "Rate Limit Exceeded"

}

}"""

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 = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

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_no_retry_connection_errors(self):

model = JsonModel()

request = HttpRequest(

HttpMockWithNonRetriableErrors(1, {'status': '200'}, '{"foo": "bar"}'),

model.response,

u'https://www.example.com/json_api_endpoint')

request._sleep = lambda _x: 0 # do nothing

request._rand = lambda: 10

with self.assertRaises(socket.error):

response = request.execute(num_retries=3)

def test_retry_connection_errors_non_resumable(self):

model = JsonModel()

request = HttpRequest(

HttpMockWithErrors(3, {'status': '200'}, '{"foo": "bar"}'),

model.response,

u'https://www.example.com/json_api_endpoint')

request._sleep = lambda _x: 0 # do nothing

request._rand = lambda: 10

response = request.execute(num_retries=3)

self.assertEqual({u'foo': u'bar'}, response)

def test_retry_connection_errors_resumable(self):

with open(datafile('small.png'), 'rb') as small_png_file:

small_png_fd = BytesIO(small_png_file.read())

upload = MediaIoBaseUpload(fd=small_png_fd, mimetype='image/png',

chunksize=500, resumable=True)

model = JsonModel()

request = HttpRequest(

HttpMockWithErrors(

3, {'status': '200', 'location': 'location'}, '{"foo": "bar"}'),

model.response,

u'https://www.example.com/file_upload',

method='POST',

resumable=upload)

request._sleep = lambda _x: 0 # do nothing

request._rand = lambda: 10

response = request.execute(num_retries=3)

self.assertEqual({u'foo': u'bar'}, response)

def test_retry(self):

num_retries = 5

resp_seq = [({'status': '500'}, '')] * (num_retries - 3)

resp_seq.append(({'status': '403'}, RATE_LIMIT_EXCEEDED_RESPONSE))

resp_seq.append(({'status': '403'}, USER_RATE_LIMIT_EXCEEDED_RESPONSE))

resp_seq.append(({'status': '429'}, ''))

resp_seq.append(({'status': '200'}, '{}'))

http = HttpMockSequence(resp_seq)

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

sleeptimes = []

request._sleep = lambda x: sleeptimes.append(x)

request._rand = lambda: 10

request.execute(num_retries=num_retries)

self.assertEqual(num_retries, len(sleeptimes))

for retry_num in range(num_retries):

self.assertEqual(10 * 2**(retry_num + 1), sleeptimes[retry_num])

def test_no_retry_succeeds(self):

num_retries = 5

resp_seq = [({'status': '200'}, '{}')] * (num_retries)

http = HttpMockSequence(resp_seq)

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

sleeptimes = []

request._sleep = lambda x: sleeptimes.append(x)

request._rand = lambda: 10

request.execute(num_retries=num_retries)

self.assertEqual(0, len(sleeptimes))

def test_no_retry_fails_fast(self):

http = HttpMockSequence([

({'status': '500'}, ''),

({'status': '200'}, '{}')

])

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

request._rand = lambda: 1.0

request._sleep = mock.MagicMock()

with self.assertRaises(HttpError):

request.execute()

request._sleep.assert_not_called()

def test_no_retry_403_not_configured_fails_fast(self):

http = HttpMockSequence([

({'status': '403'}, NOT_CONFIGURED_RESPONSE),

({'status': '200'}, '{}')

])

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

request._rand = lambda: 1.0

request._sleep = mock.MagicMock()

with self.assertRaises(HttpError):

request.execute()

request._sleep.assert_not_called()

def test_no_retry_403_fails_fast(self):

http = HttpMockSequence([

({'status': '403'}, ''),

({'status': '200'}, '{}')

])

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

request._rand = lambda: 1.0

request._sleep = mock.MagicMock()

with self.assertRaises(HttpError):

request.execute()

request._sleep.assert_not_called()

def test_no_retry_401_fails_fast(self):

http = HttpMockSequence([

({'status': '401'}, ''),

({'status': '200'}, '{}')

])

model = JsonModel()

uri = u'https://www.googleapis.com/someapi/v1/collection/?foo=bar'

method = u'POST'

request = HttpRequest(

http,

model.response,

uri,

method=method,

body=u'{}',

headers={'content-type': 'application/json'})

request._rand = lambda: 1.0

request._sleep = mock.MagicMock()

with self.assertRaises(HttpError):

request.execute()

request._sleep.assert_not_called()

class TestBatch(unittest.TestCase):

def setUp(self):

model = JsonModel()

self.request1 = HttpRequest(

None,

model.response,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='POST',

body='{}',

headers={'content-type': 'application/json'})

self.request2 = HttpRequest(

None,

model.response,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='GET',

body='',

headers={'content-type': 'application/json'})

def test_id_to_from_content_id_header(self):

batch = BatchHttpRequest()

self.assertEquals('12', batch._header_to_id(batch._id_to_header('12')))

def test_invalid_content_id_header(self):

batch = BatchHttpRequest()

self.assertRaises(BatchError, batch._header_to_id, '[foo+x]')

self.assertRaises(BatchError, batch._header_to_id, 'foo+1')

self.assertRaises(BatchError, batch._header_to_id, '<foo>')

def test_serialize_request(self):

batch = BatchHttpRequest()

request = HttpRequest(

None,

None,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='POST',

body=u'{}',

headers={'content-type': 'application/json'},

methodId=None,

resumable=None)

s = batch._serialize_request(request).splitlines()

self.assertEqual(EXPECTED.splitlines(), s)

def test_serialize_request_media_body(self):

batch = BatchHttpRequest()

f = open(datafile('small.png'), 'rb')

body = f.read()

f.close()

request = HttpRequest(

None,

None,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='POST',

body=body,

headers={'content-type': 'application/json'},

methodId=None,

resumable=None)

# Just testing it shouldn't raise an exception.

s = batch._serialize_request(request).splitlines()

def test_serialize_request_no_body(self):

batch = BatchHttpRequest()

request = HttpRequest(

None,

None,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='POST',

body=b'',

headers={'content-type': 'application/json'},

methodId=None,

resumable=None)

s = batch._serialize_request(request).splitlines()

self.assertEqual(NO_BODY_EXPECTED.splitlines(), s)

def test_serialize_get_request_no_body(self):

batch = BatchHttpRequest()

request = HttpRequest(

None,

None,

'https://www.googleapis.com/someapi/v1/collection/?foo=bar',

method='GET',

body=None,

headers={'content-type': 'application/json'},

methodId=None,

resumable=None)

s = batch._serialize_request(request).splitlines()

self.assertEqual(NO_BODY_EXPECTED_GET.splitlines(), s)

def test_deserialize_response(self):

batch = BatchHttpRequest()

resp, content = batch._deserialize_response(RESPONSE)

self.assertEqual(200, resp.status)

self.assertEqual('OK', resp.reason)

self.assertEqual(11, resp.version)