#!/usr/bin/python # # Copyright 2011 Google Inc. All Rights Reserved. # # 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. # These tests attempt to connect to Google servers. """Live integration tests of the Google Documents List API. RESOURCES: Dict of test resource data, keyed on resource type. """ __author__ = 'vicfryzel@google.com (Vic Fryzel)' import os import os.path import tempfile import time import unittest import gdata.client import gdata.data import gdata.gauth import gdata.docs.client import gdata.docs.data import gdata.test_config as conf RESOURCES = { 'document': (gdata.docs.data.DOCUMENT_LABEL, 'Text Document', 'data/test.0.doc', 'data/test.1.doc', 'application/msword', 'doc'), 'empty_document': (gdata.docs.data.DOCUMENT_LABEL, 'Empty Text Document', None, 'data/test.1.doc', 'application/msword', 'txt'), 'spreadsheet': (gdata.docs.data.SPREADSHEET_LABEL, 'Spreadsheet', 'data/test.0.csv', 'data/test.1.csv', 'text/csv', 'csv'), 'presentation': (gdata.docs.data.PRESENTATION_LABEL, 'Presentation', 'data/test.0.ppt', 'data/test.1.ppt', 'application/vnd.ms-powerpoint', 'ppt'), 'drawing': (gdata.docs.data.DRAWING_LABEL, 'Drawing', 'data/test.0.wmf', 'data/test.1.wmf', 'application/x-msmetafile', 'png'), 'pdf': (gdata.docs.data.PDF_LABEL, 'PDF', 'data/test.0.pdf', 'data/test.1.pdf', 'application/pdf', None), 'file': (gdata.docs.data.FILE_LABEL, 'File', 'data/test.0.bin', 'data/test.1.bin', 'application/octet-stream', None), 'collection': (gdata.docs.data.COLLECTION_LABEL, 'Collection A', None, None, None, None) } class DocsTestCase(unittest.TestCase): def shortDescription(self): if hasattr(self, 'resource_type'): return '%s for %s' % (self.__class__.__name__, self.resource_type) else: return self.__class__.__name__ def _delete(self, resource): try: self.client.DeleteResource(resource, permanent=True, force=True) except: pass def _delete_all(self): resources = self.client.GetAllResources( '/feeds/default/private/full?showfolders=true&showdeleted=true') for resource in resources: self._delete(resource) def _create(self): ms = None if self.resource_path is not None and self.resource_mime is not None: ms = gdata.data.MediaSource( file_path=os.path.join(os.path.dirname(__file__), self.resource_path), content_type=self.resource_mime) entry = gdata.docs.data.Resource(type=self.resource_label, title=self.resource_title) self.resource = self.client.CreateResource(entry, media=ms) def _update(self): ms = None if self.resource_alt_path is not None and self.resource_mime is not None: ms = gdata.data.MediaSource( file_path=os.path.join(os.path.dirname(__file__), self.resource_alt_path), content_type=self.resource_mime) self.resource.title.text = '%s Updated' % self.resource_title return self.client.UpdateResource(self.resource, media=ms, force=True, new_revision=True) def setUp(self): if conf.options.get_value('runlive') != 'true': raise RuntimeError('Live tests require --runlive true') self.client = gdata.docs.client.DocsClient() if conf.options.get_value('ssl') == 'false': self.client.ssl = False conf.configure_client(self.client, 'DocsTest', self.client.auth_service) conf.configure_cache(self.client, str(self.__class__)) if conf.options.get_value('clean') == 'true': self._delete_all() tries = 0 while tries < 3: try: tries += 1 self._create() break except gdata.client.RequestError: if tries >= 2: self.tearDown() raise def tearDown(self): if conf.options.get_value('runlive') == 'true': if conf.options.get_value('clean') == 'true': self._delete_all() else: try: self._delete(self.resource) except: pass conf.close_client(self.client) class ResourcesTest(DocsTestCase): def testGetAllResources(self): results = self.client.GetAllResources( '/feeds/default/private/full?showfolders=true') self.assert_(all(isinstance(item, gdata.docs.data.Resource) \ for item in results)) self.assertEqual(len(results), 1) def testGetResources(self): feed = self.client.GetResources( '/feeds/default/private/full?showfolders=true', limit=1) self.assert_(isinstance(feed, gdata.docs.data.ResourceFeed)) self.assertEqual(len(feed.entry), 1) def testGetResource(self): entry = self.client.GetResource(self.resource) self.assert_(isinstance(entry, gdata.docs.data.Resource)) self.assert_(entry.id.text is not None) self.assert_(entry.title.text is not None) self.assert_(entry.resource_id.text is not None) self.assert_(entry.title.text is not None) entry = self.client.GetResourceById(self.resource.resource_id.text) self.assert_(isinstance(entry, gdata.docs.data.Resource)) self.assert_(entry.id.text is not None) self.assert_(entry.title.text is not None) self.assert_(entry.resource_id.text is not None) self.assert_(entry.title.text is not None) entry = self.client.GetResourceById( self.resource.resource_id.text.split(':')[1]) self.assert_(isinstance(entry, gdata.docs.data.Resource)) self.assert_(entry.id.text is not None) self.assert_(entry.title.text is not None) self.assert_(entry.resource_id.text is not None) self.assert_(entry.title.text is not None) entry = self.client.GetResourceBySelfLink( self.resource.GetSelfLink().href) self.assert_(isinstance(entry, gdata.docs.data.Resource)) self.assert_(entry.id.text is not None) self.assert_(entry.title.text is not None) self.assert_(entry.resource_id.text is not None) self.assert_(entry.title.text is not None) def testMoveResource(self): entry = gdata.docs.data.Resource( type=gdata.docs.data.COLLECTION_LABEL, title='Collection B') collection = self.client.CreateResource(entry) # Start off in 0 collections self.assertEqual(len(self.resource.InCollections()), 0) # Move resource into collection entry = self.client.MoveResource(self.resource, collection) self.assertEqual(len(entry.InCollections()), 1) self.assertEqual(entry.InCollections()[0].title, collection.title.text) self.client.DeleteResource(collection, permanent=True, force=True) def testCopyResource(self): copy_title = '%s Copy' % self.resource_title # Copy only supported for document, spreadsheet, presentation types if self.resource_type in ['document', 'empty_document', 'spreadsheet', 'presentation']: copy = self.client.CopyResource(self.resource, copy_title) self.assertEqual(copy.title.text, copy_title) self.client.DeleteResource(copy, permanent=True, force=True) # TODO(vicfryzel): Expect appropriate error for drawings. elif self.resource_type != 'drawing': self.assertRaises(gdata.client.NotImplemented, self.client.CopyResource, self.resource, copy_title) def testDownloadResource(self): tmp = tempfile.mkstemp() if self.resource_type != 'collection': if self.resource_export is not None: extra_params = {'exportFormat': self.resource_export, 'format': self.resource_export} self.client.DownloadResource(self.resource, tmp[1], extra_params=extra_params) else: self.client.DownloadResource(self.resource, tmp[1]) else: # Cannot download collections self.assertRaises(ValueError, self.client.DownloadResource, self.resource, tmp[1]) # Should get a 404 entry = gdata.docs.data.Resource(type=gdata.docs.data.DOCUMENT_LABEL, title='Does Not Exist') self.assertRaises(AttributeError, self.client.DownloadResource, entry, tmp[1]) os.close(tmp[0]) os.remove(tmp[1]) def testDownloadResourceToMemory(self): if self.resource_type != 'collection': data = None if self.resource_export is not None: extra_params = {'exportFormat': self.resource_export, 'format': self.resource_export} data = self.client.DownloadResourceToMemory( self.resource, extra_params=extra_params) else: data = self.client.DownloadResourceToMemory(self.resource) if self.resource_type == 'empty_document': self.assertEqual(len(data), 3) else: self.assertNotEqual(len(data), 0) else: # Cannot download collections self.assertRaises(ValueError, self.client.DownloadResourceToMemory, self.resource) def testDelete(self): self.assertEqual(self.resource.deleted, None) self.client.DeleteResource(self.resource, force=True) self.resource = self.client.GetResource(self.resource) self.assertNotEqual(self.resource.deleted, None) self.client.DeleteResource(self.resource, permanent=True, force=True) self.assertRaises(gdata.client.RequestError, self.client.GetResource, self.resource) class AclTest(DocsTestCase): def testGetAcl(self): acl_feed = self.client.GetResourceAcl(self.resource) self.assert_(isinstance(acl_feed, gdata.docs.data.AclFeed)) self.assertEqual(len(acl_feed.entry), 1) self.assert_(isinstance(acl_feed.entry[0], gdata.docs.data.AclEntry)) self.assert_(acl_feed.entry[0].scope is not None) self.assert_(acl_feed.entry[0].role is not None) def testGetAclEntry(self): acl_feed = self.client.GetResourceAcl(self.resource) acl_entry = acl_feed.entry[0] same_acl_entry = self.client.GetAclEntry(acl_entry) self.assert_(isinstance(same_acl_entry, gdata.docs.data.AclEntry)) self.assertEqual(acl_entry.GetSelfLink().href, same_acl_entry.GetSelfLink().href) self.assertEqual(acl_entry.title.text, same_acl_entry.title.text) def testAddAclEntry(self): acl_entry_to_add = gdata.docs.data.AclEntry.GetInstance( role='writer', scope_type='default', key=True) new_acl_entry = self.client.AddAclEntry(self.resource, acl_entry_to_add) self.assertEqual(acl_entry_to_add.scope.type, new_acl_entry.scope.type) self.assertEqual(new_acl_entry.scope.value, None) # Key will always be overridden on add self.assertEqual(acl_entry_to_add.with_key.role.value, new_acl_entry.with_key.role.value) acl_feed = self.client.GetResourceAcl(self.resource) self.assert_(isinstance(acl_feed, gdata.docs.data.AclFeed)) self.assert_(isinstance(acl_feed.entry[0], gdata.docs.data.AclEntry)) self.assert_(isinstance(acl_feed.entry[1], gdata.docs.data.AclEntry)) def testUpdateAclEntry(self): acl_entry_to_add = gdata.docs.data.AclEntry.GetInstance( role='reader', scope_type='user', scope_value='jeff@example.com', key=True) other_acl_entry = gdata.docs.data.AclEntry.GetInstance( role='writer', scope_type='user', scope_value='jeff@example.com') new_acl_entry = self.client.AddAclEntry(self.resource, acl_entry_to_add) new_acl_entry.with_key = None new_acl_entry.scope = other_acl_entry.scope new_acl_entry.role = other_acl_entry.role updated_acl_entry = self.client.UpdateAclEntry(new_acl_entry) self.assertEqual(updated_acl_entry.GetSelfLink().href, new_acl_entry.GetSelfLink().href) self.assertEqual(updated_acl_entry.title.text, new_acl_entry.title.text) self.assertEqual(updated_acl_entry.scope.type, other_acl_entry.scope.type) self.assertEqual(updated_acl_entry.scope.value, other_acl_entry.scope.value) self.assertEqual(updated_acl_entry.role.value, other_acl_entry.role.value) self.assertEqual(updated_acl_entry.with_key, None) def testDeleteAclEntry(self): acl_entry_to_add = gdata.docs.data.AclEntry.GetInstance( role='writer', scope_type='user', scope_value='joe@example.com', key=True) acl_feed = self.client.GetResourceAcl(self.resource) new_acl_entry = self.client.AddAclEntry(self.resource, acl_entry_to_add) acl_feed = self.client.GetResourceAcl(self.resource) self.assert_(isinstance(acl_feed, gdata.docs.data.AclFeed)) self.assertEqual(len(acl_feed.entry), 2) self.assert_(isinstance(acl_feed.entry[0], gdata.docs.data.AclEntry)) self.assert_(isinstance(acl_feed.entry[1], gdata.docs.data.AclEntry)) self.client.DeleteAclEntry(new_acl_entry) acl_feed = self.client.GetResourceAcl(self.resource) self.assert_(isinstance(acl_feed, gdata.docs.data.AclFeed)) self.assertEqual(len(acl_feed.entry), 1) self.assert_(isinstance(acl_feed.entry[0], gdata.docs.data.AclEntry)) class RevisionsTest(DocsTestCase): def testGetRevisions(self): # There are no revisions of collections if self.resource_type != 'collection': revisions = self.client.GetRevisions(self.resource) self.assert_(isinstance(revisions, gdata.docs.data.RevisionFeed)) self.assert_(isinstance(revisions.entry[0], gdata.docs.data.Revision)) # Currently, there is a bug where new presentations have 2 revisions. if self.resource_type != 'presentation': self.assertEqual(len(revisions.entry), 1) def testGetRevision(self): # There are no revisions of collections if self.resource_type != 'collection': revisions = self.client.GetRevisions(self.resource) entry = revisions.entry[0] new_entry = self.client.GetRevision(entry) self.assertEqual(entry.GetSelfLink().href, new_entry.GetSelfLink().href) self.assertEqual(entry.title.text, new_entry.title.text) def testGetRevisionBySelfLink(self): # There are no revisions of collections if self.resource_type != 'collection': revisions = self.client.GetRevisions(self.resource) entry = revisions.entry[0] new_entry = self.client.GetRevisionBySelfLink(entry.GetSelfLink().href) self.assertEqual(entry.GetSelfLink().href, new_entry.GetSelfLink().href) self.assertEqual(entry.title.text, new_entry.title.text) def testMultipleRevisionsAndUpdateResource(self): if self.resource_type not in ['collection', 'presentation']: revisions = self.client.GetRevisions(self.resource) self.assertEqual(len(revisions.entry), 1) # Currently, there is a bug where uploaded presentations have 2 revisions. elif self.resource_type == 'presentation': revisions = self.client.GetRevisions(self.resource) self.assertEqual(len(revisions.entry), 2) # Drawings do not currently support update, thus the rest of these # tests do not yet work as expected. if self.resource_type == 'drawing': return entry = self._update() self.assertEqual(entry.title.text, '%s Updated' % self.resource_title) if self.resource_type != 'collection': revisions = self.client.GetRevisions(entry) self.assert_(isinstance(revisions, gdata.docs.data.RevisionFeed)) if self.resource_type == 'presentation': self.assertEqual(len(revisions.entry), 3) self.assert_(isinstance(revisions.entry[2], gdata.docs.data.Revision)) else: self.assertEqual(len(revisions.entry), 2) self.assert_(isinstance(revisions.entry[0], gdata.docs.data.Revision)) self.assert_(isinstance(revisions.entry[1], gdata.docs.data.Revision)) def testPublishRevision(self): if self.resource_type in ['file', 'pdf', 'collection']: return # Drawings do not currently support update, thus this test would fail. if self.resource_type == 'drawing': return entry = self._update() revisions = self.client.GetRevisions(entry) revision = self.client.PublishRevision(revisions.entry[1]) # Currently, there is a bug where uploaded presentations have 2 revisions. if self.resource_type == 'presentation': revisions = self.client.GetRevisions(entry) revision = revisions.entry[2] self.assert_(isinstance(revision, gdata.docs.data.Revision)) self.assertEqual(revision.publish.value, 'true') self.assertEqual(revision.publish_auto.value, 'false') # The rest of the tests require an Apps domain if 'gmail' in conf.options.get_value('username'): return self.assertEqual(revision.publish_outside_domain.value, 'false') # Empty documents won't have further revisions b/c content didn't change if self.resource_type == 'empty_document': return revisions = self.client.GetRevisions(entry) if self.resource_type == 'presentation': revision = self.client.PublishRevision( revisions.entry[2], publish_auto=True, publish_outside_domain=True) else: revision = self.client.PublishRevision( revisions.entry[1], publish_auto=True, publish_outside_domain=True) if self.resource_type == 'spreadsheet': revision = client.GetRevisions(entry).entry[1] self.assert_(isinstance(revision, gdata.docs.data.Revision)) self.assertEqual(revision.publish.value, 'true') self.assertEqual(revision.publish_auto.value, 'true') self.assertEqual(revision.publish_outside_domain.value, 'true') revision = self.client.GetRevision(revisions.entry[0]) self.assertEqual(revision.publish, None) self.assertEqual(revision.publish_auto, None) self.assertEqual(revision.publish_outside_domain, None) def testDownloadRevision(self): if self.resource_type == 'collection': return revisions = self.client.GetRevisions(self.resource) tmp = tempfile.mkstemp() self.client.DownloadRevision(revisions.entry[0], tmp[1]) os.close(tmp[0]) os.remove(tmp[1]) def testDeleteRevision(self): # API can only delete file revisions if self.resource_type != 'file': return entry = self._update() revisions = self.client.GetRevisions(entry) self.assertEqual(len(revisions.entry), 2) self.client.DeleteRevision(revisions.entry[1]) self.assert_(isinstance(revisions, gdata.docs.data.RevisionFeed)) self.assert_(isinstance(revisions.entry[0], gdata.docs.data.Revision)) revisions = self.client.GetRevisions(entry) self.assertEqual(len(revisions.entry), 1) class ChangesTest(DocsTestCase): def testGetChanges(self): # This test assumes that by the time this test is run, the account # being used already has a number of changes changes = self.client.GetChanges(max_results=5) self.assert_(isinstance(changes, gdata.docs.data.ChangeFeed)) self.assert_(len(changes.entry) <= 5) self.assert_(isinstance(changes.entry[0], gdata.docs.data.Change)) self._update() changes = self.client.GetChanges(changes.entry[0].changestamp.value, 5) self.assert_(isinstance(changes, gdata.docs.data.ChangeFeed)) self.assert_(len(changes.entry) <= 5) self.assert_(isinstance(changes.entry[0], gdata.docs.data.Change)) def testDeleteResourceCreatesNewChange(self): """Ensure that deleting a resource causes a new change entry.""" self._update() changes = self.client.GetChanges(max_results=1) latest = changes.entry[0].changestamp.value self._delete(self.resource) time.sleep(10) changes = self.client.GetChanges(max_results=1) self.assert_(latest < changes.entry[0].changestamp.value) class MetadataTest(DocsTestCase): def setUp(self): if conf.options.get_value('runlive') != 'true': raise RuntimeError('Live tests require --runlive true') else: self.client = gdata.docs.client.DocsClient() if conf.options.get_value('ssl') == 'true': self.client.ssl = True conf.configure_client(self.client, 'DocsTest', self.client.auth_service) conf.configure_cache(self.client, str(self.__class__)) if conf.options.get_value('clean') == 'true': self._delete_all() def tearDown(self): conf.close_client(self.client) def testMetadata(self): metadata = self.client.GetMetadata() self.assert_(isinstance(metadata, gdata.docs.data.Metadata)) self.assertNotEqual(int(metadata.quota_bytes_total.text), 0) self.assertEqual(int(metadata.quota_bytes_used.text), 0) self.assertEqual(int(metadata.quota_bytes_used_in_trash.text), 0) self.assertNotEqual(len(metadata.import_formats), 0) self.assertNotEqual(len(metadata.export_formats), 0) self.assertNotEqual(len(metadata.features), 0) self.assertNotEqual(len(metadata.max_upload_sizes), 0) def suite(): suite = unittest.TestSuite() for key, value in RESOURCES.iteritems(): for case in [ResourcesTest, AclTest, RevisionsTest, ChangesTest]: tests = unittest.TestLoader().loadTestsFromTestCase(case) for test in tests: test.resource_type = key test.resource_label = value[0] test.resource_title = value[1] test.resource_path = value[2] test.resource_alt_path = value[3] test.resource_mime = value[4] test.resource_export = value[5] suite.addTest(test) suite.addTests(unittest.TestLoader().loadTestsFromTestCase(MetadataTest)) return suite if __name__ == '__main__': unittest.TextTestRunner().run(suite())