#!/usr/bin/python # # Copyright Google 2007-2008, 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. import time import unittest import getpass import gdata.spreadsheet.text_db import gdata.spreadsheet.service __author__ = 'api.jscudder (Jeffrey Scudder)' username = '' password = '' class FactoryTest(unittest.TestCase): def setUp(self): self.client = gdata.spreadsheet.text_db.DatabaseClient() def testBadCredentials(self): try: self.client.SetCredentials('foo', 'bar') self.fail() except gdata.spreadsheet.text_db.Error, e: pass def testCreateGetAndDeleteDatabase(self): db_title = 'google_spreadsheets_db unit test 1' self.client.SetCredentials(username, password) db = self.client.CreateDatabase(db_title) # Test finding the database using the name time.sleep(5) db_list = self.client.GetDatabases(name=db_title) self.assert_(len(db_list) >= 1) if len(db_list) >= 1: self.assert_(db_list[0].entry.title.text == db_title) # Test finding the database using the spreadsheet key db_list = self.client.GetDatabases(spreadsheet_key=db.spreadsheet_key) self.assert_(len(db_list) == 1) self.assert_(db_list[0].entry.title.text == db_title) # Delete the test spreadsheet time.sleep(10) db.Delete() class DatabaseTest(unittest.TestCase): def setUp(self): client = gdata.spreadsheet.text_db.DatabaseClient(username, password) self.db = client.CreateDatabase('google_spreadsheets_db unit test 2') def tearDown(self): time.sleep(10) self.db.Delete() def testCreateGetAndDeleteTable(self): table = self.db.CreateTable('test1', ['1','2','3']) # Try to get the new table using the worksheet id. table_list = self.db.GetTables(worksheet_id=table.worksheet_id) self.assert_(len(table_list) == 1) self.assert_(table_list[0].entry.title.text, 'test1') # Try to get the table using the name table_list = self.db.GetTables(name='test1') self.assert_(len(table_list) == 1) self.assert_(table_list[0].entry.title.text, 'test1') # Delete the table table.Delete() class TableTest(unittest.TestCase): def setUp(self): client = gdata.spreadsheet.text_db.DatabaseClient(username, password) self.db = client.CreateDatabase('google_spreadsheets_db unit test 3') self.table = self.db.CreateTable('test1', ['a','b','c_d','a', 'd:e']) def tearDown(self): time.sleep(10) self.db.Delete() def testCreateGetAndDeleteRecord(self): new_record = self.table.AddRecord({'a':'test1', 'b':'test2', 'cd':'test3', 'a_2':'test4', 'de':'test5'}) # Test getting record by line number. record = self.table.GetRecord(row_number=1) self.assert_(record is not None) self.assert_(record.content['a'] == 'test1') self.assert_(record.content['b'] == 'test2') self.assert_(record.content['cd'] == 'test3') self.assert_(record.content['a_2'] == 'test4') # Test getting record using the id. record_list = self.table.GetRecord(row_id=new_record.row_id) self.assert_(record is not None) # Delete the record. time.sleep(10) new_record.Delete() def testPushPullSyncing(self): # Get two copies of the same row. first_copy = self.table.AddRecord({'a':'1', 'b':'2', 'cd':'3', 'a_2':'4', 'de':'5'}) second_copy = self.table.GetRecord(first_copy.row_id) # Make changes in the first copy first_copy.content['a'] = '7' first_copy.content['b'] = '9' # Try to get the changes before they've been committed second_copy.Pull() self.assert_(second_copy.content['a'] == '1') self.assert_(second_copy.content['b'] == '2') # Commit the changes, the content should now be different first_copy.Push() second_copy.Pull() self.assert_(second_copy.content['a'] == '7') self.assert_(second_copy.content['b'] == '9') # Make changes to the second copy, push, then try to push changes from # the first copy. first_copy.content['a'] = '10' second_copy.content['a'] = '15' first_copy.Push() try: second_copy.Push() # The second update should raise and exception due to a 409 conflict. self.fail() except gdata.spreadsheet.service.RequestError: pass except Exception, error: #TODO: Why won't the except RequestError catch this? pass def testFindRecords(self): # Add lots of test records: self.table.AddRecord({'a':'1', 'b':'2', 'cd':'3', 'a_2':'4', 'de':'5'}) self.table.AddRecord({'a':'hi', 'b':'2', 'cd':'20', 'a_2':'4', 'de':'5'}) self.table.AddRecord({'a':'2', 'b':'2', 'cd':'3'}) self.table.AddRecord({'a':'2', 'b':'2', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'hi hi hi', 'b':'2', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'"5"', 'b':'5', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'5', 'b':'5', 'cd':'15', 'de':'7'}) time.sleep(10) matches = self.table.FindRecords('a == 1') self.assert_(len(matches) == 1) self.assert_(matches[0].content['a'] == '1') self.assert_(matches[0].content['b'] == '2') matches = self.table.FindRecords('a > 1 && cd < 20') self.assert_(len(matches) == 4) matches = self.table.FindRecords('cd < de') self.assert_(len(matches) == 7) matches = self.table.FindRecords('a == b') self.assert_(len(matches) == 0) matches = self.table.FindRecords('a == 5') self.assert_(len(matches) == 1) def testIterateResultSet(self): # Populate the table with test data. self.table.AddRecord({'a':'1', 'b':'2', 'cd':'3', 'a_2':'4', 'de':'5'}) self.table.AddRecord({'a':'hi', 'b':'2', 'cd':'20', 'a_2':'4', 'de':'5'}) self.table.AddRecord({'a':'2', 'b':'2', 'cd':'3'}) self.table.AddRecord({'a':'2', 'b':'2', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'hi hi hi', 'b':'2', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'"5"', 'b':'5', 'cd':'15', 'de':'7'}) self.table.AddRecord({'a':'5', 'b':'5', 'cd':'15', 'de':'7'}) # Get the first two rows. records = self.table.GetRecords(1, 2) self.assert_(len(records) == 2) self.assert_(records[0].content['a'] == '1') self.assert_(records[1].content['a'] == 'hi') # Then get the next two rows. next_records = records.GetNext() self.assert_(len(next_records) == 2) self.assert_(next_records[0].content['a'] == '2') self.assert_(next_records[0].content['cd'] == '3') self.assert_(next_records[1].content['cd'] == '15') self.assert_(next_records[1].content['de'] == '7') def testLookupFieldsOnPreexistingTable(self): existing_table = self.db.GetTables(name='test1')[0] existing_table.LookupFields() self.assertEquals(existing_table.fields, ['a', 'b', 'cd', 'a_2', 'de']) if __name__ == '__main__': if not username: username = raw_input('Spreadsheets API | Text DB Tests\n' 'Please enter your username: ') if not password: password = getpass.getpass() unittest.main()