| # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for the gdata_lib module.""" |
| |
| from __future__ import print_function |
| |
| import getpass |
| import mox |
| import re |
| |
| import atom.service |
| import gdata.projecthosting.client as gd_ph_client |
| import gdata.spreadsheet.service |
| |
| from chromite.lib import cros_test_lib |
| from chromite.lib import gdata_lib |
| from chromite.lib import osutils |
| |
| |
| # pylint: disable=W0212,E1101 |
| |
| |
| class GdataLibTest(cros_test_lib.OutputTestCase): |
| """Tests for methods that escape/unescape strings for speadsheets.""" |
| |
| def testPrepColNameForSS(self): |
| tests = { |
| 'foo': 'foo', |
| 'Foo': 'foo', |
| 'FOO': 'foo', |
| 'foo bar': 'foobar', |
| 'Foo Bar': 'foobar', |
| 'F O O B A R': 'foobar', |
| 'Foo/Bar': 'foobar', |
| 'Foo Bar/Dude': 'foobardude', |
| 'foo/bar': 'foobar', |
| } |
| |
| for col in tests: |
| expected = tests[col] |
| self.assertEquals(expected, gdata_lib.PrepColNameForSS(col)) |
| self.assertEquals(expected, gdata_lib.PrepColNameForSS(expected)) |
| |
| def testPrepValForSS(self): |
| tests = { |
| None: None, |
| '': '', |
| 'foo': 'foo', |
| 'foo123': 'foo123', |
| '123': "'123", |
| '1.2': "'1.2", |
| } |
| |
| for val in tests: |
| expected = tests[val] |
| self.assertEquals(expected, gdata_lib.PrepValForSS(val)) |
| |
| def testPrepRowForSS(self): |
| vals = { |
| None: None, |
| '': '', |
| 'foo': 'foo', |
| 'foo123': 'foo123', |
| '123': "'123", |
| '1.2': "'1.2", |
| } |
| |
| # Create before and after rows (rowIn, rowOut). |
| rowIn = {} |
| rowOut = {} |
| col = 'a' # Column names not important. |
| for valIn in vals: |
| valOut = vals[valIn] |
| rowIn[col] = valIn |
| rowOut[col] = valOut |
| |
| col = chr(ord(col) + 1) # Change column name. |
| |
| self.assertEquals(rowOut, gdata_lib.PrepRowForSS(rowIn)) |
| |
| def testScrubValFromSS(self): |
| tests = { |
| None: None, |
| 'foo': 'foo', |
| '123': '123', |
| "'123": '123', |
| } |
| |
| for val in tests: |
| expected = tests[val] |
| self.assertEquals(expected, gdata_lib.ScrubValFromSS(val)) |
| |
| |
| class CredsTest(cros_test_lib.MockOutputTestCase): |
| """Tests related to user credentials.""" |
| |
| USER = 'somedude@chromium.org' |
| PASSWORD = 'worldsbestpassword' |
| DOCS_TOKEN = 'SomeDocsAuthToken' |
| TRACKER_TOKEN = 'SomeTrackerAuthToken' |
| |
| @osutils.TempFileDecorator |
| def testStoreLoadCreds(self): |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| creds.SetCreds(self.USER, self.PASSWORD) |
| self.assertEquals(self.USER, creds.user) |
| self.assertEquals(self.PASSWORD, creds.password) |
| self.assertTrue(creds.creds_dirty) |
| |
| creds.StoreCreds(self.tempfile) |
| self.assertEquals(self.USER, creds.user) |
| self.assertEquals(self.PASSWORD, creds.password) |
| self.assertFalse(creds.creds_dirty) |
| |
| # Clear user/password before loading from just-created file. |
| creds.user = None |
| creds.password = None |
| self.assertEquals(None, creds.user) |
| self.assertEquals(None, creds.password) |
| |
| creds.LoadCreds(self.tempfile) |
| self.assertEquals(self.USER, creds.user) |
| self.assertEquals(self.PASSWORD, creds.password) |
| self.assertFalse(creds.creds_dirty) |
| |
| @osutils.TempFileDecorator |
| def testStoreLoadToken(self): |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| creds.user = self.USER |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| creds.SetDocsAuthToken(self.DOCS_TOKEN) |
| self.assertEquals(self.DOCS_TOKEN, creds.docs_auth_token) |
| self.assertTrue(creds.token_dirty) |
| creds.SetTrackerAuthToken(self.TRACKER_TOKEN) |
| self.assertEquals(self.TRACKER_TOKEN, creds.tracker_auth_token) |
| self.assertTrue(creds.token_dirty) |
| |
| creds.StoreAuthToken(self.tempfile) |
| self.assertEquals(self.DOCS_TOKEN, creds.docs_auth_token) |
| self.assertEquals(self.TRACKER_TOKEN, creds.tracker_auth_token) |
| self.assertFalse(creds.token_dirty) |
| |
| # Clear auth_tokens before loading from just-created file. |
| creds.docs_auth_token = None |
| creds.tracker_auth_token = None |
| creds.user = None |
| |
| creds.LoadAuthToken(self.tempfile) |
| self.assertEquals(self.DOCS_TOKEN, creds.docs_auth_token) |
| self.assertEquals(self.TRACKER_TOKEN, creds.tracker_auth_token) |
| self.assertFalse(creds.token_dirty) |
| self.assertEquals(self.USER, creds.user) |
| |
| def testSetCreds(self): |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| |
| # This is the test verification. |
| creds.SetCreds(self.USER, password=self.PASSWORD) |
| self.assertEquals(self.USER, creds.user) |
| self.assertEquals(self.PASSWORD, creds.password) |
| self.assertTrue(creds.creds_dirty) |
| |
| def testSetCredsNoPassword(self): |
| # Add test-specific mocks/stubs |
| self.PatchObject(getpass, 'getpass', return_value=self.PASSWORD) |
| |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| |
| # This is the test verification. |
| creds.SetCreds(self.USER) |
| self.assertEquals(self.USER, creds.user) |
| self.assertEquals(self.PASSWORD, creds.password) |
| self.assertTrue(creds.creds_dirty) |
| |
| def testSetDocsToken(self): |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| |
| # This is the test verification. |
| creds.SetDocsAuthToken(self.DOCS_TOKEN) |
| self.assertEquals(self.DOCS_TOKEN, creds.docs_auth_token) |
| self.assertTrue(creds.token_dirty) |
| |
| def testSetTrackerToken(self): |
| # This is the replay script for the test. |
| creds = gdata_lib.Creds() |
| |
| # This is the test verification. |
| creds.SetTrackerAuthToken(self.TRACKER_TOKEN) |
| self.assertEquals(self.TRACKER_TOKEN, creds.tracker_auth_token) |
| self.assertTrue(creds.token_dirty) |
| |
| |
| class SpreadsheetRowTest(cros_test_lib.OutputTestCase): |
| """Tests related to spreadsheet row interaction.""" |
| |
| SS_ROW_OBJ = 'SSRowObj' |
| SS_ROW_NUM = 5 |
| |
| def testEmpty(self): |
| row = gdata_lib.SpreadsheetRow(self.SS_ROW_OBJ, self.SS_ROW_NUM) |
| |
| self.assertEquals(0, len(row)) |
| self.assertEquals(self.SS_ROW_OBJ, row.ss_row_obj) |
| self.assertEquals(self.SS_ROW_NUM, row.ss_row_num) |
| |
| self.assertRaises(TypeError, row, '__setitem__', 'abc', 'xyz') |
| self.assertEquals(0, len(row)) |
| self.assertFalse('abc' in row) |
| |
| def testInit(self): |
| starting_vals = {'abc': 'xyz', 'foo': 'bar'} |
| row = gdata_lib.SpreadsheetRow(self.SS_ROW_OBJ, self.SS_ROW_NUM, |
| starting_vals) |
| |
| self.assertEquals(len(starting_vals), len(row)) |
| self.assertEquals(starting_vals, row) |
| self.assertEquals(row['abc'], 'xyz') |
| self.assertTrue('abc' in row) |
| self.assertEquals(row['foo'], 'bar') |
| self.assertTrue('foo' in row) |
| |
| self.assertEquals(self.SS_ROW_OBJ, row.ss_row_obj) |
| self.assertEquals(self.SS_ROW_NUM, row.ss_row_num) |
| |
| self.assertRaises(TypeError, row, '__delitem__', 'abc') |
| self.assertEquals(len(starting_vals), len(row)) |
| self.assertTrue('abc' in row) |
| |
| |
| class SpreadsheetCommTest(cros_test_lib.MoxOutputTestCase): |
| """Test Speadsheet communication.""" |
| |
| SS_KEY = 'TheSSKey' |
| WS_NAME = 'TheWSName' |
| WS_KEY = 'TheWSKey' |
| |
| USER = 'dude' |
| PASSWORD = 'shhh' |
| TOKEN = 'authtoken' |
| |
| COLUMNS = ('greeting', 'name', 'title') |
| ROWS = ( |
| {'greeting': 'Hi', 'name': 'George', 'title': 'Mr.'}, |
| {'greeting': 'Howdy', 'name': 'Billy Bob', 'title': 'Mr.'}, |
| {'greeting': 'Yo', 'name': 'Adriane', 'title': 'Ms.'}, |
| ) |
| |
| def MockScomm(self, connect=True): |
| """Return a mocked SpreadsheetComm""" |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| |
| mocked_scomm._columns = None |
| mocked_scomm._rows = None |
| |
| if connect: |
| mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService) |
| mocked_scomm.gd_client = mocked_gdclient |
| mocked_scomm.ss_key = self.SS_KEY |
| mocked_scomm.ws_name = self.WS_NAME |
| mocked_scomm.ws_key = self.WS_KEY |
| else: |
| mocked_scomm.gd_client = None |
| mocked_scomm.ss_key = None |
| mocked_scomm.ws_name = None |
| mocked_scomm.ws_key = None |
| |
| return mocked_scomm |
| |
| def NewScomm(self, gd_client=None, connect=True): |
| """Return a non-mocked SpreadsheetComm.""" |
| scomm = gdata_lib.SpreadsheetComm() |
| scomm.gd_client = gd_client |
| |
| if connect: |
| scomm.ss_key = self.SS_KEY |
| scomm.ws_name = self.WS_NAME |
| scomm.ws_key = self.WS_KEY |
| else: |
| scomm.ss_key = None |
| scomm.ws_name = None |
| scomm.ws_key = None |
| |
| return scomm |
| |
| def GenerateCreds(self, skip_user=False, skip_token=False): |
| creds = gdata_lib.Creds() |
| if not skip_user: |
| creds.user = self.USER |
| creds.password = self.PASSWORD |
| |
| if not skip_token: |
| creds.docs_auth_token = self.TOKEN |
| |
| return creds |
| |
| def testConnect(self): |
| mocked_scomm = self.MockScomm(connect=False) |
| creds = self.GenerateCreds() |
| |
| # This is the replay script for the test. |
| mocked_scomm._Login(creds, 'chromiumos') |
| mocked_scomm.SetCurrentWorksheet(self.WS_NAME, ss_key=self.SS_KEY) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.Connect(mocked_scomm, creds, |
| self.SS_KEY, self.WS_NAME) |
| self.mox.VerifyAll() |
| |
| def testColumns(self): |
| """Test the .columns property. Testing a property gets ugly.""" |
| self.mox.StubOutWithMock(gdata.spreadsheet.service, 'CellQuery') |
| mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService) |
| scomm = self.NewScomm(gd_client=mocked_gdclient, connect=True) |
| |
| query = {'max-row': '1'} |
| |
| # Simulate a Cells feed from spreadsheet for the column row. |
| cols = [c[0].upper() + c[1:] for c in self.COLUMNS] |
| entry = [cros_test_lib.EasyAttr( |
| content=cros_test_lib.EasyAttr(text=c)) for c in cols] |
| feed = cros_test_lib.EasyAttr(entry=entry) |
| |
| # This is the replay script for the test. |
| gdata.spreadsheet.service.CellQuery().AndReturn(query) |
| mocked_gdclient.GetCellsFeed( |
| self.SS_KEY, self.WS_KEY, query=query).AndReturn(feed) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = scomm.columns |
| del scomm # Force deletion now before VerifyAll. |
| self.mox.VerifyAll() |
| |
| expected_result = self.COLUMNS |
| self.assertEquals(expected_result, result) |
| |
| def testRows(self): |
| """Test the .rows property. Testing a property gets ugly.""" |
| mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService) |
| scomm = self.NewScomm(gd_client=mocked_gdclient, connect=True) |
| |
| # Simulate a List feed from spreadsheet for all rows. |
| rows = [ |
| {'col_name': 'Joe', 'col_age': '12', 'col_zip': '12345'}, |
| {'col_name': 'Bob', 'col_age': '15', 'col_zip': '54321'}, |
| ] |
| entry = [] |
| for row in rows: |
| custom = dict((k, cros_test_lib.EasyAttr(text=v)) |
| for (k, v) in row.iteritems()) |
| entry.append(cros_test_lib.EasyAttr(custom=custom)) |
| feed = cros_test_lib.EasyAttr(entry=entry) |
| |
| # This is the replay script for the test. |
| mocked_gdclient.GetListFeed(self.SS_KEY, self.WS_KEY).AndReturn(feed) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = scomm.rows |
| del scomm # Force deletion now before VerifyAll. |
| self.mox.VerifyAll() |
| self.assertEquals(tuple(rows), result) |
| |
| # Result tuple should have spreadsheet row num as attribute on each row. |
| self.assertEquals(2, result[0].ss_row_num) |
| self.assertEquals(3, result[1].ss_row_num) |
| |
| # Result tuple should have spreadsheet row obj as attribute on each row. |
| self.assertEquals(entry[0], result[0].ss_row_obj) |
| self.assertEquals(entry[1], result[1].ss_row_obj) |
| |
| def testSetCurrentWorksheetStart(self): |
| mocked_scomm = self.MockScomm(connect=True) |
| |
| # Undo worksheet settings. |
| mocked_scomm.ss_key = None |
| mocked_scomm.ws_name = None |
| mocked_scomm.ws_key = None |
| |
| # This is the replay script for the test. |
| mocked_scomm._ClearCache() |
| mocked_scomm._GetWorksheetKey( |
| self.SS_KEY, self.WS_NAME).AndReturn(self.WS_KEY) |
| mocked_scomm._ClearCache() |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.SetCurrentWorksheet(mocked_scomm, self.WS_NAME, |
| ss_key=self.SS_KEY) |
| self.mox.VerifyAll() |
| |
| self.assertEquals(self.SS_KEY, mocked_scomm.ss_key) |
| self.assertEquals(self.WS_KEY, mocked_scomm.ws_key) |
| self.assertEquals(self.WS_NAME, mocked_scomm.ws_name) |
| |
| def testSetCurrentWorksheetRestart(self): |
| mocked_scomm = self.MockScomm(connect=True) |
| |
| other_ws_name = 'OtherWSName' |
| other_ws_key = 'OtherWSKey' |
| |
| # This is the replay script for the test. |
| mocked_scomm._GetWorksheetKey( |
| self.SS_KEY, other_ws_name).AndReturn(other_ws_key) |
| mocked_scomm._ClearCache() |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.SetCurrentWorksheet(mocked_scomm, other_ws_name) |
| self.mox.VerifyAll() |
| |
| self.assertEquals(self.SS_KEY, mocked_scomm.ss_key) |
| self.assertEquals(other_ws_key, mocked_scomm.ws_key) |
| self.assertEquals(other_ws_name, mocked_scomm.ws_name) |
| |
| def testClearCache(self): |
| rows = 'SomeRows' |
| cols = 'SomeColumns' |
| |
| scomm = self.NewScomm() |
| scomm._rows = rows |
| scomm._columns = cols |
| |
| scomm._ClearCache(keep_columns=True) |
| self.assertTrue(scomm._rows is None) |
| self.assertEquals(cols, scomm._columns) |
| |
| scomm._rows = rows |
| scomm._columns = cols |
| |
| scomm._ClearCache(keep_columns=False) |
| self.assertTrue(scomm._rows is None) |
| self.assertTrue(scomm._columns is None) |
| |
| scomm._rows = rows |
| scomm._columns = cols |
| |
| scomm._ClearCache() |
| self.assertTrue(scomm._rows is None) |
| self.assertTrue(scomm._columns is None) |
| |
| def testLoginWithUserPassword(self): |
| mocked_scomm = self.MockScomm(connect=False) |
| creds = self.GenerateCreds(skip_token=True) |
| |
| self.mox.StubOutClassWithMocks(gdata_lib, 'RetrySpreadsheetsService') |
| |
| source = 'SomeSource' |
| |
| # This is the replay script for the test. |
| mocked_gdclient = gdata_lib.RetrySpreadsheetsService() |
| mocked_gdclient.ProgrammaticLogin() |
| mocked_gdclient.GetClientLoginToken().AndReturn(self.TOKEN) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| gdata_lib.SpreadsheetComm._Login(mocked_scomm, creds, source) |
| self.mox.VerifyAll() |
| self.assertEquals(self.USER, mocked_gdclient.email) |
| self.assertEquals(self.PASSWORD, mocked_gdclient.password) |
| self.assertEquals(self.TOKEN, creds.docs_auth_token) |
| self.assertEquals(source, mocked_gdclient.source) |
| self.assertEquals(mocked_gdclient, mocked_scomm.gd_client) |
| |
| def testLoginWithToken(self): |
| mocked_scomm = self.MockScomm(connect=False) |
| creds = self.GenerateCreds(skip_user=True) |
| |
| self.mox.StubOutClassWithMocks(gdata_lib, 'RetrySpreadsheetsService') |
| |
| source = 'SomeSource' |
| |
| # This is the replay script for the test. |
| mocked_gdclient = gdata_lib.RetrySpreadsheetsService() |
| mocked_gdclient.SetClientLoginToken(creds.docs_auth_token) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| gdata_lib.SpreadsheetComm._Login(mocked_scomm, creds, source) |
| self.mox.VerifyAll() |
| self.assertFalse(hasattr(mocked_gdclient, 'email')) |
| self.assertFalse(hasattr(mocked_gdclient, 'password')) |
| self.assertEquals(source, mocked_gdclient.source) |
| self.assertEquals(mocked_gdclient, mocked_scomm.gd_client) |
| |
| def testGetWorksheetKey(self): |
| mocked_scomm = self.MockScomm() |
| |
| entrylist = [ |
| cros_test_lib.EasyAttr( |
| title=cros_test_lib.EasyAttr(text='Foo'), id='NotImportant'), |
| cros_test_lib.EasyAttr( |
| title=cros_test_lib.EasyAttr(text=self.WS_NAME), |
| id=cros_test_lib.EasyAttr(text='/some/path/%s' % self.WS_KEY)), |
| cros_test_lib.EasyAttr( |
| title=cros_test_lib.EasyAttr(text='Bar'), id='NotImportant'), |
| ] |
| feed = cros_test_lib.EasyAttr(entry=entrylist) |
| |
| # This is the replay script for the test. |
| mocked_scomm.gd_client.GetWorksheetsFeed(self.SS_KEY).AndReturn(feed) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm._GetWorksheetKey(mocked_scomm, |
| self.SS_KEY, self.WS_NAME) |
| self.mox.VerifyAll() |
| |
| def testGetColumns(self): |
| mocked_scomm = self.MockScomm() |
| mocked_scomm.columns = 'SomeColumns' |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = gdata_lib.SpreadsheetComm.GetColumns(mocked_scomm) |
| self.mox.VerifyAll() |
| self.assertEquals('SomeColumns', result) |
| |
| def testGetColumnIndex(self): |
| # Note that spreadsheet column indices start at 1. |
| mocked_scomm = self.MockScomm() |
| mocked_scomm.columns = ['these', 'are', 'column', 'names'] |
| |
| # This is the replay script for the test. |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = gdata_lib.SpreadsheetComm.GetColumnIndex(mocked_scomm, 'are') |
| self.mox.VerifyAll() |
| self.assertEquals(2, result) |
| |
| def testGetRows(self): |
| mocked_scomm = self.MockScomm() |
| rows = [] |
| for row_ix, row_dict in enumerate(self.ROWS): |
| rows.append(gdata_lib.SpreadsheetRow('SSRowObj%d' % (row_ix + 2), |
| (row_ix + 2), row_dict)) |
| mocked_scomm.rows = tuple(rows) |
| |
| # This is the replay script for the test. |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = gdata_lib.SpreadsheetComm.GetRows(mocked_scomm) |
| self.mox.VerifyAll() |
| self.assertEquals(self.ROWS, result) |
| for row_ix in xrange(len(self.ROWS)): |
| self.assertEquals(row_ix + 2, result[row_ix].ss_row_num) |
| self.assertEquals('SSRowObj%d' % (row_ix + 2), result[row_ix].ss_row_obj) |
| |
| def testGetRowCacheByCol(self): |
| mocked_scomm = self.MockScomm() |
| |
| # This is the replay script for the test. |
| mocked_scomm.GetRows().AndReturn(self.ROWS) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = gdata_lib.SpreadsheetComm.GetRowCacheByCol(mocked_scomm, 'name') |
| self.mox.VerifyAll() |
| |
| # Result is a dict of rows by the 'name' column. |
| for row in self.ROWS: |
| name = row['name'] |
| self.assertEquals(row, result[name]) |
| |
| def testGetRowCacheByColDuplicates(self): |
| mocked_scomm = self.MockScomm() |
| |
| # Create new row list with duplicates by name column. |
| rows = [] |
| for row in self.ROWS: |
| new_row = dict(row) |
| new_row['greeting'] = row['greeting'] + ' there' |
| rows.append(new_row) |
| |
| rows.extend(self.ROWS) |
| |
| # This is the replay script for the test. |
| mocked_scomm.GetRows().AndReturn(tuple(rows)) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| result = gdata_lib.SpreadsheetComm.GetRowCacheByCol(mocked_scomm, 'name') |
| self.mox.VerifyAll() |
| |
| # Result is a dict of rows by the 'name' column. In this |
| # test each result should be a list of the rows with the same |
| # value in the 'name' column. |
| num_rows = len(rows) |
| for ix in xrange(num_rows / 2): |
| row1 = rows[ix] |
| row2 = rows[ix + (num_rows / 2)] |
| |
| name = row1['name'] |
| self.assertEquals(name, row2['name']) |
| |
| expected_rows = [row1, row2] |
| self.assertEquals(expected_rows, result[name]) |
| |
| def testInsertRow(self): |
| mocked_scomm = self.MockScomm() |
| |
| row = 'TheRow' |
| |
| # Replay script |
| mocked_scomm.gd_client.InsertRow(row, mocked_scomm.ss_key, |
| mocked_scomm.ws_key) |
| mocked_scomm._ClearCache(keep_columns=True) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.InsertRow(mocked_scomm, row) |
| self.mox.VerifyAll() |
| |
| def testUpdateRowCellByCell(self): |
| mocked_scomm = self.MockScomm() |
| |
| rowIx = 5 |
| row = {'a': 123, 'b': 234, 'c': 345} |
| colIndices = {'a': 1, 'b': None, 'c': 4} |
| |
| # Replay script |
| for colName in row: |
| colIx = colIndices[colName] |
| mocked_scomm.GetColumnIndex(colName).AndReturn(colIx) |
| if colIx is not None: |
| mocked_scomm.ReplaceCellValue(rowIx, colIx, row[colName]) |
| mocked_scomm._ClearCache(keep_columns=True) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.UpdateRowCellByCell(mocked_scomm, rowIx, row) |
| self.mox.VerifyAll() |
| |
| def testDeleteRow(self): |
| mocked_scomm = self.MockScomm() |
| |
| ss_row = 'TheRow' |
| |
| # Replay script |
| mocked_scomm.gd_client.DeleteRow(ss_row) |
| mocked_scomm._ClearCache(keep_columns=True) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| gdata_lib.SpreadsheetComm.DeleteRow(mocked_scomm, ss_row) |
| self.mox.VerifyAll() |
| |
| def testReplaceCellValue(self): |
| mocked_scomm = self.MockScomm() |
| |
| rowIx = 14 |
| colIx = 4 |
| val = 'TheValue' |
| |
| # Replay script |
| mocked_scomm.gd_client.UpdateCell(rowIx, colIx, val, |
| mocked_scomm.ss_key, mocked_scomm.ws_key) |
| mocked_scomm._ClearCache(keep_columns=True) |
| self.mox.ReplayAll() |
| |
| # Verify |
| gdata_lib.SpreadsheetComm.ReplaceCellValue(mocked_scomm, rowIx, colIx, val) |
| self.mox.VerifyAll() |
| |
| def testClearCellValue(self): |
| mocked_scomm = self.MockScomm() |
| |
| rowIx = 14 |
| colIx = 4 |
| |
| # Replay script |
| mocked_scomm.ReplaceCellValue(rowIx, colIx, None) |
| self.mox.ReplayAll() |
| |
| # Verify |
| gdata_lib.SpreadsheetComm.ClearCellValue(mocked_scomm, rowIx, colIx) |
| self.mox.VerifyAll() |
| |
| |
| class IssueCommentTest(cros_test_lib.TestCase): |
| """Test creating comments.""" |
| |
| def testInit(self): |
| title = 'Greetings, Earthlings' |
| text = 'I come in peace.' |
| ic = gdata_lib.IssueComment(title, text) |
| |
| self.assertEquals(title, ic.title) |
| self.assertEquals(text, ic.text) |
| |
| |
| def createTrackerIssue(tid, labels, owner, status, content, title): |
| tissue = cros_test_lib.EasyAttr() |
| tissue.id = cros_test_lib.EasyAttr( |
| text='http://www/some/path/%d' % tid) |
| tissue.label = [cros_test_lib.EasyAttr(text=l) for l in labels] |
| tissue.owner = cros_test_lib.EasyAttr( |
| username=cros_test_lib.EasyAttr(text=owner)) |
| tissue.status = cros_test_lib.EasyAttr(text=status) |
| tissue.content = cros_test_lib.EasyAttr(text=content) |
| tissue.title = cros_test_lib.EasyAttr(text=title) |
| return tissue |
| |
| |
| class IssueTest(cros_test_lib.MoxTestCase): |
| """Test creating a bug.""" |
| |
| def testInitOverride(self): |
| owner = 'somedude@chromium.org' |
| status = 'Assigned' |
| issue = gdata_lib.Issue(owner=owner, status=status) |
| |
| self.assertEquals(owner, issue.owner) |
| self.assertEquals(status, issue.status) |
| |
| def testInitInvalidOverride(self): |
| self.assertRaises(ValueError, gdata_lib.Issue, |
| foobar='NotARealAttr') |
| |
| def testInitFromTracker(self): |
| # Need to create a dummy Tracker Issue object. |
| tissue_id = 123 |
| tissue_labels = ['Iteration-10', 'Effort-2'] |
| tissue_owner = 'thedude@chromium.org' |
| tissue_status = 'Available' |
| tissue_content = 'The summary message' |
| tissue_title = 'The Big Title' |
| |
| tissue = createTrackerIssue(tid=tissue_id, labels=tissue_labels, |
| owner=tissue_owner, status=tissue_status, |
| content=tissue_content, title=tissue_title) |
| |
| mocked_issue = self.mox.CreateMock(gdata_lib.Issue) |
| |
| # Replay script |
| mocked_issue.GetTrackerIssueComments(tissue_id, 'TheProject').AndReturn([]) |
| self.mox.ReplayAll() |
| |
| # Verify |
| gdata_lib.Issue.InitFromTracker(mocked_issue, tissue, 'TheProject') |
| self.mox.VerifyAll() |
| self.assertEquals(tissue_id, mocked_issue.id) |
| self.assertEquals(tissue_labels, mocked_issue.labels) |
| self.assertEquals(tissue_owner, mocked_issue.owner) |
| self.assertEquals(tissue_status, mocked_issue.status) |
| self.assertEquals(tissue_content, mocked_issue.summary) |
| self.assertEquals(tissue_title, mocked_issue.title) |
| self.assertEquals([], mocked_issue.comments) |
| |
| |
| class TrackerCommTest(cros_test_lib.MoxOutputTestCase): |
| """Test bug tracker communication.""" |
| |
| def testConnectEmail(self): |
| source = 'TheSource' |
| token = 'TheToken' |
| creds = gdata_lib.Creds() |
| creds.user = 'dude' |
| creds.password = 'shhh' |
| creds.tracker_auth_token = None |
| self.mox.StubOutClassWithMocks(gd_ph_client, 'ProjectHostingClient') |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| |
| def set_token(*_args, **_kwargs): |
| mocked_itclient.auth_token = cros_test_lib.EasyAttr(token_string=token) |
| |
| # Replay script |
| mocked_itclient = gd_ph_client.ProjectHostingClient() |
| mocked_itclient.ClientLogin( |
| creds.user, creds.password, source=source, service='code', |
| account_type='GOOGLE').WithSideEffects(set_token) |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| gdata_lib.TrackerComm.Connect(mocked_tcomm, creds, 'TheProject', |
| source=source) |
| self.mox.VerifyAll() |
| self.assertEquals(mocked_tcomm.it_client, mocked_itclient) |
| |
| def testConnectToken(self): |
| source = 'TheSource' |
| token = 'TheToken' |
| creds = gdata_lib.Creds() |
| creds.user = 'dude' |
| creds.password = 'shhh' |
| creds.tracker_auth_token = token |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| |
| self.mox.StubOutClassWithMocks(gd_ph_client, 'ProjectHostingClient') |
| self.mox.StubOutClassWithMocks(gdata.gauth, 'ClientLoginToken') |
| |
| # Replay script |
| mocked_itclient = gd_ph_client.ProjectHostingClient() |
| mocked_token = gdata.gauth.ClientLoginToken(token) |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| gdata_lib.TrackerComm.Connect(mocked_tcomm, creds, 'TheProject', |
| source=source) |
| self.mox.VerifyAll() |
| self.assertEquals(mocked_tcomm.it_client, mocked_itclient) |
| self.assertEquals(mocked_itclient.auth_token, mocked_token) |
| |
| def testGetTrackerIssueById(self): |
| mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient) |
| tcomm = gdata_lib.TrackerComm() |
| tcomm.it_client = mocked_itclient |
| tcomm.project_name = 'TheProject' |
| |
| self.mox.StubOutClassWithMocks(gd_ph_client, 'Query') |
| self.mox.StubOutClassWithMocks(gdata_lib, 'Issue') |
| self.mox.StubOutWithMock(gdata_lib.Issue, 'InitFromTracker') |
| |
| issue_id = 12345 |
| feed = cros_test_lib.EasyAttr(entry=['hi', 'there']) |
| |
| # Replay script |
| mocked_query = gd_ph_client.Query(issue_id=str(issue_id)) |
| mocked_itclient.get_issues( |
| 'TheProject', query=mocked_query).AndReturn(feed) |
| mocked_issue = gdata_lib.Issue() |
| mocked_issue.InitFromTracker(feed.entry[0], 'TheProject') |
| self.mox.ReplayAll() |
| |
| # Verify |
| issue = tcomm.GetTrackerIssueById(issue_id) |
| self.mox.VerifyAll() |
| self.assertEquals(mocked_issue, issue) |
| |
| def testGetTrackerIssuesByText(self): |
| author = 'TheAuthor' |
| project = 'TheProject' |
| text = "find me" |
| |
| # Set up the fake tracker issue. |
| tissue_id = 1 |
| tissue_labels = ['auto-filed'] |
| tissue_owner = 'someone@chromium.org' |
| tissue_status = 'Available' |
| tissue_content = 'find me in body' |
| tissue_title = 'find me in title' |
| |
| tissue = createTrackerIssue(tid=tissue_id, labels=tissue_labels, |
| owner=tissue_owner, status=tissue_status, |
| content=tissue_content, title=tissue_title) |
| |
| issue = gdata_lib.Issue(id=tissue_id, labels=tissue_labels, |
| owner=tissue_owner, status=tissue_status, |
| title=tissue_title, summary=tissue_content) |
| |
| # This will get called as part of Issue.InitFromTracker. |
| self.mox.StubOutWithMock(gdata_lib.Issue, 'GetTrackerIssueComments') |
| |
| mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient) |
| |
| tcomm = gdata_lib.TrackerComm() |
| tcomm.author = author |
| tcomm.it_client = mocked_itclient |
| tcomm.project_name = project |
| |
| # We expect a Query instance to be passed into get_issues. |
| # pylint: disable=E1120 |
| self.mox.StubOutClassWithMocks(gd_ph_client, 'Query') |
| |
| mocked_query = gd_ph_client.Query(text_query='%s is:open' % text) |
| feed = cros_test_lib.EasyAttr(entry=[tissue]) |
| mocked_itclient.get_issues(project, query=mocked_query).AndReturn(feed) |
| gdata_lib.Issue.GetTrackerIssueComments(1, project).AndReturn([]) |
| |
| self.mox.ReplayAll() |
| |
| issues = tcomm.GetTrackerIssuesByText(text) |
| self.assertEquals(issues, [issue]) |
| |
| def testCreateTrackerIssue(self): |
| author = 'TheAuthor' |
| mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_tcomm.author = author |
| mocked_tcomm.it_client = mocked_itclient |
| mocked_tcomm.project_name = 'TheProject' |
| |
| issue = cros_test_lib.EasyAttr(title='TheTitle', |
| summary='TheSummary', |
| status='TheStatus', |
| owner='TheOwner', |
| labels='TheLabels', |
| ccs=[]) |
| |
| # Replay script |
| issue_id = cros_test_lib.EasyAttr( |
| id=cros_test_lib.EasyAttr(text='foo/bar/123')) |
| mocked_itclient.add_issue( |
| project_name='TheProject', |
| title=issue.title, |
| content=issue.summary, |
| author=author, |
| status=issue.status, |
| owner=issue.owner, |
| labels=issue.labels, |
| ccs=issue.ccs).AndReturn(issue_id) |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = gdata_lib.TrackerComm.CreateTrackerIssue(mocked_tcomm, issue) |
| self.mox.VerifyAll() |
| self.assertEquals(123, result) |
| |
| def testAppendTrackerIssueById(self): |
| author = 'TheAuthor' |
| project_name = 'TheProject' |
| mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_tcomm.author = author |
| mocked_tcomm.it_client = mocked_itclient |
| mocked_tcomm.project_name = project_name |
| |
| issue_id = 54321 |
| comment = 'TheComment' |
| |
| # Replay script |
| mocked_itclient.update_issue(project_name=project_name, |
| issue_id=issue_id, |
| author=author, |
| comment=comment, |
| owner=None) |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = gdata_lib.TrackerComm.AppendTrackerIssueById(mocked_tcomm, |
| issue_id, comment) |
| self.mox.VerifyAll() |
| self.assertEquals(issue_id, result) |
| |
| |
| class RetrySpreadsheetsServiceTest(cros_test_lib.MoxOutputTestCase): |
| """Test Spreadsheet server retry helper.""" |
| |
| def testRequest(self): |
| """Test that calling request method invokes _RetryRequest wrapper.""" |
| # pylint: disable=W0212 |
| |
| self.mox.StubOutWithMock(gdata_lib.RetrySpreadsheetsService, |
| '_RetryRequest') |
| |
| # Use a real RetrySpreadsheetsService object rather than a mocked |
| # one, because the .request method only exists if __init__ is run. |
| # Also split up __new__ and __init__ in order to grab the original |
| # rss.request method (inherited from base class at that point). |
| rss = gdata_lib.RetrySpreadsheetsService.__new__( |
| gdata_lib.RetrySpreadsheetsService) |
| orig_request = rss.request |
| rss.__init__() |
| |
| args = ('GET', 'http://foo.bar') |
| |
| # This is the replay script for the test. |
| gdata_lib.RetrySpreadsheetsService._RetryRequest( |
| orig_request, *args).AndReturn('wrapped') |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| retval = rss.request(*args) |
| self.mox.VerifyAll() |
| self.assertEquals('wrapped', retval) |
| |
| def _TestHttpClientRetryRequest(self, statuses): |
| """Test retry logic in http_client request during ProgrammaticLogin. |
| |
| |statuses| is list of http codes to simulate, where 200 means success. |
| """ |
| expect_success = statuses[-1] == 200 |
| |
| self.mox.StubOutWithMock(atom.http.ProxiedHttpClient, 'request') |
| rss = gdata_lib.RetrySpreadsheetsService() |
| |
| args = ('POST', 'https://www.google.com/accounts/ClientLogin') |
| def _read(): |
| return 'Some response text' |
| |
| # This is the replay script for the test. |
| # Simulate the return codes in statuses. |
| for status in statuses: |
| retstatus = cros_test_lib.EasyAttr(status=status, read=_read) |
| atom.http.ProxiedHttpClient.request( |
| *args, data=mox.IgnoreArg(), |
| headers=mox.IgnoreArg()).AndReturn(retstatus) |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| if expect_success: |
| rss.ProgrammaticLogin() |
| else: |
| self.assertRaises(gdata.service.Error, rss.ProgrammaticLogin) |
| self.mox.VerifyAll() |
| |
| if not expect_success: |
| # Retries did not help, request still failed. |
| regexp = re.compile(r'^Giving up on HTTP request') |
| self.AssertOutputContainsWarning(regexp=regexp) |
| elif len(statuses) > 1: |
| # Warning expected if retries were needed. |
| self.AssertOutputContainsWarning() |
| else: |
| # First try worked, expect no warnings. |
| self.AssertOutputContainsWarning(invert=True) |
| |
| def testHttpClientRetryRequest(self): |
| self._TestHttpClientRetryRequest([200]) |
| |
| def testHttpClientRetryRequest403(self): |
| self._TestHttpClientRetryRequest([403, 200]) |
| |
| def testHttpClientRetryRequest403x2(self): |
| self._TestHttpClientRetryRequest([403, 403, 200]) |
| |
| def testHttpClientRetryRequest403x3(self): |
| self._TestHttpClientRetryRequest([403, 403, 403, 200]) |
| |
| def testHttpClientRetryRequest403x4(self): |
| self._TestHttpClientRetryRequest([403, 403, 403, 403, 200]) |
| |
| def testHttpClientRetryRequest403x5(self): |
| # This one should exhaust the retries. |
| self._TestHttpClientRetryRequest([403, 403, 403, 403, 403]) |
| |
| def _TestRetryRequest(self, statuses): |
| """Test retry logic for request method. |
| |
| |statuses| is list of http codes to simulate, where 200 means success. |
| """ |
| expect_success = statuses[-1] == 200 |
| expected_status_index = len(statuses) - 1 if expect_success else 0 |
| |
| mocked_ss = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService) |
| args = ('GET', 'http://foo.bar') |
| |
| # This is the replay script for the test. |
| for ix, status in enumerate(statuses): |
| # Add index of status to track which status the request function is |
| # returning. It is expected to return the last return status if |
| # successful (retries or not), but first return status if failed. |
| retval = cros_test_lib.EasyAttr(status=status, index=ix) |
| mocked_ss.request(*args).AndReturn(retval) |
| |
| self.mox.ReplayAll() |
| |
| # This is the test verification. |
| with self.OutputCapturer(): |
| # pylint: disable=W0212 |
| rval = gdata_lib.RetrySpreadsheetsService._RetryRequest(mocked_ss, |
| mocked_ss.request, |
| *args) |
| self.mox.VerifyAll() |
| self.assertEquals(statuses[expected_status_index], rval.status) |
| self.assertEquals(expected_status_index, rval.index) |
| |
| if not expect_success: |
| # Retries did not help, request still failed. |
| regexp = re.compile(r'^Giving up on HTTP request') |
| self.AssertOutputContainsWarning(regexp=regexp) |
| elif expected_status_index > 0: |
| # Warning expected if retries were needed. |
| self.AssertOutputContainsWarning() |
| else: |
| # First try worked, expect no warnings. |
| self.AssertOutputContainsWarning(invert=True) |
| |
| def testRetryRequest(self): |
| self._TestRetryRequest([200]) |
| |
| def testRetryRequest403(self): |
| self._TestRetryRequest([403, 200]) |
| |
| def testRetryRequest403x2(self): |
| self._TestRetryRequest([403, 403, 200]) |
| |
| def testRetryRequest403x3(self): |
| self._TestRetryRequest([403, 403, 403, 200]) |
| |
| def testRetryRequest403x4(self): |
| self._TestRetryRequest([403, 403, 403, 403, 200]) |
| |
| def testRetryRequest403x5(self): |
| # This one should exhaust the retries. |
| self._TestRetryRequest([403, 403, 403, 403, 403]) |
