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
22import six
23
24__author__ = "jcgregorio@google.com (Joe Gregorio)"
25
26import copy
27import json
28import os
29import platform
30import unittest2 as unittest
31import httplib2
32import googleapiclient.model
33
34from googleapiclient import __version__
35from googleapiclient.errors import HttpError
36from googleapiclient.model import JsonModel
37
38from six.moves.urllib.parse import parse_qs
39
40
41class Model(unittest.TestCase):
42    def test_json_no_body(self):
43        model = JsonModel(data_wrapper=False)
44
45        headers = {}
46        path_params = {}
47        query_params = {}
48        body = None
49
50        headers, unused_params, query, body = model.request(
51            headers, path_params, query_params, body
52        )
53
54        self.assertEqual(headers["accept"], "application/json")
55        self.assertTrue("content-type" not in headers)
56        self.assertNotEqual(query, "")
57        self.assertEqual(body, None)
58
59    def test_json_body(self):
60        model = JsonModel(data_wrapper=False)
61
62        headers = {}
63        path_params = {}
64        query_params = {}
65        body = {}
66
67        headers, unused_params, query, body = model.request(
68            headers, path_params, query_params, body
69        )
70
71        self.assertEqual(headers["accept"], "application/json")
72        self.assertEqual(headers["content-type"], "application/json")
73        self.assertNotEqual(query, "")
74        self.assertEqual(body, "{}")
75
76    def test_json_body_data_wrapper(self):
77        model = JsonModel(data_wrapper=True)
78
79        headers = {}
80        path_params = {}
81        query_params = {}
82        body = {}
83
84        headers, unused_params, query, body = model.request(
85            headers, path_params, query_params, body
86        )
87
88        self.assertEqual(headers["accept"], "application/json")
89        self.assertEqual(headers["content-type"], "application/json")
90        self.assertNotEqual(query, "")
91        self.assertEqual(body, '{"data": {}}')
92
93    def test_json_body_default_data(self):
94        """Test that a 'data' wrapper doesn't get added if one is already present."""
95        model = JsonModel(data_wrapper=True)
96
97        headers = {}
98        path_params = {}
99        query_params = {}
100        body = {"data": "foo"}
101
102        headers, unused_params, query, body = model.request(
103            headers, path_params, query_params, body
104        )
105
106        self.assertEqual(headers["accept"], "application/json")
107        self.assertEqual(headers["content-type"], "application/json")
108        self.assertNotEqual(query, "")
109        self.assertEqual(body, '{"data": "foo"}')
110
111    def test_json_build_query(self):
112        model = JsonModel(data_wrapper=False)
113
114        headers = {}
115        path_params = {}
116        query_params = {
117            "foo": 1,
118            "bar": u"\N{COMET}",
119            "baz": ["fe", "fi", "fo", "fum"],  # Repeated parameters
120            "qux": [],
121        }
122        body = {}
123
124        headers, unused_params, query, body = model.request(
125            headers, path_params, query_params, body
126        )
127
128        self.assertEqual(headers["accept"], "application/json")
129        self.assertEqual(headers["content-type"], "application/json")
130
131        query_dict = parse_qs(query[1:])
132        self.assertEqual(query_dict["foo"], ["1"])
133        if six.PY3:
134            # Python 3, no need to encode
135            self.assertEqual(query_dict["bar"], [u"\N{COMET}"])
136        else:
137            # Python 2, encode string
138            self.assertEqual(query_dict["bar"], [u"\N{COMET}".encode("utf-8")])
139        self.assertEqual(query_dict["baz"], ["fe", "fi", "fo", "fum"])
140        self.assertTrue("qux" not in query_dict)
141        self.assertEqual(body, "{}")
142
143    def test_user_agent(self):
144        model = JsonModel(data_wrapper=False)
145
146        headers = {"user-agent": "my-test-app/1.23.4"}
147        path_params = {}
148        query_params = {}
149        body = {}
150
151        headers, unused_params, unused_query, body = model.request(
152            headers, path_params, query_params, body
153        )
154
155        self.assertEqual(headers["user-agent"], "my-test-app/1.23.4 (gzip)")
156
157    def test_x_goog_api_client(self):
158        model = JsonModel(data_wrapper=False)
159
160        # test header composition for cloud clients that wrap discovery
161        headers = {"x-goog-api-client": "gccl/1.23.4"}
162        path_params = {}
163        query_params = {}
164        body = {}
165
166        headers, unused_params, unused_query, body = model.request(
167            headers, path_params, query_params, body
168        )
169
170        self.assertEqual(
171            headers["x-goog-api-client"],
172            "gccl/1.23.4"
173            + " gdcl/"
174            + __version__
175            + " gl-python/"
176            + platform.python_version(),
177        )
178
179    def test_bad_response(self):
180        model = JsonModel(data_wrapper=False)
181        resp = httplib2.Response({"status": "401"})
182        resp.reason = "Unauthorized"
183        content = b'{"error": {"message": "not authorized"}}'
184
185        try:
186            content = model.response(resp, content)
187            self.fail("Should have thrown an exception")
188        except HttpError as e:
189            self.assertTrue("not authorized" in str(e))
190
191        resp["content-type"] = "application/json"
192
193        try:
194            content = model.response(resp, content)
195            self.fail("Should have thrown an exception")
196        except HttpError as e:
197            self.assertTrue("not authorized" in str(e))
198
199    def test_good_response(self):
200        model = JsonModel(data_wrapper=True)
201        resp = httplib2.Response({"status": "200"})
202        resp.reason = "OK"
203        content = '{"data": "is good"}'
204
205        content = model.response(resp, content)
206        self.assertEqual(content, "is good")
207
208    def test_good_response_wo_data(self):
209        model = JsonModel(data_wrapper=False)
210        resp = httplib2.Response({"status": "200"})
211        resp.reason = "OK"
212        content = '{"foo": "is good"}'
213
214        content = model.response(resp, content)
215        self.assertEqual(content, {"foo": "is good"})
216
217    def test_good_response_wo_data_str(self):
218        model = JsonModel(data_wrapper=False)
219        resp = httplib2.Response({"status": "200"})
220        resp.reason = "OK"
221        content = '"data goes here"'
222
223        content = model.response(resp, content)
224        self.assertEqual(content, "data goes here")
225
226    def test_no_content_response(self):
227        model = JsonModel(data_wrapper=False)
228        resp = httplib2.Response({"status": "204"})
229        resp.reason = "No Content"
230        content = ""
231
232        content = model.response(resp, content)
233        self.assertEqual(content, {})
234
235    def test_logging(self):
236        class MockLogging(object):
237            def __init__(self):
238                self.info_record = []
239                self.debug_record = []
240
241            def info(self, message, *args):
242                self.info_record.append(message % args)
243
244            def debug(self, message, *args):
245                self.debug_record.append(message % args)
246
247        class MockResponse(dict):
248            def __init__(self, items):
249                super(MockResponse, self).__init__()
250                self.status = items["status"]
251                for key, value in six.iteritems(items):
252                    self[key] = value
253
254        old_logging = googleapiclient.model.LOGGER
255        googleapiclient.model.LOGGER = MockLogging()
256        googleapiclient.model.dump_request_response = True
257        model = JsonModel()
258        request_body = {"field1": "value1", "field2": "value2"}
259        body_string = model.request({}, {}, {}, request_body)[-1]
260        json_body = json.loads(body_string)
261        self.assertEqual(request_body, json_body)
262
263        response = {
264            "status": 200,
265            "response_field_1": "response_value_1",
266            "response_field_2": "response_value_2",
267        }
268        response_body = model.response(MockResponse(response), body_string)
269        self.assertEqual(request_body, response_body)
270        self.assertEqual(
271            googleapiclient.model.LOGGER.info_record[:2],
272            ["--request-start--", "-headers-start-"],
273        )
274        self.assertTrue(
275            "response_field_1: response_value_1"
276            in googleapiclient.model.LOGGER.info_record
277        )
278        self.assertTrue(
279            "response_field_2: response_value_2"
280            in googleapiclient.model.LOGGER.info_record
281        )
282        self.assertEqual(
283            json.loads(googleapiclient.model.LOGGER.info_record[-2]), request_body
284        )
285        self.assertEqual(
286            googleapiclient.model.LOGGER.info_record[-1], "--response-end--"
287        )
288        googleapiclient.model.LOGGER = old_logging
289
290    def test_no_data_wrapper_deserialize(self):
291        model = JsonModel(data_wrapper=False)
292        resp = httplib2.Response({"status": "200"})
293        resp.reason = "OK"
294        content = '{"data": "is good"}'
295        content = model.response(resp, content)
296        self.assertEqual(content, {"data": "is good"})
297
298    def test_data_wrapper_deserialize(self):
299        model = JsonModel(data_wrapper=True)
300        resp = httplib2.Response({"status": "200"})
301        resp.reason = "OK"
302        content = '{"data": "is good"}'
303        content = model.response(resp, content)
304        self.assertEqual(content, "is good")
305
306    def test_data_wrapper_deserialize_nodata(self):
307        model = JsonModel(data_wrapper=True)
308        resp = httplib2.Response({"status": "200"})
309        resp.reason = "OK"
310        content = '{"atad": "is good"}'
311        content = model.response(resp, content)
312        self.assertEqual(content, {"atad": "is good"})
313
314
315if __name__ == "__main__":
316    unittest.main()
317