tests/test_json_model.py

#!/usr/bin/python2.4
#
# Copyright 2010 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""JSON Model tests

Unit tests for the JSON model.
"""

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

import copy
import gflags
import os
import unittest
import httplib2
import apiclient.model

from apiclient.anyjson import simplejson
from apiclient.errors import HttpError
from apiclient.model import JsonModel
from apiclient.model import LoggingJsonModel

FLAGS = gflags.FLAGS

# Python 2.5 requires different modules
try:
  from urlparse import parse_qs
except ImportError:
  from cgi import parse_qs


class Model(unittest.TestCase):
  def test_json_no_body(self):
    model = JsonModel(data_wrapper=False)

    headers = {}
    path_params = {}
    query_params = {}
    body = None

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

  def test_json_body(self):
    model = JsonModel(data_wrapper=False)

    headers = {}
    path_params = {}
    query_params = {}
    body = {}

    headers, 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, '{}')

  def test_json_body_data_wrapper(self):
    model = JsonModel(data_wrapper=True)

    headers = {}
    path_params = {}
    query_params = {}
    body = {}

    headers, 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": {}}')

  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, 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"}')

  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, 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'])
    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, params, query, body = model.request(headers, path_params, query_params, body)

    self.assertEqual(headers['user-agent'], 'my-test-app/1.23.4 google-api-python-client/1.0')

  def test_bad_response(self):
    model = JsonModel(data_wrapper=False)
    resp = httplib2.Response({'status': '401'})
    resp.reason = 'Unauthorized'
    content = '{"error": {"message": "not authorized"}}'

    try:
      content = model.response(resp, content)
      self.fail('Should have thrown an exception')
    except HttpError, e:
      self.assertTrue('Unauthorized' in str(e))

    resp['content-type'] = 'application/json'

    try:
      content = model.response(resp, content)
      self.fail('Should have thrown an exception')
    except HttpError, 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')


class LoggingModel(unittest.TestCase):

  def test_logging_json_model(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 items.iteritems():
          self[key] = value
    apiclient.model.logging = MockLogging()
    apiclient.model.FLAGS = copy.deepcopy(FLAGS)
    apiclient.model.FLAGS.dump_request_response = True
    model = LoggingJsonModel()
    request_body = {
        'field1': 'value1',
        'field2': 'value2'
        }
    body_string = model.request({}, {}, {}, request_body)[-1]
    json_body = simplejson.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(apiclient.model.logging.info_record[:2],
                     ['--request-start--',
                      '-headers-start-'])
    self.assertTrue('response_field_1: response_value_1' in
                    apiclient.model.logging.info_record)
    self.assertTrue('response_field_2: response_value_2' in
                    apiclient.model.logging.info_record)
    self.assertEqual(simplejson.loads(apiclient.model.logging.info_record[-2]),
                     request_body)
    self.assertEqual(apiclient.model.logging.info_record[-1],
                     '--response-end--')



if __name__ == '__main__':
  unittest.main()