gdata_test.py

#!/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__ = 'j.s@google.com (Jeff Scudder)'


import unittest
try:
  from xml.etree import ElementTree
except ImportError:
  from elementtree import ElementTree
import gdata
import atom
from gdata import test_data
import gdata.test_config as conf


class StartIndexTest(unittest.TestCase):
  
  def setUp(self):
    self.start_index = gdata.StartIndex()
    
  def testToAndFromString(self):
    self.start_index.text = '1'
    self.assert_(self.start_index.text == '1')
    new_start_index = gdata.StartIndexFromString(self.start_index.ToString())
    self.assert_(self.start_index.text == new_start_index.text)
    
    
class ItemsPerPageTest(unittest.TestCase):
  
  def setUp(self):
    self.items_per_page = gdata.ItemsPerPage()
    
  def testToAndFromString(self):
    self.items_per_page.text = '10'
    self.assert_(self.items_per_page.text == '10')
    new_items_per_page = gdata.ItemsPerPageFromString(
         self.items_per_page.ToString())
    self.assert_(self.items_per_page.text == new_items_per_page.text)


class GDataEntryTest(unittest.TestCase):

  def testIdShouldBeCleaned(self):
    entry = gdata.GDataEntryFromString(test_data.XML_ENTRY_1)
    element_tree = ElementTree.fromstring(test_data.XML_ENTRY_1)
    self.assert_(element_tree.findall(
        '{http://www.w3.org/2005/Atom}id')[0].text != entry.id.text)
    self.assert_(entry.id.text == 'http://www.google.com/test/id/url')
    
  def testGeneratorShouldBeCleaned(self):
    feed = gdata.GDataFeedFromString(test_data.GBASE_FEED)
    element_tree = ElementTree.fromstring(test_data.GBASE_FEED)
    self.assert_(element_tree.findall('{http://www.w3.org/2005/Atom}generator'
        )[0].text != feed.generator.text)
    self.assert_(feed.generator.text == 'GoogleBase')

  def testAllowsEmptyId(self):
    entry = gdata.GDataEntry()
    try:
      entry.id = atom.Id()
    except AttributeError:
      self.fail('Empty id should not raise an attribute error.')


class LinkFinderTest(unittest.TestCase):
  
  def setUp(self):
    self.entry = gdata.GDataEntryFromString(test_data.XML_ENTRY_1)

  def testLinkFinderGetsLicenseLink(self):
    self.assertEquals(isinstance(self.entry.GetLicenseLink(), atom.Link), 
                      True)
    self.assertEquals(self.entry.GetLicenseLink().href, 
        'http://creativecommons.org/licenses/by-nc/2.5/rdf')
    self.assertEquals(self.entry.GetLicenseLink().rel, 'license')

  def testLinkFinderGetsAlternateLink(self):
    self.assertEquals(isinstance(self.entry.GetAlternateLink(), atom.Link), 
                                 True)
    self.assertEquals(self.entry.GetAlternateLink().href, 
        'http://www.provider-host.com/123456789')
    self.assertEquals(self.entry.GetAlternateLink().rel, 'alternate')


class GDataFeedTest(unittest.TestCase):

  def testCorrectConversionToElementTree(self):
    test_feed = gdata.GDataFeedFromString(test_data.GBASE_FEED)
    self.assert_(test_feed.total_results is not None)
    element_tree = test_feed._ToElementTree()
    feed = element_tree.find('{http://www.w3.org/2005/Atom}feed')
    self.assert_(element_tree.find(
        '{http://a9.com/-/spec/opensearchrss/1.0/}totalResults') is not None)

  def testAllowsEmptyId(self):
    feed = gdata.GDataFeed()
    try:
      feed.id = atom.Id()
    except AttributeError:
      self.fail('Empty id should not raise an attribute error.')


class BatchEntryTest(unittest.TestCase):

  def testCorrectConversionFromAndToString(self):
    batch_entry = gdata.BatchEntryFromString(test_data.BATCH_ENTRY)
    
    self.assertEquals(batch_entry.batch_id.text, 'itemB')
    self.assertEquals(batch_entry.id.text, 
                      'http://www.google.com/base/feeds/items/'
                      '2173859253842813008')
    self.assertEquals(batch_entry.batch_operation.type, 'insert')
    self.assertEquals(batch_entry.batch_status.code, '201')
    self.assertEquals(batch_entry.batch_status.reason, 'Created')
    
    new_entry = gdata.BatchEntryFromString(str(batch_entry))

    self.assertEquals(batch_entry.batch_id.text, new_entry.batch_id.text)
    self.assertEquals(batch_entry.id.text, new_entry.id.text)
    self.assertEquals(batch_entry.batch_operation.type, 
                      new_entry.batch_operation.type)
    self.assertEquals(batch_entry.batch_status.code, 
                      new_entry.batch_status.code)
    self.assertEquals(batch_entry.batch_status.reason, 
                      new_entry.batch_status.reason)


class BatchFeedTest(unittest.TestCase):

  def setUp(self):
    self.batch_feed = gdata.BatchFeed()
    self.example_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/1'), text='This is a test')

  def testConvertRequestFeed(self):
    batch_feed = gdata.BatchFeedFromString(test_data.BATCH_FEED_REQUEST)

    self.assertEquals(len(batch_feed.entry), 4)
    for entry in batch_feed.entry:
      self.assert_(isinstance(entry, gdata.BatchEntry))
    self.assertEquals(batch_feed.title.text, 'My Batch Feed')

    new_feed = gdata.BatchFeedFromString(str(batch_feed))

    self.assertEquals(len(new_feed.entry), 4)
    for entry in new_feed.entry:
      self.assert_(isinstance(entry, gdata.BatchEntry))
    self.assertEquals(new_feed.title.text, 'My Batch Feed')

  def testConvertResultFeed(self):
    batch_feed = gdata.BatchFeedFromString(test_data.BATCH_FEED_RESULT)
    
    self.assertEquals(len(batch_feed.entry), 4)
    for entry in batch_feed.entry:
      self.assert_(isinstance(entry, gdata.BatchEntry))
      if entry.id.text == ('http://www.google.com/base/feeds/items/'
                           '2173859253842813008'):
        self.assertEquals(entry.batch_operation.type, 'insert')
        self.assertEquals(entry.batch_id.text, 'itemB')
        self.assertEquals(entry.batch_status.code, '201')
        self.assertEquals(entry.batch_status.reason, 'Created')
    self.assertEquals(batch_feed.title.text, 'My Batch')

    new_feed = gdata.BatchFeedFromString(str(batch_feed))
    
    self.assertEquals(len(new_feed.entry), 4)
    for entry in new_feed.entry:
      self.assert_(isinstance(entry, gdata.BatchEntry))
      if entry.id.text == ('http://www.google.com/base/feeds/items/'
                           '2173859253842813008'):
        self.assertEquals(entry.batch_operation.type, 'insert')
        self.assertEquals(entry.batch_id.text, 'itemB')
        self.assertEquals(entry.batch_status.code, '201')
        self.assertEquals(entry.batch_status.reason, 'Created')
    self.assertEquals(new_feed.title.text, 'My Batch')

  def testAddBatchEntry(self):
    try:
      self.batch_feed.AddBatchEntry(batch_id_string='a')
      self.fail('AddBatchEntry with neither entry or URL should raise Error')
    except gdata.MissingRequiredParameters:
      pass

    new_entry = self.batch_feed.AddBatchEntry(
        id_url_string='http://example.com/1')
    self.assertEquals(len(self.batch_feed.entry), 1)
    self.assertEquals(self.batch_feed.entry[0].id.text, 
                      'http://example.com/1')
    self.assertEquals(self.batch_feed.entry[0].batch_id.text, '0')
    self.assertEquals(new_entry.id.text, 'http://example.com/1')
    self.assertEquals(new_entry.batch_id.text, '0')

    to_add = gdata.BatchEntry(atom_id=atom.Id(text='originalId'))
    new_entry = self.batch_feed.AddBatchEntry(entry=to_add, 
                                              batch_id_string='foo')
    self.assertEquals(new_entry.batch_id.text, 'foo')
    self.assertEquals(new_entry.id.text, 'originalId')

    to_add = gdata.BatchEntry(atom_id=atom.Id(text='originalId'), 
                              batch_id=gdata.BatchId(text='bar'))
    new_entry = self.batch_feed.AddBatchEntry(entry=to_add, 
                                              id_url_string='newId',
                                              batch_id_string='foo')
    self.assertEquals(new_entry.batch_id.text, 'foo')
    self.assertEquals(new_entry.id.text, 'originalId')

    to_add = gdata.BatchEntry(atom_id=atom.Id(text='originalId'), 
                              batch_id=gdata.BatchId(text='bar'))
    new_entry = self.batch_feed.AddBatchEntry(entry=to_add, 
                                              id_url_string='newId')
    self.assertEquals(new_entry.batch_id.text, 'bar')
    self.assertEquals(new_entry.id.text, 'originalId')

    to_add = gdata.BatchEntry(atom_id=atom.Id(text='originalId'), 
                              batch_id=gdata.BatchId(text='bar'),
                              batch_operation=gdata.BatchOperation(
                                  op_type=gdata.BATCH_INSERT))
    self.assertEquals(to_add.batch_operation.type, gdata.BATCH_INSERT)
    new_entry = self.batch_feed.AddBatchEntry(entry=to_add, 
        id_url_string='newId', batch_id_string='foo', 
        operation_string=gdata.BATCH_UPDATE)
    self.assertEquals(new_entry.batch_operation.type, gdata.BATCH_UPDATE)


  def testAddInsert(self):
    
    first_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/1'), text='This is a test1')
    self.batch_feed.AddInsert(first_entry)
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type, 
                      gdata.BATCH_INSERT)
    self.assertEquals(self.batch_feed.entry[0].batch_id.text, '0')

    second_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/2'), text='This is a test2')
    self.batch_feed.AddInsert(second_entry, batch_id_string='foo')
    self.assertEquals(self.batch_feed.entry[1].batch_operation.type, 
                      gdata.BATCH_INSERT)
    self.assertEquals(self.batch_feed.entry[1].batch_id.text, 'foo')

    
    third_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/3'), text='This is a test3')
    third_entry.batch_operation = gdata.BatchOperation(
        op_type=gdata.BATCH_DELETE)
    # Add an entry with a delete operation already assigned.
    self.batch_feed.AddInsert(third_entry)
    # The batch entry should not have the original operation, it should 
    # have been changed to an insert.
    self.assertEquals(self.batch_feed.entry[2].batch_operation.type, 
                      gdata.BATCH_INSERT)
    self.assertEquals(self.batch_feed.entry[2].batch_id.text, '2')

  def testAddDelete(self):
    # Try deleting an entry
    delete_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/1'), text='This is a test')
    self.batch_feed.AddDelete(entry=delete_entry)
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type, 
                      gdata.BATCH_DELETE)
    self.assertEquals(self.batch_feed.entry[0].id.text, 
                      'http://example.com/1')
    self.assertEquals(self.batch_feed.entry[0].text, 'This is a test') 

    # Try deleting a URL
    self.batch_feed.AddDelete(url_string='http://example.com/2')
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type, 
                      gdata.BATCH_DELETE)
    self.assertEquals(self.batch_feed.entry[1].id.text, 
                      'http://example.com/2')
    self.assert_(self.batch_feed.entry[1].text is None) 

  def testAddQuery(self):
    # Try querying with an existing batch entry
    delete_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/1'))
    self.batch_feed.AddQuery(entry=delete_entry)
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type,
                      gdata.BATCH_QUERY)
    self.assertEquals(self.batch_feed.entry[0].id.text,
                      'http://example.com/1')

    # Try querying a URL
    self.batch_feed.AddQuery(url_string='http://example.com/2')
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type,
                      gdata.BATCH_QUERY)
    self.assertEquals(self.batch_feed.entry[1].id.text,
                      'http://example.com/2')

  def testAddUpdate(self):
    # Try updating an entry
    delete_entry = gdata.BatchEntry(
        atom_id=atom.Id(text='http://example.com/1'), text='This is a test')
    self.batch_feed.AddUpdate(entry=delete_entry)
    self.assertEquals(self.batch_feed.entry[0].batch_operation.type,
                      gdata.BATCH_UPDATE)
    self.assertEquals(self.batch_feed.entry[0].id.text,
                      'http://example.com/1')
    self.assertEquals(self.batch_feed.entry[0].text, 'This is a test')


class ExtendedPropertyTest(unittest.TestCase):
  
  def testXmlBlobRoundTrip(self):
    ep = gdata.ExtendedProperty(name='blobby')
    ep.SetXmlBlob('<some_xml attr="test"/>')
    extension = ep.GetXmlBlobExtensionElement()
    self.assertEquals(extension.tag, 'some_xml')
    self.assert_(extension.namespace is None)
    self.assertEquals(extension.attributes['attr'], 'test')

    ep2 = gdata.ExtendedPropertyFromString(ep.ToString())

    extension = ep2.GetXmlBlobExtensionElement()
    self.assertEquals(extension.tag, 'some_xml')
    self.assert_(extension.namespace is None)
    self.assertEquals(extension.attributes['attr'], 'test')

  def testGettersShouldReturnNoneWithNoBlob(self):
    ep = gdata.ExtendedProperty(name='no blob')
    self.assert_(ep.GetXmlBlobExtensionElement() is None)
    self.assert_(ep.GetXmlBlobString() is None)

  def testGettersReturnCorrectTypes(self):
    ep = gdata.ExtendedProperty(name='has blob')
    ep.SetXmlBlob('<some_xml attr="test"/>')
    self.assert_(isinstance(ep.GetXmlBlobExtensionElement(), 
        atom.ExtensionElement))
    self.assert_(isinstance(ep.GetXmlBlobString(), str))


class FeedLinkTest(unittest.TestCase):
  
  def testCorrectFromStringType(self):
    link = gdata.FeedLinkFromString(
        '<feedLink xmlns="http://schemas.google.com/g/2005" countHint="5"/>')
    self.assert_(isinstance(link, gdata.FeedLink))
    self.assertEqual(link.count_hint, '5')


def suite():
  return conf.build_suite([StartIndexTest, StartIndexTest, GDataEntryTest,
      LinkFinderTest, GDataFeedTest, BatchEntryTest, BatchFeedTest,
      ExtendedPropertyTest, FeedLinkTest])


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