#!/usr/bin/python
#
# Copyright (C) 2006 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.

__author__ = 'api.jscudder (Jeff Scudder)'

import unittest
import getpass
try:
  from xml.etree import ElementTree
except ImportError:
  from elementtree import ElementTree
import gdata.service
import gdata
import gdata.auth
import atom
import atom.service
import atom.token_store
import os.path
from gdata import test_data
import atom.mock_http
import atom.mock_http_core


username = ''
password = ''
test_image_location = '../testimage.jpg'
test_image_name = 'testimage.jpg'


class GDataServiceMediaUnitTest(unittest.TestCase):

  def setUp(self):
    self.gd_client = gdata.service.GDataService()
    self.gd_client.email = username
    self.gd_client.password = password
    self.gd_client.service = 'lh2'
    self.gd_client.source = 'GDataService Media "Unit" Tests'
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    
    # create a test album
    gd_entry = gdata.GDataEntry()
    gd_entry.title = atom.Title(text='GData Test Album')
    gd_entry.category.append(atom.Category(
        scheme='http://schemas.google.com/g/2005#kind',
        term='http://schemas.google.com/photos/2007#album'))
    
    self.album_entry = self.gd_client.Post(gd_entry, 
        'http://picasaweb.google.com/data/feed/api/user/' + username)
    
  def tearDown(self):
    album_entry = self.gd_client.Get(self.album_entry.id.text)
    self.gd_client.Delete(album_entry.GetEditLink().href)

  def testSourceGeneratesUserAgentHeader(self):
    self.gd_client.source = 'GoogleInc-ServiceUnitTest-1'
    self.assert_(self.gd_client.additional_headers['User-Agent'].startswith(
        'GoogleInc-ServiceUnitTest-1 GData-Python'))
    
  def testMedia1(self):
    # Create media-only
    ms = gdata.MediaSource()
    ms.setFile(test_image_location, 'image/jpeg')
    media_entry = self.gd_client.Post(None, 
        self.album_entry.GetFeedLink().href, media_source = ms)
    self.assert_(media_entry is not None)
    self.assert_(isinstance(media_entry, gdata.GDataEntry))
    self.assert_(media_entry.IsMedia())
    
    # Update media & metadata
    ms = gdata.MediaSource()
    ms.setFile(test_image_location, 'image/jpeg')
    media_entry.summary = atom.Summary(text='Test Image')
    media_entry2 = self.gd_client.Put(media_entry,
        media_entry.GetEditLink().href, media_source = ms)
    self.assert_(media_entry2 is not None)
    self.assert_(isinstance(media_entry2, gdata.GDataEntry))
    self.assert_(media_entry2.IsMedia())
    self.assert_(media_entry2.summary.text == 'Test Image')
    
    # Read media binary
    imageSource = self.gd_client.GetMedia(media_entry2.GetMediaURL())
    self.assert_(isinstance(imageSource, gdata.MediaSource))
    self.assert_(imageSource.content_type == 'image/jpeg')
    self.assert_(imageSource.content_length)
    
    imageData = imageSource.file_handle.read()
    self.assert_(imageData)
    
    # Delete entry
    response = self.gd_client.Delete(media_entry2.GetEditLink().href)
    self.assert_(response)
    
  def testMedia2(self):
    # Create media & metadata
    ms = gdata.MediaSource()
    ms.setFile(test_image_location, 'image/jpeg')
    new_media_entry = gdata.GDataEntry()
    new_media_entry.title = atom.Title(text='testimage1.jpg')
    new_media_entry.summary = atom.Summary(text='Test Image')
    new_media_entry.category.append(atom.Category(scheme = 
        'http://schemas.google.com/g/2005#kind', term = 
        'http://schemas.google.com/photos/2007#photo'))
    media_entry = self.gd_client.Post(new_media_entry,
        self.album_entry.GetFeedLink().href, media_source = ms)
    self.assert_(media_entry is not None)
    self.assert_(isinstance(media_entry, gdata.GDataEntry))
    self.assert_(media_entry.IsMedia())
    self.assert_(media_entry.summary.text == 'Test Image')
    
    # Update media only
    ms = gdata.MediaSource()
    ms.setFile(test_image_location, 'image/jpeg')
    media_entry = self.gd_client.Put(None, media_entry.GetEditMediaLink().href,
        media_source = ms)
    self.assert_(media_entry is not None)
    self.assert_(isinstance(media_entry, gdata.GDataEntry))
    self.assert_(media_entry.IsMedia())
    
    # Delete entry
    response = self.gd_client.Delete(media_entry.GetEditLink().href)
    self.assert_(response)

  def testMediaConstructorDefaults(self):

    ms = gdata.MediaSource()
    ms.setFile(test_image_location, 'image/jpeg')
    
    self.assert_(ms is not None)
    self.assert_(isinstance(ms, gdata.MediaSource))
    self.assertEquals(ms.file_name, test_image_name)
    self.assertEquals(ms.content_type, 'image/jpeg')
  
  def testMediaConstructorWithFilePath(self):

    ms = gdata.MediaSource(file_path=test_image_location,
                           content_type='image/jpeg')
    
    self.assert_(ms is not None)
    self.assert_(isinstance(ms, gdata.MediaSource))
    self.assertEquals(ms.file_name, test_image_name)
    self.assertEquals(ms.content_type, 'image/jpeg')
   
  def testMediaConstructorWithFileHandle(self):

    fh = open(test_image_location, 'r')
    len = os.path.getsize(test_image_location)
    ms = gdata.MediaSource(fh, 'image/jpeg', len, file_name=test_image_location)
    
    self.assert_(ms is not None)
    self.assert_(isinstance(ms, gdata.MediaSource))
    self.assertEquals(ms.file_name, test_image_location)
    self.assertEquals(ms.content_type, 'image/jpeg')


class GDataServiceUnitTest(unittest.TestCase):
  
  def setUp(self):
    self.gd_client = gdata.service.GDataService()
    self.gd_client.email = username
    self.gd_client.password = password
    self.gd_client.service = 'gbase'
    self.gd_client.source = 'GDataClient "Unit" Tests'

  def testProperties(self):
    email_string = 'Test Email'
    password_string = 'Passwd'

    self.gd_client.email = email_string
    self.assertEquals(self.gd_client.email, email_string)
    self.gd_client.password = password_string
    self.assertEquals(self.gd_client.password, password_string)

  def testCorrectLogin(self):
    try:
      self.gd_client.ProgrammaticLogin()
      self.assert_(isinstance(
          self.gd_client.token_store.find_token(
              'http://base.google.com/base/feeds/'),
          gdata.auth.ClientLoginToken))
      self.assert_(self.gd_client.captcha_token is None)
      self.assert_(self.gd_client.captcha_url is None)
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')

  def testDefaultHttpClient(self):
    self.assert_(isinstance(self.gd_client.http_client, 
                            atom.http.HttpClient))


  def testGet(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.additional_headers = {'X-Google-Key': 
                                               'ABQIAAAAoLioN3buSs9KqIIq9V' +
                                               'mkFxT2yXp_ZAY8_ufC3CFXhHIE' +
                                               '1NvwkxRK8C1Q8OWhsWA2AIKv-c' +
                                               'VKlVrNhQ'}
    self.gd_client.server = 'base.google.com'
    result = self.gd_client.Get('/base/feeds/snippets?bq=digital+camera')
    self.assert_(result is not None)
    self.assert_(isinstance(result, atom.Feed))

  def testGetWithAuthentication(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.additional_headers = {'X-Google-Key':
                                               'ABQIAAAAoLioN3buSs9KqIIq9V' +
                                               'mkFxT2yXp_ZAY8_ufC3CFXhHIE' +
                                               '1NvwkxRK8C1Q8OWhsWA2AIKv-c' +
                                               'VKlVrNhQ'}
    self.gd_client.server = 'base.google.com'
    result = self.gd_client.Get('/base/feeds/items?bq=digital+camera')
    self.assert_(result is not None)
    self.assert_(isinstance(result, atom.Feed))

  def testGetEntry(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.server = 'base.google.com'
    try:
      result = self.gd_client.GetEntry('/base/feeds/items?bq=digital+camera')
      self.fail(
          'Result from server in GetEntry should have raised an exception')
    except gdata.service.UnexpectedReturnType:
      pass

  def testGetFeed(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.server = 'base.google.com'
    result = self.gd_client.GetFeed('/base/feeds/items?bq=digital+camera')
    self.assert_(result is not None)
    self.assert_(isinstance(result, atom.Feed))

  def testGetWithResponseTransformer(self):
    # Query Google Base and interpret the results as a GBaseSnippetFeed.
    feed = self.gd_client.Get(
        'http://www.google.com/base/feeds/snippets?bq=digital+camera',
        converter=gdata.base.GBaseSnippetFeedFromString)
    self.assertEquals(isinstance(feed, gdata.base.GBaseSnippetFeed), True)

  def testPostPutAndDelete(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.additional_headers = {'X-Google-Key':
                                               'ABQIAAAAoLioN3buSs9KqIIq9V' +
                                               'mkFxT2yXp_ZAY8_ufC3CFXhHIE' +
                                               '1NvwkxRK8C1Q8OWhsWA2AIKv-c' +
                                               'VKlVrNhQ'}
    self.gd_client.server = 'base.google.com'

    # Insert a new item
    response = self.gd_client.Post(test_data.TEST_BASE_ENTRY, 
        '/base/feeds/items')
    self.assert_(response is not None)
    self.assert_(isinstance(response, atom.Entry))
    self.assert_(response.category[0].term == 'products')

    # Find the item id of the created item
    item_id = response.id.text.lstrip(
        'http://www.google.com/base/feeds/items/')
    self.assert_(item_id is not None)
    
    updated_xml = gdata.base.GBaseItemFromString(test_data.TEST_BASE_ENTRY)
    # Change one of the labels in the item
    updated_xml.label[2].text = 'beach ball'
    # Update the item
    response = self.gd_client.Put(updated_xml, 
        '/base/feeds/items/%s' % item_id)
    self.assert_(response is not None)
    new_base_item = gdata.base.GBaseItemFromString(str(response))
    self.assert_(isinstance(new_base_item, atom.Entry))
    
    # Delete the item the test just created.
    response = self.gd_client.Delete('/base/feeds/items/%s' % item_id)
    self.assert_(response)

  def testPostPutAndDeleteWithConverters(self):
    try:
      self.gd_client.ProgrammaticLogin()
    except gdata.service.CaptchaRequired:
      self.fail('Required Captcha')
    except gdata.service.BadAuthentication:
      self.fail('Bad Authentication')
    except gdata.service.Error:
      self.fail('Login Error')
    self.gd_client.additional_headers = {'X-Google-Key':
                                               'ABQIAAAAoLioN3buSs9KqIIq9V' +
                                               'mkFxT2yXp_ZAY8_ufC3CFXhHIE' +
                                               '1NvwkxRK8C1Q8OWhsWA2AIKv-c' +
                                               'VKlVrNhQ'}
    self.gd_client.server = 'base.google.com'

    # Insert a new item
    response = self.gd_client.Post(test_data.TEST_BASE_ENTRY,
        '/base/feeds/items', converter=gdata.base.GBaseItemFromString)
    self.assert_(response is not None)
    self.assert_(isinstance(response, atom.Entry))
    self.assert_(isinstance(response, gdata.base.GBaseItem))
    self.assert_(response.category[0].term == 'products')

    updated_xml = gdata.base.GBaseItemFromString(test_data.TEST_BASE_ENTRY)
    # Change one of the labels in the item
    updated_xml.label[2].text = 'beach ball'
    # Update the item
    response = self.gd_client.Put(updated_xml,
        response.id.text,
        converter=gdata.base.GBaseItemFromString)
    self.assertEquals(response is not None, True)
    self.assertEquals(isinstance(response, gdata.base.GBaseItem), True)

    # Delete the item the test just created.
    response = self.gd_client.Delete(response.id.text)
    self.assert_(response)

  def testCaptchaUrlGeneration(self):
    # Populate the mock server with a pairing for a ClientLogin request to a
    # CAPTCHA challenge.
    mock_client = atom.mock_http.MockHttpClient()
    captcha_response = atom.mock_http.MockResponse(
        body="""Url=http://www.google.com/login/captcha
Error=CaptchaRequired
CaptchaToken=DQAAAGgAdkI1LK9
CaptchaUrl=Captcha?ctoken=HiteT4b0Bk5Xg18_AcVoP6-yFkHPibe7O9EqxeiI7lUSN
""", status=403, reason='Access Forbidden')
    mock_client.add_response(captcha_response, 'POST', 
        'https://www.google.com/accounts/ClientLogin')

    # Set the exising client's handler so that it will make requests to the
    # mock service instead of the real server.
    self.gd_client.http_client = mock_client

    try:
      self.gd_client.ProgrammaticLogin()
      self.fail('Login attempt should have caused a CAPTCHA challenge.')
    except gdata.service.CaptchaRequired, error:
      self.assertEquals(self.gd_client.captcha_url, 
          ('https://www.google.com/accounts/Captcha?ctoken=HiteT4b0Bk5Xg18_'
           'AcVoP6-yFkHPibe7O9EqxeiI7lUSN'))


class DeleteWithUrlParamsTest(unittest.TestCase):

  def setUp(self):
    self.gd_client = gdata.service.GDataService()
    # Set the client to echo the request back in the response.
    self.gd_client.http_client.v2_http_client = (
        atom.mock_http_core.SettableHttpClient(200, 'OK', '', {}))
    
  def testDeleteWithUrlParams(self):
    self.assert_(self.gd_client.Delete('http://example.com/test', 
        {'TestHeader': '123'}, {'urlParam1': 'a', 'urlParam2': 'test'}))
    request = self.gd_client.http_client.v2_http_client.last_request
    self.assertEqual(request.uri.host, 'example.com')
    self.assertEqual(request.uri.path, '/test')
    self.assertEqual(request.uri.query, 
        {'urlParam1': 'a', 'urlParam2': 'test'})

  def testDeleteWithSessionId(self):
    self.gd_client._SetSessionId('test_session_id')
    self.assert_(self.gd_client.Delete('http://example.com/test', 
        {'TestHeader': '123'}, {'urlParam1': 'a', 'urlParam2': 'test'}))
    request = self.gd_client.http_client.v2_http_client.last_request
    self.assertEqual(request.uri.host, 'example.com')
    self.assertEqual(request.uri.path, '/test')
    self.assertEqual(request.uri.query, {'urlParam1': 'a', 
        'urlParam2': 'test', 'gsessionid': 'test_session_id'})
      

class QueryTest(unittest.TestCase):

  def setUp(self):
    self.query = gdata.service.Query()

  def testQueryShouldBehaveLikeDict(self):
    try:
      self.query['zap']
      self.fail()
    except KeyError:
      pass
    self.query['zap'] = 'x'
    self.assert_(self.query['zap'] == 'x')

  def testContructorShouldRejectBadInputs(self):
    test_q = gdata.service.Query(params=[1,2,3,4])
    self.assert_(len(test_q.keys()) == 0)

  def testTextQueryProperty(self):
    self.assert_(self.query.text_query is None)
    self.query['q'] = 'test1'
    self.assert_(self.query.text_query == 'test1')
    self.query.text_query = 'test2'
    self.assert_(self.query.text_query == 'test2')

  def testOrderByQueryProperty(self): 
    self.assert_(self.query.orderby is None) 
    self.query['orderby'] = 'updated' 
    self.assert_(self.query.orderby == 'updated') 
    self.query.orderby = 'starttime' 
    self.assert_(self.query.orderby == 'starttime') 

  def testQueryShouldProduceExampleUris(self):
    self.query.feed = '/base/feeds/snippets'
    self.query.text_query = 'This is a test'
    self.assert_(self.query.ToUri() == '/base/feeds/snippets?q=This+is+a+test')

  def testCategoriesFormattedCorrectly(self):
    self.query.feed = '/x'
    self.query.categories.append('Fritz')
    self.query.categories.append('Laurie')
    self.assert_(self.query.ToUri() == '/x/-/Fritz/Laurie')
    # The query's feed should not have been changed
    self.assert_(self.query.feed == '/x')
    self.assert_(self.query.ToUri() == '/x/-/Fritz/Laurie')

  def testCategoryQueriesShouldEscapeOrSymbols(self):
    self.query.feed = '/x'
    self.query.categories.append('Fritz|Laurie')
    self.assert_(self.query.ToUri() == '/x/-/Fritz%7CLaurie')

  def testTypeCoercionOnIntParams(self):
    self.query.feed = '/x'
    self.query.max_results = 10
    self.query.start_index = 5
    self.assert_(isinstance(self.query.max_results, str))
    self.assert_(isinstance(self.query.start_index, str))
    self.assertEquals(self.query['max-results'], '10')
    self.assertEquals(self.query['start-index'], '5')

  def testPassInCategoryListToConstructor(self):
    query = gdata.service.Query(feed='/feed/sample', categories=['foo', 'bar',
            'eggs|spam'])
    url = query.ToUri()
    self.assert_(url.find('/foo') > -1)
    self.assert_(url.find('/bar') > -1)
    self.assert_(url.find('/eggs%7Cspam') > -1)


class GetNextPageInFeedTest(unittest.TestCase):

  def setUp(self):
    self.gd_client = gdata.service.GDataService()

  def testGetNextPage(self):
    feed = self.gd_client.Get(
        'http://www.google.com/base/feeds/snippets?max-results=2',
        converter=gdata.base.GBaseSnippetFeedFromString)
    self.assert_(len(feed.entry) > 0)
    first_id = feed.entry[0].id.text
    feed2 = self.gd_client.GetNext(feed)
    self.assert_(len(feed2.entry) > 0)
    next_id = feed2.entry[0].id.text
    self.assert_(first_id != next_id)
    self.assert_(feed2.__class__ == feed.__class__)


class ScopeLookupTest(unittest.TestCase):

  def testLookupScopes(self):
    scopes = gdata.service.lookup_scopes('cl')
    self.assertEquals(scopes, gdata.service.CLIENT_LOGIN_SCOPES['cl'])
    scopes = gdata.service.lookup_scopes(None)
    self.assert_(scopes is None)
    scopes = gdata.service.lookup_scopes('UNKNOWN_SERVICE')
    self.assert_(scopes is None)


class TokenLookupTest(unittest.TestCase):

  def setUp(self):
    self.client = gdata.service.GDataService()

  def testSetAndGetClientLoginTokenWithNoService(self):
    self.assert_(self.client.auth_token is None)
    self.client.SetClientLoginToken('foo')
    self.assert_(self.client.auth_token is None)
    self.assert_(self.client.token_store.find_token(
        atom.token_store.SCOPE_ALL) is not None)
    self.assertEquals(self.client.GetClientLoginToken(), 'foo')
    self.client.SetClientLoginToken('foo2')
    self.assertEquals(self.client.GetClientLoginToken(), 'foo2')

  def testSetAndGetClientLoginTokenWithService(self):
    self.client.service = 'cp'
    self.client.SetClientLoginToken('bar')
    self.assertEquals(self.client.GetClientLoginToken(), 'bar')
    # Changing the service should cause the token to no longer be found.
    self.client.service = 'gbase'
    self.client.current_token = None
    self.assert_(self.client.GetClientLoginToken() is None)

  def testSetAndGetClientLoginTokenWithScopes(self):
    scopes = gdata.service.CLIENT_LOGIN_SCOPES['cl'][:]
    scopes.extend(gdata.service.CLIENT_LOGIN_SCOPES['gbase'])
    self.client.SetClientLoginToken('baz', scopes=scopes)
    self.client.current_token = None
    self.assert_(self.client.GetClientLoginToken() is None)
    self.client.service = 'cl'
    self.assertEquals(self.client.GetClientLoginToken(), 'baz')
    self.client.service = 'gbase'
    self.assertEquals(self.client.GetClientLoginToken(), 'baz')
    self.client.service = 'wise'
    self.assert_(self.client.GetClientLoginToken() is None)

  def testLookupUsingTokenStore(self):
    scopes = gdata.service.CLIENT_LOGIN_SCOPES['cl'][:]
    scopes.extend(gdata.service.CLIENT_LOGIN_SCOPES['gbase'])
    self.client.SetClientLoginToken('baz', scopes=scopes)
    token = self.client.token_store.find_token(
        'http://www.google.com/calendar/feeds/foo')
    self.assertEquals(token.get_token_string(), 'baz')
    self.assertEquals(token.auth_header, '%s%s' % (
        gdata.auth.PROGRAMMATIC_AUTH_LABEL, 'baz'))
    token = self.client.token_store.find_token(
        'http://www.google.com/calendar/')
    self.assert_(isinstance(token, gdata.auth.ClientLoginToken) == False)
    token = self.client.token_store.find_token(
        'http://www.google.com/base/feeds/snippets')
    self.assertEquals(token.get_token_string(), 'baz')


if __name__ == '__main__':
  print ('GData Service Media Unit Tests\nNOTE: Please run these tests only '
         'with a test  account. The tests may delete or update your data.')
  username = raw_input('Please enter your username: ')
  password = getpass.getpass()
  unittest.main()