tests/gdata_tests/client_test.py

#!/usr/bin/env python
#
# Copyright (C) 2008, 2009 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.


# This module is used for version 2 of the Google Data APIs.


__author__ = 'j.s@google.com (Jeff Scudder)'


import unittest
import gdata.client
import gdata.gauth
import gdata.data
import atom.mock_http_core
import StringIO


class ClientLoginTest(unittest.TestCase):

  def test_token_request(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(200, 'OK', 
        'SID=DQAAAGgA...7Zg8CTN\n'
        'LSID=DQAAAGsA...lk8BBbG\n'
        'Auth=DQAAAGgA...dk3fA5N', {'Content-Type': 'text/plain'})
    token = client.request_client_login_token('email', 'pw', 'cp', 'test')
    self.assert_(isinstance(token, gdata.gauth.ClientLoginToken))
    self.assertEqual(token.token_string, 'DQAAAGgA...dk3fA5N')

    # Test a server response without a ClientLogin token.`
    client.http_client.set_response(200, 'OK', 'SID=12345\nLSID=34567', {})
    self.assertRaises(gdata.client.ClientLoginTokenMissing,
        client.request_client_login_token, 'email', 'pw', '', '')

    # Test a 302 redirect from the server on a login request.
    client.http_client.set_response(302, 'ignored', '', {})
    # TODO: change the exception class to one in gdata.client.
    self.assertRaises(gdata.client.BadAuthenticationServiceURL,
        client.request_client_login_token, 'email', 'pw', '', '')

    # Test a CAPTCHA challenge from the server
    client.http_client.set_response(403, 'Access Forbidden', 
        'Url=http://www.google.com/login/captcha\n'
        'Error=CaptchaRequired\n'
        'CaptchaToken=DQAAAGgA...dkI1LK9\n'
        # TODO: verify this sample CAPTCHA URL matches an
        # actual challenge from the server.
        'CaptchaUrl=Captcha?ctoken=HiteT4bVoP6-yFkHPibe7O9EqxeiI7lUSN', {})
    try:
      token = client.request_client_login_token('email', 'pw', '', '')
      self.fail('should raise a CaptchaChallenge on a 403 with a '
                'CaptchRequired error.')
    except gdata.client.CaptchaChallenge, challenge:
      self.assertEquals(challenge.captcha_url, 
          'http://www.google.com/accounts/'
          'Captcha?ctoken=HiteT4bVoP6-yFkHPibe7O9EqxeiI7lUSN')
      self.assertEquals(challenge.captcha_token, 'DQAAAGgA...dkI1LK9')

    # Test an unexpected response, a 404 for example.
    client.http_client.set_response(404, 'ignored', '', {})
    self.assertRaises(gdata.client.ClientLoginFailed,
        client.request_client_login_token, 'email', 'pw', '', '')

  def test_client_login(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(200, 'OK', 
        'SID=DQAAAGgA...7Zg8CTN\n'
        'LSID=DQAAAGsA...lk8BBbG\n'
        'Auth=DQAAAGgA...dk3fA5N', {'Content-Type': 'text/plain'})
    client.client_login('me@example.com', 'password', 'wise', 'unit test')
    self.assert_(isinstance(client.auth_token, gdata.gauth.ClientLoginToken))
    self.assertEqual(client.auth_token.token_string, 'DQAAAGgA...dk3fA5N')


class AuthSubTest(unittest.TestCase):

  def test_get_and_upgrade_token(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(200, 'OK', 
        'Token=UpgradedTokenVal\n'
        'Extra data', {'Content-Type': 'text/plain'})

    page_url = 'http://example.com/showcalendar.html?token=CKF50YzIHxCTKMAg'

    client.auth_token = gdata.gauth.AuthSubToken.from_url(page_url)

    self.assert_(isinstance(client.auth_token, gdata.gauth.AuthSubToken))
    self.assertEqual(client.auth_token.token_string, 'CKF50YzIHxCTKMAg')

    upgraded = client.upgrade_token()

    self.assert_(isinstance(client.auth_token, gdata.gauth.AuthSubToken))
    self.assertEqual(client.auth_token.token_string, 'UpgradedTokenVal')
    self.assertEqual(client.auth_token, upgraded)

    # Ensure passing in a token returns without modifying client's auth_token.
    client.http_client.set_response(200, 'OK', 'Token=4567', {})
    upgraded = client.upgrade_token(
        gdata.gauth.AuthSubToken.from_url('?token=1234'))
    self.assertEqual(upgraded.token_string, '4567')
    self.assertEqual(client.auth_token.token_string, 'UpgradedTokenVal')
    self.assertNotEqual(client.auth_token, upgraded)

    # Test exception cases
    client.auth_token = None
    self.assertRaises(gdata.client.UnableToUpgradeToken, client.upgrade_token,
                      None)
    self.assertRaises(gdata.client.UnableToUpgradeToken, client.upgrade_token)

  def test_revoke_token(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(
        200, 'OK', '', {})
    page_url = 'http://example.com/showcalendar.html?token=CKF50YzIHxCTKMAg'
    client.auth_token = gdata.gauth.AuthSubToken.from_url(page_url)
    deleted = client.revoke_token()
    self.assert_(deleted)
    self.assertEqual(
        client.http_client.last_request.headers['Authorization'],
        'AuthSub token=CKF50YzIHxCTKMAg')


class OAuthTest(unittest.TestCase):

  def test_hmac_flow(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(
        200, 'OK', 'oauth_token=ab3cd9j4ks7&oauth_token_secret=ZXhhbXBsZS',
        {})
    request_token = client.get_oauth_token(
        ['http://example.com/service'], 'http://example.net/myapp',
        'consumer', consumer_secret='secret')
    # Check that the response was correctly parsed.
    self.assertEqual(request_token.token, 'ab3cd9j4ks7')
    self.assertEqual(request_token.token_secret, 'ZXhhbXBsZS')
    self.assertEqual(request_token.auth_state, gdata.gauth.REQUEST_TOKEN)

    # Also check the Authorization header which was sent in the request.
    auth_header = client.http_client.last_request.headers['Authorization']
    self.assert_('OAuth' in auth_header)
    self.assert_(
        'oauth_callback="http%3A%2F%2Fexample.net%2Fmyapp"' in auth_header)
    self.assert_('oauth_version="1.0"' in auth_header)
    self.assert_('oauth_signature_method="HMAC-SHA1"' in auth_header)
    self.assert_('oauth_consumer_key="consumer"' in auth_header)

    # Check generation of the authorization URL.
    authorize_url = request_token.generate_authorization_url()
    self.assert_(str(authorize_url).startswith(
        'https://www.google.com/accounts/OAuthAuthorizeToken'))
    self.assert_('oauth_token=ab3cd9j4ks7' in str(authorize_url))

    # Check that the token information from the browser's URL is parsed.
    redirected_url = (
        'http://example.net/myapp?oauth_token=CKF5zz&oauth_verifier=Xhhbas')
    gdata.gauth.authorize_request_token(request_token, redirected_url)
    self.assertEqual(request_token.token, 'CKF5zz')
    self.assertEqual(request_token.verifier, 'Xhhbas')
    self.assertEqual(request_token.auth_state,
                     gdata.gauth.AUTHORIZED_REQUEST_TOKEN)

    # Check that the token upgrade response was correctly parsed.
    client.http_client.set_response(
        200, 'OK', 'oauth_token=3cd9Fj417&oauth_token_secret=Xhrh6bXBs', {})
    access_token = client.get_access_token(request_token)
    self.assertEqual(request_token.token, '3cd9Fj417')
    self.assertEqual(request_token.token_secret, 'Xhrh6bXBs')
    self.assert_(request_token.verifier is None)
    self.assertEqual(request_token.auth_state, gdata.gauth.ACCESS_TOKEN)
    self.assertEqual(request_token.token, access_token.token)
    self.assertEqual(request_token.token_secret, access_token.token_secret)
    self.assert_(access_token.verifier is None)
    self.assertEqual(request_token.auth_state, access_token.auth_state)

    # Also check the Authorization header which was sent in the request.
    auth_header = client.http_client.last_request.headers['Authorization']
    self.assert_('OAuth' in auth_header)
    self.assert_('oauth_callback="' not in auth_header)
    self.assert_('oauth_version="1.0"' in auth_header)
    self.assert_('oauth_verifier="Xhhbas"' in auth_header)
    self.assert_('oauth_signature_method="HMAC-SHA1"' in auth_header)
    self.assert_('oauth_consumer_key="consumer"' in auth_header)


class RequestTest(unittest.TestCase):

  def test_simple_request(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.EchoHttpClient()
    response = client.request('GET', 'https://example.com/test')
    self.assertEqual(response.getheader('Echo-Host'), 'example.com:None')
    self.assertEqual(response.getheader('Echo-Uri'), '/test')
    self.assertEqual(response.getheader('Echo-Scheme'), 'https')
    self.assertEqual(response.getheader('Echo-Method'), 'GET')

    http_request = atom.http_core.HttpRequest(
        uri=atom.http_core.Uri(scheme='http', host='example.net', port=8080),
        method='POST', headers={'X': 1})
    http_request.add_body_part('test', 'text/plain')
    response = client.request(http_request=http_request)
    self.assertEqual(response.getheader('Echo-Host'), 'example.net:8080')
    # A Uri with path set to None should default to /.
    self.assertEqual(response.getheader('Echo-Uri'), '/')
    self.assertEqual(response.getheader('Echo-Scheme'), 'http')
    self.assertEqual(response.getheader('Echo-Method'), 'POST')
    self.assertEqual(response.getheader('Content-Type'), 'text/plain')
    self.assertEqual(response.getheader('X'), '1')
    self.assertEqual(response.read(), 'test')

    # Use the same request object from above, but overwrite the request path
    # by passing in a URI.
    response = client.request(uri='/new/path?p=1', http_request=http_request)
    self.assertEqual(response.getheader('Echo-Host'), 'example.net:8080')
    self.assertEqual(response.getheader('Echo-Uri'), '/new/path?p=1')
    self.assertEqual(response.getheader('Echo-Scheme'), 'http')
    self.assertEqual(response.read(), 'test')

  def test_gdata_version_header(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.EchoHttpClient()

    response = client.request('GET', 'http://example.com')
    self.assertEqual(response.getheader('GData-Version'), None)

    client.api_version = '2'
    response = client.request('GET', 'http://example.com')
    self.assertEqual(response.getheader('GData-Version'), '2')

  def test_redirects(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.MockHttpClient()
    # Add the redirect response for the initial request.
    first_request = atom.http_core.HttpRequest('http://example.com/1', 
                                               'POST')
    client.http_client.add_response(first_request, 302, None, 
        {'Location': 'http://example.com/1?gsessionid=12'})
    second_request = atom.http_core.HttpRequest(
        'http://example.com/1?gsessionid=12', 'POST')
    client.http_client.AddResponse(second_request, 200, 'OK', body='Done')

    response = client.Request('POST', 'http://example.com/1')
    self.assertEqual(response.status, 200)
    self.assertEqual(response.reason, 'OK')
    self.assertEqual(response.read(), 'Done')

    redirect_loop_request = atom.http_core.HttpRequest(
        'http://example.com/2?gsessionid=loop', 'PUT')
    client.http_client.add_response(redirect_loop_request, 302, None, 
        {'Location': 'http://example.com/2?gsessionid=loop'})
    try:
      response = client.request(method='PUT', uri='http://example.com/2?gsessionid=loop')
      self.fail('Loop URL should have redirected forever.')
    except gdata.client.RedirectError, err:
      self.assert_(str(err).startswith('Too many redirects from server'))

  def test_lowercase_location(self):
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.MockHttpClient()
    # Add the redirect response for the initial request.
    first_request = atom.http_core.HttpRequest('http://example.com/1', 
                                               'POST')
    # In some environments, notably App Engine, the HTTP headers which come
    # back from a server will be normalized to all lowercase.
    client.http_client.add_response(first_request, 302, None, 
        {'location': 'http://example.com/1?gsessionid=12'})
    second_request = atom.http_core.HttpRequest(
        'http://example.com/1?gsessionid=12', 'POST')
    client.http_client.AddResponse(second_request, 200, 'OK', body='Done')

    response = client.Request('POST', 'http://example.com/1')
    self.assertEqual(response.status, 200)
    self.assertEqual(response.reason, 'OK')
    self.assertEqual(response.read(), 'Done')

  def test_exercise_exceptions(self):
    # TODO
    pass

  def test_converter_vs_desired_class(self):

    def bad_converter(string):
      return 1
  
    class TestClass(atom.core.XmlElement):
      _qname = '{http://www.w3.org/2005/Atom}entry'
    
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.EchoHttpClient()
    test_entry = gdata.data.GDEntry()
    result = client.post(test_entry, 'http://example.com')
    self.assert_(isinstance(result, gdata.data.GDEntry))
    result = client.post(test_entry, 'http://example.com', converter=bad_converter)
    self.assertEquals(result, 1)
    result = client.post(test_entry, 'http://example.com', desired_class=TestClass)
    self.assert_(isinstance(result, TestClass))


class QueryTest(unittest.TestCase):

  def test_query_modifies_request(self):
    request = atom.http_core.HttpRequest()
    gdata.client.Query(
        text_query='foo', categories=['a', 'b']).modify_request(request)
    # categories param from above is named category in URL
    self.assertEqual(request.uri.query, {'q': 'foo', 'category': 'a,b'})

  def test_client_uses_query_modification(self):
    """If the Query is passed as an unexpected param it should apply"""
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.EchoHttpClient()
    query = gdata.client.Query(max_results=7)

    client.http_client = atom.mock_http_core.SettableHttpClient(
        201, 'CREATED', gdata.data.GDEntry().ToString(), {})
    response = client.get('https://example.com/foo', a_random_param=query)
    self.assertEqual(
        client.http_client.last_request.uri.query['max-results'], '7')


class VersionConversionTest(unittest.TestCase):

  def test_use_default_version(self):
    self.assertEquals(gdata.client.get_xml_version(None), 1)

  def test_str_to_int_version(self):
    self.assertEquals(gdata.client.get_xml_version('1'), 1)
    self.assertEquals(gdata.client.get_xml_version('2'), 2)
    self.assertEquals(gdata.client.get_xml_version('2.1.2'), 2)
    self.assertEquals(gdata.client.get_xml_version('10.4'), 10)


class UpdateTest(unittest.TestCase):
  """Test Update/PUT"""

  def test_update_uri_editlink(self):
    """Test that the PUT uri is grabbed from the entry's edit link"""
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(
        200, 'OK', gdata.data.GDEntry().ToString(), {})
    entry = gdata.data.GDEntry()
    entry.link.append(atom.data.Link(rel='edit', href='https://example.com/edit'))
    response = client.update(entry)
    request = client.http_client.last_request
    self.assertEqual(str(client.http_client.last_request.uri),
                     'https://example.com/edit')

  def test_update_uri(self):
    """Test that when passed, a uri overrides the entry's edit link"""
    client = gdata.client.GDClient()
    client.http_client = atom.mock_http_core.SettableHttpClient(
        200, 'OK', gdata.data.GDEntry().ToString(), {})
    entry = gdata.data.GDEntry()
    entry.link.append(atom.data.Link(rel='edit', href='https://example.com/edit'))
    response = client.update(entry, uri='https://example.com/test')
    self.assertEqual(str(client.http_client.last_request.uri),
                     'https://example.com/test')


def suite():
  return unittest.TestSuite((unittest.makeSuite(ClientLoginTest, 'test'),
                             unittest.makeSuite(AuthSubTest, 'test'),
                             unittest.makeSuite(OAuthTest, 'test'),
                             unittest.makeSuite(RequestTest, 'test'),
                             unittest.makeSuite(VersionConversionTest, 'test'),
                             unittest.makeSuite(QueryTest, 'test'),
                             unittest.makeSuite(UpdateTest, 'test')))


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