chore: blacken (#772)
diff --git a/tests/__init__.py b/tests/__init__.py
index d88353f..b4f509d 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -13,12 +13,12 @@
"""Test Package set up."""
from __future__ import absolute_import
-__author__ = 'afshar@google.com (Ali Afshar)'
+__author__ = "afshar@google.com (Ali Afshar)"
from googleapiclient import _helpers as util
def setup_package():
- """Run on testing package."""
- util.positional_parameters_enforcement = 'EXCEPTION'
+ """Run on testing package."""
+ util.positional_parameters_enforcement = "EXCEPTION"
diff --git a/tests/test__auth.py b/tests/test__auth.py
index d43fe68..b65ed81 100644
--- a/tests/test__auth.py
+++ b/tests/test__auth.py
@@ -33,9 +33,8 @@
_auth.HAS_OAUTH2CLIENT = True
def test_default_credentials(self):
- with mock.patch('google.auth.default', autospec=True) as default:
- default.return_value = (
- mock.sentinel.credentials, mock.sentinel.project)
+ with mock.patch("google.auth.default", autospec=True) as default:
+ default.return_value = (mock.sentinel.credentials, mock.sentinel.project)
credentials = _auth.default_credentials()
@@ -50,8 +49,8 @@
def test_with_scopes_scoped(self):
class CredentialsWithScopes(
- google.auth.credentials.Credentials,
- google.auth.credentials.Scoped):
+ google.auth.credentials.Credentials, google.auth.credentials.Scoped
+ ):
pass
credentials = mock.Mock(spec=CredentialsWithScopes)
@@ -68,9 +67,7 @@
authorized_http = _auth.authorized_http(credentials)
- self.assertIsInstance(
- authorized_http,
- google_auth_httplib2.AuthorizedHttp)
+ self.assertIsInstance(authorized_http, google_auth_httplib2.AuthorizedHttp)
self.assertEqual(authorized_http.credentials, credentials)
self.assertIsInstance(authorized_http.http, httplib2.Http)
self.assertIsInstance(authorized_http.http.timeout, int)
@@ -88,7 +85,8 @@
def test_default_credentials(self):
default_patch = mock.patch(
- 'oauth2client.client.GoogleCredentials.get_application_default')
+ "oauth2client.client.GoogleCredentials.get_application_default"
+ )
with default_patch as default:
default.return_value = mock.sentinel.credentials
@@ -128,7 +126,6 @@
class TestAuthWithoutAuth(unittest.TestCase):
-
def setUp(self):
_auth.HAS_GOOGLE_AUTH = False
_auth.HAS_OAUTH2CLIENT = False
diff --git a/tests/test__helpers.py b/tests/test__helpers.py
index e33ea71..90c75ef 100644
--- a/tests/test__helpers.py
+++ b/tests/test__helpers.py
@@ -25,10 +25,8 @@
class PositionalTests(unittest.TestCase):
-
def test_usage(self):
- _helpers.positional_parameters_enforcement = (
- _helpers.POSITIONAL_EXCEPTION)
+ _helpers.positional_parameters_enforcement = _helpers.POSITIONAL_EXCEPTION
# 1 positional arg, 1 keyword-only arg.
@_helpers.positional(1)
@@ -60,10 +58,9 @@
with self.assertRaises(TypeError):
function3(1, 2)
- @mock.patch('googleapiclient._helpers.logger')
+ @mock.patch("googleapiclient._helpers.logger")
def test_enforcement_warning(self, mock_logger):
- _helpers.positional_parameters_enforcement = (
- _helpers.POSITIONAL_WARNING)
+ _helpers.positional_parameters_enforcement = _helpers.POSITIONAL_WARNING
@_helpers.positional(1)
def function(pos, kwonly=None):
@@ -72,7 +69,7 @@
self.assertTrue(function(1, 2))
self.assertTrue(mock_logger.warning.called)
- @mock.patch('googleapiclient._helpers.logger')
+ @mock.patch("googleapiclient._helpers.logger")
def test_enforcement_ignore(self, mock_logger):
_helpers.positional_parameters_enforcement = _helpers.POSITIONAL_IGNORE
@@ -85,24 +82,22 @@
class AddQueryParameterTests(unittest.TestCase):
-
def test__add_query_parameter(self):
+ self.assertEqual(_helpers._add_query_parameter("/action", "a", None), "/action")
self.assertEqual(
- _helpers._add_query_parameter('/action', 'a', None),
- '/action')
+ _helpers._add_query_parameter("/action", "a", "b"), "/action?a=b"
+ )
self.assertEqual(
- _helpers._add_query_parameter('/action', 'a', 'b'),
- '/action?a=b')
- self.assertEqual(
- _helpers._add_query_parameter('/action?a=b', 'a', 'c'),
- '/action?a=c')
+ _helpers._add_query_parameter("/action?a=b", "a", "c"), "/action?a=c"
+ )
# Order is non-deterministic.
self.assertIn(
- _helpers._add_query_parameter('/action?a=b', 'c', 'd'),
- ['/action?a=b&c=d', '/action?c=d&a=b'])
+ _helpers._add_query_parameter("/action?a=b", "c", "d"),
+ ["/action?a=b&c=d", "/action?c=d&a=b"],
+ )
self.assertEqual(
- _helpers._add_query_parameter('/action', 'a', ' ='),
- '/action?a=+%3D')
+ _helpers._add_query_parameter("/action", "a", " ="), "/action?a=+%3D"
+ )
def assertUrisEqual(testcase, expected, actual):
@@ -123,39 +118,37 @@
class Test_update_query_params(unittest.TestCase):
-
def test_update_query_params_no_params(self):
- uri = 'http://www.google.com'
- updated = _helpers.update_query_params(uri, {'a': 'b'})
- self.assertEqual(updated, uri + '?a=b')
+ uri = "http://www.google.com"
+ updated = _helpers.update_query_params(uri, {"a": "b"})
+ self.assertEqual(updated, uri + "?a=b")
def test_update_query_params_existing_params(self):
- uri = 'http://www.google.com?x=y'
- updated = _helpers.update_query_params(uri, {'a': 'b', 'c': 'd&'})
- hardcoded_update = uri + '&a=b&c=d%26'
+ uri = "http://www.google.com?x=y"
+ updated = _helpers.update_query_params(uri, {"a": "b", "c": "d&"})
+ hardcoded_update = uri + "&a=b&c=d%26"
assertUrisEqual(self, updated, hardcoded_update)
def test_update_query_params_replace_param(self):
- base_uri = 'http://www.google.com'
- uri = base_uri + '?x=a'
- updated = _helpers.update_query_params(uri, {'x': 'b', 'y': 'c'})
- hardcoded_update = base_uri + '?x=b&y=c'
+ base_uri = "http://www.google.com"
+ uri = base_uri + "?x=a"
+ updated = _helpers.update_query_params(uri, {"x": "b", "y": "c"})
+ hardcoded_update = base_uri + "?x=b&y=c"
assertUrisEqual(self, updated, hardcoded_update)
def test_update_query_params_repeated_params(self):
- uri = 'http://www.google.com?x=a&x=b'
+ uri = "http://www.google.com?x=a&x=b"
with self.assertRaises(ValueError):
- _helpers.update_query_params(uri, {'a': 'c'})
+ _helpers.update_query_params(uri, {"a": "c"})
class Test_parse_unique_urlencoded(unittest.TestCase):
-
def test_without_repeats(self):
- content = 'a=b&c=d'
+ content = "a=b&c=d"
result = _helpers.parse_unique_urlencoded(content)
- self.assertEqual(result, {'a': 'b', 'c': 'd'})
+ self.assertEqual(result, {"a": "b", "c": "d"})
def test_with_repeats(self):
- content = 'a=b&a=d'
+ content = "a=b&a=d"
with self.assertRaises(ValueError):
_helpers.parse_unique_urlencoded(content)
diff --git a/tests/test_channel.py b/tests/test_channel.py
index 4141353..8a46bf0 100644
--- a/tests/test_channel.py
+++ b/tests/test_channel.py
@@ -1,7 +1,7 @@
"""Notification channels tests."""
from __future__ import absolute_import
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
import unittest2 as unittest
import datetime
@@ -11,115 +11,128 @@
class TestChannel(unittest.TestCase):
- def test_basic(self):
- ch = channel.Channel('web_hook', 'myid', 'mytoken',
- 'http://example.org/callback',
- expiration=0,
- params={'extra': 'info'},
- resource_id='the_resource_id',
- resource_uri='http://example.com/resource_1')
+ def test_basic(self):
+ ch = channel.Channel(
+ "web_hook",
+ "myid",
+ "mytoken",
+ "http://example.org/callback",
+ expiration=0,
+ params={"extra": "info"},
+ resource_id="the_resource_id",
+ resource_uri="http://example.com/resource_1",
+ )
- # Converting to a body.
- body = ch.body()
- self.assertEqual('http://example.org/callback', body['address'])
- self.assertEqual('myid', body['id'])
- self.assertEqual('missing', body.get('expiration', 'missing'))
- self.assertEqual('info', body['params']['extra'])
- self.assertEqual('the_resource_id', body['resourceId'])
- self.assertEqual('http://example.com/resource_1', body['resourceUri'])
- self.assertEqual('web_hook', body['type'])
+ # Converting to a body.
+ body = ch.body()
+ self.assertEqual("http://example.org/callback", body["address"])
+ self.assertEqual("myid", body["id"])
+ self.assertEqual("missing", body.get("expiration", "missing"))
+ self.assertEqual("info", body["params"]["extra"])
+ self.assertEqual("the_resource_id", body["resourceId"])
+ self.assertEqual("http://example.com/resource_1", body["resourceUri"])
+ self.assertEqual("web_hook", body["type"])
- # Converting to a body with expiration set.
- ch.expiration = 1
- body = ch.body()
- self.assertEqual(1, body.get('expiration', 'missing'))
+ # Converting to a body with expiration set.
+ ch.expiration = 1
+ body = ch.body()
+ self.assertEqual(1, body.get("expiration", "missing"))
- # Converting to a body after updating with a response body.
- ch.update({
- 'resourceId': 'updated_res_id',
- 'resourceUri': 'updated_res_uri',
- 'some_random_parameter': 2,
- })
+ # Converting to a body after updating with a response body.
+ ch.update(
+ {
+ "resourceId": "updated_res_id",
+ "resourceUri": "updated_res_uri",
+ "some_random_parameter": 2,
+ }
+ )
- body = ch.body()
- self.assertEqual('http://example.org/callback', body['address'])
- self.assertEqual('myid', body['id'])
- self.assertEqual(1, body.get('expiration', 'missing'))
- self.assertEqual('info', body['params']['extra'])
- self.assertEqual('updated_res_id', body['resourceId'])
- self.assertEqual('updated_res_uri', body['resourceUri'])
- self.assertEqual('web_hook', body['type'])
+ body = ch.body()
+ self.assertEqual("http://example.org/callback", body["address"])
+ self.assertEqual("myid", body["id"])
+ self.assertEqual(1, body.get("expiration", "missing"))
+ self.assertEqual("info", body["params"]["extra"])
+ self.assertEqual("updated_res_id", body["resourceId"])
+ self.assertEqual("updated_res_uri", body["resourceUri"])
+ self.assertEqual("web_hook", body["type"])
- def test_new_webhook_channel(self):
- ch = channel.new_webhook_channel('http://example.com/callback')
- self.assertEqual(0, ch.expiration)
- self.assertEqual('http://example.com/callback', ch.address)
- self.assertEqual(None, ch.params)
+ def test_new_webhook_channel(self):
+ ch = channel.new_webhook_channel("http://example.com/callback")
+ self.assertEqual(0, ch.expiration)
+ self.assertEqual("http://example.com/callback", ch.address)
+ self.assertEqual(None, ch.params)
- # New channel with an obviously wrong expiration time.
- ch = channel.new_webhook_channel(
- 'http://example.com/callback',
- expiration=datetime.datetime(1965, 1, 1))
- self.assertEqual(0, ch.expiration)
+ # New channel with an obviously wrong expiration time.
+ ch = channel.new_webhook_channel(
+ "http://example.com/callback", expiration=datetime.datetime(1965, 1, 1)
+ )
+ self.assertEqual(0, ch.expiration)
- # New channel with an expiration time.
- ch = channel.new_webhook_channel(
- 'http://example.com/callback',
- expiration=datetime.datetime(1970, 1, 1, second=5))
- self.assertEqual(5000, ch.expiration)
- self.assertEqual('http://example.com/callback', ch.address)
- self.assertEqual(None, ch.params)
+ # New channel with an expiration time.
+ ch = channel.new_webhook_channel(
+ "http://example.com/callback",
+ expiration=datetime.datetime(1970, 1, 1, second=5),
+ )
+ self.assertEqual(5000, ch.expiration)
+ self.assertEqual("http://example.com/callback", ch.address)
+ self.assertEqual(None, ch.params)
- # New channel with an expiration time and params.
- ch = channel.new_webhook_channel(
- 'http://example.com/callback',
- expiration=datetime.datetime(1970, 1, 1, second=5, microsecond=1000),
- params={'some':'stuff'})
- self.assertEqual(5001, ch.expiration)
- self.assertEqual('http://example.com/callback', ch.address)
- self.assertEqual({'some': 'stuff'}, ch.params)
+ # New channel with an expiration time and params.
+ ch = channel.new_webhook_channel(
+ "http://example.com/callback",
+ expiration=datetime.datetime(1970, 1, 1, second=5, microsecond=1000),
+ params={"some": "stuff"},
+ )
+ self.assertEqual(5001, ch.expiration)
+ self.assertEqual("http://example.com/callback", ch.address)
+ self.assertEqual({"some": "stuff"}, ch.params)
class TestNotification(unittest.TestCase):
- def test_basic(self):
- n = channel.Notification(12, 'sync', 'http://example.org',
- 'http://example.org/v1')
+ def test_basic(self):
+ n = channel.Notification(
+ 12, "sync", "http://example.org", "http://example.org/v1"
+ )
- self.assertEqual(12, n.message_number)
- self.assertEqual('sync', n.state)
- self.assertEqual('http://example.org', n.resource_uri)
- self.assertEqual('http://example.org/v1', n.resource_id)
+ self.assertEqual(12, n.message_number)
+ self.assertEqual("sync", n.state)
+ self.assertEqual("http://example.org", n.resource_uri)
+ self.assertEqual("http://example.org/v1", n.resource_id)
- def test_notification_from_headers(self):
- headers = {
- 'X-GoOG-CHANNEL-ID': 'myid',
- 'X-Goog-MESSAGE-NUMBER': '1',
- 'X-Goog-rESOURCE-STATE': 'sync',
- 'X-Goog-reSOURCE-URI': 'http://example.com/',
- 'X-Goog-resOURCE-ID': 'http://example.com/resource_1',
+ def test_notification_from_headers(self):
+ headers = {
+ "X-GoOG-CHANNEL-ID": "myid",
+ "X-Goog-MESSAGE-NUMBER": "1",
+ "X-Goog-rESOURCE-STATE": "sync",
+ "X-Goog-reSOURCE-URI": "http://example.com/",
+ "X-Goog-resOURCE-ID": "http://example.com/resource_1",
}
- ch = channel.Channel('web_hook', 'myid', 'mytoken',
- 'http://example.org/callback',
- expiration=0,
- params={'extra': 'info'},
- resource_id='the_resource_id',
- resource_uri='http://example.com/resource_1')
+ ch = channel.Channel(
+ "web_hook",
+ "myid",
+ "mytoken",
+ "http://example.org/callback",
+ expiration=0,
+ params={"extra": "info"},
+ resource_id="the_resource_id",
+ resource_uri="http://example.com/resource_1",
+ )
- # Good test case.
- n = channel.notification_from_headers(ch, headers)
- self.assertEqual('http://example.com/resource_1', n.resource_id)
- self.assertEqual('http://example.com/', n.resource_uri)
- self.assertEqual('sync', n.state)
- self.assertEqual(1, n.message_number)
+ # Good test case.
+ n = channel.notification_from_headers(ch, headers)
+ self.assertEqual("http://example.com/resource_1", n.resource_id)
+ self.assertEqual("http://example.com/", n.resource_uri)
+ self.assertEqual("sync", n.state)
+ self.assertEqual(1, n.message_number)
- # Detect id mismatch.
- ch.id = 'different_id'
- try:
- n = channel.notification_from_headers(ch, headers)
- self.fail('Should have raised exception')
- except errors.InvalidNotificationError:
- pass
+ # Detect id mismatch.
+ ch.id = "different_id"
+ try:
+ n = channel.notification_from_headers(ch, headers)
+ self.fail("Should have raised exception")
+ except errors.InvalidNotificationError:
+ pass
- # Set the id back to a correct value.
- ch.id = 'myid'
+ # Set the id back to a correct value.
+ ch.id = "myid"
diff --git a/tests/test_discovery.py b/tests/test_discovery.py
index b41051a..f85035e 100644
--- a/tests/test_discovery.py
+++ b/tests/test_discovery.py
@@ -23,7 +23,7 @@
from __future__ import absolute_import
import six
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
from six import BytesIO, StringIO
from six.moves.urllib.parse import urlparse, parse_qs
@@ -84,1448 +84,1593 @@
import uritemplate
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
def assertUrisEqual(testcase, expected, actual):
- """Test that URIs are the same, up to reordering of query parameters."""
- expected = urlparse(expected)
- actual = urlparse(actual)
- testcase.assertEqual(expected.scheme, actual.scheme)
- testcase.assertEqual(expected.netloc, actual.netloc)
- testcase.assertEqual(expected.path, actual.path)
- testcase.assertEqual(expected.params, actual.params)
- testcase.assertEqual(expected.fragment, actual.fragment)
- expected_query = parse_qs(expected.query)
- actual_query = parse_qs(actual.query)
- for name in list(expected_query.keys()):
- testcase.assertEqual(expected_query[name], actual_query[name])
- for name in list(actual_query.keys()):
- testcase.assertEqual(expected_query[name], actual_query[name])
+ """Test that URIs are the same, up to reordering of query parameters."""
+ expected = urlparse(expected)
+ actual = urlparse(actual)
+ testcase.assertEqual(expected.scheme, actual.scheme)
+ testcase.assertEqual(expected.netloc, actual.netloc)
+ testcase.assertEqual(expected.path, actual.path)
+ testcase.assertEqual(expected.params, actual.params)
+ testcase.assertEqual(expected.fragment, actual.fragment)
+ expected_query = parse_qs(expected.query)
+ actual_query = parse_qs(actual.query)
+ for name in list(expected_query.keys()):
+ testcase.assertEqual(expected_query[name], actual_query[name])
+ for name in list(actual_query.keys()):
+ testcase.assertEqual(expected_query[name], actual_query[name])
def datafile(filename):
- return os.path.join(DATA_DIR, filename)
+ return os.path.join(DATA_DIR, filename)
class SetupHttplib2(unittest.TestCase):
-
- def test_retries(self):
- # Merely loading googleapiclient.discovery should set the RETRIES to 1.
- self.assertEqual(1, httplib2.RETRIES)
+ def test_retries(self):
+ # Merely loading googleapiclient.discovery should set the RETRIES to 1.
+ self.assertEqual(1, httplib2.RETRIES)
class Utilities(unittest.TestCase):
+ def setUp(self):
+ with open(datafile("zoo.json"), "r") as fh:
+ self.zoo_root_desc = json.loads(fh.read())
+ self.zoo_get_method_desc = self.zoo_root_desc["methods"]["query"]
+ self.zoo_animals_resource = self.zoo_root_desc["resources"]["animals"]
+ self.zoo_insert_method_desc = self.zoo_animals_resource["methods"]["insert"]
+ self.zoo_schema = Schemas(self.zoo_root_desc)
- def setUp(self):
- with open(datafile('zoo.json'), 'r') as fh:
- self.zoo_root_desc = json.loads(fh.read())
- self.zoo_get_method_desc = self.zoo_root_desc['methods']['query']
- self.zoo_animals_resource = self.zoo_root_desc['resources']['animals']
- self.zoo_insert_method_desc = self.zoo_animals_resource['methods']['insert']
- self.zoo_schema = Schemas(self.zoo_root_desc)
+ def test_key2param(self):
+ self.assertEqual("max_results", key2param("max-results"))
+ self.assertEqual("x007_bond", key2param("007-bond"))
- def test_key2param(self):
- self.assertEqual('max_results', key2param('max-results'))
- self.assertEqual('x007_bond', key2param('007-bond'))
+ def _base_fix_up_parameters_test(self, method_desc, http_method, root_desc, schema):
+ self.assertEqual(method_desc["httpMethod"], http_method)
- def _base_fix_up_parameters_test(
- self, method_desc, http_method, root_desc, schema):
- self.assertEqual(method_desc['httpMethod'], http_method)
+ method_desc_copy = copy.deepcopy(method_desc)
+ self.assertEqual(method_desc, method_desc_copy)
- method_desc_copy = copy.deepcopy(method_desc)
- self.assertEqual(method_desc, method_desc_copy)
+ parameters = _fix_up_parameters(
+ method_desc_copy, root_desc, http_method, schema
+ )
- parameters = _fix_up_parameters(method_desc_copy, root_desc, http_method,
- schema)
+ self.assertNotEqual(method_desc, method_desc_copy)
- self.assertNotEqual(method_desc, method_desc_copy)
+ for param_name in STACK_QUERY_PARAMETERS:
+ self.assertEqual(
+ STACK_QUERY_PARAMETER_DEFAULT_VALUE, parameters[param_name]
+ )
- for param_name in STACK_QUERY_PARAMETERS:
- self.assertEqual(STACK_QUERY_PARAMETER_DEFAULT_VALUE,
- parameters[param_name])
+ for param_name, value in six.iteritems(root_desc.get("parameters", {})):
+ self.assertEqual(value, parameters[param_name])
- for param_name, value in six.iteritems(root_desc.get('parameters', {})):
- self.assertEqual(value, parameters[param_name])
+ return parameters
- return parameters
+ def test_fix_up_parameters_get(self):
+ parameters = self._base_fix_up_parameters_test(
+ self.zoo_get_method_desc, "GET", self.zoo_root_desc, self.zoo_schema
+ )
+ # Since http_method is 'GET'
+ self.assertFalse("body" in parameters)
- def test_fix_up_parameters_get(self):
- parameters = self._base_fix_up_parameters_test(
- self.zoo_get_method_desc, 'GET', self.zoo_root_desc, self.zoo_schema)
- # Since http_method is 'GET'
- self.assertFalse('body' in parameters)
+ def test_fix_up_parameters_insert(self):
+ parameters = self._base_fix_up_parameters_test(
+ self.zoo_insert_method_desc, "POST", self.zoo_root_desc, self.zoo_schema
+ )
+ body = {"description": "The request body.", "type": "object", "$ref": "Animal"}
+ self.assertEqual(parameters["body"], body)
- def test_fix_up_parameters_insert(self):
- parameters = self._base_fix_up_parameters_test(
- self.zoo_insert_method_desc, 'POST', self.zoo_root_desc, self.zoo_schema)
- body = {
- 'description': 'The request body.',
- 'type': 'object',
- '$ref': 'Animal',
- }
- self.assertEqual(parameters['body'], body)
-
- def test_fix_up_parameters_check_body(self):
- dummy_root_desc = {}
- dummy_schema = {
- 'Request': {
- 'properties': {
- "description": "Required. Dummy parameter.",
- "type": "string"
+ def test_fix_up_parameters_check_body(self):
+ dummy_root_desc = {}
+ dummy_schema = {
+ "Request": {
+ "properties": {
+ "description": "Required. Dummy parameter.",
+ "type": "string",
+ }
+ }
}
- }
- }
- no_payload_http_method = 'DELETE'
- with_payload_http_method = 'PUT'
+ no_payload_http_method = "DELETE"
+ with_payload_http_method = "PUT"
- invalid_method_desc = {'response': 'Who cares'}
- valid_method_desc = {
- 'request': {
- 'key1': 'value1',
- 'key2': 'value2',
- '$ref': 'Request'
- }
- }
+ invalid_method_desc = {"response": "Who cares"}
+ valid_method_desc = {
+ "request": {"key1": "value1", "key2": "value2", "$ref": "Request"}
+ }
- parameters = _fix_up_parameters(invalid_method_desc, dummy_root_desc,
- no_payload_http_method, dummy_schema)
- self.assertFalse('body' in parameters)
+ parameters = _fix_up_parameters(
+ invalid_method_desc, dummy_root_desc, no_payload_http_method, dummy_schema
+ )
+ self.assertFalse("body" in parameters)
- parameters = _fix_up_parameters(valid_method_desc, dummy_root_desc,
- no_payload_http_method, dummy_schema)
- self.assertFalse('body' in parameters)
+ parameters = _fix_up_parameters(
+ valid_method_desc, dummy_root_desc, no_payload_http_method, dummy_schema
+ )
+ self.assertFalse("body" in parameters)
- parameters = _fix_up_parameters(invalid_method_desc, dummy_root_desc,
- with_payload_http_method, dummy_schema)
- self.assertFalse('body' in parameters)
+ parameters = _fix_up_parameters(
+ invalid_method_desc, dummy_root_desc, with_payload_http_method, dummy_schema
+ )
+ self.assertFalse("body" in parameters)
- parameters = _fix_up_parameters(valid_method_desc, dummy_root_desc,
- with_payload_http_method, dummy_schema)
- body = {
- 'description': 'The request body.',
- 'type': 'object',
- '$ref': 'Request',
- 'key1': 'value1',
- 'key2': 'value2',
- }
- self.assertEqual(parameters['body'], body)
+ parameters = _fix_up_parameters(
+ valid_method_desc, dummy_root_desc, with_payload_http_method, dummy_schema
+ )
+ body = {
+ "description": "The request body.",
+ "type": "object",
+ "$ref": "Request",
+ "key1": "value1",
+ "key2": "value2",
+ }
+ self.assertEqual(parameters["body"], body)
- def test_fix_up_parameters_optional_body(self):
- # Request with no parameters
- dummy_schema = {'Request': {'properties': {}}}
- method_desc = {'request': {'$ref': 'Request'}}
+ def test_fix_up_parameters_optional_body(self):
+ # Request with no parameters
+ dummy_schema = {"Request": {"properties": {}}}
+ method_desc = {"request": {"$ref": "Request"}}
- parameters = _fix_up_parameters(method_desc, {}, 'POST', dummy_schema)
+ parameters = _fix_up_parameters(method_desc, {}, "POST", dummy_schema)
- def _base_fix_up_method_description_test(
- self, method_desc, initial_parameters, final_parameters,
- final_accept, final_max_size, final_media_path_url):
- fake_root_desc = {'rootUrl': 'http://root/',
- 'servicePath': 'fake/'}
- fake_path_url = 'fake-path/'
+ def _base_fix_up_method_description_test(
+ self,
+ method_desc,
+ initial_parameters,
+ final_parameters,
+ final_accept,
+ final_max_size,
+ final_media_path_url,
+ ):
+ fake_root_desc = {"rootUrl": "http://root/", "servicePath": "fake/"}
+ fake_path_url = "fake-path/"
- accept, max_size, media_path_url = _fix_up_media_upload(
- method_desc, fake_root_desc, fake_path_url, initial_parameters)
- self.assertEqual(accept, final_accept)
- self.assertEqual(max_size, final_max_size)
- self.assertEqual(media_path_url, final_media_path_url)
- self.assertEqual(initial_parameters, final_parameters)
+ accept, max_size, media_path_url = _fix_up_media_upload(
+ method_desc, fake_root_desc, fake_path_url, initial_parameters
+ )
+ self.assertEqual(accept, final_accept)
+ self.assertEqual(max_size, final_max_size)
+ self.assertEqual(media_path_url, final_media_path_url)
+ self.assertEqual(initial_parameters, final_parameters)
- def test_fix_up_media_upload_no_initial_invalid(self):
- invalid_method_desc = {'response': 'Who cares'}
- self._base_fix_up_method_description_test(invalid_method_desc, {}, {},
- [], 0, None)
+ def test_fix_up_media_upload_no_initial_invalid(self):
+ invalid_method_desc = {"response": "Who cares"}
+ self._base_fix_up_method_description_test(
+ invalid_method_desc, {}, {}, [], 0, None
+ )
- def test_fix_up_media_upload_no_initial_valid_minimal(self):
- valid_method_desc = {'mediaUpload': {'accept': []}}
- final_parameters = {'media_body': MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
- 'media_mime_type': MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE}
- self._base_fix_up_method_description_test(
- valid_method_desc, {}, final_parameters, [], 0,
- 'http://root/upload/fake/fake-path/')
+ def test_fix_up_media_upload_no_initial_valid_minimal(self):
+ valid_method_desc = {"mediaUpload": {"accept": []}}
+ final_parameters = {
+ "media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
+ "media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,
+ }
+ self._base_fix_up_method_description_test(
+ valid_method_desc,
+ {},
+ final_parameters,
+ [],
+ 0,
+ "http://root/upload/fake/fake-path/",
+ )
- def test_fix_up_media_upload_no_initial_valid_full(self):
- valid_method_desc = {'mediaUpload': {'accept': ['*/*'], 'maxSize': '10GB'}}
- final_parameters = {'media_body': MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
- 'media_mime_type': MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE}
- ten_gb = 10 * 2**30
- self._base_fix_up_method_description_test(
- valid_method_desc, {}, final_parameters, ['*/*'],
- ten_gb, 'http://root/upload/fake/fake-path/')
+ def test_fix_up_media_upload_no_initial_valid_full(self):
+ valid_method_desc = {"mediaUpload": {"accept": ["*/*"], "maxSize": "10GB"}}
+ final_parameters = {
+ "media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
+ "media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,
+ }
+ ten_gb = 10 * 2 ** 30
+ self._base_fix_up_method_description_test(
+ valid_method_desc,
+ {},
+ final_parameters,
+ ["*/*"],
+ ten_gb,
+ "http://root/upload/fake/fake-path/",
+ )
- def test_fix_up_media_upload_with_initial_invalid(self):
- invalid_method_desc = {'response': 'Who cares'}
- initial_parameters = {'body': {}}
- self._base_fix_up_method_description_test(
- invalid_method_desc, initial_parameters,
- initial_parameters, [], 0, None)
+ def test_fix_up_media_upload_with_initial_invalid(self):
+ invalid_method_desc = {"response": "Who cares"}
+ initial_parameters = {"body": {}}
+ self._base_fix_up_method_description_test(
+ invalid_method_desc, initial_parameters, initial_parameters, [], 0, None
+ )
- def test_fix_up_media_upload_with_initial_valid_minimal(self):
- valid_method_desc = {'mediaUpload': {'accept': []}}
- initial_parameters = {'body': {}}
- final_parameters = {'body': {},
- 'media_body': MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
- 'media_mime_type': MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE}
- self._base_fix_up_method_description_test(
- valid_method_desc, initial_parameters, final_parameters, [], 0,
- 'http://root/upload/fake/fake-path/')
+ def test_fix_up_media_upload_with_initial_valid_minimal(self):
+ valid_method_desc = {"mediaUpload": {"accept": []}}
+ initial_parameters = {"body": {}}
+ final_parameters = {
+ "body": {},
+ "media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
+ "media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,
+ }
+ self._base_fix_up_method_description_test(
+ valid_method_desc,
+ initial_parameters,
+ final_parameters,
+ [],
+ 0,
+ "http://root/upload/fake/fake-path/",
+ )
- def test_fix_up_media_upload_with_initial_valid_full(self):
- valid_method_desc = {'mediaUpload': {'accept': ['*/*'], 'maxSize': '10GB'}}
- initial_parameters = {'body': {}}
- final_parameters = {'body': {},
- 'media_body': MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
- 'media_mime_type': MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE}
- ten_gb = 10 * 2**30
- self._base_fix_up_method_description_test(
- valid_method_desc, initial_parameters, final_parameters, ['*/*'],
- ten_gb, 'http://root/upload/fake/fake-path/')
+ def test_fix_up_media_upload_with_initial_valid_full(self):
+ valid_method_desc = {"mediaUpload": {"accept": ["*/*"], "maxSize": "10GB"}}
+ initial_parameters = {"body": {}}
+ final_parameters = {
+ "body": {},
+ "media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,
+ "media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,
+ }
+ ten_gb = 10 * 2 ** 30
+ self._base_fix_up_method_description_test(
+ valid_method_desc,
+ initial_parameters,
+ final_parameters,
+ ["*/*"],
+ ten_gb,
+ "http://root/upload/fake/fake-path/",
+ )
- def test_fix_up_method_description_get(self):
- result = _fix_up_method_description(self.zoo_get_method_desc,
- self.zoo_root_desc, self.zoo_schema)
- path_url = 'query'
- http_method = 'GET'
- method_id = 'bigquery.query'
- accept = []
- max_size = 0
- media_path_url = None
- self.assertEqual(result, (path_url, http_method, method_id, accept,
- max_size, media_path_url))
+ def test_fix_up_method_description_get(self):
+ result = _fix_up_method_description(
+ self.zoo_get_method_desc, self.zoo_root_desc, self.zoo_schema
+ )
+ path_url = "query"
+ http_method = "GET"
+ method_id = "bigquery.query"
+ accept = []
+ max_size = 0
+ media_path_url = None
+ self.assertEqual(
+ result, (path_url, http_method, method_id, accept, max_size, media_path_url)
+ )
- def test_fix_up_method_description_insert(self):
- result = _fix_up_method_description(self.zoo_insert_method_desc,
- self.zoo_root_desc, self.zoo_schema)
- path_url = 'animals'
- http_method = 'POST'
- method_id = 'zoo.animals.insert'
- accept = ['image/png']
- max_size = 1024
- media_path_url = 'https://www.googleapis.com/upload/zoo/v1/animals'
- self.assertEqual(result, (path_url, http_method, method_id, accept,
- max_size, media_path_url))
+ def test_fix_up_method_description_insert(self):
+ result = _fix_up_method_description(
+ self.zoo_insert_method_desc, self.zoo_root_desc, self.zoo_schema
+ )
+ path_url = "animals"
+ http_method = "POST"
+ method_id = "zoo.animals.insert"
+ accept = ["image/png"]
+ max_size = 1024
+ media_path_url = "https://www.googleapis.com/upload/zoo/v1/animals"
+ self.assertEqual(
+ result, (path_url, http_method, method_id, accept, max_size, media_path_url)
+ )
- def test_urljoin(self):
- # We want to exhaustively test various URL combinations.
- simple_bases = ['https://www.googleapis.com', 'https://www.googleapis.com/']
- long_urls = ['foo/v1/bar:custom?alt=json', '/foo/v1/bar:custom?alt=json']
+ def test_urljoin(self):
+ # We want to exhaustively test various URL combinations.
+ simple_bases = ["https://www.googleapis.com", "https://www.googleapis.com/"]
+ long_urls = ["foo/v1/bar:custom?alt=json", "/foo/v1/bar:custom?alt=json"]
- long_bases = [
- 'https://www.googleapis.com/foo/v1',
- 'https://www.googleapis.com/foo/v1/',
- ]
- simple_urls = ['bar:custom?alt=json', '/bar:custom?alt=json']
+ long_bases = [
+ "https://www.googleapis.com/foo/v1",
+ "https://www.googleapis.com/foo/v1/",
+ ]
+ simple_urls = ["bar:custom?alt=json", "/bar:custom?alt=json"]
- final_url = 'https://www.googleapis.com/foo/v1/bar:custom?alt=json'
- for base, url in itertools.product(simple_bases, long_urls):
- self.assertEqual(final_url, _urljoin(base, url))
- for base, url in itertools.product(long_bases, simple_urls):
- self.assertEqual(final_url, _urljoin(base, url))
+ final_url = "https://www.googleapis.com/foo/v1/bar:custom?alt=json"
+ for base, url in itertools.product(simple_bases, long_urls):
+ self.assertEqual(final_url, _urljoin(base, url))
+ for base, url in itertools.product(long_bases, simple_urls):
+ self.assertEqual(final_url, _urljoin(base, url))
+ def test_ResourceMethodParameters_zoo_get(self):
+ parameters = ResourceMethodParameters(self.zoo_get_method_desc)
- def test_ResourceMethodParameters_zoo_get(self):
- parameters = ResourceMethodParameters(self.zoo_get_method_desc)
+ param_types = {
+ "a": "any",
+ "b": "boolean",
+ "e": "string",
+ "er": "string",
+ "i": "integer",
+ "n": "number",
+ "o": "object",
+ "q": "string",
+ "rr": "string",
+ }
+ keys = list(param_types.keys())
+ self.assertEqual(parameters.argmap, dict((key, key) for key in keys))
+ self.assertEqual(parameters.required_params, [])
+ self.assertEqual(sorted(parameters.repeated_params), ["er", "rr"])
+ self.assertEqual(parameters.pattern_params, {"rr": "[a-z]+"})
+ self.assertEqual(
+ sorted(parameters.query_params),
+ ["a", "b", "e", "er", "i", "n", "o", "q", "rr"],
+ )
+ self.assertEqual(parameters.path_params, set())
+ self.assertEqual(parameters.param_types, param_types)
+ enum_params = {"e": ["foo", "bar"], "er": ["one", "two", "three"]}
+ self.assertEqual(parameters.enum_params, enum_params)
- param_types = {'a': 'any',
- 'b': 'boolean',
- 'e': 'string',
- 'er': 'string',
- 'i': 'integer',
- 'n': 'number',
- 'o': 'object',
- 'q': 'string',
- 'rr': 'string'}
- keys = list(param_types.keys())
- self.assertEqual(parameters.argmap, dict((key, key) for key in keys))
- self.assertEqual(parameters.required_params, [])
- self.assertEqual(sorted(parameters.repeated_params), ['er', 'rr'])
- self.assertEqual(parameters.pattern_params, {'rr': '[a-z]+'})
- self.assertEqual(sorted(parameters.query_params),
- ['a', 'b', 'e', 'er', 'i', 'n', 'o', 'q', 'rr'])
- self.assertEqual(parameters.path_params, set())
- self.assertEqual(parameters.param_types, param_types)
- enum_params = {'e': ['foo', 'bar'],
- 'er': ['one', 'two', 'three']}
- self.assertEqual(parameters.enum_params, enum_params)
+ def test_ResourceMethodParameters_zoo_animals_patch(self):
+ method_desc = self.zoo_animals_resource["methods"]["patch"]
+ parameters = ResourceMethodParameters(method_desc)
- def test_ResourceMethodParameters_zoo_animals_patch(self):
- method_desc = self.zoo_animals_resource['methods']['patch']
- parameters = ResourceMethodParameters(method_desc)
-
- param_types = {'name': 'string'}
- keys = list(param_types.keys())
- self.assertEqual(parameters.argmap, dict((key, key) for key in keys))
- self.assertEqual(parameters.required_params, ['name'])
- self.assertEqual(parameters.repeated_params, [])
- self.assertEqual(parameters.pattern_params, {})
- self.assertEqual(parameters.query_params, [])
- self.assertEqual(parameters.path_params, set(['name']))
- self.assertEqual(parameters.param_types, param_types)
- self.assertEqual(parameters.enum_params, {})
+ param_types = {"name": "string"}
+ keys = list(param_types.keys())
+ self.assertEqual(parameters.argmap, dict((key, key) for key in keys))
+ self.assertEqual(parameters.required_params, ["name"])
+ self.assertEqual(parameters.repeated_params, [])
+ self.assertEqual(parameters.pattern_params, {})
+ self.assertEqual(parameters.query_params, [])
+ self.assertEqual(parameters.path_params, set(["name"]))
+ self.assertEqual(parameters.param_types, param_types)
+ self.assertEqual(parameters.enum_params, {})
class DiscoveryErrors(unittest.TestCase):
+ def test_tests_should_be_run_with_strict_positional_enforcement(self):
+ try:
+ plus = build("plus", "v1", None)
+ self.fail("should have raised a TypeError exception over missing http=.")
+ except TypeError:
+ pass
- def test_tests_should_be_run_with_strict_positional_enforcement(self):
- try:
- plus = build('plus', 'v1', None)
- self.fail("should have raised a TypeError exception over missing http=.")
- except TypeError:
- pass
+ def test_failed_to_parse_discovery_json(self):
+ self.http = HttpMock(datafile("malformed.json"), {"status": "200"})
+ try:
+ plus = build("plus", "v1", http=self.http, cache_discovery=False)
+ self.fail("should have raised an exception over malformed JSON.")
+ except InvalidJsonError:
+ pass
- def test_failed_to_parse_discovery_json(self):
- self.http = HttpMock(datafile('malformed.json'), {'status': '200'})
- try:
- plus = build('plus', 'v1', http=self.http, cache_discovery=False)
- self.fail("should have raised an exception over malformed JSON.")
- except InvalidJsonError:
- pass
+ def test_unknown_api_name_or_version(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "404"}, open(datafile("zoo.json"), "rb").read()),
+ ({"status": "404"}, open(datafile("zoo.json"), "rb").read()),
+ ]
+ )
+ with self.assertRaises(UnknownApiNameOrVersion):
+ plus = build("plus", "v1", http=http, cache_discovery=False)
- def test_unknown_api_name_or_version(self):
- http = HttpMockSequence([
- ({'status': '404'}, open(datafile('zoo.json'), 'rb').read()),
- ({'status': '404'}, open(datafile('zoo.json'), 'rb').read()),
- ])
- with self.assertRaises(UnknownApiNameOrVersion):
- plus = build('plus', 'v1', http=http, cache_discovery=False)
-
- def test_credentials_and_http_mutually_exclusive(self):
- http = HttpMock(datafile('plus.json'), {'status': '200'})
- with self.assertRaises(ValueError):
- build(
- 'plus', 'v1', http=http, credentials=mock.sentinel.credentials)
+ def test_credentials_and_http_mutually_exclusive(self):
+ http = HttpMock(datafile("plus.json"), {"status": "200"})
+ with self.assertRaises(ValueError):
+ build("plus", "v1", http=http, credentials=mock.sentinel.credentials)
class DiscoveryFromDocument(unittest.TestCase):
- MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.Credentials)
+ MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.Credentials)
- def test_can_build_from_local_document(self):
- discovery = open(datafile('plus.json')).read()
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/",
- credentials=self.MOCK_CREDENTIALS)
- self.assertTrue(plus is not None)
- self.assertTrue(hasattr(plus, 'activities'))
+ def test_can_build_from_local_document(self):
+ discovery = open(datafile("plus.json")).read()
+ plus = build_from_document(
+ discovery,
+ base="https://www.googleapis.com/",
+ credentials=self.MOCK_CREDENTIALS,
+ )
+ self.assertTrue(plus is not None)
+ self.assertTrue(hasattr(plus, "activities"))
- def test_can_build_from_local_deserialized_document(self):
- discovery = open(datafile('plus.json')).read()
- discovery = json.loads(discovery)
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/",
- credentials=self.MOCK_CREDENTIALS)
- self.assertTrue(plus is not None)
- self.assertTrue(hasattr(plus, 'activities'))
+ def test_can_build_from_local_deserialized_document(self):
+ discovery = open(datafile("plus.json")).read()
+ discovery = json.loads(discovery)
+ plus = build_from_document(
+ discovery,
+ base="https://www.googleapis.com/",
+ credentials=self.MOCK_CREDENTIALS,
+ )
+ self.assertTrue(plus is not None)
+ self.assertTrue(hasattr(plus, "activities"))
- def test_building_with_base_remembers_base(self):
- discovery = open(datafile('plus.json')).read()
+ def test_building_with_base_remembers_base(self):
+ discovery = open(datafile("plus.json")).read()
- base = "https://www.example.com/"
- plus = build_from_document(
- discovery, base=base, credentials=self.MOCK_CREDENTIALS)
- self.assertEquals("https://www.googleapis.com/plus/v1/", plus._baseUrl)
+ base = "https://www.example.com/"
+ plus = build_from_document(
+ discovery, base=base, credentials=self.MOCK_CREDENTIALS
+ )
+ self.assertEquals("https://www.googleapis.com/plus/v1/", plus._baseUrl)
- def test_building_with_optional_http_with_authorization(self):
- discovery = open(datafile('plus.json')).read()
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/",
- credentials=self.MOCK_CREDENTIALS)
+ def test_building_with_optional_http_with_authorization(self):
+ discovery = open(datafile("plus.json")).read()
+ plus = build_from_document(
+ discovery,
+ base="https://www.googleapis.com/",
+ credentials=self.MOCK_CREDENTIALS,
+ )
- # plus service requires Authorization, hence we expect to see AuthorizedHttp object here
- self.assertIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)
- self.assertIsInstance(plus._http.http, httplib2.Http)
- self.assertIsInstance(plus._http.http.timeout, int)
- self.assertGreater(plus._http.http.timeout, 0)
+ # plus service requires Authorization, hence we expect to see AuthorizedHttp object here
+ self.assertIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)
+ self.assertIsInstance(plus._http.http, httplib2.Http)
+ self.assertIsInstance(plus._http.http.timeout, int)
+ self.assertGreater(plus._http.http.timeout, 0)
- def test_building_with_optional_http_with_no_authorization(self):
- discovery = open(datafile('plus.json')).read()
- # Cleanup auth field, so we would use plain http client
- discovery = json.loads(discovery)
- discovery['auth'] = {}
- discovery = json.dumps(discovery)
+ def test_building_with_optional_http_with_no_authorization(self):
+ discovery = open(datafile("plus.json")).read()
+ # Cleanup auth field, so we would use plain http client
+ discovery = json.loads(discovery)
+ discovery["auth"] = {}
+ discovery = json.dumps(discovery)
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/",
- credentials=None)
- # plus service requires Authorization
- self.assertIsInstance(plus._http, httplib2.Http)
- self.assertIsInstance(plus._http.timeout, int)
- self.assertGreater(plus._http.timeout, 0)
+ plus = build_from_document(
+ discovery, base="https://www.googleapis.com/", credentials=None
+ )
+ # plus service requires Authorization
+ self.assertIsInstance(plus._http, httplib2.Http)
+ self.assertIsInstance(plus._http.timeout, int)
+ self.assertGreater(plus._http.timeout, 0)
- def test_building_with_explicit_http(self):
- http = HttpMock()
- discovery = open(datafile('plus.json')).read()
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/", http=http)
- self.assertEquals(plus._http, http)
+ def test_building_with_explicit_http(self):
+ http = HttpMock()
+ discovery = open(datafile("plus.json")).read()
+ plus = build_from_document(
+ discovery, base="https://www.googleapis.com/", http=http
+ )
+ self.assertEquals(plus._http, http)
- def test_building_with_developer_key_skips_adc(self):
- discovery = open(datafile('plus.json')).read()
- plus = build_from_document(
- discovery, base="https://www.googleapis.com/", developerKey='123')
- self.assertIsInstance(plus._http, httplib2.Http)
- # It should not be an AuthorizedHttp, because that would indicate that
- # application default credentials were used.
- self.assertNotIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)
+ def test_building_with_developer_key_skips_adc(self):
+ discovery = open(datafile("plus.json")).read()
+ plus = build_from_document(
+ discovery, base="https://www.googleapis.com/", developerKey="123"
+ )
+ self.assertIsInstance(plus._http, httplib2.Http)
+ # It should not be an AuthorizedHttp, because that would indicate that
+ # application default credentials were used.
+ self.assertNotIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)
class DiscoveryFromHttp(unittest.TestCase):
- def setUp(self):
- self.old_environ = os.environ.copy()
+ def setUp(self):
+ self.old_environ = os.environ.copy()
- def tearDown(self):
- os.environ = self.old_environ
+ def tearDown(self):
+ os.environ = self.old_environ
- def test_userip_is_added_to_discovery_uri(self):
- # build() will raise an HttpError on a 400, use this to pick the request uri
- # out of the raised exception.
- os.environ['REMOTE_ADDR'] = '10.0.0.1'
- try:
- http = HttpMockSequence([
- ({'status': '400'}, open(datafile('zoo.json'), 'rb').read()),
- ])
- zoo = build('zoo', 'v1', http=http, developerKey=None,
- discoveryServiceUrl='http://example.com')
- self.fail('Should have raised an exception.')
- except HttpError as e:
- self.assertEqual(e.uri, 'http://example.com?userIp=10.0.0.1')
+ def test_userip_is_added_to_discovery_uri(self):
+ # build() will raise an HttpError on a 400, use this to pick the request uri
+ # out of the raised exception.
+ os.environ["REMOTE_ADDR"] = "10.0.0.1"
+ try:
+ http = HttpMockSequence(
+ [({"status": "400"}, open(datafile("zoo.json"), "rb").read())]
+ )
+ zoo = build(
+ "zoo",
+ "v1",
+ http=http,
+ developerKey=None,
+ discoveryServiceUrl="http://example.com",
+ )
+ self.fail("Should have raised an exception.")
+ except HttpError as e:
+ self.assertEqual(e.uri, "http://example.com?userIp=10.0.0.1")
- def test_userip_missing_is_not_added_to_discovery_uri(self):
- # build() will raise an HttpError on a 400, use this to pick the request uri
- # out of the raised exception.
- try:
- http = HttpMockSequence([
- ({'status': '400'}, open(datafile('zoo.json'), 'rb').read()),
- ])
- zoo = build('zoo', 'v1', http=http, developerKey=None,
- discoveryServiceUrl='http://example.com')
- self.fail('Should have raised an exception.')
- except HttpError as e:
- self.assertEqual(e.uri, 'http://example.com')
+ def test_userip_missing_is_not_added_to_discovery_uri(self):
+ # build() will raise an HttpError on a 400, use this to pick the request uri
+ # out of the raised exception.
+ try:
+ http = HttpMockSequence(
+ [({"status": "400"}, open(datafile("zoo.json"), "rb").read())]
+ )
+ zoo = build(
+ "zoo",
+ "v1",
+ http=http,
+ developerKey=None,
+ discoveryServiceUrl="http://example.com",
+ )
+ self.fail("Should have raised an exception.")
+ except HttpError as e:
+ self.assertEqual(e.uri, "http://example.com")
- def test_key_is_added_to_discovery_uri(self):
- # build() will raise an HttpError on a 400, use this to pick the request uri
- # out of the raised exception.
- try:
- http = HttpMockSequence([
- ({'status': '400'}, open(datafile('zoo.json'), 'rb').read()),
- ])
- zoo = build('zoo', 'v1', http=http, developerKey='foo',
- discoveryServiceUrl='http://example.com')
- self.fail('Should have raised an exception.')
- except HttpError as e:
- self.assertEqual(e.uri, 'http://example.com?key=foo')
+ def test_key_is_added_to_discovery_uri(self):
+ # build() will raise an HttpError on a 400, use this to pick the request uri
+ # out of the raised exception.
+ try:
+ http = HttpMockSequence(
+ [({"status": "400"}, open(datafile("zoo.json"), "rb").read())]
+ )
+ zoo = build(
+ "zoo",
+ "v1",
+ http=http,
+ developerKey="foo",
+ discoveryServiceUrl="http://example.com",
+ )
+ self.fail("Should have raised an exception.")
+ except HttpError as e:
+ self.assertEqual(e.uri, "http://example.com?key=foo")
- def test_discovery_loading_from_v2_discovery_uri(self):
- http = HttpMockSequence([
- ({'status': '404'}, 'Not found'),
- ({'status': '200'}, open(datafile('zoo.json'), 'rb').read()),
- ])
- zoo = build('zoo', 'v1', http=http, cache_discovery=False)
- self.assertTrue(hasattr(zoo, 'animals'))
+ def test_discovery_loading_from_v2_discovery_uri(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "404"}, "Not found"),
+ ({"status": "200"}, open(datafile("zoo.json"), "rb").read()),
+ ]
+ )
+ zoo = build("zoo", "v1", http=http, cache_discovery=False)
+ self.assertTrue(hasattr(zoo, "animals"))
+
class DiscoveryFromAppEngineCache(unittest.TestCase):
- def test_appengine_memcache(self):
- # Hack module import
- self.orig_import = __import__
- self.mocked_api = mock.MagicMock()
+ def test_appengine_memcache(self):
+ # Hack module import
+ self.orig_import = __import__
+ self.mocked_api = mock.MagicMock()
- def import_mock(name, *args, **kwargs):
- if name == 'google.appengine.api':
- return self.mocked_api
- return self.orig_import(name, *args, **kwargs)
+ def import_mock(name, *args, **kwargs):
+ if name == "google.appengine.api":
+ return self.mocked_api
+ return self.orig_import(name, *args, **kwargs)
- import_fullname = '__builtin__.__import__'
- if sys.version_info[0] >= 3:
- import_fullname = 'builtins.__import__'
+ import_fullname = "__builtin__.__import__"
+ if sys.version_info[0] >= 3:
+ import_fullname = "builtins.__import__"
- with mock.patch(import_fullname, side_effect=import_mock):
- namespace = 'google-api-client'
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
+ with mock.patch(import_fullname, side_effect=import_mock):
+ namespace = "google-api-client"
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
- self.mocked_api.memcache.get.return_value = None
+ self.mocked_api.memcache.get.return_value = None
- plus = build('plus', 'v1', http=self.http)
+ plus = build("plus", "v1", http=self.http)
- # memcache.get is called once
- url = 'https://www.googleapis.com/discovery/v1/apis/plus/v1/rest'
- self.mocked_api.memcache.get.assert_called_once_with(url,
- namespace=namespace)
+ # memcache.get is called once
+ url = "https://www.googleapis.com/discovery/v1/apis/plus/v1/rest"
+ self.mocked_api.memcache.get.assert_called_once_with(
+ url, namespace=namespace
+ )
- # memcache.set is called once
- with open(datafile('plus.json')) as f:
- content = f.read()
- self.mocked_api.memcache.set.assert_called_once_with(
- url, content, time=DISCOVERY_DOC_MAX_AGE, namespace=namespace)
+ # memcache.set is called once
+ with open(datafile("plus.json")) as f:
+ content = f.read()
+ self.mocked_api.memcache.set.assert_called_once_with(
+ url, content, time=DISCOVERY_DOC_MAX_AGE, namespace=namespace
+ )
- # Returns the cached content this time.
- self.mocked_api.memcache.get.return_value = content
+ # Returns the cached content this time.
+ self.mocked_api.memcache.get.return_value = content
- # Make sure the contents are returned from the cache.
- # (Otherwise it should through an error)
- self.http = HttpMock(None, {'status': '200'})
+ # Make sure the contents are returned from the cache.
+ # (Otherwise it should through an error)
+ self.http = HttpMock(None, {"status": "200"})
- plus = build('plus', 'v1', http=self.http)
+ plus = build("plus", "v1", http=self.http)
- # memcache.get is called twice
- self.mocked_api.memcache.get.assert_has_calls(
- [mock.call(url, namespace=namespace),
- mock.call(url, namespace=namespace)])
+ # memcache.get is called twice
+ self.mocked_api.memcache.get.assert_has_calls(
+ [
+ mock.call(url, namespace=namespace),
+ mock.call(url, namespace=namespace),
+ ]
+ )
- # memcahce.set is called just once
- self.mocked_api.memcache.set.assert_called_once_with(
- url, content, time=DISCOVERY_DOC_MAX_AGE,namespace=namespace)
+ # memcahce.set is called just once
+ self.mocked_api.memcache.set.assert_called_once_with(
+ url, content, time=DISCOVERY_DOC_MAX_AGE, namespace=namespace
+ )
class DictCache(Cache):
- def __init__(self):
- self.d = {}
- def get(self, url):
- return self.d.get(url, None)
- def set(self, url, content):
- self.d[url] = content
- def contains(self, url):
- return url in self.d
+ def __init__(self):
+ self.d = {}
+
+ def get(self, url):
+ return self.d.get(url, None)
+
+ def set(self, url, content):
+ self.d[url] = content
+
+ def contains(self, url):
+ return url in self.d
class DiscoveryFromFileCache(unittest.TestCase):
- def test_file_based_cache(self):
- cache = mock.Mock(wraps=DictCache())
- with mock.patch('googleapiclient.discovery_cache.autodetect',
- return_value=cache):
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
+ def test_file_based_cache(self):
+ cache = mock.Mock(wraps=DictCache())
+ with mock.patch(
+ "googleapiclient.discovery_cache.autodetect", return_value=cache
+ ):
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
- plus = build('plus', 'v1', http=self.http)
+ plus = build("plus", "v1", http=self.http)
- # cache.get is called once
- url = 'https://www.googleapis.com/discovery/v1/apis/plus/v1/rest'
- cache.get.assert_called_once_with(url)
+ # cache.get is called once
+ url = "https://www.googleapis.com/discovery/v1/apis/plus/v1/rest"
+ cache.get.assert_called_once_with(url)
- # cache.set is called once
- with open(datafile('plus.json')) as f:
- content = f.read()
- cache.set.assert_called_once_with(url, content)
+ # cache.set is called once
+ with open(datafile("plus.json")) as f:
+ content = f.read()
+ cache.set.assert_called_once_with(url, content)
- # Make sure there is a cache entry for the plus v1 discovery doc.
- self.assertTrue(cache.contains(url))
+ # Make sure there is a cache entry for the plus v1 discovery doc.
+ self.assertTrue(cache.contains(url))
- # Make sure the contents are returned from the cache.
- # (Otherwise it should through an error)
- self.http = HttpMock(None, {'status': '200'})
+ # Make sure the contents are returned from the cache.
+ # (Otherwise it should through an error)
+ self.http = HttpMock(None, {"status": "200"})
- plus = build('plus', 'v1', http=self.http)
+ plus = build("plus", "v1", http=self.http)
- # cache.get is called twice
- cache.get.assert_has_calls([mock.call(url), mock.call(url)])
+ # cache.get is called twice
+ cache.get.assert_has_calls([mock.call(url), mock.call(url)])
- # cahce.set is called just once
- cache.set.assert_called_once_with(url, content)
+ # cahce.set is called just once
+ cache.set.assert_called_once_with(url, content)
class Discovery(unittest.TestCase):
+ def test_method_error_checking(self):
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
+ plus = build("plus", "v1", http=self.http)
- def test_method_error_checking(self):
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
- plus = build('plus', 'v1', http=self.http)
+ # Missing required parameters
+ try:
+ plus.activities().list()
+ self.fail()
+ except TypeError as e:
+ self.assertTrue("Missing" in str(e))
- # Missing required parameters
- try:
- plus.activities().list()
- self.fail()
- except TypeError as e:
- self.assertTrue('Missing' in str(e))
+ # Missing required parameters even if supplied as None.
+ try:
+ plus.activities().list(collection=None, userId=None)
+ self.fail()
+ except TypeError as e:
+ self.assertTrue("Missing" in str(e))
- # Missing required parameters even if supplied as None.
- try:
- plus.activities().list(collection=None, userId=None)
- self.fail()
- except TypeError as e:
- self.assertTrue('Missing' in str(e))
+ # Parameter doesn't match regex
+ try:
+ plus.activities().list(collection="not_a_collection_name", userId="me")
+ self.fail()
+ except TypeError as e:
+ self.assertTrue("not an allowed value" in str(e))
- # Parameter doesn't match regex
- try:
- plus.activities().list(collection='not_a_collection_name', userId='me')
- self.fail()
- except TypeError as e:
- self.assertTrue('not an allowed value' in str(e))
+ # Unexpected parameter
+ try:
+ plus.activities().list(flubber=12)
+ self.fail()
+ except TypeError as e:
+ self.assertTrue("unexpected" in str(e))
- # Unexpected parameter
- try:
- plus.activities().list(flubber=12)
- self.fail()
- except TypeError as e:
- self.assertTrue('unexpected' in str(e))
+ def _check_query_types(self, request):
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["q"], ["foo"])
+ self.assertEqual(q["i"], ["1"])
+ self.assertEqual(q["n"], ["1.0"])
+ self.assertEqual(q["b"], ["false"])
+ self.assertEqual(q["a"], ["[1, 2, 3]"])
+ self.assertEqual(q["o"], ["{'a': 1}"])
+ self.assertEqual(q["e"], ["bar"])
- def _check_query_types(self, request):
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['q'], ['foo'])
- self.assertEqual(q['i'], ['1'])
- self.assertEqual(q['n'], ['1.0'])
- self.assertEqual(q['b'], ['false'])
- self.assertEqual(q['a'], ['[1, 2, 3]'])
- self.assertEqual(q['o'], ['{\'a\': 1}'])
- self.assertEqual(q['e'], ['bar'])
+ def test_type_coercion(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
- def test_type_coercion(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
+ request = zoo.query(
+ q="foo", i=1.0, n=1.0, b=0, a=[1, 2, 3], o={"a": 1}, e="bar"
+ )
+ self._check_query_types(request)
+ request = zoo.query(
+ q="foo", i=1, n=1, b=False, a=[1, 2, 3], o={"a": 1}, e="bar"
+ )
+ self._check_query_types(request)
- request = zoo.query(
- q="foo", i=1.0, n=1.0, b=0, a=[1,2,3], o={'a':1}, e='bar')
- self._check_query_types(request)
- request = zoo.query(
- q="foo", i=1, n=1, b=False, a=[1,2,3], o={'a':1}, e='bar')
- self._check_query_types(request)
+ request = zoo.query(
+ q="foo", i="1", n="1", b="", a=[1, 2, 3], o={"a": 1}, e="bar", er="two"
+ )
- request = zoo.query(
- q="foo", i="1", n="1", b="", a=[1,2,3], o={'a':1}, e='bar', er='two')
+ request = zoo.query(
+ q="foo",
+ i="1",
+ n="1",
+ b="",
+ a=[1, 2, 3],
+ o={"a": 1},
+ e="bar",
+ er=["one", "three"],
+ rr=["foo", "bar"],
+ )
+ self._check_query_types(request)
- request = zoo.query(
- q="foo", i="1", n="1", b="", a=[1,2,3], o={'a':1}, e='bar',
- er=['one', 'three'], rr=['foo', 'bar'])
- self._check_query_types(request)
+ # Five is right out.
+ self.assertRaises(TypeError, zoo.query, er=["one", "five"])
- # Five is right out.
- self.assertRaises(TypeError, zoo.query, er=['one', 'five'])
+ def test_optional_stack_query_parameters(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.query(trace="html", fields="description")
- def test_optional_stack_query_parameters(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.query(trace='html', fields='description')
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["trace"], ["html"])
+ self.assertEqual(q["fields"], ["description"])
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['trace'], ['html'])
- self.assertEqual(q['fields'], ['description'])
+ def test_string_params_value_of_none_get_dropped(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.query(trace=None, fields="description")
- def test_string_params_value_of_none_get_dropped(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.query(trace=None, fields='description')
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertFalse("trace" in q)
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertFalse('trace' in q)
+ def test_model_added_query_parameters(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.animals().get(name="Lion")
- def test_model_added_query_parameters(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.animals().get(name='Lion')
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["alt"], ["json"])
+ self.assertEqual(request.headers["accept"], "application/json")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['alt'], ['json'])
- self.assertEqual(request.headers['accept'], 'application/json')
+ def test_fallback_to_raw_model(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.animals().getmedia(name="Lion")
- def test_fallback_to_raw_model(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.animals().getmedia(name='Lion')
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertTrue("alt" not in q)
+ self.assertEqual(request.headers["accept"], "*/*")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertTrue('alt' not in q)
- self.assertEqual(request.headers['accept'], '*/*')
+ def test_patch(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.animals().patch(name="lion", body='{"description": "foo"}')
- def test_patch(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.animals().patch(name='lion', body='{"description": "foo"}')
+ self.assertEqual(request.method, "PATCH")
- self.assertEqual(request.method, 'PATCH')
+ def test_batch_request_from_discovery(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ # zoo defines a batchPath
+ zoo = build("zoo", "v1", http=self.http)
+ batch_request = zoo.new_batch_http_request()
+ self.assertEqual(
+ batch_request._batch_uri, "https://www.googleapis.com/batchZoo"
+ )
- def test_batch_request_from_discovery(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- # zoo defines a batchPath
- zoo = build('zoo', 'v1', http=self.http)
- batch_request = zoo.new_batch_http_request()
- self.assertEqual(batch_request._batch_uri,
- "https://www.googleapis.com/batchZoo")
+ def test_batch_request_from_default(self):
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
+ # plus does not define a batchPath
+ plus = build("plus", "v1", http=self.http)
+ batch_request = plus.new_batch_http_request()
+ self.assertEqual(batch_request._batch_uri, "https://www.googleapis.com/batch")
- def test_batch_request_from_default(self):
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
- # plus does not define a batchPath
- plus = build('plus', 'v1', http=self.http)
- batch_request = plus.new_batch_http_request()
- self.assertEqual(batch_request._batch_uri,
- "https://www.googleapis.com/batch")
+ def test_tunnel_patch(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "200"}, open(datafile("zoo.json"), "rb").read()),
+ ({"status": "200"}, "echo_request_headers_as_json"),
+ ]
+ )
+ http = tunnel_patch(http)
+ zoo = build("zoo", "v1", http=http, cache_discovery=False)
+ resp = zoo.animals().patch(name="lion", body='{"description": "foo"}').execute()
- def test_tunnel_patch(self):
- http = HttpMockSequence([
- ({'status': '200'}, open(datafile('zoo.json'), 'rb').read()),
- ({'status': '200'}, 'echo_request_headers_as_json'),
- ])
- http = tunnel_patch(http)
- zoo = build('zoo', 'v1', http=http, cache_discovery=False)
- resp = zoo.animals().patch(
- name='lion', body='{"description": "foo"}').execute()
+ self.assertTrue("x-http-method-override" in resp)
- self.assertTrue('x-http-method-override' in resp)
+ def test_plus_resources(self):
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
+ plus = build("plus", "v1", http=self.http)
+ self.assertTrue(getattr(plus, "activities"))
+ self.assertTrue(getattr(plus, "people"))
- def test_plus_resources(self):
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
- plus = build('plus', 'v1', http=self.http)
- self.assertTrue(getattr(plus, 'activities'))
- self.assertTrue(getattr(plus, 'people'))
+ def test_oauth2client_credentials(self):
+ credentials = mock.Mock(spec=GoogleCredentials)
+ credentials.create_scoped_required.return_value = False
- def test_oauth2client_credentials(self):
- credentials = mock.Mock(spec=GoogleCredentials)
- credentials.create_scoped_required.return_value = False
+ discovery = open(datafile("plus.json")).read()
+ service = build_from_document(discovery, credentials=credentials)
+ self.assertEqual(service._http, credentials.authorize.return_value)
- discovery = open(datafile('plus.json')).read()
- service = build_from_document(discovery, credentials=credentials)
- self.assertEqual(service._http, credentials.authorize.return_value)
+ def test_google_auth_credentials(self):
+ credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+ discovery = open(datafile("plus.json")).read()
+ service = build_from_document(discovery, credentials=credentials)
- def test_google_auth_credentials(self):
- credentials = mock.Mock(spec=google.auth.credentials.Credentials)
- discovery = open(datafile('plus.json')).read()
- service = build_from_document(discovery, credentials=credentials)
+ self.assertIsInstance(service._http, google_auth_httplib2.AuthorizedHttp)
+ self.assertEqual(service._http.credentials, credentials)
- self.assertIsInstance(service._http, google_auth_httplib2.AuthorizedHttp)
- self.assertEqual(service._http.credentials, credentials)
+ def test_no_scopes_no_credentials(self):
+ # Zoo doesn't have scopes
+ discovery = open(datafile("zoo.json")).read()
+ service = build_from_document(discovery)
+ # Should be an ordinary httplib2.Http instance and not AuthorizedHttp.
+ self.assertIsInstance(service._http, httplib2.Http)
- def test_no_scopes_no_credentials(self):
- # Zoo doesn't have scopes
- discovery = open(datafile('zoo.json')).read()
- service = build_from_document(discovery)
- # Should be an ordinary httplib2.Http instance and not AuthorizedHttp.
- self.assertIsInstance(service._http, httplib2.Http)
+ def test_full_featured(self):
+ # Zoo should exercise all discovery facets
+ # and should also have no future.json file.
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ self.assertTrue(getattr(zoo, "animals"))
- def test_full_featured(self):
- # Zoo should exercise all discovery facets
- # and should also have no future.json file.
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- self.assertTrue(getattr(zoo, 'animals'))
+ request = zoo.animals().list(name="bat", projection="full")
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["name"], ["bat"])
+ self.assertEqual(q["projection"], ["full"])
- request = zoo.animals().list(name='bat', projection="full")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['name'], ['bat'])
- self.assertEqual(q['projection'], ['full'])
+ def test_nested_resources(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ self.assertTrue(getattr(zoo, "animals"))
+ request = zoo.my().favorites().list(max_results="5")
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["max-results"], ["5"])
- def test_nested_resources(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- self.assertTrue(getattr(zoo, 'animals'))
- request = zoo.my().favorites().list(max_results="5")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['max-results'], ['5'])
+ @unittest.skipIf(six.PY3, "print is not a reserved name in Python 3")
+ def test_methods_with_reserved_names(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ self.assertTrue(getattr(zoo, "animals"))
+ request = zoo.global_().print_().assert_(max_results="5")
+ parsed = urlparse(request.uri)
+ self.assertEqual(parsed[2], "/zoo/v1/global/print/assert")
- @unittest.skipIf(six.PY3, 'print is not a reserved name in Python 3')
- def test_methods_with_reserved_names(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- self.assertTrue(getattr(zoo, 'animals'))
- request = zoo.global_().print_().assert_(max_results="5")
- parsed = urlparse(request.uri)
- self.assertEqual(parsed[2], '/zoo/v1/global/print/assert')
+ def test_top_level_functions(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ self.assertTrue(getattr(zoo, "query"))
+ request = zoo.query(q="foo")
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["q"], ["foo"])
- def test_top_level_functions(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- self.assertTrue(getattr(zoo, 'query'))
- request = zoo.query(q="foo")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['q'], ['foo'])
+ def test_simple_media_uploads(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ doc = getattr(zoo.animals().insert, "__doc__")
+ self.assertTrue("media_body" in doc)
- def test_simple_media_uploads(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- doc = getattr(zoo.animals().insert, '__doc__')
- self.assertTrue('media_body' in doc)
+ def test_simple_media_upload_no_max_size_provided(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
+ request = zoo.animals().crossbreed(media_body=datafile("small.png"))
+ self.assertEquals("image/png", request.headers["content-type"])
+ self.assertEquals(b"PNG", request.body[1:4])
- def test_simple_media_upload_no_max_size_provided(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
- request = zoo.animals().crossbreed(media_body=datafile('small.png'))
- self.assertEquals('image/png', request.headers['content-type'])
- self.assertEquals(b'PNG', request.body[1:4])
+ def test_simple_media_raise_correct_exceptions(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_simple_media_raise_correct_exceptions(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ try:
+ zoo.animals().insert(media_body=datafile("smiley.png"))
+ self.fail("should throw exception if media is too large.")
+ except MediaUploadSizeError:
+ pass
- try:
- zoo.animals().insert(media_body=datafile('smiley.png'))
- self.fail("should throw exception if media is too large.")
- except MediaUploadSizeError:
- pass
+ try:
+ zoo.animals().insert(media_body=datafile("small.jpg"))
+ self.fail("should throw exception if mimetype is unacceptable.")
+ except UnacceptableMimeTypeError:
+ pass
- try:
- zoo.animals().insert(media_body=datafile('small.jpg'))
- self.fail("should throw exception if mimetype is unacceptable.")
- except UnacceptableMimeTypeError:
- pass
+ def test_simple_media_good_upload(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_simple_media_good_upload(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ request = zoo.animals().insert(media_body=datafile("small.png"))
+ self.assertEquals("image/png", request.headers["content-type"])
+ self.assertEquals(b"PNG", request.body[1:4])
+ assertUrisEqual(
+ self,
+ "https://www.googleapis.com/upload/zoo/v1/animals?uploadType=media&alt=json",
+ request.uri,
+ )
- request = zoo.animals().insert(media_body=datafile('small.png'))
- self.assertEquals('image/png', request.headers['content-type'])
- self.assertEquals(b'PNG', request.body[1:4])
- assertUrisEqual(self,
- 'https://www.googleapis.com/upload/zoo/v1/animals?uploadType=media&alt=json',
- request.uri)
+ def test_simple_media_unknown_mimetype(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_simple_media_unknown_mimetype(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ try:
+ zoo.animals().insert(media_body=datafile("small-png"))
+ self.fail("should throw exception if mimetype is unknown.")
+ except UnknownFileType:
+ pass
- try:
- zoo.animals().insert(media_body=datafile('small-png'))
- self.fail("should throw exception if mimetype is unknown.")
- except UnknownFileType:
- pass
+ request = zoo.animals().insert(
+ media_body=datafile("small-png"), media_mime_type="image/png"
+ )
+ self.assertEquals("image/png", request.headers["content-type"])
+ self.assertEquals(b"PNG", request.body[1:4])
+ assertUrisEqual(
+ self,
+ "https://www.googleapis.com/upload/zoo/v1/animals?uploadType=media&alt=json",
+ request.uri,
+ )
- request = zoo.animals().insert(media_body=datafile('small-png'),
- media_mime_type='image/png')
- self.assertEquals('image/png', request.headers['content-type'])
- self.assertEquals(b'PNG', request.body[1:4])
- assertUrisEqual(self,
- 'https://www.googleapis.com/upload/zoo/v1/animals?uploadType=media&alt=json',
- request.uri)
+ def test_multipart_media_raise_correct_exceptions(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_multipart_media_raise_correct_exceptions(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ try:
+ zoo.animals().insert(media_body=datafile("smiley.png"), body={})
+ self.fail("should throw exception if media is too large.")
+ except MediaUploadSizeError:
+ pass
- try:
- zoo.animals().insert(media_body=datafile('smiley.png'), body={})
- self.fail("should throw exception if media is too large.")
- except MediaUploadSizeError:
- pass
+ try:
+ zoo.animals().insert(media_body=datafile("small.jpg"), body={})
+ self.fail("should throw exception if mimetype is unacceptable.")
+ except UnacceptableMimeTypeError:
+ pass
- try:
- zoo.animals().insert(media_body=datafile('small.jpg'), body={})
- self.fail("should throw exception if mimetype is unacceptable.")
- except UnacceptableMimeTypeError:
- pass
+ def test_multipart_media_good_upload(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_multipart_media_good_upload(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ request = zoo.animals().insert(media_body=datafile("small.png"), body={})
+ self.assertTrue(request.headers["content-type"].startswith("multipart/related"))
+ with open(datafile("small.png"), "rb") as f:
+ contents = f.read()
+ boundary = re.match(b"--=+([^=]+)", request.body).group(1)
+ self.assertEqual(
+ request.body.rstrip(b"\n"), # Python 2.6 does not add a trailing \n
+ b"--==============="
+ + boundary
+ + b"==\n"
+ + b"Content-Type: application/json\n"
+ + b"MIME-Version: 1.0\n\n"
+ + b'{"data": {}}\n'
+ + b"--==============="
+ + boundary
+ + b"==\n"
+ + b"Content-Type: image/png\n"
+ + b"MIME-Version: 1.0\n"
+ + b"Content-Transfer-Encoding: binary\n\n"
+ + contents
+ + b"\n--==============="
+ + boundary
+ + b"==--",
+ )
+ assertUrisEqual(
+ self,
+ "https://www.googleapis.com/upload/zoo/v1/animals?uploadType=multipart&alt=json",
+ request.uri,
+ )
- request = zoo.animals().insert(media_body=datafile('small.png'), body={})
- self.assertTrue(request.headers['content-type'].startswith(
- 'multipart/related'))
- with open(datafile('small.png'), 'rb') as f:
- contents = f.read()
- boundary = re.match(b'--=+([^=]+)', request.body).group(1)
- self.assertEqual(
- request.body.rstrip(b"\n"), # Python 2.6 does not add a trailing \n
- b'--===============' + boundary + b'==\n' +
- b'Content-Type: application/json\n' +
- b'MIME-Version: 1.0\n\n' +
- b'{"data": {}}\n' +
- b'--===============' + boundary + b'==\n' +
- b'Content-Type: image/png\n' +
- b'MIME-Version: 1.0\n' +
- b'Content-Transfer-Encoding: binary\n\n' +
- contents +
- b'\n--===============' + boundary + b'==--')
- assertUrisEqual(self,
- 'https://www.googleapis.com/upload/zoo/v1/animals?uploadType=multipart&alt=json',
- request.uri)
+ def test_media_capable_method_without_media(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_media_capable_method_without_media(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ request = zoo.animals().insert(body={})
+ self.assertTrue(request.headers["content-type"], "application/json")
- request = zoo.animals().insert(body={})
- self.assertTrue(request.headers['content-type'], 'application/json')
+ def test_resumable_multipart_media_good_upload(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_resumable_multipart_media_good_upload(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ media_upload = MediaFileUpload(datafile("small.png"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body={})
+ self.assertTrue(request.headers["content-type"].startswith("application/json"))
+ self.assertEquals('{"data": {}}', request.body)
+ self.assertEquals(media_upload, request.resumable)
- media_upload = MediaFileUpload(datafile('small.png'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body={})
- self.assertTrue(request.headers['content-type'].startswith(
- 'application/json'))
- self.assertEquals('{"data": {}}', request.body)
- self.assertEquals(media_upload, request.resumable)
+ self.assertEquals("image/png", request.resumable.mimetype())
- self.assertEquals('image/png', request.resumable.mimetype())
+ self.assertNotEquals(request.body, None)
+ self.assertEquals(request.resumable_uri, None)
- self.assertNotEquals(request.body, None)
- self.assertEquals(request.resumable_uri, None)
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "308", "location": "http://upload.example.com/2"}, ""),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/3",
+ "range": "0-12",
+ },
+ "",
+ ),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/4",
+ "range": "0-%d" % (media_upload.size() - 2),
+ },
+ "",
+ ),
+ ({"status": "200"}, '{"foo": "bar"}'),
+ ]
+ )
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/2'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/3',
- 'range': '0-12'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/4',
- 'range': '0-%d' % (media_upload.size() - 2)}, ''),
- ({'status': '200'}, '{"foo": "bar"}'),
- ])
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(None, body)
+ self.assertTrue(isinstance(status, MediaUploadProgress))
+ self.assertEquals(0, status.resumable_progress)
- status, body = request.next_chunk(http=http)
- self.assertEquals(None, body)
- self.assertTrue(isinstance(status, MediaUploadProgress))
- self.assertEquals(0, status.resumable_progress)
+ # Two requests should have been made and the resumable_uri should have been
+ # updated for each one.
+ self.assertEquals(request.resumable_uri, "http://upload.example.com/2")
+ self.assertEquals(media_upload, request.resumable)
+ self.assertEquals(0, request.resumable_progress)
- # Two requests should have been made and the resumable_uri should have been
- # updated for each one.
- self.assertEquals(request.resumable_uri, 'http://upload.example.com/2')
- self.assertEquals(media_upload, request.resumable)
- self.assertEquals(0, request.resumable_progress)
-
- # This next chuck call should upload the first chunk
- status, body = request.next_chunk(http=http)
- self.assertEquals(request.resumable_uri, 'http://upload.example.com/3')
- self.assertEquals(media_upload, request.resumable)
- self.assertEquals(13, request.resumable_progress)
+ # This next chuck call should upload the first chunk
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(request.resumable_uri, "http://upload.example.com/3")
+ self.assertEquals(media_upload, request.resumable)
+ self.assertEquals(13, request.resumable_progress)
- # This call will upload the next chunk
- status, body = request.next_chunk(http=http)
- self.assertEquals(request.resumable_uri, 'http://upload.example.com/4')
- self.assertEquals(media_upload.size()-1, request.resumable_progress)
- self.assertEquals('{"data": {}}', request.body)
+ # This call will upload the next chunk
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(request.resumable_uri, "http://upload.example.com/4")
+ self.assertEquals(media_upload.size() - 1, request.resumable_progress)
+ self.assertEquals('{"data": {}}', request.body)
- # Final call to next_chunk should complete the upload.
- status, body = request.next_chunk(http=http)
- self.assertEquals(body, {"foo": "bar"})
- self.assertEquals(status, None)
+ # Final call to next_chunk should complete the upload.
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(body, {"foo": "bar"})
+ self.assertEquals(status, None)
+ def test_resumable_media_good_upload(self):
+ """Not a multipart upload."""
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_resumable_media_good_upload(self):
- """Not a multipart upload."""
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ media_upload = MediaFileUpload(datafile("small.png"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
+ self.assertEquals(media_upload, request.resumable)
- media_upload = MediaFileUpload(datafile('small.png'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body=None)
- self.assertEquals(media_upload, request.resumable)
+ self.assertEquals("image/png", request.resumable.mimetype())
- self.assertEquals('image/png', request.resumable.mimetype())
+ self.assertEquals(request.body, None)
+ self.assertEquals(request.resumable_uri, None)
- self.assertEquals(request.body, None)
- self.assertEquals(request.resumable_uri, None)
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/2",
+ "range": "0-12",
+ },
+ "",
+ ),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/3",
+ "range": "0-%d" % (media_upload.size() - 2),
+ },
+ "",
+ ),
+ ({"status": "200"}, '{"foo": "bar"}'),
+ ]
+ )
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/2',
- 'range': '0-12'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/3',
- 'range': '0-%d' % (media_upload.size() - 2)}, ''),
- ({'status': '200'}, '{"foo": "bar"}'),
- ])
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(None, body)
+ self.assertTrue(isinstance(status, MediaUploadProgress))
+ self.assertEquals(13, status.resumable_progress)
+
+ # Two requests should have been made and the resumable_uri should have been
+ # updated for each one.
+ self.assertEquals(request.resumable_uri, "http://upload.example.com/2")
- status, body = request.next_chunk(http=http)
- self.assertEquals(None, body)
- self.assertTrue(isinstance(status, MediaUploadProgress))
- self.assertEquals(13, status.resumable_progress)
+ self.assertEquals(media_upload, request.resumable)
+ self.assertEquals(13, request.resumable_progress)
- # Two requests should have been made and the resumable_uri should have been
- # updated for each one.
- self.assertEquals(request.resumable_uri, 'http://upload.example.com/2')
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(request.resumable_uri, "http://upload.example.com/3")
+ self.assertEquals(media_upload.size() - 1, request.resumable_progress)
+ self.assertEquals(request.body, None)
- self.assertEquals(media_upload, request.resumable)
- self.assertEquals(13, request.resumable_progress)
+ # Final call to next_chunk should complete the upload.
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(body, {"foo": "bar"})
+ self.assertEquals(status, None)
- status, body = request.next_chunk(http=http)
- self.assertEquals(request.resumable_uri, 'http://upload.example.com/3')
- self.assertEquals(media_upload.size()-1, request.resumable_progress)
- self.assertEquals(request.body, None)
+ def test_resumable_media_good_upload_from_execute(self):
+ """Not a multipart upload."""
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- # Final call to next_chunk should complete the upload.
- status, body = request.next_chunk(http=http)
- self.assertEquals(body, {"foo": "bar"})
- self.assertEquals(status, None)
+ media_upload = MediaFileUpload(datafile("small.png"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
+ assertUrisEqual(
+ self,
+ "https://www.googleapis.com/upload/zoo/v1/animals?uploadType=resumable&alt=json",
+ request.uri,
+ )
- def test_resumable_media_good_upload_from_execute(self):
- """Not a multipart upload."""
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/2",
+ "range": "0-12",
+ },
+ "",
+ ),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/3",
+ "range": "0-%d" % media_upload.size(),
+ },
+ "",
+ ),
+ ({"status": "200"}, '{"foo": "bar"}'),
+ ]
+ )
- media_upload = MediaFileUpload(datafile('small.png'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body=None)
- assertUrisEqual(self,
- 'https://www.googleapis.com/upload/zoo/v1/animals?uploadType=resumable&alt=json',
- request.uri)
+ body = request.execute(http=http)
+ self.assertEquals(body, {"foo": "bar"})
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/2',
- 'range': '0-12'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/3',
- 'range': '0-%d' % media_upload.size()}, ''),
- ({'status': '200'}, '{"foo": "bar"}'),
- ])
+ def test_resumable_media_fail_unknown_response_code_first_request(self):
+ """Not a multipart upload."""
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- body = request.execute(http=http)
- self.assertEquals(body, {"foo": "bar"})
+ media_upload = MediaFileUpload(datafile("small.png"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
- def test_resumable_media_fail_unknown_response_code_first_request(self):
- """Not a multipart upload."""
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ http = HttpMockSequence(
+ [({"status": "400", "location": "http://upload.example.com"}, "")]
+ )
- media_upload = MediaFileUpload(datafile('small.png'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body=None)
+ try:
+ request.execute(http=http)
+ self.fail("Should have raised ResumableUploadError.")
+ except ResumableUploadError as e:
+ self.assertEqual(400, e.resp.status)
- http = HttpMockSequence([
- ({'status': '400',
- 'location': 'http://upload.example.com'}, ''),
- ])
+ def test_resumable_media_fail_unknown_response_code_subsequent_request(self):
+ """Not a multipart upload."""
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- try:
- request.execute(http=http)
- self.fail('Should have raised ResumableUploadError.')
- except ResumableUploadError as e:
- self.assertEqual(400, e.resp.status)
+ media_upload = MediaFileUpload(datafile("small.png"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
- def test_resumable_media_fail_unknown_response_code_subsequent_request(self):
- """Not a multipart upload."""
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "400"}, ""),
+ ]
+ )
- media_upload = MediaFileUpload(datafile('small.png'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body=None)
+ self.assertRaises(HttpError, request.execute, http=http)
+ self.assertTrue(request._in_error_state)
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '400'}, ''),
- ])
+ http = HttpMockSequence(
+ [
+ ({"status": "308", "range": "0-5"}, ""),
+ ({"status": "308", "range": "0-6"}, ""),
+ ]
+ )
- self.assertRaises(HttpError, request.execute, http=http)
- self.assertTrue(request._in_error_state)
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(
+ status.resumable_progress,
+ 7,
+ "Should have first checked length and then tried to PUT more.",
+ )
+ self.assertFalse(request._in_error_state)
- http = HttpMockSequence([
- ({'status': '308',
- 'range': '0-5'}, ''),
- ({'status': '308',
- 'range': '0-6'}, ''),
- ])
+ # Put it back in an error state.
+ http = HttpMockSequence([({"status": "400"}, "")])
+ self.assertRaises(HttpError, request.execute, http=http)
+ self.assertTrue(request._in_error_state)
- status, body = request.next_chunk(http=http)
- self.assertEquals(status.resumable_progress, 7,
- 'Should have first checked length and then tried to PUT more.')
- self.assertFalse(request._in_error_state)
+ # Pretend the last request that 400'd actually succeeded.
+ http = HttpMockSequence([({"status": "200"}, '{"foo": "bar"}')])
+ status, body = request.next_chunk(http=http)
+ self.assertEqual(body, {"foo": "bar"})
- # Put it back in an error state.
- http = HttpMockSequence([
- ({'status': '400'}, ''),
- ])
- self.assertRaises(HttpError, request.execute, http=http)
- self.assertTrue(request._in_error_state)
+ def test_media_io_base_stream_unlimited_chunksize_resume(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- # Pretend the last request that 400'd actually succeeded.
- http = HttpMockSequence([
- ({'status': '200'}, '{"foo": "bar"}'),
- ])
- status, body = request.next_chunk(http=http)
- self.assertEqual(body, {'foo': 'bar'})
+ # Set up a seekable stream and try to upload in single chunk.
+ fd = BytesIO(b'01234"56789"')
+ media_upload = MediaIoBaseUpload(
+ fd=fd, mimetype="text/plain", chunksize=-1, resumable=True
+ )
- def test_media_io_base_stream_unlimited_chunksize_resume(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
- # Set up a seekable stream and try to upload in single chunk.
- fd = BytesIO(b'01234"56789"')
- media_upload = MediaIoBaseUpload(
- fd=fd, mimetype='text/plain', chunksize=-1, resumable=True)
+ # The single chunk fails, restart at the right point.
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/2",
+ "range": "0-4",
+ },
+ "",
+ ),
+ ({"status": "200"}, "echo_request_body"),
+ ]
+ )
- request = zoo.animals().insert(media_body=media_upload, body=None)
+ body = request.execute(http=http)
+ self.assertEqual("56789", body)
- # The single chunk fails, restart at the right point.
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/2',
- 'range': '0-4'}, ''),
- ({'status': '200'}, 'echo_request_body'),
- ])
+ def test_media_io_base_stream_chunksize_resume(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- body = request.execute(http=http)
- self.assertEqual('56789', body)
+ # Set up a seekable stream and try to upload in chunks.
+ fd = BytesIO(b"0123456789")
+ media_upload = MediaIoBaseUpload(
+ fd=fd, mimetype="text/plain", chunksize=5, resumable=True
+ )
- def test_media_io_base_stream_chunksize_resume(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
- # Set up a seekable stream and try to upload in chunks.
- fd = BytesIO(b'0123456789')
- media_upload = MediaIoBaseUpload(
- fd=fd, mimetype='text/plain', chunksize=5, resumable=True)
+ # The single chunk fails, pull the content sent out of the exception.
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "400"}, "echo_request_body"),
+ ]
+ )
- request = zoo.animals().insert(media_body=media_upload, body=None)
+ try:
+ body = request.execute(http=http)
+ except HttpError as e:
+ self.assertEqual(b"01234", e.content)
- # The single chunk fails, pull the content sent out of the exception.
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '400'}, 'echo_request_body'),
- ])
+ def test_resumable_media_handle_uploads_of_unknown_size(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "200"}, "echo_request_headers_as_json"),
+ ]
+ )
- try:
- body = request.execute(http=http)
- except HttpError as e:
- self.assertEqual(b'01234', e.content)
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_resumable_media_handle_uploads_of_unknown_size(self):
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '200'}, 'echo_request_headers_as_json'),
- ])
+ # Create an upload that doesn't know the full size of the media.
+ class IoBaseUnknownLength(MediaUpload):
+ def chunksize(self):
+ return 10
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ def mimetype(self):
+ return "image/png"
- # Create an upload that doesn't know the full size of the media.
- class IoBaseUnknownLength(MediaUpload):
- def chunksize(self):
- return 10
+ def size(self):
+ return None
- def mimetype(self):
- return 'image/png'
+ def resumable(self):
+ return True
- def size(self):
- return None
+ def getbytes(self, begin, length):
+ return "0123456789"
- def resumable(self):
- return True
+ upload = IoBaseUnknownLength()
- def getbytes(self, begin, length):
- return '0123456789'
+ request = zoo.animals().insert(media_body=upload, body=None)
+ status, body = request.next_chunk(http=http)
+ self.assertEqual(body, {"Content-Range": "bytes 0-9/*", "Content-Length": "10"})
- upload = IoBaseUnknownLength()
+ def test_resumable_media_no_streaming_on_unsupported_platforms(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- request = zoo.animals().insert(media_body=upload, body=None)
- status, body = request.next_chunk(http=http)
- self.assertEqual(body, {
- 'Content-Range': 'bytes 0-9/*',
- 'Content-Length': '10',
- })
+ class IoBaseHasStream(MediaUpload):
+ def chunksize(self):
+ return 10
- def test_resumable_media_no_streaming_on_unsupported_platforms(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ def mimetype(self):
+ return "image/png"
- class IoBaseHasStream(MediaUpload):
- def chunksize(self):
- return 10
+ def size(self):
+ return None
- def mimetype(self):
- return 'image/png'
+ def resumable(self):
+ return True
- def size(self):
- return None
+ def getbytes(self, begin, length):
+ return "0123456789"
- def resumable(self):
- return True
+ def has_stream(self):
+ return True
- def getbytes(self, begin, length):
- return '0123456789'
+ def stream(self):
+ raise NotImplementedError()
- def has_stream(self):
- return True
+ upload = IoBaseHasStream()
- def stream(self):
- raise NotImplementedError()
+ orig_version = sys.version_info
- upload = IoBaseHasStream()
+ sys.version_info = (2, 6, 5, "final", 0)
- orig_version = sys.version_info
+ request = zoo.animals().insert(media_body=upload, body=None)
- sys.version_info = (2, 6, 5, 'final', 0)
+ # This should raise an exception because stream() will be called.
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "200"}, "echo_request_headers_as_json"),
+ ]
+ )
- request = zoo.animals().insert(media_body=upload, body=None)
+ self.assertRaises(NotImplementedError, request.next_chunk, http=http)
- # This should raise an exception because stream() will be called.
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '200'}, 'echo_request_headers_as_json'),
- ])
+ sys.version_info = orig_version
- self.assertRaises(NotImplementedError, request.next_chunk, http=http)
+ def test_resumable_media_handle_uploads_of_unknown_size_eof(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "200"}, "echo_request_headers_as_json"),
+ ]
+ )
- sys.version_info = orig_version
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_resumable_media_handle_uploads_of_unknown_size_eof(self):
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '200'}, 'echo_request_headers_as_json'),
- ])
+ fd = BytesIO(b"data goes here")
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ # Create an upload that doesn't know the full size of the media.
+ upload = MediaIoBaseUpload(
+ fd=fd, mimetype="image/png", chunksize=15, resumable=True
+ )
- fd = BytesIO(b'data goes here')
+ request = zoo.animals().insert(media_body=upload, body=None)
+ status, body = request.next_chunk(http=http)
+ self.assertEqual(
+ body, {"Content-Range": "bytes 0-13/14", "Content-Length": "14"}
+ )
- # Create an upload that doesn't know the full size of the media.
- upload = MediaIoBaseUpload(
- fd=fd, mimetype='image/png', chunksize=15, resumable=True)
+ def test_resumable_media_handle_resume_of_upload_of_unknown_size(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ ({"status": "400"}, ""),
+ ]
+ )
- request = zoo.animals().insert(media_body=upload, body=None)
- status, body = request.next_chunk(http=http)
- self.assertEqual(body, {
- 'Content-Range': 'bytes 0-13/14',
- 'Content-Length': '14',
- })
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- def test_resumable_media_handle_resume_of_upload_of_unknown_size(self):
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '400'}, ''),
- ])
+ # Create an upload that doesn't know the full size of the media.
+ fd = BytesIO(b"data goes here")
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ upload = MediaIoBaseUpload(
+ fd=fd, mimetype="image/png", chunksize=500, resumable=True
+ )
- # Create an upload that doesn't know the full size of the media.
- fd = BytesIO(b'data goes here')
+ request = zoo.animals().insert(media_body=upload, body=None)
- upload = MediaIoBaseUpload(
- fd=fd, mimetype='image/png', chunksize=500, resumable=True)
+ # Put it in an error state.
+ self.assertRaises(HttpError, request.next_chunk, http=http)
- request = zoo.animals().insert(media_body=upload, body=None)
+ http = HttpMockSequence(
+ [({"status": "400", "range": "0-5"}, "echo_request_headers_as_json")]
+ )
+ try:
+ # Should resume the upload by first querying the status of the upload.
+ request.next_chunk(http=http)
+ except HttpError as e:
+ expected = {"Content-Range": "bytes */14", "content-length": "0"}
+ self.assertEqual(
+ expected,
+ json.loads(e.content.decode("utf-8")),
+ "Should send an empty body when requesting the current upload status.",
+ )
- # Put it in an error state.
- self.assertRaises(HttpError, request.next_chunk, http=http)
+ def test_pickle(self):
+ sorted_resource_keys = [
+ "_baseUrl",
+ "_developerKey",
+ "_dynamic_attrs",
+ "_http",
+ "_model",
+ "_requestBuilder",
+ "_resourceDesc",
+ "_rootDesc",
+ "_schema",
+ "animals",
+ "global_",
+ "load",
+ "loadNoTemplate",
+ "my",
+ "new_batch_http_request",
+ "query",
+ "scopedAnimals",
+ ]
- http = HttpMockSequence([
- ({'status': '400',
- 'range': '0-5'}, 'echo_request_headers_as_json'),
- ])
- try:
- # Should resume the upload by first querying the status of the upload.
- request.next_chunk(http=http)
- except HttpError as e:
- expected = {
- 'Content-Range': 'bytes */14',
- 'content-length': '0'
- }
- self.assertEqual(expected, json.loads(e.content.decode('utf-8')),
- 'Should send an empty body when requesting the current upload status.')
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ self.assertEqual(sorted(zoo.__dict__.keys()), sorted_resource_keys)
- def test_pickle(self):
- sorted_resource_keys = ['_baseUrl',
- '_developerKey',
- '_dynamic_attrs',
- '_http',
- '_model',
- '_requestBuilder',
- '_resourceDesc',
- '_rootDesc',
- '_schema',
- 'animals',
- 'global_',
- 'load',
- 'loadNoTemplate',
- 'my',
- 'new_batch_http_request',
- 'query',
- 'scopedAnimals']
+ pickled_zoo = pickle.dumps(zoo)
+ new_zoo = pickle.loads(pickled_zoo)
+ self.assertEqual(sorted(new_zoo.__dict__.keys()), sorted_resource_keys)
+ self.assertTrue(hasattr(new_zoo, "animals"))
+ self.assertTrue(callable(new_zoo.animals))
+ self.assertTrue(hasattr(new_zoo, "global_"))
+ self.assertTrue(callable(new_zoo.global_))
+ self.assertTrue(hasattr(new_zoo, "load"))
+ self.assertTrue(callable(new_zoo.load))
+ self.assertTrue(hasattr(new_zoo, "loadNoTemplate"))
+ self.assertTrue(callable(new_zoo.loadNoTemplate))
+ self.assertTrue(hasattr(new_zoo, "my"))
+ self.assertTrue(callable(new_zoo.my))
+ self.assertTrue(hasattr(new_zoo, "query"))
+ self.assertTrue(callable(new_zoo.query))
+ self.assertTrue(hasattr(new_zoo, "scopedAnimals"))
+ self.assertTrue(callable(new_zoo.scopedAnimals))
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- self.assertEqual(sorted(zoo.__dict__.keys()), sorted_resource_keys)
+ self.assertEqual(sorted(zoo._dynamic_attrs), sorted(new_zoo._dynamic_attrs))
+ self.assertEqual(zoo._baseUrl, new_zoo._baseUrl)
+ self.assertEqual(zoo._developerKey, new_zoo._developerKey)
+ self.assertEqual(zoo._requestBuilder, new_zoo._requestBuilder)
+ self.assertEqual(zoo._resourceDesc, new_zoo._resourceDesc)
+ self.assertEqual(zoo._rootDesc, new_zoo._rootDesc)
+ # _http, _model and _schema won't be equal since we will get new
+ # instances upon un-pickling
- pickled_zoo = pickle.dumps(zoo)
- new_zoo = pickle.loads(pickled_zoo)
- self.assertEqual(sorted(new_zoo.__dict__.keys()), sorted_resource_keys)
- self.assertTrue(hasattr(new_zoo, 'animals'))
- self.assertTrue(callable(new_zoo.animals))
- self.assertTrue(hasattr(new_zoo, 'global_'))
- self.assertTrue(callable(new_zoo.global_))
- self.assertTrue(hasattr(new_zoo, 'load'))
- self.assertTrue(callable(new_zoo.load))
- self.assertTrue(hasattr(new_zoo, 'loadNoTemplate'))
- self.assertTrue(callable(new_zoo.loadNoTemplate))
- self.assertTrue(hasattr(new_zoo, 'my'))
- self.assertTrue(callable(new_zoo.my))
- self.assertTrue(hasattr(new_zoo, 'query'))
- self.assertTrue(callable(new_zoo.query))
- self.assertTrue(hasattr(new_zoo, 'scopedAnimals'))
- self.assertTrue(callable(new_zoo.scopedAnimals))
+ def _dummy_zoo_request(self):
+ with open(os.path.join(DATA_DIR, "zoo.json"), "rU") as fh:
+ zoo_contents = fh.read()
- self.assertEqual(sorted(zoo._dynamic_attrs), sorted(new_zoo._dynamic_attrs))
- self.assertEqual(zoo._baseUrl, new_zoo._baseUrl)
- self.assertEqual(zoo._developerKey, new_zoo._developerKey)
- self.assertEqual(zoo._requestBuilder, new_zoo._requestBuilder)
- self.assertEqual(zoo._resourceDesc, new_zoo._resourceDesc)
- self.assertEqual(zoo._rootDesc, new_zoo._rootDesc)
- # _http, _model and _schema won't be equal since we will get new
- # instances upon un-pickling
+ zoo_uri = uritemplate.expand(DISCOVERY_URI, {"api": "zoo", "apiVersion": "v1"})
+ if "REMOTE_ADDR" in os.environ:
+ zoo_uri = util._add_query_parameter(
+ zoo_uri, "userIp", os.environ["REMOTE_ADDR"]
+ )
- def _dummy_zoo_request(self):
- with open(os.path.join(DATA_DIR, 'zoo.json'), 'rU') as fh:
- zoo_contents = fh.read()
+ http = build_http()
+ original_request = http.request
- zoo_uri = uritemplate.expand(DISCOVERY_URI,
- {'api': 'zoo', 'apiVersion': 'v1'})
- if 'REMOTE_ADDR' in os.environ:
- zoo_uri = util._add_query_parameter(zoo_uri, 'userIp',
- os.environ['REMOTE_ADDR'])
+ def wrapped_request(uri, method="GET", *args, **kwargs):
+ if uri == zoo_uri:
+ return httplib2.Response({"status": "200"}), zoo_contents
+ return original_request(uri, method=method, *args, **kwargs)
- http = build_http()
- original_request = http.request
- def wrapped_request(uri, method='GET', *args, **kwargs):
- if uri == zoo_uri:
- return httplib2.Response({'status': '200'}), zoo_contents
- return original_request(uri, method=method, *args, **kwargs)
- http.request = wrapped_request
- return http
+ http.request = wrapped_request
+ return http
- def _dummy_token(self):
- access_token = 'foo'
- client_id = 'some_client_id'
- client_secret = 'cOuDdkfjxxnv+'
- refresh_token = '1/0/a.df219fjls0'
- token_expiry = datetime.datetime.utcnow()
- user_agent = 'refresh_checker/1.0'
- return OAuth2Credentials(
- access_token, client_id, client_secret,
- refresh_token, token_expiry, GOOGLE_TOKEN_URI,
- user_agent)
+ def _dummy_token(self):
+ access_token = "foo"
+ client_id = "some_client_id"
+ client_secret = "cOuDdkfjxxnv+"
+ refresh_token = "1/0/a.df219fjls0"
+ token_expiry = datetime.datetime.utcnow()
+ user_agent = "refresh_checker/1.0"
+ return OAuth2Credentials(
+ access_token,
+ client_id,
+ client_secret,
+ refresh_token,
+ token_expiry,
+ GOOGLE_TOKEN_URI,
+ user_agent,
+ )
- def test_pickle_with_credentials(self):
- credentials = self._dummy_token()
- http = self._dummy_zoo_request()
- http = credentials.authorize(http)
- self.assertTrue(hasattr(http.request, 'credentials'))
+ def test_pickle_with_credentials(self):
+ credentials = self._dummy_token()
+ http = self._dummy_zoo_request()
+ http = credentials.authorize(http)
+ self.assertTrue(hasattr(http.request, "credentials"))
- zoo = build('zoo', 'v1', http=http)
- pickled_zoo = pickle.dumps(zoo)
- new_zoo = pickle.loads(pickled_zoo)
- self.assertEqual(sorted(zoo.__dict__.keys()),
- sorted(new_zoo.__dict__.keys()))
- new_http = new_zoo._http
- self.assertFalse(hasattr(new_http.request, 'credentials'))
+ zoo = build("zoo", "v1", http=http)
+ pickled_zoo = pickle.dumps(zoo)
+ new_zoo = pickle.loads(pickled_zoo)
+ self.assertEqual(sorted(zoo.__dict__.keys()), sorted(new_zoo.__dict__.keys()))
+ new_http = new_zoo._http
+ self.assertFalse(hasattr(new_http.request, "credentials"))
- def test_resumable_media_upload_no_content(self):
- self.http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=self.http)
+ def test_resumable_media_upload_no_content(self):
+ self.http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=self.http)
- media_upload = MediaFileUpload(datafile('empty'), resumable=True)
- request = zoo.animals().insert(media_body=media_upload, body=None)
+ media_upload = MediaFileUpload(datafile("empty"), resumable=True)
+ request = zoo.animals().insert(media_body=media_upload, body=None)
- self.assertEquals(media_upload, request.resumable)
- self.assertEquals(request.body, None)
- self.assertEquals(request.resumable_uri, None)
+ self.assertEquals(media_upload, request.resumable)
+ self.assertEquals(request.body, None)
+ self.assertEquals(request.resumable_uri, None)
- http = HttpMockSequence([
- ({'status': '200',
- 'location': 'http://upload.example.com'}, ''),
- ({'status': '308',
- 'location': 'http://upload.example.com/2',
- 'range': '0-0'}, ''),
- ])
+ http = HttpMockSequence(
+ [
+ ({"status": "200", "location": "http://upload.example.com"}, ""),
+ (
+ {
+ "status": "308",
+ "location": "http://upload.example.com/2",
+ "range": "0-0",
+ },
+ "",
+ ),
+ ]
+ )
- status, body = request.next_chunk(http=http)
- self.assertEquals(None, body)
- self.assertTrue(isinstance(status, MediaUploadProgress))
- self.assertEquals(0, status.progress())
+ status, body = request.next_chunk(http=http)
+ self.assertEquals(None, body)
+ self.assertTrue(isinstance(status, MediaUploadProgress))
+ self.assertEquals(0, status.progress())
class Next(unittest.TestCase):
+ def test_next_successful_none_on_no_next_page_token(self):
+ self.http = HttpMock(datafile("tasks.json"), {"status": "200"})
+ tasks = build("tasks", "v1", http=self.http)
+ request = tasks.tasklists().list()
+ self.assertEqual(None, tasks.tasklists().list_next(request, {}))
- def test_next_successful_none_on_no_next_page_token(self):
- self.http = HttpMock(datafile('tasks.json'), {'status': '200'})
- tasks = build('tasks', 'v1', http=self.http)
- request = tasks.tasklists().list()
- self.assertEqual(None, tasks.tasklists().list_next(request, {}))
+ def test_next_successful_none_on_empty_page_token(self):
+ self.http = HttpMock(datafile("tasks.json"), {"status": "200"})
+ tasks = build("tasks", "v1", http=self.http)
+ request = tasks.tasklists().list()
+ next_request = tasks.tasklists().list_next(request, {"nextPageToken": ""})
+ self.assertEqual(None, next_request)
- def test_next_successful_none_on_empty_page_token(self):
- self.http = HttpMock(datafile('tasks.json'), {'status': '200'})
- tasks = build('tasks', 'v1', http=self.http)
- request = tasks.tasklists().list()
- next_request = tasks.tasklists().list_next(
- request, {'nextPageToken': ''})
- self.assertEqual(None, next_request)
+ def test_next_successful_with_next_page_token(self):
+ self.http = HttpMock(datafile("tasks.json"), {"status": "200"})
+ tasks = build("tasks", "v1", http=self.http)
+ request = tasks.tasklists().list()
+ next_request = tasks.tasklists().list_next(request, {"nextPageToken": "123abc"})
+ parsed = list(urlparse(next_request.uri))
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["pageToken"][0], "123abc")
- def test_next_successful_with_next_page_token(self):
- self.http = HttpMock(datafile('tasks.json'), {'status': '200'})
- tasks = build('tasks', 'v1', http=self.http)
- request = tasks.tasklists().list()
- next_request = tasks.tasklists().list_next(
- request, {'nextPageToken': '123abc'})
- parsed = list(urlparse(next_request.uri))
- q = parse_qs(parsed[4])
- self.assertEqual(q['pageToken'][0], '123abc')
+ def test_next_successful_with_next_page_token_alternate_name(self):
+ self.http = HttpMock(datafile("bigquery.json"), {"status": "200"})
+ bigquery = build("bigquery", "v2", http=self.http)
+ request = bigquery.tabledata().list(datasetId="", projectId="", tableId="")
+ next_request = bigquery.tabledata().list_next(request, {"pageToken": "123abc"})
+ parsed = list(urlparse(next_request.uri))
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["pageToken"][0], "123abc")
- def test_next_successful_with_next_page_token_alternate_name(self):
- self.http = HttpMock(datafile('bigquery.json'), {'status': '200'})
- bigquery = build('bigquery', 'v2', http=self.http)
- request = bigquery.tabledata().list(datasetId='', projectId='', tableId='')
- next_request = bigquery.tabledata().list_next(
- request, {'pageToken': '123abc'})
- parsed = list(urlparse(next_request.uri))
- q = parse_qs(parsed[4])
- self.assertEqual(q['pageToken'][0], '123abc')
+ def test_next_successful_with_next_page_token_in_body(self):
+ self.http = HttpMock(datafile("logging.json"), {"status": "200"})
+ logging = build("logging", "v2", http=self.http)
+ request = logging.entries().list(body={})
+ next_request = logging.entries().list_next(request, {"nextPageToken": "123abc"})
+ body = JsonModel().deserialize(next_request.body)
+ self.assertEqual(body["pageToken"], "123abc")
- def test_next_successful_with_next_page_token_in_body(self):
- self.http = HttpMock(datafile('logging.json'), {'status': '200'})
- logging = build('logging', 'v2', http=self.http)
- request = logging.entries().list(body={})
- next_request = logging.entries().list_next(
- request, {'nextPageToken': '123abc'})
- body = JsonModel().deserialize(next_request.body)
- self.assertEqual(body['pageToken'], '123abc')
+ def test_next_with_method_with_no_properties(self):
+ self.http = HttpMock(datafile("latitude.json"), {"status": "200"})
+ service = build("latitude", "v1", http=self.http)
+ service.currentLocation().get()
- def test_next_with_method_with_no_properties(self):
- self.http = HttpMock(datafile('latitude.json'), {'status': '200'})
- service = build('latitude', 'v1', http=self.http)
- service.currentLocation().get()
+ def test_next_nonexistent_with_no_next_page_token(self):
+ self.http = HttpMock(datafile("drive.json"), {"status": "200"})
+ drive = build("drive", "v3", http=self.http)
+ drive.changes().watch(body={})
+ self.assertFalse(callable(getattr(drive.changes(), "watch_next", None)))
- def test_next_nonexistent_with_no_next_page_token(self):
- self.http = HttpMock(datafile('drive.json'), {'status': '200'})
- drive = build('drive', 'v3', http=self.http)
- drive.changes().watch(body={})
- self.assertFalse(callable(getattr(drive.changes(), 'watch_next', None)))
-
- def test_next_successful_with_next_page_token_required(self):
- self.http = HttpMock(datafile('drive.json'), {'status': '200'})
- drive = build('drive', 'v3', http=self.http)
- request = drive.changes().list(pageToken='startPageToken')
- next_request = drive.changes().list_next(
- request, {'nextPageToken': '123abc'})
- parsed = list(urlparse(next_request.uri))
- q = parse_qs(parsed[4])
- self.assertEqual(q['pageToken'][0], '123abc')
+ def test_next_successful_with_next_page_token_required(self):
+ self.http = HttpMock(datafile("drive.json"), {"status": "200"})
+ drive = build("drive", "v3", http=self.http)
+ request = drive.changes().list(pageToken="startPageToken")
+ next_request = drive.changes().list_next(request, {"nextPageToken": "123abc"})
+ parsed = list(urlparse(next_request.uri))
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["pageToken"][0], "123abc")
class MediaGet(unittest.TestCase):
+ def test_get_media(self):
+ http = HttpMock(datafile("zoo.json"), {"status": "200"})
+ zoo = build("zoo", "v1", http=http)
+ request = zoo.animals().get_media(name="Lion")
- def test_get_media(self):
- http = HttpMock(datafile('zoo.json'), {'status': '200'})
- zoo = build('zoo', 'v1', http=http)
- request = zoo.animals().get_media(name='Lion')
+ parsed = urlparse(request.uri)
+ q = parse_qs(parsed[4])
+ self.assertEqual(q["alt"], ["media"])
+ self.assertEqual(request.headers["accept"], "*/*")
- parsed = urlparse(request.uri)
- q = parse_qs(parsed[4])
- self.assertEqual(q['alt'], ['media'])
- self.assertEqual(request.headers['accept'], '*/*')
-
- http = HttpMockSequence([
- ({'status': '200'}, 'standing in for media'),
- ])
- response = request.execute(http=http)
- self.assertEqual(b'standing in for media', response)
+ http = HttpMockSequence([({"status": "200"}, "standing in for media")])
+ response = request.execute(http=http)
+ self.assertEqual(b"standing in for media", response)
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_discovery_cache.py b/tests/test_discovery_cache.py
index 1786406..2c3efd8 100644
--- a/tests/test_discovery_cache.py
+++ b/tests/test_discovery_cache.py
@@ -26,33 +26,38 @@
from googleapiclient.discovery_cache.base import Cache
try:
- from googleapiclient.discovery_cache.file_cache import Cache as FileCache
+ from googleapiclient.discovery_cache.file_cache import Cache as FileCache
except ImportError:
- FileCache = None
+ FileCache = None
-@unittest.skipIf(FileCache is None, 'FileCache unavailable.')
+@unittest.skipIf(FileCache is None, "FileCache unavailable.")
class FileCacheTest(unittest.TestCase):
- @mock.patch('googleapiclient.discovery_cache.file_cache.FILENAME',
- new='google-api-python-client-discovery-doc-tests.cache')
- def test_expiration(self):
- def future_now():
- return datetime.datetime.now() + datetime.timedelta(
- seconds=DISCOVERY_DOC_MAX_AGE)
- mocked_datetime = mock.MagicMock()
- mocked_datetime.datetime.now.side_effect = future_now
- cache = FileCache(max_age=DISCOVERY_DOC_MAX_AGE)
- first_url = 'url-1'
- first_url_content = 'url-1-content'
- cache.set(first_url, first_url_content)
+ @mock.patch(
+ "googleapiclient.discovery_cache.file_cache.FILENAME",
+ new="google-api-python-client-discovery-doc-tests.cache",
+ )
+ def test_expiration(self):
+ def future_now():
+ return datetime.datetime.now() + datetime.timedelta(
+ seconds=DISCOVERY_DOC_MAX_AGE
+ )
- # Make sure the content is cached.
- self.assertEqual(first_url_content, cache.get(first_url))
+ mocked_datetime = mock.MagicMock()
+ mocked_datetime.datetime.now.side_effect = future_now
+ cache = FileCache(max_age=DISCOVERY_DOC_MAX_AGE)
+ first_url = "url-1"
+ first_url_content = "url-1-content"
+ cache.set(first_url, first_url_content)
- # Simulate another `set` call in the future date.
- with mock.patch('googleapiclient.discovery_cache.file_cache.datetime',
- new=mocked_datetime):
- cache.set('url-2', 'url-2-content')
-
- # Make sure the content is expired
- self.assertEqual(None, cache.get(first_url))
+ # Make sure the content is cached.
+ self.assertEqual(first_url_content, cache.get(first_url))
+
+ # Simulate another `set` call in the future date.
+ with mock.patch(
+ "googleapiclient.discovery_cache.file_cache.datetime", new=mocked_datetime
+ ):
+ cache.set("url-2", "url-2-content")
+
+ # Make sure the content is expired
+ self.assertEqual(None, cache.get(first_url))
diff --git a/tests/test_errors.py b/tests/test_errors.py
index e4d2f09..b0d1e43 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -18,7 +18,7 @@
"""
from __future__ import absolute_import
-__author__ = 'afshar@google.com (Ali Afshar)'
+__author__ = "afshar@google.com (Ali Afshar)"
import unittest2 as unittest
@@ -47,55 +47,68 @@
}
"""
-def fake_response(data, headers, reason='Ok'):
- response = httplib2.Response(headers)
- response.reason = reason
- return response, data
+
+def fake_response(data, headers, reason="Ok"):
+ response = httplib2.Response(headers)
+ response.reason = reason
+ return response, data
class Error(unittest.TestCase):
- """Test handling of error bodies."""
+ """Test handling of error bodies."""
- def test_json_body(self):
- """Test a nicely formed, expected error response."""
- resp, content = fake_response(JSON_ERROR_CONTENT,
- {'status':'400', 'content-type': 'application/json'},
- reason='Failed')
- error = HttpError(resp, content, uri='http://example.org')
- self.assertEqual(str(error), '<HttpError 400 when requesting http://example.org returned "country is required". Details: "error details">')
+ def test_json_body(self):
+ """Test a nicely formed, expected error response."""
+ resp, content = fake_response(
+ JSON_ERROR_CONTENT,
+ {"status": "400", "content-type": "application/json"},
+ reason="Failed",
+ )
+ error = HttpError(resp, content, uri="http://example.org")
+ self.assertEqual(
+ str(error),
+ '<HttpError 400 when requesting http://example.org returned "country is required". Details: "error details">',
+ )
- def test_bad_json_body(self):
- """Test handling of bodies with invalid json."""
- resp, content = fake_response(b'{',
- { 'status':'400', 'content-type': 'application/json'},
- reason='Failed')
- error = HttpError(resp, content)
- self.assertEqual(str(error), '<HttpError 400 "Failed">')
+ def test_bad_json_body(self):
+ """Test handling of bodies with invalid json."""
+ resp, content = fake_response(
+ b"{", {"status": "400", "content-type": "application/json"}, reason="Failed"
+ )
+ error = HttpError(resp, content)
+ self.assertEqual(str(error), '<HttpError 400 "Failed">')
- def test_with_uri(self):
- """Test handling of passing in the request uri."""
- resp, content = fake_response(b'{',
- {'status':'400', 'content-type': 'application/json'},
- reason='Failure')
- error = HttpError(resp, content, uri='http://example.org')
- self.assertEqual(str(error), '<HttpError 400 when requesting http://example.org returned "Failure">')
+ def test_with_uri(self):
+ """Test handling of passing in the request uri."""
+ resp, content = fake_response(
+ b"{",
+ {"status": "400", "content-type": "application/json"},
+ reason="Failure",
+ )
+ error = HttpError(resp, content, uri="http://example.org")
+ self.assertEqual(
+ str(error),
+ '<HttpError 400 when requesting http://example.org returned "Failure">',
+ )
- def test_missing_message_json_body(self):
- """Test handling of bodies with missing expected 'message' element."""
- resp, content = fake_response(b'{}',
- {'status':'400', 'content-type': 'application/json'},
- reason='Failed')
- error = HttpError(resp, content)
- self.assertEqual(str(error), '<HttpError 400 "Failed">')
+ def test_missing_message_json_body(self):
+ """Test handling of bodies with missing expected 'message' element."""
+ resp, content = fake_response(
+ b"{}",
+ {"status": "400", "content-type": "application/json"},
+ reason="Failed",
+ )
+ error = HttpError(resp, content)
+ self.assertEqual(str(error), '<HttpError 400 "Failed">')
- def test_non_json(self):
- """Test handling of non-JSON bodies"""
- resp, content = fake_response(b'}NOT OK', {'status':'400'})
- error = HttpError(resp, content)
- self.assertEqual(str(error), '<HttpError 400 "Ok">')
+ def test_non_json(self):
+ """Test handling of non-JSON bodies"""
+ resp, content = fake_response(b"}NOT OK", {"status": "400"})
+ error = HttpError(resp, content)
+ self.assertEqual(str(error), '<HttpError 400 "Ok">')
- def test_missing_reason(self):
- """Test an empty dict with a missing resp.reason."""
- resp, content = fake_response(b'}NOT OK', {'status': '400'}, reason=None)
- error = HttpError(resp, content)
- self.assertEqual(str(error), '<HttpError 400 "">')
+ def test_missing_reason(self):
+ """Test an empty dict with a missing resp.reason."""
+ resp, content = fake_response(b"}NOT OK", {"status": "400"}, reason=None)
+ error = HttpError(resp, content)
+ self.assertEqual(str(error), '<HttpError 400 "">')
diff --git a/tests/test_http.py b/tests/test_http.py
index b92e63f..1b0caa5 100644
--- a/tests/test_http.py
+++ b/tests/test_http.py
@@ -21,7 +21,7 @@
from __future__ import absolute_import
from six.moves import range
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
from six import PY3
from six import BytesIO, StringIO
@@ -63,594 +63,607 @@
class MockCredentials(Credentials):
- """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
+ """Mock class for all Credentials objects."""
- @property
- def access_token(self):
- return self._bearer_token
+ 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_expired(self):
- return self._access_token_expired
+ @property
+ def access_token(self):
+ return self._bearer_token
- def authorize(self, http):
- self._authorized += 1
+ @property
+ def access_token_expired(self):
+ return self._access_token_expired
- request_orig = http.request
+ def authorize(self, http):
+ self._authorized += 1
- # 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)
+ request_orig = http.request
- resp, content = request_orig(uri, method, body, headers,
- redirections, connection_type)
+ # 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)
- return resp, content
+ resp, content = request_orig(
+ uri, method, body, headers, redirections, connection_type
+ )
- # Replace the request method with our own closure.
- http.request = new_request
+ return resp, content
- # Set credentials as a property of the request method.
- setattr(http.request, 'credentials', self)
+ # Replace the request method with our own closure.
+ http.request = new_request
- return http
+ # Set credentials as a property of the request method.
+ setattr(http.request, "credentials", self)
- def refresh(self, http):
- self._refreshed += 1
+ return http
- def apply(self, headers):
- self._applied += 1
- headers['authorization'] = self._bearer_token + ' ' + str(self._refreshed)
+ 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 __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: # initial == 2
- raise ssl.SSLError()
- if self.num_errors == 3: # initial == 4
- raise httplib2.ServerNotFoundError()
- else: # initial != 2,4
- if self.num_errors == 2:
- # first try a broken pipe error (#218)
- ex = socket.error()
- ex.errno = socket.errno.EPIPE
+ def request(self, *args, **kwargs):
+ if not self.num_errors:
+ return httplib2.Response(self.success_json), self.success_data
else:
- # Initialize the timeout error code to the platform's error code.
- try:
- # For Windows:
- ex = socket.error()
- ex.errno = socket.errno.WSAETIMEDOUT
- except AttributeError:
- # For Linux/Mac:
- if PY3:
- ex = socket.timeout()
- else:
- ex = socket.error()
- ex.errno = socket.errno.ETIMEDOUT
- # Now raise the correct error.
- raise ex
+ self.num_errors -= 1
+ if self.num_errors == 1: # initial == 2
+ raise ssl.SSLError()
+ if self.num_errors == 3: # initial == 4
+ raise httplib2.ServerNotFoundError()
+ else: # initial != 2,4
+ if self.num_errors == 2:
+ # first try a broken pipe error (#218)
+ ex = socket.error()
+ ex.errno = socket.errno.EPIPE
+ else:
+ # Initialize the timeout error code to the platform's error code.
+ try:
+ # For Windows:
+ ex = socket.error()
+ ex.errno = socket.errno.WSAETIMEDOUT
+ except AttributeError:
+ # For Linux/Mac:
+ if PY3:
+ ex = socket.timeout()
+ else:
+ ex = socket.error()
+ 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 __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
+ 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')
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
def datafile(filename):
- return os.path.join(DATA_DIR, filename)
+ return os.path.join(DATA_DIR, filename)
+
def _postproc_none(*kwargs):
- pass
+ pass
class TestUserAgent(unittest.TestCase):
+ def test_set_user_agent(self):
+ http = HttpMockSequence([({"status": "200"}, "echo_request_headers")])
- 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"])
- 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")])
- 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'])
+ 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)
+ if PY3:
+ with mock.patch("builtins.open", return_value=opener):
+ upload = MediaFileUpload(datafile("test_close"), mimetype="text/plain")
+ else:
+ with mock.patch("__builtin__.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_closes_fd_in___del__(self):
- file_desc = mock.Mock(spec=io.TextIOWrapper)
- opener = mock.mock_open(file_desc)
- if PY3:
- with mock.patch('builtins.open', return_value=opener):
- upload = MediaFileUpload(datafile('test_close'), mimetype='text/plain')
- else:
- with mock.patch('__builtin__.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())
- 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())
- 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))
- 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)
- 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))
- 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_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_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,
+ )
- 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)
- 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({'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)
+ 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_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_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")
- 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
- 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()
- @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()
- 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_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_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_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
+ )
- 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 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),
+ ({"status": "403"}, RATE_LIMIT_EXCEEDED_RESPONSE),
+ ({"status": "429"}, ""),
+ ({"status": "200"}, "{}"),
+ ]
+ )
- # 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),
- ({'status': '403'}, RATE_LIMIT_EXCEEDED_RESPONSE),
- ({'status': '429'}, ''),
- ({'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
+ )
- 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
- 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)
- 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 = BytesIO(b"i am png")
+ upload = MediaIoBaseUpload(
+ fd=fd, mimetype="image/png", chunksize=500, resumable=True
+ )
- def test_media_io_base_next_chunk_no_retry_403_not_configured(self):
- fd = 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"}, "{}")]
+ )
- http = HttpMockSequence([
- ({'status': '403'}, NOT_CONFIGURED_RESPONSE),
- ({'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
+ )
- 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)
+ request._rand = lambda: 1.0
+ request._sleep = mock.MagicMock()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
-
- with self.assertRaises(HttpError):
- request.execute(num_retries=3)
- request._sleep.assert_not_called()
+ with self.assertRaises(HttpError):
+ request.execute(num_retries=3)
+ request._sleep.assert_not_called()
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 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)
- 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)
- 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)
- 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()
- 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)
- 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()
- 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)
- 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_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)
- 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"
- self.request.headers['Cache-Control'] = 'no-store'
+ download = MediaIoBaseDownload(fd=self.fd, request=self.request, chunksize=3)
- download = MediaIoBaseDownload(
- fd=self.fd, request=self.request, chunksize=3)
+ self.assertEqual(download._headers.get("Cache-Control"), "no-store")
- self.assertEqual(download._headers.get('Cache-Control'), 'no-store')
+ status, done = download.next_chunk()
- status, done = download.next_chunk()
+ result = json.loads(self.fd.getvalue().decode("utf-8"))
- 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
- # 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")
- self.assertEqual(result.get("Cache-Control"), "no-store")
+ download._fd = self.fd = BytesIO()
+ status, done = download.next_chunk()
- download._fd = self.fd = BytesIO()
- status, done = download.next_chunk()
+ result = json.loads(self.fd.getvalue().decode("utf-8"))
+ self.assertEqual(result.get("Cache-Control"), "no-store")
- 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"),
+ ]
+ )
- 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)
- download = MediaIoBaseDownload(
- fd=self.fd, request=self.request, chunksize=3)
+ status, done = download.next_chunk()
- status, done = download.next_chunk()
+ self.assertEqual("https://secure.example.net/lion", download._uri)
- self.assertEqual('https://secure.example.net/lion', download._uri)
+ def test_media_io_base_download_handle_4xx(self):
+ self.request.http = HttpMockSequence([({"status": "400"}, "")])
- 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)
- 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
- 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")]
+ )
- # 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()
- status, done = download.next_chunk()
+ self.assertEqual(self.fd.getvalue(), b"123")
- self.assertEqual(self.fd.getvalue(), b'123')
+ def test_media_io_base_download_retries_connection_errors(self):
+ self.request.http = HttpMockWithErrors(
+ 4, {"status": "200", "content-range": "0-2/3"}, b"123"
+ )
- def test_media_io_base_download_retries_connection_errors(self):
- self.request.http = HttpMockWithErrors(
- 4, {'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
- 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=4)
- status, done = download.next_chunk(num_retries=4)
+ self.assertEqual(self.fd.getvalue(), b"123")
+ self.assertEqual(True, done)
- 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"),
+ ]
+ )
- 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)
- 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)
- 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
- # 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)
- status, done = download.next_chunk(num_retries=3)
+ # Check for exponential backoff using the rand function above.
+ self.assertEqual([20, 40, 80], sleeptimes)
- # 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)
- 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)]
- # Reset time.sleep stub.
- del sleeptimes[0:len(sleeptimes)]
+ status, done = download.next_chunk(num_retries=3)
- status, done = download.next_chunk(num_retries=3)
+ # Check for exponential backoff using the rand function above.
+ self.assertEqual([20, 40, 80], sleeptimes)
- # 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)
- 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"")]
+ )
- 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)
- 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)
- 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()
- 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())
- 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")])
- 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)
- 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)
- 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()
- 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())
+ 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
@@ -829,746 +842,813 @@
}
]"""
-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 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_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_empty_content_type(self):
- """Test for #284"""
- http = HttpMock(None, headers={'status': 200})
- uri = u'https://www.googleapis.com/someapi/v1/upload/?foo=bar'
- method = u'POST'
- request = HttpRequest(
- http,
- _postproc_none,
- uri,
- method=method,
- headers={'content-type': ''})
- request.execute()
- self.assertEqual('', http.headers.get('content-type'))
+ def test_empty_content_type(self):
+ """Test for #284"""
+ http = HttpMock(None, headers={"status": 200})
+ uri = u"https://www.googleapis.com/someapi/v1/upload/?foo=bar"
+ method = u"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,
- 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_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(4, {"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=4)
+ self.assertEqual({u"foo": u"bar"}, response)
- def test_retry_connection_errors_non_resumable(self):
- model = JsonModel()
- request = HttpRequest(
- HttpMockWithErrors(4, {'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=4)
- 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()
- 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(
+ 4, {"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=4)
+ self.assertEqual({u"foo": u"bar"}, response)
- request = HttpRequest(
- HttpMockWithErrors(
- 4, {'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=4)
- 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"}, "{}"))
- 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"},
+ )
- 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
- sleeptimes = []
- request._sleep = lambda x: sleeptimes.append(x)
- request._rand = lambda: 10
+ request.execute(num_retries=num_retries)
- 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])
- 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)
- 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"},
+ )
- 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
- sleeptimes = []
- request._sleep = lambda x: sleeptimes.append(x)
- request._rand = lambda: 10
+ request.execute(num_retries=num_retries)
- request.execute(num_retries=num_retries)
+ self.assertEqual(0, len(sleeptimes))
- 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"},
+ )
- 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()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
+ with self.assertRaises(HttpError):
+ request.execute()
+ request._sleep.assert_not_called()
- 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"},
+ )
- 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()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
+ with self.assertRaises(HttpError):
+ request.execute()
+ request._sleep.assert_not_called()
- 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"},
+ )
- 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()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
+ with self.assertRaises(HttpError):
+ request.execute()
+ request._sleep.assert_not_called()
- 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"},
+ )
- 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()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
+ with self.assertRaises(HttpError):
+ request.execute()
+ request._sleep.assert_not_called()
- with self.assertRaises(HttpError):
- request.execute()
- request._sleep.assert_not_called()
+ def test_no_retry_403_list_fails(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "403"}, LIST_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"},
+ )
- def test_no_retry_403_list_fails(self):
- http = HttpMockSequence([
- ({'status': '403'}, LIST_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()
- request._rand = lambda: 1.0
- request._sleep = mock.MagicMock()
+ with self.assertRaises(HttpError):
+ request.execute()
+ request._sleep.assert_not_called()
- 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"},
+ )
- 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"},
+ )
- 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_id_to_from_content_id_header(self):
- batch = BatchHttpRequest()
- self.assertEquals('12', batch._header_to_id(batch._id_to_header('12')))
+ 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_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_media_body(self):
+ batch = BatchHttpRequest()
+ f = open(datafile("small.png"), "rb")
+ body = f.read()
+ f.close()
- 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)
+ 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_media_body(self):
- batch = BatchHttpRequest()
- f = open(datafile('small.png'), 'rb')
- body = f.read()
- f.close()
+ 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)
- 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_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_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_deserialize_response(self):
+ batch = BatchHttpRequest()
+ resp, content = batch._deserialize_response(RESPONSE)
- 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)
+ self.assertEqual(200, resp.status)
+ self.assertEqual("OK", resp.reason)
+ self.assertEqual(11, resp.version)
+ self.assertEqual('{"answer": 42}', content)
- def test_deserialize_response(self):
- batch = BatchHttpRequest()
- resp, content = batch._deserialize_response(RESPONSE)
+ def test_new_id(self):
+ batch = BatchHttpRequest()
- self.assertEqual(200, resp.status)
- self.assertEqual('OK', resp.reason)
- self.assertEqual(11, resp.version)
- self.assertEqual('{"answer": 42}', content)
+ id_ = batch._new_id()
+ self.assertEqual("1", id_)
- def test_new_id(self):
- batch = BatchHttpRequest()
+ id_ = batch._new_id()
+ self.assertEqual("2", id_)
- id_ = batch._new_id()
- self.assertEqual('1', id_)
+ batch.add(self.request1, request_id="3")
- id_ = batch._new_id()
- self.assertEqual('2', id_)
+ id_ = batch._new_id()
+ self.assertEqual("4", id_)
- batch.add(self.request1, request_id='3')
+ def test_add(self):
+ batch = BatchHttpRequest()
+ batch.add(self.request1, request_id="1")
+ self.assertRaises(KeyError, batch.add, self.request1, request_id="1")
- id_ = batch._new_id()
- self.assertEqual('4', id_)
+ def test_add_fail_for_over_limit(self):
+ from googleapiclient.http import MAX_BATCH_LIMIT
- def test_add(self):
- batch = BatchHttpRequest()
- batch.add(self.request1, request_id='1')
- self.assertRaises(KeyError, batch.add, self.request1, request_id='1')
-
- def test_add_fail_for_over_limit(self):
- from googleapiclient.http import MAX_BATCH_LIMIT
-
- batch = BatchHttpRequest()
- for i in range(0, MAX_BATCH_LIMIT):
- batch.add(HttpRequest(
- None,
- None,
- 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
- method='POST',
- body='{}',
- headers={'content-type': 'application/json'})
- )
- self.assertRaises(BatchError, batch.add, self.request1)
-
- def test_add_fail_for_resumable(self):
- batch = BatchHttpRequest()
-
- upload = MediaFileUpload(
- datafile('small.png'), chunksize=500, resumable=True)
- self.request1.resumable = upload
- with self.assertRaises(BatchError) as batch_error:
- batch.add(self.request1, request_id='1')
- str(batch_error.exception)
-
- def test_execute_empty_batch_no_http(self):
- batch = BatchHttpRequest()
- ret = batch.execute()
- self.assertEqual(None, ret)
-
- def test_execute(self):
- batch = BatchHttpRequest()
- callbacks = Callbacks()
-
- batch.add(self.request1, callback=callbacks.f)
- batch.add(self.request2, callback=callbacks.f)
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_RESPONSE),
- ])
- batch.execute(http=http)
- self.assertEqual({'foo': 42}, callbacks.responses['1'])
- self.assertEqual(None, callbacks.exceptions['1'])
- self.assertEqual({'baz': 'qux'}, callbacks.responses['2'])
- self.assertEqual(None, callbacks.exceptions['2'])
-
- def test_execute_request_body(self):
- batch = BatchHttpRequest()
-
- batch.add(self.request1)
- batch.add(self.request2)
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- 'echo_request_body'),
- ])
- try:
- batch.execute(http=http)
- self.fail('Should raise exception')
- except BatchError as e:
- boundary, _ = e.content.split(None, 1)
- self.assertEqual('--', boundary[:2])
- parts = e.content.split(boundary)
- self.assertEqual(4, len(parts))
- self.assertEqual('', parts[0])
- self.assertEqual('--', parts[3].rstrip())
- header = parts[1].splitlines()[1]
- self.assertEqual('Content-Type: application/http', header)
-
- def test_execute_request_body_with_custom_long_request_ids(self):
- batch = BatchHttpRequest()
-
- batch.add(self.request1, request_id='abc'*20)
- batch.add(self.request2, request_id='def'*20)
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- 'echo_request_body'),
- ])
- try:
- batch.execute(http=http)
- self.fail('Should raise exception')
- except BatchError as e:
- boundary, _ = e.content.split(None, 1)
- self.assertEqual('--', boundary[:2])
- parts = e.content.split(boundary)
- self.assertEqual(4, len(parts))
- self.assertEqual('', parts[0])
- self.assertEqual('--', parts[3].rstrip())
- for partindex, request_id in ((1, 'abc'*20), (2, 'def'*20)):
- lines = parts[partindex].splitlines()
- for n, line in enumerate(lines):
- if line.startswith('Content-ID:'):
- # assert correct header folding
- self.assertTrue(line.endswith('+'), line)
- header_continuation = lines[n+1]
- self.assertEqual(
- header_continuation,
- ' %s>' % request_id,
- header_continuation
+ batch = BatchHttpRequest()
+ for i in range(0, MAX_BATCH_LIMIT):
+ batch.add(
+ HttpRequest(
+ None,
+ None,
+ "https://www.googleapis.com/someapi/v1/collection/?foo=bar",
+ method="POST",
+ body="{}",
+ headers={"content-type": "application/json"},
+ )
)
+ self.assertRaises(BatchError, batch.add, self.request1)
- def test_execute_initial_refresh_oauth2(self):
- batch = BatchHttpRequest()
- callbacks = Callbacks()
- cred = MockCredentials('Foo', expired=True)
+ def test_add_fail_for_resumable(self):
+ batch = BatchHttpRequest()
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_SINGLE_RESPONSE),
- ])
+ upload = MediaFileUpload(datafile("small.png"), chunksize=500, resumable=True)
+ self.request1.resumable = upload
+ with self.assertRaises(BatchError) as batch_error:
+ batch.add(self.request1, request_id="1")
+ str(batch_error.exception)
- cred.authorize(http)
+ def test_execute_empty_batch_no_http(self):
+ batch = BatchHttpRequest()
+ ret = batch.execute()
+ self.assertEqual(None, ret)
- batch.add(self.request1, callback=callbacks.f)
- batch.execute(http=http)
+ def test_execute(self):
+ batch = BatchHttpRequest()
+ callbacks = Callbacks()
- self.assertEqual({'foo': 42}, callbacks.responses['1'])
- self.assertIsNone(callbacks.exceptions['1'])
+ batch.add(self.request1, callback=callbacks.f)
+ batch.add(self.request2, callback=callbacks.f)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_RESPONSE,
+ )
+ ]
+ )
+ batch.execute(http=http)
+ self.assertEqual({"foo": 42}, callbacks.responses["1"])
+ self.assertEqual(None, callbacks.exceptions["1"])
+ self.assertEqual({"baz": "qux"}, callbacks.responses["2"])
+ self.assertEqual(None, callbacks.exceptions["2"])
- self.assertEqual(1, cred._refreshed)
+ def test_execute_request_body(self):
+ batch = BatchHttpRequest()
- self.assertEqual(1, cred._authorized)
+ batch.add(self.request1)
+ batch.add(self.request2)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ "echo_request_body",
+ )
+ ]
+ )
+ try:
+ batch.execute(http=http)
+ self.fail("Should raise exception")
+ except BatchError as e:
+ boundary, _ = e.content.split(None, 1)
+ self.assertEqual("--", boundary[:2])
+ parts = e.content.split(boundary)
+ self.assertEqual(4, len(parts))
+ self.assertEqual("", parts[0])
+ self.assertEqual("--", parts[3].rstrip())
+ header = parts[1].splitlines()[1]
+ self.assertEqual("Content-Type: application/http", header)
- self.assertEqual(1, cred._applied)
+ def test_execute_request_body_with_custom_long_request_ids(self):
+ batch = BatchHttpRequest()
- def test_execute_refresh_and_retry_on_401(self):
- batch = BatchHttpRequest()
- callbacks = Callbacks()
- cred_1 = MockCredentials('Foo')
- cred_2 = MockCredentials('Bar')
+ batch.add(self.request1, request_id="abc" * 20)
+ batch.add(self.request2, request_id="def" * 20)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ "echo_request_body",
+ )
+ ]
+ )
+ try:
+ batch.execute(http=http)
+ self.fail("Should raise exception")
+ except BatchError as e:
+ boundary, _ = e.content.split(None, 1)
+ self.assertEqual("--", boundary[:2])
+ parts = e.content.split(boundary)
+ self.assertEqual(4, len(parts))
+ self.assertEqual("", parts[0])
+ self.assertEqual("--", parts[3].rstrip())
+ for partindex, request_id in ((1, "abc" * 20), (2, "def" * 20)):
+ lines = parts[partindex].splitlines()
+ for n, line in enumerate(lines):
+ if line.startswith("Content-ID:"):
+ # assert correct header folding
+ self.assertTrue(line.endswith("+"), line)
+ header_continuation = lines[n + 1]
+ self.assertEqual(
+ header_continuation,
+ " %s>" % request_id,
+ header_continuation,
+ )
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_RESPONSE_WITH_401),
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_SINGLE_RESPONSE),
- ])
+ def test_execute_initial_refresh_oauth2(self):
+ batch = BatchHttpRequest()
+ callbacks = Callbacks()
+ cred = MockCredentials("Foo", expired=True)
- creds_http_1 = HttpMockSequence([])
- cred_1.authorize(creds_http_1)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_SINGLE_RESPONSE,
+ )
+ ]
+ )
- creds_http_2 = HttpMockSequence([])
- cred_2.authorize(creds_http_2)
+ cred.authorize(http)
- self.request1.http = creds_http_1
- self.request2.http = creds_http_2
+ batch.add(self.request1, callback=callbacks.f)
+ batch.execute(http=http)
- batch.add(self.request1, callback=callbacks.f)
- batch.add(self.request2, callback=callbacks.f)
- batch.execute(http=http)
+ self.assertEqual({"foo": 42}, callbacks.responses["1"])
+ self.assertIsNone(callbacks.exceptions["1"])
- self.assertEqual({'foo': 42}, callbacks.responses['1'])
- self.assertEqual(None, callbacks.exceptions['1'])
- self.assertEqual({'baz': 'qux'}, callbacks.responses['2'])
- self.assertEqual(None, callbacks.exceptions['2'])
+ self.assertEqual(1, cred._refreshed)
- self.assertEqual(1, cred_1._refreshed)
- self.assertEqual(0, cred_2._refreshed)
+ self.assertEqual(1, cred._authorized)
- self.assertEqual(1, cred_1._authorized)
- self.assertEqual(1, cred_2._authorized)
+ self.assertEqual(1, cred._applied)
- self.assertEqual(1, cred_2._applied)
- self.assertEqual(2, cred_1._applied)
+ def test_execute_refresh_and_retry_on_401(self):
+ batch = BatchHttpRequest()
+ callbacks = Callbacks()
+ cred_1 = MockCredentials("Foo")
+ cred_2 = MockCredentials("Bar")
- def test_http_errors_passed_to_callback(self):
- batch = BatchHttpRequest()
- callbacks = Callbacks()
- cred_1 = MockCredentials('Foo')
- cred_2 = MockCredentials('Bar')
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_RESPONSE_WITH_401,
+ ),
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_SINGLE_RESPONSE,
+ ),
+ ]
+ )
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_RESPONSE_WITH_401),
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_RESPONSE_WITH_401),
- ])
+ creds_http_1 = HttpMockSequence([])
+ cred_1.authorize(creds_http_1)
- creds_http_1 = HttpMockSequence([])
- cred_1.authorize(creds_http_1)
+ creds_http_2 = HttpMockSequence([])
+ cred_2.authorize(creds_http_2)
- creds_http_2 = HttpMockSequence([])
- cred_2.authorize(creds_http_2)
+ self.request1.http = creds_http_1
+ self.request2.http = creds_http_2
- self.request1.http = creds_http_1
- self.request2.http = creds_http_2
+ batch.add(self.request1, callback=callbacks.f)
+ batch.add(self.request2, callback=callbacks.f)
+ batch.execute(http=http)
- batch.add(self.request1, callback=callbacks.f)
- batch.add(self.request2, callback=callbacks.f)
- batch.execute(http=http)
+ self.assertEqual({"foo": 42}, callbacks.responses["1"])
+ self.assertEqual(None, callbacks.exceptions["1"])
+ self.assertEqual({"baz": "qux"}, callbacks.responses["2"])
+ self.assertEqual(None, callbacks.exceptions["2"])
- self.assertEqual(None, callbacks.responses['1'])
- self.assertEqual(401, callbacks.exceptions['1'].resp.status)
- self.assertEqual(
- 'Authorization Required', callbacks.exceptions['1'].resp.reason)
- self.assertEqual({u'baz': u'qux'}, callbacks.responses['2'])
- self.assertEqual(None, callbacks.exceptions['2'])
+ self.assertEqual(1, cred_1._refreshed)
+ self.assertEqual(0, cred_2._refreshed)
- def test_execute_global_callback(self):
- callbacks = Callbacks()
- batch = BatchHttpRequest(callback=callbacks.f)
+ self.assertEqual(1, cred_1._authorized)
+ self.assertEqual(1, cred_2._authorized)
- batch.add(self.request1)
- batch.add(self.request2)
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_RESPONSE),
- ])
- batch.execute(http=http)
- self.assertEqual({'foo': 42}, callbacks.responses['1'])
- self.assertEqual({'baz': 'qux'}, callbacks.responses['2'])
+ self.assertEqual(1, cred_2._applied)
+ self.assertEqual(2, cred_1._applied)
- def test_execute_batch_http_error(self):
- callbacks = Callbacks()
- batch = BatchHttpRequest(callback=callbacks.f)
+ def test_http_errors_passed_to_callback(self):
+ batch = BatchHttpRequest()
+ callbacks = Callbacks()
+ cred_1 = MockCredentials("Foo")
+ cred_2 = MockCredentials("Bar")
- batch.add(self.request1)
- batch.add(self.request2)
- http = HttpMockSequence([
- ({'status': '200',
- 'content-type': 'multipart/mixed; boundary="batch_foobarbaz"'},
- BATCH_ERROR_RESPONSE),
- ])
- batch.execute(http=http)
- self.assertEqual({'foo': 42}, callbacks.responses['1'])
- expected = ('<HttpError 403 when requesting '
- 'https://www.googleapis.com/someapi/v1/collection/?foo=bar returned '
- '"Access Not Configured">')
- self.assertEqual(expected, str(callbacks.exceptions['2']))
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_RESPONSE_WITH_401,
+ ),
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_RESPONSE_WITH_401,
+ ),
+ ]
+ )
+
+ creds_http_1 = HttpMockSequence([])
+ cred_1.authorize(creds_http_1)
+
+ creds_http_2 = HttpMockSequence([])
+ cred_2.authorize(creds_http_2)
+
+ self.request1.http = creds_http_1
+ self.request2.http = creds_http_2
+
+ batch.add(self.request1, callback=callbacks.f)
+ batch.add(self.request2, callback=callbacks.f)
+ batch.execute(http=http)
+
+ self.assertEqual(None, callbacks.responses["1"])
+ self.assertEqual(401, callbacks.exceptions["1"].resp.status)
+ self.assertEqual(
+ "Authorization Required", callbacks.exceptions["1"].resp.reason
+ )
+ self.assertEqual({u"baz": u"qux"}, callbacks.responses["2"])
+ self.assertEqual(None, callbacks.exceptions["2"])
+
+ def test_execute_global_callback(self):
+ callbacks = Callbacks()
+ batch = BatchHttpRequest(callback=callbacks.f)
+
+ batch.add(self.request1)
+ batch.add(self.request2)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_RESPONSE,
+ )
+ ]
+ )
+ batch.execute(http=http)
+ self.assertEqual({"foo": 42}, callbacks.responses["1"])
+ self.assertEqual({"baz": "qux"}, callbacks.responses["2"])
+
+ def test_execute_batch_http_error(self):
+ callbacks = Callbacks()
+ batch = BatchHttpRequest(callback=callbacks.f)
+
+ batch.add(self.request1)
+ batch.add(self.request2)
+ http = HttpMockSequence(
+ [
+ (
+ {
+ "status": "200",
+ "content-type": 'multipart/mixed; boundary="batch_foobarbaz"',
+ },
+ BATCH_ERROR_RESPONSE,
+ )
+ ]
+ )
+ batch.execute(http=http)
+ self.assertEqual({"foo": 42}, callbacks.responses["1"])
+ expected = (
+ "<HttpError 403 when requesting "
+ "https://www.googleapis.com/someapi/v1/collection/?foo=bar returned "
+ '"Access Not Configured">'
+ )
+ self.assertEqual(expected, str(callbacks.exceptions["2"]))
class TestRequestUriTooLong(unittest.TestCase):
+ def test_turn_get_into_post(self):
+ def _postproc(resp, content):
+ return content
- def test_turn_get_into_post(self):
+ http = HttpMockSequence(
+ [
+ ({"status": "200"}, "echo_request_body"),
+ ({"status": "200"}, "echo_request_headers"),
+ ]
+ )
- def _postproc(resp, content):
- return content
+ # Send a long query parameter.
+ query = {"q": "a" * MAX_URI_LENGTH + "?&"}
+ req = HttpRequest(
+ http,
+ _postproc,
+ "http://example.com?" + urlencode(query),
+ method="GET",
+ body=None,
+ headers={},
+ methodId="foo",
+ resumable=None,
+ )
- http = HttpMockSequence([
- ({'status': '200'},
- 'echo_request_body'),
- ({'status': '200'},
- 'echo_request_headers'),
- ])
+ # Query parameters should be sent in the body.
+ response = req.execute()
+ self.assertEqual(b"q=" + b"a" * MAX_URI_LENGTH + b"%3F%26", response)
- # Send a long query parameter.
- query = {
- 'q': 'a' * MAX_URI_LENGTH + '?&'
- }
- req = HttpRequest(
- http,
- _postproc,
- 'http://example.com?' + urlencode(query),
- method='GET',
- body=None,
- headers={},
- methodId='foo',
- resumable=None)
-
- # Query parameters should be sent in the body.
- response = req.execute()
- self.assertEqual(b'q=' + b'a' * MAX_URI_LENGTH + b'%3F%26', response)
-
- # Extra headers should be set.
- response = req.execute()
- self.assertEqual('GET', response['x-http-method-override'])
- self.assertEqual(str(MAX_URI_LENGTH + 8), response['content-length'])
- self.assertEqual(
- 'application/x-www-form-urlencoded', response['content-type'])
+ # Extra headers should be set.
+ response = req.execute()
+ self.assertEqual("GET", response["x-http-method-override"])
+ self.assertEqual(str(MAX_URI_LENGTH + 8), response["content-length"])
+ self.assertEqual("application/x-www-form-urlencoded", response["content-type"])
class TestStreamSlice(unittest.TestCase):
- """Test _StreamSlice."""
+ """Test _StreamSlice."""
- def setUp(self):
- self.stream = BytesIO(b'0123456789')
+ def setUp(self):
+ self.stream = BytesIO(b"0123456789")
- def test_read(self):
- s = _StreamSlice(self.stream, 0, 4)
- self.assertEqual(b'', s.read(0))
- self.assertEqual(b'0', s.read(1))
- self.assertEqual(b'123', s.read())
+ def test_read(self):
+ s = _StreamSlice(self.stream, 0, 4)
+ self.assertEqual(b"", s.read(0))
+ self.assertEqual(b"0", s.read(1))
+ self.assertEqual(b"123", s.read())
- def test_read_too_much(self):
- s = _StreamSlice(self.stream, 1, 4)
- self.assertEqual(b'1234', s.read(6))
+ def test_read_too_much(self):
+ s = _StreamSlice(self.stream, 1, 4)
+ self.assertEqual(b"1234", s.read(6))
- def test_read_all(self):
- s = _StreamSlice(self.stream, 2, 1)
- self.assertEqual(b'2', s.read(-1))
+ def test_read_all(self):
+ s = _StreamSlice(self.stream, 2, 1)
+ self.assertEqual(b"2", s.read(-1))
class TestResponseCallback(unittest.TestCase):
- """Test adding callbacks to responses."""
+ """Test adding callbacks to responses."""
- def test_ensure_response_callback(self):
- m = JsonModel()
- request = HttpRequest(
- None,
- m.response,
- 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
- method='POST',
- body='{}',
- headers={'content-type': 'application/json'})
- h = HttpMockSequence([ ({'status': 200}, '{}')])
- responses = []
- def _on_response(resp, responses=responses):
- responses.append(resp)
- request.add_response_callback(_on_response)
- request.execute(http=h)
- self.assertEqual(1, len(responses))
+ def test_ensure_response_callback(self):
+ m = JsonModel()
+ request = HttpRequest(
+ None,
+ m.response,
+ "https://www.googleapis.com/someapi/v1/collection/?foo=bar",
+ method="POST",
+ body="{}",
+ headers={"content-type": "application/json"},
+ )
+ h = HttpMockSequence([({"status": 200}, "{}")])
+ responses = []
+
+ def _on_response(resp, responses=responses):
+ responses.append(resp)
+
+ request.add_response_callback(_on_response)
+ request.execute(http=h)
+ self.assertEqual(1, len(responses))
class TestHttpMock(unittest.TestCase):
- def test_default_response_headers(self):
- http = HttpMock(datafile('zoo.json'))
- resp, content = http.request("http://example.com")
- self.assertEqual(resp.status, 200)
+ def test_default_response_headers(self):
+ http = HttpMock(datafile("zoo.json"))
+ resp, content = http.request("http://example.com")
+ self.assertEqual(resp.status, 200)
- def test_error_response(self):
- http = HttpMock(datafile('bad_request.json'), {'status': '400'})
- model = JsonModel()
- request = HttpRequest(
- http,
- model.response,
- 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
- method='GET',
- headers={})
- self.assertRaises(HttpError, request.execute)
+ def test_error_response(self):
+ http = HttpMock(datafile("bad_request.json"), {"status": "400"})
+ model = JsonModel()
+ request = HttpRequest(
+ http,
+ model.response,
+ "https://www.googleapis.com/someapi/v1/collection/?foo=bar",
+ method="GET",
+ headers={},
+ )
+ self.assertRaises(HttpError, request.execute)
class TestHttpBuild(unittest.TestCase):
- original_socket_default_timeout = None
+ original_socket_default_timeout = None
- @classmethod
- def setUpClass(cls):
- cls.original_socket_default_timeout = socket.getdefaulttimeout()
+ @classmethod
+ def setUpClass(cls):
+ cls.original_socket_default_timeout = socket.getdefaulttimeout()
- @classmethod
- def tearDownClass(cls):
- socket.setdefaulttimeout(cls.original_socket_default_timeout)
+ @classmethod
+ def tearDownClass(cls):
+ socket.setdefaulttimeout(cls.original_socket_default_timeout)
- def test_build_http_sets_default_timeout_if_none_specified(self):
- socket.setdefaulttimeout(None)
- http = build_http()
- self.assertIsInstance(http.timeout, int)
- self.assertGreater(http.timeout, 0)
+ def test_build_http_sets_default_timeout_if_none_specified(self):
+ socket.setdefaulttimeout(None)
+ http = build_http()
+ self.assertIsInstance(http.timeout, int)
+ self.assertGreater(http.timeout, 0)
- def test_build_http_default_timeout_can_be_overridden(self):
- socket.setdefaulttimeout(1.5)
- http = build_http()
- self.assertAlmostEqual(http.timeout, 1.5, delta=0.001)
+ def test_build_http_default_timeout_can_be_overridden(self):
+ socket.setdefaulttimeout(1.5)
+ http = build_http()
+ self.assertAlmostEqual(http.timeout, 1.5, delta=0.001)
- def test_build_http_default_timeout_can_be_set_to_zero(self):
- socket.setdefaulttimeout(0)
- http = build_http()
- self.assertEquals(http.timeout, 0)
+ def test_build_http_default_timeout_can_be_set_to_zero(self):
+ socket.setdefaulttimeout(0)
+ http = build_http()
+ self.assertEquals(http.timeout, 0)
-if __name__ == '__main__':
- logging.getLogger().setLevel(logging.ERROR)
- unittest.main()
+if __name__ == "__main__":
+ logging.getLogger().setLevel(logging.ERROR)
+ unittest.main()
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index 006eb47..0064f3f 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -21,7 +21,7 @@
from __future__ import absolute_import
import six
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
import copy
import json
@@ -39,256 +39,278 @@
class Model(unittest.TestCase):
- def test_json_no_body(self):
- model = JsonModel(data_wrapper=False)
+ def test_json_no_body(self):
+ model = JsonModel(data_wrapper=False)
- headers = {}
- path_params = {}
- query_params = {}
- body = None
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = None
- headers, unused_params, query, body = model.request(
- headers, path_params, query_params, body)
+ headers, unused_params, query, body = model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/json')
- self.assertTrue('content-type' not in headers)
- self.assertNotEqual(query, '')
- self.assertEqual(body, None)
+ self.assertEqual(headers["accept"], "application/json")
+ self.assertTrue("content-type" not in headers)
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, None)
- def test_json_body(self):
- model = JsonModel(data_wrapper=False)
+ def test_json_body(self):
+ model = JsonModel(data_wrapper=False)
- headers = {}
- path_params = {}
- query_params = {}
- body = {}
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = {}
- headers, unused_params, query, body = model.request(
- headers, path_params, query_params, body)
+ headers, unused_params, query, body = model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/json')
- self.assertEqual(headers['content-type'], 'application/json')
- self.assertNotEqual(query, '')
- self.assertEqual(body, '{}')
+ self.assertEqual(headers["accept"], "application/json")
+ self.assertEqual(headers["content-type"], "application/json")
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, "{}")
- def test_json_body_data_wrapper(self):
- model = JsonModel(data_wrapper=True)
+ def test_json_body_data_wrapper(self):
+ model = JsonModel(data_wrapper=True)
- headers = {}
- path_params = {}
- query_params = {}
- body = {}
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = {}
- headers, unused_params, query, body = model.request(
- headers, path_params, query_params, body)
+ headers, unused_params, query, body = model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/json')
- self.assertEqual(headers['content-type'], 'application/json')
- self.assertNotEqual(query, '')
- self.assertEqual(body, '{"data": {}}')
+ self.assertEqual(headers["accept"], "application/json")
+ self.assertEqual(headers["content-type"], "application/json")
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, '{"data": {}}')
- def test_json_body_default_data(self):
- """Test that a 'data' wrapper doesn't get added if one is already present."""
- model = JsonModel(data_wrapper=True)
+ def test_json_body_default_data(self):
+ """Test that a 'data' wrapper doesn't get added if one is already present."""
+ model = JsonModel(data_wrapper=True)
- headers = {}
- path_params = {}
- query_params = {}
- body = {'data': 'foo'}
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = {"data": "foo"}
- headers, unused_params, query, body = model.request(
- headers, path_params, query_params, body)
+ headers, unused_params, query, body = model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/json')
- self.assertEqual(headers['content-type'], 'application/json')
- self.assertNotEqual(query, '')
- self.assertEqual(body, '{"data": "foo"}')
+ self.assertEqual(headers["accept"], "application/json")
+ self.assertEqual(headers["content-type"], "application/json")
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, '{"data": "foo"}')
- def test_json_build_query(self):
- model = JsonModel(data_wrapper=False)
+ def test_json_build_query(self):
+ model = JsonModel(data_wrapper=False)
- headers = {}
- path_params = {}
- query_params = {'foo': 1, 'bar': u'\N{COMET}',
- 'baz': ['fe', 'fi', 'fo', 'fum'], # Repeated parameters
- 'qux': []}
- body = {}
-
- headers, unused_params, query, body = model.request(
- headers, path_params, query_params, body)
-
- self.assertEqual(headers['accept'], 'application/json')
- self.assertEqual(headers['content-type'], 'application/json')
-
- query_dict = parse_qs(query[1:])
- self.assertEqual(query_dict['foo'], ['1'])
- if six.PY3:
- # Python 3, no need to encode
- self.assertEqual(query_dict['bar'], [u'\N{COMET}'])
- else:
- # Python 2, encode string
- self.assertEqual(query_dict['bar'], [u'\N{COMET}'.encode('utf-8')])
- self.assertEqual(query_dict['baz'], ['fe', 'fi', 'fo', 'fum'])
- self.assertTrue('qux' not in query_dict)
- self.assertEqual(body, '{}')
-
- def test_user_agent(self):
- model = JsonModel(data_wrapper=False)
-
- headers = {'user-agent': 'my-test-app/1.23.4'}
- path_params = {}
- query_params = {}
- body = {}
-
- headers, unused_params, unused_query, body = model.request(
- headers, path_params, query_params, body)
-
- self.assertEqual(headers['user-agent'],
- 'my-test-app/1.23.4 (gzip)')
-
- def test_x_goog_api_client(self):
- model = JsonModel(data_wrapper=False)
-
- # test header composition for cloud clients that wrap discovery
- headers = {'x-goog-api-client': 'gccl/1.23.4'}
- path_params = {}
- query_params = {}
- body = {}
-
- headers, unused_params, unused_query, body = model.request(
- headers, path_params, query_params, body)
-
- self.assertEqual(headers['x-goog-api-client'],
- 'gccl/1.23.4' + ' gdcl/' + __version__ + ' gl-python/' + platform.python_version())
-
- def test_bad_response(self):
- model = JsonModel(data_wrapper=False)
- resp = httplib2.Response({'status': '401'})
- resp.reason = 'Unauthorized'
- content = b'{"error": {"message": "not authorized"}}'
-
- try:
- content = model.response(resp, content)
- self.fail('Should have thrown an exception')
- except HttpError as e:
- self.assertTrue('not authorized' in str(e))
-
- resp['content-type'] = 'application/json'
-
- try:
- content = model.response(resp, content)
- self.fail('Should have thrown an exception')
- except HttpError as e:
- self.assertTrue('not authorized' in str(e))
-
- def test_good_response(self):
- model = JsonModel(data_wrapper=True)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '{"data": "is good"}'
-
- content = model.response(resp, content)
- self.assertEqual(content, 'is good')
-
- def test_good_response_wo_data(self):
- model = JsonModel(data_wrapper=False)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '{"foo": "is good"}'
-
- content = model.response(resp, content)
- self.assertEqual(content, {'foo': 'is good'})
-
- def test_good_response_wo_data_str(self):
- model = JsonModel(data_wrapper=False)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '"data goes here"'
-
- content = model.response(resp, content)
- self.assertEqual(content, 'data goes here')
-
- def test_no_content_response(self):
- model = JsonModel(data_wrapper=False)
- resp = httplib2.Response({'status': '204'})
- resp.reason = 'No Content'
- content = ''
-
- content = model.response(resp, content)
- self.assertEqual(content, {})
-
- def test_logging(self):
- class MockLogging(object):
- def __init__(self):
- self.info_record = []
- self.debug_record = []
- def info(self, message, *args):
- self.info_record.append(message % args)
-
- def debug(self, message, *args):
- self.debug_record.append(message % args)
-
- class MockResponse(dict):
- def __init__(self, items):
- super(MockResponse, self).__init__()
- self.status = items['status']
- for key, value in six.iteritems(items):
- self[key] = value
- old_logging = googleapiclient.model.LOGGER
- googleapiclient.model.LOGGER = MockLogging()
- googleapiclient.model.dump_request_response = True
- model = JsonModel()
- request_body = {
- 'field1': 'value1',
- 'field2': 'value2'
+ headers = {}
+ path_params = {}
+ query_params = {
+ "foo": 1,
+ "bar": u"\N{COMET}",
+ "baz": ["fe", "fi", "fo", "fum"], # Repeated parameters
+ "qux": [],
}
- body_string = model.request({}, {}, {}, request_body)[-1]
- json_body = json.loads(body_string)
- self.assertEqual(request_body, json_body)
+ body = {}
- response = {'status': 200,
- 'response_field_1': 'response_value_1',
- 'response_field_2': 'response_value_2'}
- response_body = model.response(MockResponse(response), body_string)
- self.assertEqual(request_body, response_body)
- self.assertEqual(googleapiclient.model.LOGGER.info_record[:2],
- ['--request-start--',
- '-headers-start-'])
- self.assertTrue('response_field_1: response_value_1' in
- googleapiclient.model.LOGGER.info_record)
- self.assertTrue('response_field_2: response_value_2' in
- googleapiclient.model.LOGGER.info_record)
- self.assertEqual(json.loads(googleapiclient.model.LOGGER.info_record[-2]),
- request_body)
- self.assertEqual(googleapiclient.model.LOGGER.info_record[-1],
- '--response-end--')
- googleapiclient.model.LOGGER = old_logging
+ headers, unused_params, query, body = model.request(
+ headers, path_params, query_params, body
+ )
- def test_no_data_wrapper_deserialize(self):
- model = JsonModel(data_wrapper=False)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '{"data": "is good"}'
- content = model.response(resp, content)
- self.assertEqual(content, {'data': 'is good'})
+ self.assertEqual(headers["accept"], "application/json")
+ self.assertEqual(headers["content-type"], "application/json")
- def test_data_wrapper_deserialize(self):
- model = JsonModel(data_wrapper=True)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '{"data": "is good"}'
- content = model.response(resp, content)
- self.assertEqual(content, 'is good')
+ query_dict = parse_qs(query[1:])
+ self.assertEqual(query_dict["foo"], ["1"])
+ if six.PY3:
+ # Python 3, no need to encode
+ self.assertEqual(query_dict["bar"], [u"\N{COMET}"])
+ else:
+ # Python 2, encode string
+ self.assertEqual(query_dict["bar"], [u"\N{COMET}".encode("utf-8")])
+ self.assertEqual(query_dict["baz"], ["fe", "fi", "fo", "fum"])
+ self.assertTrue("qux" not in query_dict)
+ self.assertEqual(body, "{}")
- def test_data_wrapper_deserialize_nodata(self):
- model = JsonModel(data_wrapper=True)
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = '{"atad": "is good"}'
- content = model.response(resp, content)
- self.assertEqual(content, {'atad': 'is good'})
+ def test_user_agent(self):
+ model = JsonModel(data_wrapper=False)
+
+ headers = {"user-agent": "my-test-app/1.23.4"}
+ path_params = {}
+ query_params = {}
+ body = {}
+
+ headers, unused_params, unused_query, body = model.request(
+ headers, path_params, query_params, body
+ )
+
+ self.assertEqual(headers["user-agent"], "my-test-app/1.23.4 (gzip)")
+
+ def test_x_goog_api_client(self):
+ model = JsonModel(data_wrapper=False)
+
+ # test header composition for cloud clients that wrap discovery
+ headers = {"x-goog-api-client": "gccl/1.23.4"}
+ path_params = {}
+ query_params = {}
+ body = {}
+
+ headers, unused_params, unused_query, body = model.request(
+ headers, path_params, query_params, body
+ )
+
+ self.assertEqual(
+ headers["x-goog-api-client"],
+ "gccl/1.23.4"
+ + " gdcl/"
+ + __version__
+ + " gl-python/"
+ + platform.python_version(),
+ )
+
+ def test_bad_response(self):
+ model = JsonModel(data_wrapper=False)
+ resp = httplib2.Response({"status": "401"})
+ resp.reason = "Unauthorized"
+ content = b'{"error": {"message": "not authorized"}}'
+
+ try:
+ content = model.response(resp, content)
+ self.fail("Should have thrown an exception")
+ except HttpError as e:
+ self.assertTrue("not authorized" in str(e))
+
+ resp["content-type"] = "application/json"
+
+ try:
+ content = model.response(resp, content)
+ self.fail("Should have thrown an exception")
+ except HttpError as e:
+ self.assertTrue("not authorized" in str(e))
+
+ def test_good_response(self):
+ model = JsonModel(data_wrapper=True)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '{"data": "is good"}'
+
+ content = model.response(resp, content)
+ self.assertEqual(content, "is good")
+
+ def test_good_response_wo_data(self):
+ model = JsonModel(data_wrapper=False)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '{"foo": "is good"}'
+
+ content = model.response(resp, content)
+ self.assertEqual(content, {"foo": "is good"})
+
+ def test_good_response_wo_data_str(self):
+ model = JsonModel(data_wrapper=False)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '"data goes here"'
+
+ content = model.response(resp, content)
+ self.assertEqual(content, "data goes here")
+
+ def test_no_content_response(self):
+ model = JsonModel(data_wrapper=False)
+ resp = httplib2.Response({"status": "204"})
+ resp.reason = "No Content"
+ content = ""
+
+ content = model.response(resp, content)
+ self.assertEqual(content, {})
+
+ def test_logging(self):
+ class MockLogging(object):
+ def __init__(self):
+ self.info_record = []
+ self.debug_record = []
+
+ def info(self, message, *args):
+ self.info_record.append(message % args)
+
+ def debug(self, message, *args):
+ self.debug_record.append(message % args)
+
+ class MockResponse(dict):
+ def __init__(self, items):
+ super(MockResponse, self).__init__()
+ self.status = items["status"]
+ for key, value in six.iteritems(items):
+ self[key] = value
+
+ old_logging = googleapiclient.model.LOGGER
+ googleapiclient.model.LOGGER = MockLogging()
+ googleapiclient.model.dump_request_response = True
+ model = JsonModel()
+ request_body = {"field1": "value1", "field2": "value2"}
+ body_string = model.request({}, {}, {}, request_body)[-1]
+ json_body = json.loads(body_string)
+ self.assertEqual(request_body, json_body)
+
+ response = {
+ "status": 200,
+ "response_field_1": "response_value_1",
+ "response_field_2": "response_value_2",
+ }
+ response_body = model.response(MockResponse(response), body_string)
+ self.assertEqual(request_body, response_body)
+ self.assertEqual(
+ googleapiclient.model.LOGGER.info_record[:2],
+ ["--request-start--", "-headers-start-"],
+ )
+ self.assertTrue(
+ "response_field_1: response_value_1"
+ in googleapiclient.model.LOGGER.info_record
+ )
+ self.assertTrue(
+ "response_field_2: response_value_2"
+ in googleapiclient.model.LOGGER.info_record
+ )
+ self.assertEqual(
+ json.loads(googleapiclient.model.LOGGER.info_record[-2]), request_body
+ )
+ self.assertEqual(
+ googleapiclient.model.LOGGER.info_record[-1], "--response-end--"
+ )
+ googleapiclient.model.LOGGER = old_logging
+
+ def test_no_data_wrapper_deserialize(self):
+ model = JsonModel(data_wrapper=False)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '{"data": "is good"}'
+ content = model.response(resp, content)
+ self.assertEqual(content, {"data": "is good"})
+
+ def test_data_wrapper_deserialize(self):
+ model = JsonModel(data_wrapper=True)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '{"data": "is good"}'
+ content = model.response(resp, content)
+ self.assertEqual(content, "is good")
+
+ def test_data_wrapper_deserialize_nodata(self):
+ model = JsonModel(data_wrapper=True)
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = '{"atad": "is good"}'
+ content = model.response(resp, content)
+ self.assertEqual(content, {"atad": "is good"})
-
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_mocks.py b/tests/test_mocks.py
index a456b9e..f020f6b 100644
--- a/tests/test_mocks.py
+++ b/tests/test_mocks.py
@@ -20,7 +20,7 @@
"""
from __future__ import absolute_import
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
import httplib2
import os
@@ -34,118 +34,133 @@
from googleapiclient.http import HttpMock
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
+
def datafile(filename):
- return os.path.join(DATA_DIR, filename)
+ return os.path.join(DATA_DIR, filename)
class Mocks(unittest.TestCase):
- def setUp(self):
- self.http = HttpMock(datafile('plus.json'), {'status': '200'})
- self.zoo_http = HttpMock(datafile('zoo.json'), {'status': '200'})
+ def setUp(self):
+ self.http = HttpMock(datafile("plus.json"), {"status": "200"})
+ self.zoo_http = HttpMock(datafile("zoo.json"), {"status": "200"})
- def test_default_response(self):
- requestBuilder = RequestMockBuilder({})
- plus = build('plus', 'v1', http=self.http, requestBuilder=requestBuilder)
- activity = plus.activities().get(activityId='tag:blah').execute()
- self.assertEqual({}, activity)
+ def test_default_response(self):
+ requestBuilder = RequestMockBuilder({})
+ plus = build("plus", "v1", http=self.http, requestBuilder=requestBuilder)
+ activity = plus.activities().get(activityId="tag:blah").execute()
+ self.assertEqual({}, activity)
- def test_simple_response(self):
- requestBuilder = RequestMockBuilder({
- 'plus.activities.get': (None, '{"foo": "bar"}')
- })
- plus = build('plus', 'v1', http=self.http, requestBuilder=requestBuilder)
+ def test_simple_response(self):
+ requestBuilder = RequestMockBuilder(
+ {"plus.activities.get": (None, '{"foo": "bar"}')}
+ )
+ plus = build("plus", "v1", http=self.http, requestBuilder=requestBuilder)
- activity = plus.activities().get(activityId='tag:blah').execute()
- self.assertEqual({"foo": "bar"}, activity)
+ activity = plus.activities().get(activityId="tag:blah").execute()
+ self.assertEqual({"foo": "bar"}, activity)
- def test_unexpected_call(self):
- requestBuilder = RequestMockBuilder({}, check_unexpected=True)
+ def test_unexpected_call(self):
+ requestBuilder = RequestMockBuilder({}, check_unexpected=True)
- plus = build('plus', 'v1', http=self.http, requestBuilder=requestBuilder)
+ plus = build("plus", "v1", http=self.http, requestBuilder=requestBuilder)
- try:
- plus.activities().get(activityId='tag:blah').execute()
- self.fail('UnexpectedMethodError should have been raised')
- except UnexpectedMethodError:
- pass
+ try:
+ plus.activities().get(activityId="tag:blah").execute()
+ self.fail("UnexpectedMethodError should have been raised")
+ except UnexpectedMethodError:
+ pass
- def test_simple_unexpected_body(self):
- requestBuilder = RequestMockBuilder({
- 'zoo.animals.insert': (None, '{"data": {"foo": "bar"}}', None)
- })
- zoo = build('zoo', 'v1', http=self.zoo_http, requestBuilder=requestBuilder)
+ def test_simple_unexpected_body(self):
+ requestBuilder = RequestMockBuilder(
+ {"zoo.animals.insert": (None, '{"data": {"foo": "bar"}}', None)}
+ )
+ zoo = build("zoo", "v1", http=self.zoo_http, requestBuilder=requestBuilder)
- try:
- zoo.animals().insert(body='{}').execute()
- self.fail('UnexpectedBodyError should have been raised')
- except UnexpectedBodyError:
- pass
+ try:
+ zoo.animals().insert(body="{}").execute()
+ self.fail("UnexpectedBodyError should have been raised")
+ except UnexpectedBodyError:
+ pass
- def test_simple_expected_body(self):
- requestBuilder = RequestMockBuilder({
- 'zoo.animals.insert': (None, '{"data": {"foo": "bar"}}', '{}')
- })
- zoo = build('zoo', 'v1', http=self.zoo_http, requestBuilder=requestBuilder)
+ def test_simple_expected_body(self):
+ requestBuilder = RequestMockBuilder(
+ {"zoo.animals.insert": (None, '{"data": {"foo": "bar"}}', "{}")}
+ )
+ zoo = build("zoo", "v1", http=self.zoo_http, requestBuilder=requestBuilder)
- try:
- zoo.animals().insert(body='').execute()
- self.fail('UnexpectedBodyError should have been raised')
- except UnexpectedBodyError:
- pass
+ try:
+ zoo.animals().insert(body="").execute()
+ self.fail("UnexpectedBodyError should have been raised")
+ except UnexpectedBodyError:
+ pass
- def test_simple_wrong_body(self):
- requestBuilder = RequestMockBuilder({
- 'zoo.animals.insert': (None, '{"data": {"foo": "bar"}}',
- '{"data": {"foo": "bar"}}')
- })
- zoo = build('zoo', 'v1', http=self.zoo_http, requestBuilder=requestBuilder)
+ def test_simple_wrong_body(self):
+ requestBuilder = RequestMockBuilder(
+ {
+ "zoo.animals.insert": (
+ None,
+ '{"data": {"foo": "bar"}}',
+ '{"data": {"foo": "bar"}}',
+ )
+ }
+ )
+ zoo = build("zoo", "v1", http=self.zoo_http, requestBuilder=requestBuilder)
- try:
- zoo.animals().insert(
- body='{"data": {"foo": "blah"}}').execute()
- self.fail('UnexpectedBodyError should have been raised')
- except UnexpectedBodyError:
- pass
+ try:
+ zoo.animals().insert(body='{"data": {"foo": "blah"}}').execute()
+ self.fail("UnexpectedBodyError should have been raised")
+ except UnexpectedBodyError:
+ pass
- def test_simple_matching_str_body(self):
- requestBuilder = RequestMockBuilder({
- 'zoo.animals.insert': (None, '{"data": {"foo": "bar"}}',
- '{"data": {"foo": "bar"}}')
- })
- zoo = build('zoo', 'v1', http=self.zoo_http, requestBuilder=requestBuilder)
+ def test_simple_matching_str_body(self):
+ requestBuilder = RequestMockBuilder(
+ {
+ "zoo.animals.insert": (
+ None,
+ '{"data": {"foo": "bar"}}',
+ '{"data": {"foo": "bar"}}',
+ )
+ }
+ )
+ zoo = build("zoo", "v1", http=self.zoo_http, requestBuilder=requestBuilder)
- activity = zoo.animals().insert(
- body={'data': {'foo': 'bar'}}).execute()
- self.assertEqual({'foo': 'bar'}, activity)
+ activity = zoo.animals().insert(body={"data": {"foo": "bar"}}).execute()
+ self.assertEqual({"foo": "bar"}, activity)
- def test_simple_matching_dict_body(self):
- requestBuilder = RequestMockBuilder({
- 'zoo.animals.insert': (None, '{"data": {"foo": "bar"}}',
- {'data': {'foo': 'bar'}})
- })
- zoo = build('zoo', 'v1', http=self.zoo_http, requestBuilder=requestBuilder)
+ def test_simple_matching_dict_body(self):
+ requestBuilder = RequestMockBuilder(
+ {
+ "zoo.animals.insert": (
+ None,
+ '{"data": {"foo": "bar"}}',
+ {"data": {"foo": "bar"}},
+ )
+ }
+ )
+ zoo = build("zoo", "v1", http=self.zoo_http, requestBuilder=requestBuilder)
- activity = zoo.animals().insert(
- body={'data': {'foo': 'bar'}}).execute()
- self.assertEqual({'foo': 'bar'}, activity)
+ activity = zoo.animals().insert(body={"data": {"foo": "bar"}}).execute()
+ self.assertEqual({"foo": "bar"}, activity)
- def test_errors(self):
- errorResponse = httplib2.Response({'status': 500, 'reason': 'Server Error'})
- requestBuilder = RequestMockBuilder({
- 'plus.activities.list': (errorResponse, b'{}')
- })
- plus = build('plus', 'v1', http=self.http, requestBuilder=requestBuilder)
+ def test_errors(self):
+ errorResponse = httplib2.Response({"status": 500, "reason": "Server Error"})
+ requestBuilder = RequestMockBuilder(
+ {"plus.activities.list": (errorResponse, b"{}")}
+ )
+ plus = build("plus", "v1", http=self.http, requestBuilder=requestBuilder)
- try:
- activity = plus.activities().list(collection='public', userId='me').execute()
- self.fail('An exception should have been thrown')
- except HttpError as e:
- self.assertEqual(b'{}', e.content)
- self.assertEqual(500, e.resp.status)
- self.assertEqual('Server Error', e.resp.reason)
+ try:
+ activity = (
+ plus.activities().list(collection="public", userId="me").execute()
+ )
+ self.fail("An exception should have been thrown")
+ except HttpError as e:
+ self.assertEqual(b"{}", e.content)
+ self.assertEqual(500, e.resp.status)
+ self.assertEqual("Server Error", e.resp.reason)
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_model.py b/tests/test_model.py
index 6506cfc..c3c936c 100644
--- a/tests/test_model.py
+++ b/tests/test_model.py
@@ -21,7 +21,7 @@
"""
from __future__ import absolute_import
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
import unittest2 as unittest
@@ -31,42 +31,43 @@
TEST_CASES = [
# (message, original, modified, expected)
- ("Remove an item from an object",
- {'a': 1, 'b': 2}, {'a': 1}, {'b': None}),
- ("Add an item to an object",
- {'a': 1}, {'a': 1, 'b': 2}, {'b': 2}),
- ("No changes",
- {'a': 1, 'b': 2}, {'a': 1, 'b': 2}, {}),
- ("Empty objects",
- {}, {}, {}),
- ("Modify an item in an object",
- {'a': 1, 'b': 2}, {'a': 1, 'b': 3}, {'b': 3}),
- ("Change an array",
- {'a': 1, 'b': [2, 3]}, {'a': 1, 'b': [2]}, {'b': [2]}),
- ("Modify a nested item",
- {'a': 1, 'b': {'foo':'bar', 'baz': 'qux'}},
- {'a': 1, 'b': {'foo':'bar', 'baz': 'qaax'}},
- {'b': {'baz': 'qaax'}}),
- ("Modify a nested array",
- {'a': 1, 'b': [{'foo':'bar', 'baz': 'qux'}]},
- {'a': 1, 'b': [{'foo':'bar', 'baz': 'qaax'}]},
- {'b': [{'foo':'bar', 'baz': 'qaax'}]}),
- ("Remove item from a nested array",
- {'a': 1, 'b': [{'foo':'bar', 'baz': 'qux'}]},
- {'a': 1, 'b': [{'foo':'bar'}]},
- {'b': [{'foo':'bar'}]}),
- ("Remove a nested item",
- {'a': 1, 'b': {'foo':'bar', 'baz': 'qux'}},
- {'a': 1, 'b': {'foo':'bar'}},
- {'b': {'baz': None}})
+ ("Remove an item from an object", {"a": 1, "b": 2}, {"a": 1}, {"b": None}),
+ ("Add an item to an object", {"a": 1}, {"a": 1, "b": 2}, {"b": 2}),
+ ("No changes", {"a": 1, "b": 2}, {"a": 1, "b": 2}, {}),
+ ("Empty objects", {}, {}, {}),
+ ("Modify an item in an object", {"a": 1, "b": 2}, {"a": 1, "b": 3}, {"b": 3}),
+ ("Change an array", {"a": 1, "b": [2, 3]}, {"a": 1, "b": [2]}, {"b": [2]}),
+ (
+ "Modify a nested item",
+ {"a": 1, "b": {"foo": "bar", "baz": "qux"}},
+ {"a": 1, "b": {"foo": "bar", "baz": "qaax"}},
+ {"b": {"baz": "qaax"}},
+ ),
+ (
+ "Modify a nested array",
+ {"a": 1, "b": [{"foo": "bar", "baz": "qux"}]},
+ {"a": 1, "b": [{"foo": "bar", "baz": "qaax"}]},
+ {"b": [{"foo": "bar", "baz": "qaax"}]},
+ ),
+ (
+ "Remove item from a nested array",
+ {"a": 1, "b": [{"foo": "bar", "baz": "qux"}]},
+ {"a": 1, "b": [{"foo": "bar"}]},
+ {"b": [{"foo": "bar"}]},
+ ),
+ (
+ "Remove a nested item",
+ {"a": 1, "b": {"foo": "bar", "baz": "qux"}},
+ {"a": 1, "b": {"foo": "bar"}},
+ {"b": {"baz": None}},
+ ),
]
class TestPatch(unittest.TestCase):
-
- def test_patch(self):
- for (msg, orig, mod, expected_patch) in TEST_CASES:
- self.assertEqual(expected_patch, makepatch(orig, mod), msg=msg)
+ def test_patch(self):
+ for (msg, orig, mod, expected_patch) in TEST_CASES:
+ self.assertEqual(expected_patch, makepatch(orig, mod), msg=msg)
class TestBaseModel(unittest.TestCase):
@@ -74,16 +75,16 @@
model = BaseModel()
test_cases = [
- ('hello', 'world', '?hello=world'),
- ('hello', u'world', '?hello=world'),
- ('hello', '세계', '?hello=%EC%84%B8%EA%B3%84'),
- ('hello', u'세계', '?hello=%EC%84%B8%EA%B3%84'),
- ('hello', 'こんにちは', '?hello=%E3%81%93%E3%82%93%E3%81%AB%E3%81%A1%E3%81%AF'),
- ('hello', u'こんにちは', '?hello=%E3%81%93%E3%82%93%E3%81%AB%E3%81%A1%E3%81%AF'),
- ('hello', '你好', '?hello=%E4%BD%A0%E5%A5%BD'),
- ('hello', u'你好', '?hello=%E4%BD%A0%E5%A5%BD'),
- ('hello', 'مرحبا', '?hello=%D9%85%D8%B1%D8%AD%D8%A8%D8%A7'),
- ('hello', u'مرحبا', '?hello=%D9%85%D8%B1%D8%AD%D8%A8%D8%A7')
+ ("hello", "world", "?hello=world"),
+ ("hello", u"world", "?hello=world"),
+ ("hello", "세계", "?hello=%EC%84%B8%EA%B3%84"),
+ ("hello", u"세계", "?hello=%EC%84%B8%EA%B3%84"),
+ ("hello", "こんにちは", "?hello=%E3%81%93%E3%82%93%E3%81%AB%E3%81%A1%E3%81%AF"),
+ ("hello", u"こんにちは", "?hello=%E3%81%93%E3%82%93%E3%81%AB%E3%81%A1%E3%81%AF"),
+ ("hello", "你好", "?hello=%E4%BD%A0%E5%A5%BD"),
+ ("hello", u"你好", "?hello=%E4%BD%A0%E5%A5%BD"),
+ ("hello", "مرحبا", "?hello=%D9%85%D8%B1%D8%AD%D8%A8%D8%A7"),
+ ("hello", u"مرحبا", "?hello=%D9%85%D8%B1%D8%AD%D8%A8%D8%A7"),
]
for case in test_cases:
@@ -91,5 +92,5 @@
self.assertEqual(expect, model._build_query({key: value}))
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_protobuf_model.py b/tests/test_protobuf_model.py
index 465d120..78caf4e 100644
--- a/tests/test_protobuf_model.py
+++ b/tests/test_protobuf_model.py
@@ -20,7 +20,7 @@
"""
from __future__ import absolute_import
-__author__ = 'mmcdonald@google.com (Matt McDonald)'
+__author__ = "mmcdonald@google.com (Matt McDonald)"
import unittest2 as unittest
import httplib2
@@ -33,68 +33,70 @@
class MockProtocolBuffer(object):
- def __init__(self, data=None):
- self.data = data
+ def __init__(self, data=None):
+ self.data = data
- def __eq__(self, other):
- return self.data == other.data
+ def __eq__(self, other):
+ return self.data == other.data
- @classmethod
- def FromString(cls, string):
- return cls(string)
+ @classmethod
+ def FromString(cls, string):
+ return cls(string)
- def SerializeToString(self):
- return self.data
+ def SerializeToString(self):
+ return self.data
class Model(unittest.TestCase):
- def setUp(self):
- self.model = ProtocolBufferModel(MockProtocolBuffer)
+ def setUp(self):
+ self.model = ProtocolBufferModel(MockProtocolBuffer)
- def test_no_body(self):
- headers = {}
- path_params = {}
- query_params = {}
- body = None
+ def test_no_body(self):
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = None
- headers, params, query, body = self.model.request(
- headers, path_params, query_params, body)
+ headers, params, query, body = self.model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/x-protobuf')
- self.assertTrue('content-type' not in headers)
- self.assertNotEqual(query, '')
- self.assertEqual(body, None)
+ self.assertEqual(headers["accept"], "application/x-protobuf")
+ self.assertTrue("content-type" not in headers)
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, None)
- def test_body(self):
- headers = {}
- path_params = {}
- query_params = {}
- body = MockProtocolBuffer('data')
+ def test_body(self):
+ headers = {}
+ path_params = {}
+ query_params = {}
+ body = MockProtocolBuffer("data")
- headers, params, query, body = self.model.request(
- headers, path_params, query_params, body)
+ headers, params, query, body = self.model.request(
+ headers, path_params, query_params, body
+ )
- self.assertEqual(headers['accept'], 'application/x-protobuf')
- self.assertEqual(headers['content-type'], 'application/x-protobuf')
- self.assertNotEqual(query, '')
- self.assertEqual(body, 'data')
+ self.assertEqual(headers["accept"], "application/x-protobuf")
+ self.assertEqual(headers["content-type"], "application/x-protobuf")
+ self.assertNotEqual(query, "")
+ self.assertEqual(body, "data")
- def test_good_response(self):
- resp = httplib2.Response({'status': '200'})
- resp.reason = 'OK'
- content = 'data'
+ def test_good_response(self):
+ resp = httplib2.Response({"status": "200"})
+ resp.reason = "OK"
+ content = "data"
- content = self.model.response(resp, content)
- self.assertEqual(content, MockProtocolBuffer('data'))
+ content = self.model.response(resp, content)
+ self.assertEqual(content, MockProtocolBuffer("data"))
- def test_no_content_response(self):
- resp = httplib2.Response({'status': '204'})
- resp.reason = 'No Content'
- content = ''
+ def test_no_content_response(self):
+ resp = httplib2.Response({"status": "204"})
+ resp.reason = "No Content"
+ content = ""
- content = self.model.response(resp, content)
- self.assertEqual(content, MockProtocolBuffer())
+ content = self.model.response(resp, content)
+ self.assertEqual(content, MockProtocolBuffer())
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_schema.py b/tests/test_schema.py
index c1216a5..1732d85 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -15,7 +15,7 @@
"""Unit tests for googleapiclient.schema."""
from __future__ import absolute_import
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+__author__ = "jcgregorio@google.com (Joe Gregorio)"
import json
import os
@@ -24,11 +24,12 @@
from googleapiclient.schema import Schemas
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
+DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
def datafile(filename):
- return os.path.join(DATA_DIR, filename)
+ return os.path.join(DATA_DIR, filename)
+
LOAD_FEED = """{
"items": [
@@ -46,113 +47,131 @@
"kind": "zoo#loadFeed",
}"""
+
class SchemasTest(unittest.TestCase):
- def setUp(self):
- f = open(datafile('zoo.json'))
- discovery = f.read()
- f.close()
- discovery = json.loads(discovery)
- self.sc = Schemas(discovery)
+ def setUp(self):
+ f = open(datafile("zoo.json"))
+ discovery = f.read()
+ f.close()
+ discovery = json.loads(discovery)
+ self.sc = Schemas(discovery)
- def test_basic_formatting(self):
- self.assertEqual(sorted(LOAD_FEED.splitlines()),
- sorted(self.sc.prettyPrintByName('LoadFeed').splitlines()))
+ def test_basic_formatting(self):
+ self.assertEqual(
+ sorted(LOAD_FEED.splitlines()),
+ sorted(self.sc.prettyPrintByName("LoadFeed").splitlines()),
+ )
- def test_empty_edge_case(self):
- self.assertTrue('Unknown type' in self.sc.prettyPrintSchema({}))
+ def test_empty_edge_case(self):
+ self.assertTrue("Unknown type" in self.sc.prettyPrintSchema({}))
- def test_simple_object(self):
- self.assertEqual({}, eval(self.sc.prettyPrintSchema({'type': 'object'})))
+ def test_simple_object(self):
+ self.assertEqual({}, eval(self.sc.prettyPrintSchema({"type": "object"})))
- def test_string(self):
- self.assertEqual(type(""), type(eval(self.sc.prettyPrintSchema({'type':
- 'string'}))))
+ def test_string(self):
+ self.assertEqual(
+ type(""), type(eval(self.sc.prettyPrintSchema({"type": "string"})))
+ )
- def test_integer(self):
- self.assertEqual(type(20), type(eval(self.sc.prettyPrintSchema({'type':
- 'integer'}))))
+ def test_integer(self):
+ self.assertEqual(
+ type(20), type(eval(self.sc.prettyPrintSchema({"type": "integer"})))
+ )
- def test_number(self):
- self.assertEqual(type(1.2), type(eval(self.sc.prettyPrintSchema({'type':
- 'number'}))))
+ def test_number(self):
+ self.assertEqual(
+ type(1.2), type(eval(self.sc.prettyPrintSchema({"type": "number"})))
+ )
- def test_boolean(self):
- self.assertEqual(type(True), type(eval(self.sc.prettyPrintSchema({'type':
- 'boolean'}))))
+ def test_boolean(self):
+ self.assertEqual(
+ type(True), type(eval(self.sc.prettyPrintSchema({"type": "boolean"})))
+ )
- def test_string_default(self):
- self.assertEqual('foo', eval(self.sc.prettyPrintSchema({'type':
- 'string', 'default': 'foo'})))
+ def test_string_default(self):
+ self.assertEqual(
+ "foo", eval(self.sc.prettyPrintSchema({"type": "string", "default": "foo"}))
+ )
- def test_integer_default(self):
- self.assertEqual(20, eval(self.sc.prettyPrintSchema({'type':
- 'integer', 'default': 20})))
+ def test_integer_default(self):
+ self.assertEqual(
+ 20, eval(self.sc.prettyPrintSchema({"type": "integer", "default": 20}))
+ )
- def test_number_default(self):
- self.assertEqual(1.2, eval(self.sc.prettyPrintSchema({'type':
- 'number', 'default': 1.2})))
+ def test_number_default(self):
+ self.assertEqual(
+ 1.2, eval(self.sc.prettyPrintSchema({"type": "number", "default": 1.2}))
+ )
- def test_boolean_default(self):
- self.assertEqual(False, eval(self.sc.prettyPrintSchema({'type':
- 'boolean', 'default': False})))
+ def test_boolean_default(self):
+ self.assertEqual(
+ False,
+ eval(self.sc.prettyPrintSchema({"type": "boolean", "default": False})),
+ )
- def test_null(self):
- self.assertEqual(None, eval(self.sc.prettyPrintSchema({'type': 'null'})))
+ def test_null(self):
+ self.assertEqual(None, eval(self.sc.prettyPrintSchema({"type": "null"})))
- def test_any(self):
- self.assertEqual('', eval(self.sc.prettyPrintSchema({'type': 'any'})))
+ def test_any(self):
+ self.assertEqual("", eval(self.sc.prettyPrintSchema({"type": "any"})))
- def test_array(self):
- self.assertEqual([{}], eval(self.sc.prettyPrintSchema({'type': 'array',
- 'items': {'type': 'object'}})))
+ def test_array(self):
+ self.assertEqual(
+ [{}],
+ eval(
+ self.sc.prettyPrintSchema(
+ {"type": "array", "items": {"type": "object"}}
+ )
+ ),
+ )
- def test_nested_references(self):
- feed = {
- 'items': [ {
- 'photo': {
- 'hash': 'A String',
- 'hashAlgorithm': 'A String',
- 'filename': 'A String',
- 'type': 'A String',
- 'size': 42
- },
- 'kind': 'zoo#animal',
- 'etag': 'A String',
- 'name': 'A String'
- }
- ],
- 'kind': 'zoo#animalFeed',
- 'etag': 'A String'
- }
+ def test_nested_references(self):
+ feed = {
+ "items": [
+ {
+ "photo": {
+ "hash": "A String",
+ "hashAlgorithm": "A String",
+ "filename": "A String",
+ "type": "A String",
+ "size": 42,
+ },
+ "kind": "zoo#animal",
+ "etag": "A String",
+ "name": "A String",
+ }
+ ],
+ "kind": "zoo#animalFeed",
+ "etag": "A String",
+ }
- self.assertEqual(feed, eval(self.sc.prettyPrintByName('AnimalFeed')))
+ self.assertEqual(feed, eval(self.sc.prettyPrintByName("AnimalFeed")))
- def test_additional_properties(self):
- items = {
- 'animals': {
- 'a_key': {
- 'photo': {
- 'hash': 'A String',
- 'hashAlgorithm': 'A String',
- 'filename': 'A String',
- 'type': 'A String',
- 'size': 42
- },
- 'kind': 'zoo#animal',
- 'etag': 'A String',
- 'name': 'A String'
- }
- },
- 'kind': 'zoo#animalMap',
- 'etag': 'A String'
- }
+ def test_additional_properties(self):
+ items = {
+ "animals": {
+ "a_key": {
+ "photo": {
+ "hash": "A String",
+ "hashAlgorithm": "A String",
+ "filename": "A String",
+ "type": "A String",
+ "size": 42,
+ },
+ "kind": "zoo#animal",
+ "etag": "A String",
+ "name": "A String",
+ }
+ },
+ "kind": "zoo#animalMap",
+ "etag": "A String",
+ }
- self.assertEqual(items, eval(self.sc.prettyPrintByName('AnimalMap')))
+ self.assertEqual(items, eval(self.sc.prettyPrintByName("AnimalMap")))
- def test_unknown_name(self):
- self.assertRaises(KeyError,
- self.sc.prettyPrintByName, 'UknownSchemaThing')
+ def test_unknown_name(self):
+ self.assertRaises(KeyError, self.sc.prettyPrintByName, "UknownSchemaThing")
-if __name__ == '__main__':
- unittest.main()
+if __name__ == "__main__":
+ unittest.main()