1#!/usr/bin/env python 2# 3# Copyright 2014 Google Inc. All Rights Reserved. 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17"""JSON Model tests 18 19Unit tests for the JSON model. 20""" 21from __future__ import absolute_import 22 23__author__ = "jcgregorio@google.com (Joe Gregorio)" 24 25import io 26import httplib2 27import json 28import platform 29import unittest 30import urllib 31 32import googleapiclient.model 33 34from googleapiclient import version as googleapiclient_version 35from googleapiclient.errors import HttpError 36from googleapiclient.model import JsonModel 37 38_LIBRARY_VERSION = googleapiclient_version.__version__ 39CSV_TEXT_MOCK = 'column1,column2,column3\nstring1,1.2,string2' 40 41 42class Model(unittest.TestCase): 43 def test_json_no_body(self): 44 model = JsonModel(data_wrapper=False) 45 46 headers = {} 47 path_params = {} 48 query_params = {} 49 body = None 50 51 headers, unused_params, query, body = model.request( 52 headers, path_params, query_params, body 53 ) 54 55 self.assertEqual(headers["accept"], "application/json") 56 self.assertTrue("content-type" not in headers) 57 self.assertNotEqual(query, "") 58 self.assertEqual(body, None) 59 60 def test_json_body(self): 61 model = JsonModel(data_wrapper=False) 62 63 headers = {} 64 path_params = {} 65 query_params = {} 66 body = {} 67 68 headers, unused_params, query, body = model.request( 69 headers, path_params, query_params, body 70 ) 71 72 self.assertEqual(headers["accept"], "application/json") 73 self.assertEqual(headers["content-type"], "application/json") 74 self.assertNotEqual(query, "") 75 self.assertEqual(body, "{}") 76 77 def test_json_body_data_wrapper(self): 78 model = JsonModel(data_wrapper=True) 79 80 headers = {} 81 path_params = {} 82 query_params = {} 83 body = {} 84 85 headers, unused_params, query, body = model.request( 86 headers, path_params, query_params, body 87 ) 88 89 self.assertEqual(headers["accept"], "application/json") 90 self.assertEqual(headers["content-type"], "application/json") 91 self.assertNotEqual(query, "") 92 self.assertEqual(body, '{"data": {}}') 93 94 def test_json_body_default_data(self): 95 """Test that a 'data' wrapper doesn't get added if one is already present.""" 96 model = JsonModel(data_wrapper=True) 97 98 headers = {} 99 path_params = {} 100 query_params = {} 101 body = {"data": "foo"} 102 103 headers, unused_params, query, body = model.request( 104 headers, path_params, query_params, body 105 ) 106 107 self.assertEqual(headers["accept"], "application/json") 108 self.assertEqual(headers["content-type"], "application/json") 109 self.assertNotEqual(query, "") 110 self.assertEqual(body, '{"data": "foo"}') 111 112 def test_json_build_query(self): 113 model = JsonModel(data_wrapper=False) 114 115 headers = {} 116 path_params = {} 117 query_params = { 118 "foo": 1, 119 "bar": u"\N{COMET}", 120 "baz": ["fe", "fi", "fo", "fum"], # Repeated parameters 121 "qux": [], 122 } 123 body = {} 124 125 headers, unused_params, query, body = model.request( 126 headers, path_params, query_params, body 127 ) 128 129 self.assertEqual(headers["accept"], "application/json") 130 self.assertEqual(headers["content-type"], "application/json") 131 132 query_dict = urllib.parse.parse_qs(query[1:]) 133 self.assertEqual(query_dict["foo"], ["1"]) 134 self.assertEqual(query_dict["bar"], [u"\N{COMET}"]) 135 self.assertEqual(query_dict["baz"], ["fe", "fi", "fo", "fum"]) 136 self.assertTrue("qux" not in query_dict) 137 self.assertEqual(body, "{}") 138 139 def test_user_agent(self): 140 model = JsonModel(data_wrapper=False) 141 142 headers = {"user-agent": "my-test-app/1.23.4"} 143 path_params = {} 144 query_params = {} 145 body = {} 146 147 headers, unused_params, unused_query, body = model.request( 148 headers, path_params, query_params, body 149 ) 150 151 self.assertEqual(headers["user-agent"], "my-test-app/1.23.4 (gzip)") 152 153 def test_x_goog_api_client(self): 154 model = JsonModel(data_wrapper=False) 155 156 # test header composition for cloud clients that wrap discovery 157 headers = {"x-goog-api-client": "gccl/1.23.4"} 158 path_params = {} 159 query_params = {} 160 body = {} 161 162 headers, unused_params, unused_query, body = model.request( 163 headers, path_params, query_params, body 164 ) 165 166 self.assertEqual( 167 headers["x-goog-api-client"], 168 "gccl/1.23.4" 169 + " gdcl/" 170 + _LIBRARY_VERSION 171 + " gl-python/" 172 + platform.python_version(), 173 ) 174 175 def test_bad_response(self): 176 model = JsonModel(data_wrapper=False) 177 resp = httplib2.Response({"status": "401"}) 178 resp.reason = "Unauthorized" 179 content = b'{"error": {"message": "not authorized"}}' 180 181 try: 182 content = model.response(resp, content) 183 self.fail("Should have thrown an exception") 184 except HttpError as e: 185 self.assertTrue("not authorized" in str(e)) 186 187 resp["content-type"] = "application/json" 188 189 try: 190 content = model.response(resp, content) 191 self.fail("Should have thrown an exception") 192 except HttpError as e: 193 self.assertTrue("not authorized" in str(e)) 194 195 def test_good_response(self): 196 model = JsonModel(data_wrapper=True) 197 resp = httplib2.Response({"status": "200"}) 198 resp.reason = "OK" 199 content = '{"data": "is good"}' 200 201 content = model.response(resp, content) 202 self.assertEqual(content, "is good") 203 204 def test_good_response_wo_data(self): 205 model = JsonModel(data_wrapper=False) 206 resp = httplib2.Response({"status": "200"}) 207 resp.reason = "OK" 208 content = '{"foo": "is good"}' 209 210 content = model.response(resp, content) 211 self.assertEqual(content, {"foo": "is good"}) 212 213 def test_good_response_wo_data_str(self): 214 model = JsonModel(data_wrapper=False) 215 resp = httplib2.Response({"status": "200"}) 216 resp.reason = "OK" 217 content = '"data goes here"' 218 219 content = model.response(resp, content) 220 self.assertEqual(content, "data goes here") 221 222 def test_no_content_response(self): 223 model = JsonModel(data_wrapper=False) 224 resp = httplib2.Response({"status": "204"}) 225 resp.reason = "No Content" 226 content = "" 227 228 content = model.response(resp, content) 229 self.assertEqual(content, {}) 230 231 def test_logging(self): 232 class MockLogging(object): 233 def __init__(self): 234 self.info_record = [] 235 self.debug_record = [] 236 237 def info(self, message, *args): 238 self.info_record.append(message % args) 239 240 def debug(self, message, *args): 241 self.debug_record.append(message % args) 242 243 class MockResponse(dict): 244 def __init__(self, items): 245 super(MockResponse, self).__init__() 246 self.status = items["status"] 247 for key, value in items.items(): 248 self[key] = value 249 250 old_logging = googleapiclient.model.LOGGER 251 googleapiclient.model.LOGGER = MockLogging() 252 googleapiclient.model.dump_request_response = True 253 model = JsonModel() 254 request_body = {"field1": "value1", "field2": "value2"} 255 body_string = model.request({}, {}, {}, request_body)[-1] 256 json_body = json.loads(body_string) 257 self.assertEqual(request_body, json_body) 258 259 response = { 260 "status": 200, 261 "response_field_1": "response_value_1", 262 "response_field_2": "response_value_2", 263 } 264 response_body = model.response(MockResponse(response), body_string) 265 self.assertEqual(request_body, response_body) 266 self.assertEqual( 267 googleapiclient.model.LOGGER.info_record[:2], 268 ["--request-start--", "-headers-start-"], 269 ) 270 self.assertTrue( 271 "response_field_1: response_value_1" 272 in googleapiclient.model.LOGGER.info_record 273 ) 274 self.assertTrue( 275 "response_field_2: response_value_2" 276 in googleapiclient.model.LOGGER.info_record 277 ) 278 self.assertEqual( 279 json.loads(googleapiclient.model.LOGGER.info_record[-2]), request_body 280 ) 281 self.assertEqual( 282 googleapiclient.model.LOGGER.info_record[-1], "--response-end--" 283 ) 284 googleapiclient.model.LOGGER = old_logging 285 286 def test_no_data_wrapper_deserialize(self): 287 model = JsonModel(data_wrapper=False) 288 resp = httplib2.Response({"status": "200"}) 289 resp.reason = "OK" 290 content = '{"data": "is good"}' 291 content = model.response(resp, content) 292 self.assertEqual(content, {"data": "is good"}) 293 294 def test_no_data_wrapper_deserialize_text_format(self): 295 model = JsonModel(data_wrapper=False) 296 resp = httplib2.Response({"status": "200"}) 297 resp.reason = "OK" 298 content = CSV_TEXT_MOCK 299 content = model.response(resp, content) 300 self.assertEqual(content, CSV_TEXT_MOCK) 301 302 def test_no_data_wrapper_deserialize_raise_type_error(self): 303 buffer = io.StringIO() 304 buffer.write('String buffer') 305 model = JsonModel(data_wrapper=False) 306 resp = httplib2.Response({"status": "500"}) 307 resp.reason = "The JSON object must be str, bytes or bytearray, not StringIO" 308 content = buffer 309 with self.assertRaises(TypeError): 310 model.response(resp, content) 311 312 def test_data_wrapper_deserialize(self): 313 model = JsonModel(data_wrapper=True) 314 resp = httplib2.Response({"status": "200"}) 315 resp.reason = "OK" 316 content = '{"data": "is good"}' 317 content = model.response(resp, content) 318 self.assertEqual(content, "is good") 319 320 def test_data_wrapper_deserialize_nodata(self): 321 model = JsonModel(data_wrapper=True) 322 resp = httplib2.Response({"status": "200"}) 323 resp.reason = "OK" 324 content = '{"atad": "is good"}' 325 content = model.response(resp, content) 326 self.assertEqual(content, {"atad": "is good"}) 327 328 329if __name__ == "__main__": 330 unittest.main() 331