#!/usr/bin/python
#
# Copyright (C) 2008 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.jhartmann@gmail.com (Jochen Hartmann)'

import getpass
import time
import StringIO
import random
import unittest
import atom
import gdata.youtube
import gdata.youtube.service

YOUTUBE_TEST_CLIENT_ID = 'ytapi-pythonclientlibrary_servicetest'


class YouTubeServiceTest(unittest.TestCase):

  def setUp(self):
    self.client = gdata.youtube.service.YouTubeService()
    self.client.email = username
    self.client.password = password
    self.client.source = YOUTUBE_TEST_CLIENT_ID
    self.client.developer_key = developer_key
    self.client.client_id = YOUTUBE_TEST_CLIENT_ID
    self.client.ProgrammaticLogin()

  def testRetrieveVideoFeed(self):
    feed = self.client.GetYouTubeVideoFeed(
        'https://gdata.youtube.com/feeds/api/standardfeeds/recently_featured');
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)
    for entry in feed.entry:
      self.assert_(entry.title.text != '')

  def testRetrieveTopRatedVideoFeed(self):
    feed = self.client.GetTopRatedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveMostViewedVideoFeed(self):
    feed = self.client.GetMostViewedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveRecentlyFeaturedVideoFeed(self):
    feed = self.client.GetRecentlyFeaturedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveWatchOnMobileVideoFeed(self):
    feed = self.client.GetWatchOnMobileVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveTopFavoritesVideoFeed(self):
    feed = self.client.GetTopFavoritesVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveMostRecentVideoFeed(self):
    feed = self.client.GetMostRecentVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveMostDiscussedVideoFeed(self):
    feed = self.client.GetMostDiscussedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveMostLinkedVideoFeed(self):
    feed = self.client.GetMostLinkedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveMostRespondedVideoFeed(self):
    feed = self.client.GetMostRespondedVideoFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 10)

  def testRetrieveVideoEntryByUri(self):
    entry = self.client.GetYouTubeVideoEntry(
        'https://gdata.youtube.com/feeds/videos/Ncakifd_16k')
    self.assert_(isinstance(entry, gdata.youtube.YouTubeVideoEntry))
    self.assert_(entry.title.text != '')

  def testRetrieveVideoEntryByVideoId(self):
    entry = self.client.GetYouTubeVideoEntry(video_id='Ncakifd_16k')
    self.assert_(isinstance(entry, gdata.youtube.YouTubeVideoEntry))
    self.assert_(entry.title.text != '')

  def testRetrieveUserVideosbyUri(self):
    feed = self.client.GetYouTubeUserFeed(
        'https://gdata.youtube.com/feeds/users/gdpython/uploads')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveUserVideosbyUsername(self):
    feed = self.client.GetYouTubeUserFeed(username='gdpython')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testSearchWithVideoQuery(self):
    query = gdata.youtube.service.YouTubeVideoQuery()
    query.vq = 'google'
    query.max_results = 8
    feed = self.client.YouTubeQuery(query)
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assertEquals(len(feed.entry), 8)

  def testDirectVideoUploadStatusUpdateAndDeletion(self):
    self.assertEquals(self.client.developer_key, developer_key)
    self.assertEquals(self.client.client_id, YOUTUBE_TEST_CLIENT_ID)
    self.assertEquals(self.client.additional_headers['X-GData-Key'],
        'key=' + developer_key)
    self.assertEquals(self.client.additional_headers['X-Gdata-Client'],
        YOUTUBE_TEST_CLIENT_ID)

    test_video_title = 'my cool video ' + str(random.randint(1000,5000))
    test_video_description = 'description ' + str(random.randint(1000,5000))

    my_media_group = gdata.media.Group(
      title = gdata.media.Title(text=test_video_title),
      description = gdata.media.Description(description_type='plain',
                                            text=test_video_description),
      keywords = gdata.media.Keywords(text='video, foo'),
      category = gdata.media.Category(
          text='Autos',
          scheme='http://gdata.youtube.com/schemas/2007/categories.cat',
          label='Autos'),
      player=None
    )
    self.assert_(isinstance(my_media_group, gdata.media.Group))

    # Set Geo location to 37,-122 lat, long
    where = gdata.geo.Where()
    where.set_location((37.0,-122.0))
    
    video_entry = gdata.youtube.YouTubeVideoEntry(media=my_media_group,
                                                  geo=where)
    
    self.assert_(isinstance(video_entry, gdata.youtube.YouTubeVideoEntry))

    new_entry = self.client.InsertVideoEntry(video_entry, video_file_location)
    self.assert_(isinstance(new_entry, gdata.youtube.YouTubeVideoEntry))
    self.assertEquals(new_entry.title.text, test_video_title)
    self.assertEquals(new_entry.media.description.text, test_video_description)
    self.assert_(new_entry.id.text)

    # check upload status also
    upload_status = self.client.CheckUploadStatus(new_entry)
    self.assert_(upload_status[0] != '')

    # test updating entry meta-data
    new_video_description = 'description ' + str(random.randint(1000,5000))
    new_entry.media.description.text = new_video_description

    updated_entry = self.client.UpdateVideoEntry(new_entry)

    self.assert_(isinstance(updated_entry, gdata.youtube.YouTubeVideoEntry))
    self.assertEquals(updated_entry.media.description.text,
        new_video_description)

    # sleep for 10 seconds
    time.sleep(10)

    # test to delete the entry
    value = self.client.DeleteVideoEntry(updated_entry)

    if not value:
      # sleep more and try again
      time.sleep(20)
      # test to delete the entry
      value = self.client.DeleteVideoEntry(updated_entry)

    self.assert_(value == True)

  def testDirectVideoUploadWithDeveloperTags(self):
    self.assertEquals(self.client.developer_key, developer_key)
    self.assertEquals(self.client.client_id, YOUTUBE_TEST_CLIENT_ID)
    self.assertEquals(self.client.additional_headers['X-GData-Key'],
        'key=' + developer_key)
    self.assertEquals(self.client.additional_headers['X-Gdata-Client'],
        YOUTUBE_TEST_CLIENT_ID)

    test_video_title = 'my cool video ' + str(random.randint(1000,5000))
    test_video_description = 'description ' + str(random.randint(1000,5000))

    test_developer_tag_01 = 'tag' + str(random.randint(1000,5000))
    test_developer_tag_02 = 'tag' + str(random.randint(1000,5000))
    test_developer_tag_03 = 'tag' + str(random.randint(1000,5000)) 

    my_media_group = gdata.media.Group(
      title = gdata.media.Title(text=test_video_title),
      description = gdata.media.Description(description_type='plain',
                                            text=test_video_description),
      keywords = gdata.media.Keywords(text='video, foo'),
      category = [gdata.media.Category(
          text='Autos',
          scheme='http://gdata.youtube.com/schemas/2007/categories.cat',
          label='Autos')],
      player=None
    )

    self.assert_(isinstance(my_media_group, gdata.media.Group))

    video_entry = gdata.youtube.YouTubeVideoEntry(media=my_media_group)
    original_developer_tags = [test_developer_tag_01, test_developer_tag_02,
                               test_developer_tag_03]

    dev_tags = video_entry.AddDeveloperTags(original_developer_tags)

    for dev_tag in dev_tags:
      self.assert_(dev_tag.text in original_developer_tags) 

    self.assert_(isinstance(video_entry, gdata.youtube.YouTubeVideoEntry))

    new_entry = self.client.InsertVideoEntry(video_entry, video_file_location)

    self.assert_(isinstance(new_entry, gdata.youtube.YouTubeVideoEntry))
    self.assertEquals(new_entry.title.text, test_video_title)
    self.assertEquals(new_entry.media.description.text, test_video_description)
    self.assert_(new_entry.id.text)

    developer_tags_from_new_entry = new_entry.GetDeveloperTags()
    for dev_tag in developer_tags_from_new_entry:
      self.assert_(dev_tag.text in original_developer_tags) 

    self.assertEquals(len(developer_tags_from_new_entry),
        len(original_developer_tags))

    # sleep for 10 seconds
    time.sleep(10)

    # test to delete the entry
    value = self.client.DeleteVideoEntry(new_entry)

    if not value:
      # sleep more and try again
      time.sleep(20)
      # test to delete the entry
      value = self.client.DeleteVideoEntry(new_entry)

    self.assert_(value == True)

  def testBrowserBasedVideoUpload(self):
    self.assertEquals(self.client.developer_key, developer_key)
    self.assertEquals(self.client.client_id, YOUTUBE_TEST_CLIENT_ID)
    self.assertEquals(self.client.additional_headers['X-GData-Key'],
        'key=' + developer_key)
    self.assertEquals(self.client.additional_headers['X-Gdata-Client'],
        YOUTUBE_TEST_CLIENT_ID)
    test_video_title = 'my cool video ' + str(random.randint(1000,5000))
    test_video_description = 'description ' + str(random.randint(1000,5000))

    my_media_group = gdata.media.Group(
      title = gdata.media.Title(text=test_video_title),
      description = gdata.media.Description(description_type='plain',
                                            text=test_video_description),
      keywords = gdata.media.Keywords(text='video, foo'),
      category = gdata.media.Category(
          text='Autos',
          scheme='http://gdata.youtube.com/schemas/2007/categories.cat',
          label='Autos'),
      player=None
    )
    self.assert_(isinstance(my_media_group, gdata.media.Group))

    video_entry = gdata.youtube.YouTubeVideoEntry(media=my_media_group)
    self.assert_(isinstance(video_entry, gdata.youtube.YouTubeVideoEntry))

    response = self.client.GetFormUploadToken(video_entry)
    self.assert_(response[0].startswith(
        'https://uploads.gdata.youtube.com/action/FormDataUpload/'))
    self.assert_(len(response[0]) > 55)
    self.assert_(len(response[1]) > 100)

  def testRetrieveRelatedVideoFeedByUri(self):
    feed = self.client.GetYouTubeRelatedVideoFeed(
        'https://gdata.youtube.com/feeds/videos/Ncakifd_16k/related')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveRelatedVideoFeedById(self):
    feed = self.client.GetYouTubeRelatedVideoFeed(video_id = 'Ncakifd_16k')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveResponseVideoFeedByUri(self):
    feed = self.client.GetYouTubeVideoResponseFeed(
        'https://gdata.youtube.com/feeds/videos/Ncakifd_16k/responses')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoResponseFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveResponseVideoFeedById(self):
    feed = self.client.GetYouTubeVideoResponseFeed(video_id='Ncakifd_16k')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoResponseFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveVideoCommentFeedByUri(self):
    feed = self.client.GetYouTubeVideoCommentFeed(
        'https://gdata.youtube.com/feeds/api/videos/Ncakifd_16k/comments')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoCommentFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveVideoCommentFeedByVideoId(self):
    feed = self.client.GetYouTubeVideoCommentFeed(video_id='Ncakifd_16k')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoCommentFeed))
    self.assert_(len(feed.entry) > 0)

  def testAddComment(self):
    video_id = '9g6buYJTt_g'
    video_entry = self.client.GetYouTubeVideoEntry(video_id=video_id)
    random_comment_text = 'test_comment_' + str(random.randint(1000,50000))
    self.client.AddComment(comment_text=random_comment_text,
                           video_entry=video_entry)
    comment_feed = self.client.GetYouTubeVideoCommentFeed(video_id=video_id)
    comment_found = False
    for item in comment_feed.entry:
      if (item.content.text == random_comment_text):
        comment_found = True
    self.assertEquals(comment_found, True)

  def testAddRating(self):
    video_id_to_rate = 'Ncakifd_16k'
    video_entry = self.client.GetYouTubeVideoEntry(video_id=video_id_to_rate)
    response = self.client.AddRating(3, video_entry)
    self.assert_(isinstance(response, gdata.GDataEntry))

  def testRetrievePlaylistFeedByUri(self):
    feed = self.client.GetYouTubePlaylistFeed(
        'https://gdata.youtube.com/feeds/users/gdpython/playlists')
    self.assert_(isinstance(feed, gdata.youtube.YouTubePlaylistFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrievePlaylistListFeedByUsername(self):
    feed = self.client.GetYouTubePlaylistFeed(username='gdpython')
    self.assert_(isinstance(feed, gdata.youtube.YouTubePlaylistFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrievePlaylistVideoFeed(self):
    feed = self.client.GetYouTubePlaylistVideoFeed(
        'https://gdata.youtube.com/feeds/api/playlists/BCB3BB96DF51B505')
    self.assert_(isinstance(feed, gdata.youtube.YouTubePlaylistVideoFeed))
    self.assert_(len(feed.entry) > 0)
    self.assert_(isinstance(feed.entry[0],
        gdata.youtube.YouTubePlaylistVideoEntry))

  def testAddUpdateAndDeletePlaylist(self):
    test_playlist_title = 'my test playlist ' + str(random.randint(1000,3000))
    test_playlist_description = 'test playlist '
    response = self.client.AddPlaylist(test_playlist_title,
                                       test_playlist_description)
    self.assert_(isinstance(response, gdata.youtube.YouTubePlaylistEntry))

    new_playlist_title = 'my updated playlist ' + str(random.randint(1000,4000))
    new_playlist_description = 'my updated playlist '
    playlist_entry_id = response.id.text.split('/')[-1]

    updated_playlist = self.client.UpdatePlaylist(playlist_entry_id,
                                                  new_playlist_title,
                                                  new_playlist_description)

    playlist_feed = self.client.GetYouTubePlaylistFeed()

    update_successful = False

    for playlist_entry in playlist_feed.entry:
      if playlist_entry.title.text == new_playlist_title:
        update_successful = True
        break

    self.assertEquals(update_successful, True)

    # wait
    time.sleep(10)
    # delete it
    playlist_uri = updated_playlist.id.text
    response = self.client.DeletePlaylist(playlist_uri)
    self.assertEquals(response, True)

  def testAddUpdateAndDeletePrivatePlaylist(self):
    test_playlist_title = 'my test playlist ' + str(random.randint(1000,3000))
    test_playlist_description = 'test playlist '
    response = self.client.AddPlaylist(test_playlist_title,
                                       test_playlist_description,
                                       playlist_private=True)
    self.assert_(isinstance(response, gdata.youtube.YouTubePlaylistEntry))

    new_playlist_title = 'my updated playlist ' + str(random.randint(1000,4000))
    new_playlist_description = 'my updated playlist '
    playlist_entry_id = response.id.text.split('/')[-1]

    updated_playlist = self.client.UpdatePlaylist(playlist_entry_id,
                                                  new_playlist_title,
                                                  new_playlist_description,
                                                  playlist_private=True)

    playlist_feed = self.client.GetYouTubePlaylistFeed()

    update_successful = False
    playlist_still_private = False
    for playlist_entry in playlist_feed.entry:
      if playlist_entry.title.text == new_playlist_title:
        update_successful = True
        if playlist_entry.private is not None:
          playlist_still_private = True

    self.assertEquals(update_successful, True)
    self.assertEquals(playlist_still_private, True)

    # wait
    time.sleep(10)
    # delete it
    playlist_uri = updated_playlist.id.text
    response = self.client.DeletePlaylist(playlist_uri)
    self.assertEquals(response, True)

  def testAddEditAndDeleteVideoFromPlaylist(self):
    test_playlist_title = 'my test playlist ' + str(random.randint(1000,3000))
    test_playlist_description = 'test playlist '
    response = self.client.AddPlaylist(test_playlist_title,
                                       test_playlist_description)
    self.assert_(isinstance(response, gdata.youtube.YouTubePlaylistEntry))

    custom_video_title = 'my test video on my test playlist'
    custom_video_description = 'this is a test video on my test playlist'
    video_id = 'Ncakifd_16k'
    playlist_uri = response.feed_link[0].href
    time.sleep(10)
    response = self.client.AddPlaylistVideoEntryToPlaylist(
        playlist_uri, video_id, custom_video_title, custom_video_description)

    self.assert_(isinstance(response, gdata.youtube.YouTubePlaylistVideoEntry))

    playlist_entry_id = response.id.text.split('/')[-1]
    playlist_uri = response.id.text.split(playlist_entry_id)[0][:-1]
    new_video_title = 'video number ' + str(random.randint(1000,3000))
    new_video_description = 'test video'
    time.sleep(10)
    response = self.client.UpdatePlaylistVideoEntryMetaData(
        playlist_uri,
        playlist_entry_id,
        new_video_title,
        new_video_description,
        1)
    self.assert_(isinstance(response, gdata.youtube.YouTubePlaylistVideoEntry))
    time.sleep(10)

    playlist_entry_id = response.id.text.split('/')[-1]
    # remove video from playlist
    response = self.client.DeletePlaylistVideoEntry(playlist_uri,
                                                    playlist_entry_id)
    self.assertEquals(response, True)

    time.sleep(10)
    # delete the playlist
    response = self.client.DeletePlaylist(playlist_uri)
    self.assertEquals(response, True)

  def testRetrieveSubscriptionFeedByUri(self):
    feed = self.client.GetYouTubeSubscriptionFeed(
        'https://gdata.youtube.com/feeds/users/gdpython/subscriptions')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeSubscriptionFeed))
    self.assert_(len(feed.entry) == 3)

    subscription_to_channel_found = False
    subscription_to_favorites_found = False
    subscription_to_query_found = False
    all_types_found = False

    for entry in feed.entry:
      self.assert_(isinstance(entry, gdata.youtube.YouTubeSubscriptionEntry))
      subscription_type = entry.GetSubscriptionType()
      if subscription_type == 'channel':
        subscription_to_channel_found = True
      elif subscription_type == 'favorites':
        subscription_to_favorites_found = True
      elif subscription_type == 'query':
        subscription_to_query_found = True

    if (subscription_to_channel_found and subscription_to_favorites_found and
        subscription_to_query_found):
      all_types_found = True

    self.assertEquals(all_types_found, True)

  def testRetrieveSubscriptionFeedByUsername(self):
    feed = self.client.GetYouTubeSubscriptionFeed(username='gdpython')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeSubscriptionFeed))
    self.assert_(len(feed.entry) == 3)

    subscription_to_channel_found = False
    subscription_to_favorites_found = False
    subscription_to_query_found = False
    all_types_found = False

    for entry in feed.entry:
      self.assert_(isinstance(entry, gdata.youtube.YouTubeSubscriptionEntry))
      subscription_type = entry.GetSubscriptionType()
      if subscription_type == 'channel':
        subscription_to_channel_found = True
      elif subscription_type == 'favorites':
        subscription_to_favorites_found = True
      elif subscription_type == 'query':
        subscription_to_query_found = True

    if (subscription_to_channel_found and subscription_to_favorites_found and
        subscription_to_query_found):
      all_types_found = True
    self.assertEquals(all_types_found, True)

  def testRetrieveUserProfileByUri(self):
    user = self.client.GetYouTubeUserEntry(
        'https://gdata.youtube.com/feeds/users/gdpython')
    self.assert_(isinstance(user, gdata.youtube.YouTubeUserEntry))
    self.assertEquals(user.location.text, 'US')

  def testRetrieveUserProfileByUsername(self):
    user = self.client.GetYouTubeUserEntry(username='gdpython')
    self.assert_(isinstance(user, gdata.youtube.YouTubeUserEntry))
    self.assertEquals(user.location.text, 'US')

  def testRetrieveUserFavoritesFeed(self):
    feed = self.client.GetUserFavoritesFeed(username='gdpython')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testRetrieveDefaultUserFavoritesFeed(self):
    feed = self.client.GetUserFavoritesFeed()
    self.assert_(isinstance(feed, gdata.youtube.YouTubeVideoFeed))
    self.assert_(len(feed.entry) > 0)

  def testAddAndDeleteVideoFromFavorites(self):
    video_id = 'Ncakifd_16k'
    video_entry = self.client.GetYouTubeVideoEntry(video_id=video_id)
    response = self.client.AddVideoEntryToFavorites(video_entry)
    self.assert_(isinstance(response, gdata.GDataEntry))
    time.sleep(10)
    response = self.client.DeleteVideoEntryFromFavorites(video_id)
    self.assertEquals(response, True)

  def testRetrieveContactFeedByUri(self):
    feed = self.client.GetYouTubeContactFeed(
        'https://gdata.youtube.com/feeds/users/gdpython/contacts')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeContactFeed))
    self.assertEquals(len(feed.entry), 1)

  def testRetrieveContactFeedByUsername(self):
    feed = self.client.GetYouTubeContactFeed(username='gdpython')
    self.assert_(isinstance(feed, gdata.youtube.YouTubeContactFeed))
    self.assertEquals(len(feed.entry), 1)

if __name__ == '__main__':
  print ('NOTE: 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()
  developer_key = raw_input('Please enter your developer key: ')
  video_file_location = raw_input(
      'Please enter the absolute path to a video file: ')
  unittest.main()