src/third_party/chromite/lib/gdata_lib_unittest.py

   1 #!/usr/bin/python
   2 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """Unit tests for the gdata_lib module."""
   7
   8 import getpass
   9 import mox
  10 import os
  11 import re
  12 import sys
  13
  14 # pylint: disable=F0401
  15 import atom.service
  16 import gdata.projecthosting.client as gd_ph_client
  17 import gdata.spreadsheet.service
  18 # pylint: enable=F0401
  19
  20 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
  21     os.path.abspath(__file__)))))
  22
  23 from chromite.lib import cros_test_lib
  24 from chromite.lib import gdata_lib
  25 from chromite.lib import osutils
  26
  27 # pylint: disable=W0201,W0212,E1101,R0904
  28
  29
  30 class GdataLibTest(cros_test_lib.OutputTestCase):
  31   """Tests for methods that escape/unescape strings for speadsheets."""
  32
  33   def testPrepColNameForSS(self):
  34     tests = {
  35       'foo': 'foo',
  36       'Foo': 'foo',
  37       'FOO': 'foo',
  38       'foo bar': 'foobar',
  39       'Foo Bar': 'foobar',
  40       'F O O B A R': 'foobar',
  41       'Foo/Bar': 'foobar',
  42       'Foo Bar/Dude': 'foobardude',
  43       'foo/bar': 'foobar',
  44       }
  45
  46     for col in tests:
  47       expected = tests[col]
  48       self.assertEquals(expected, gdata_lib.PrepColNameForSS(col))
  49       self.assertEquals(expected, gdata_lib.PrepColNameForSS(expected))
  50
  51   def testPrepValForSS(self):
  52     tests = {
  53       None: None,
  54       '': '',
  55       'foo': 'foo',
  56       'foo123': 'foo123',
  57       '123': "'123",
  58       '1.2': "'1.2",
  59       }
  60
  61     for val in tests:
  62       expected = tests[val]
  63       self.assertEquals(expected, gdata_lib.PrepValForSS(val))
  64
  65   def testPrepRowForSS(self):
  66     vals = {
  67       None: None,
  68       '': '',
  69       'foo': 'foo',
  70       'foo123': 'foo123',
  71       '123': "'123",
  72       '1.2': "'1.2",
  73       }
  74
  75     # Create before and after rows (rowIn, rowOut).
  76     rowIn = {}
  77     rowOut = {}
  78     col = 'a' # Column names not important.
  79     for valIn in vals:
  80       valOut = vals[valIn]
  81       rowIn[col] = valIn
  82       rowOut[col] = valOut
  83
  84       col = chr(ord(col) + 1) # Change column name.
  85
  86     self.assertEquals(rowOut, gdata_lib.PrepRowForSS(rowIn))
  87
  88   def testScrubValFromSS(self):
  89     tests = {
  90       None: None,
  91       'foo': 'foo',
  92       '123': '123',
  93       "'123": '123',
  94       }
  95
  96     for val in tests:
  97       expected = tests[val]
  98       self.assertEquals(expected, gdata_lib.ScrubValFromSS(val))
  99
 100
 101 class CredsTest(cros_test_lib.MoxOutputTestCase):
 102   """Tests related to user credentials."""
 103
 104   USER = 'somedude@chromium.org'
 105   PASSWORD = 'worldsbestpassword'
 106   DOCS_TOKEN = 'SomeDocsAuthToken'
 107   TRACKER_TOKEN = 'SomeTrackerAuthToken'
 108
 109   @osutils.TempFileDecorator
 110   def testStoreLoadCreds(self):
 111     # This is the replay script for the test.
 112     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 113     self.mox.ReplayAll()
 114
 115     # This is the test verification.
 116     with self.OutputCapturer():
 117       gdata_lib.Creds.SetCreds(mocked_creds, self.USER, self.PASSWORD)
 118       self.assertEquals(self.USER, mocked_creds.user)
 119       self.assertEquals(self.PASSWORD, mocked_creds.password)
 120       self.assertTrue(mocked_creds.creds_dirty)
 121
 122       gdata_lib.Creds.StoreCreds(mocked_creds, self.tempfile)
 123       self.assertEquals(self.USER, mocked_creds.user)
 124       self.assertEquals(self.PASSWORD, mocked_creds.password)
 125       self.assertFalse(mocked_creds.creds_dirty)
 126
 127       # Clear user/password before loading from just-created file.
 128       mocked_creds.user = None
 129       mocked_creds.password = None
 130       self.assertEquals(None, mocked_creds.user)
 131       self.assertEquals(None, mocked_creds.password)
 132
 133       gdata_lib.Creds.LoadCreds(mocked_creds, self.tempfile)
 134       self.assertEquals(self.USER, mocked_creds.user)
 135       self.assertEquals(self.PASSWORD, mocked_creds.password)
 136       self.assertFalse(mocked_creds.creds_dirty)
 137
 138     self.mox.VerifyAll()
 139
 140   @osutils.TempFileDecorator
 141   def testStoreLoadToken(self):
 142     # This is the replay script for the test.
 143     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 144     mocked_creds.user = self.USER
 145     self.mox.ReplayAll()
 146
 147     # This is the test verification.
 148     with self.OutputCapturer():
 149       gdata_lib.Creds.SetDocsAuthToken(mocked_creds, self.DOCS_TOKEN)
 150       self.assertEquals(self.DOCS_TOKEN, mocked_creds.docs_auth_token)
 151       self.assertTrue(mocked_creds.token_dirty)
 152       gdata_lib.Creds.SetTrackerAuthToken(mocked_creds, self.TRACKER_TOKEN)
 153       self.assertEquals(self.TRACKER_TOKEN, mocked_creds.tracker_auth_token)
 154       self.assertTrue(mocked_creds.token_dirty)
 155
 156       gdata_lib.Creds.StoreAuthToken(mocked_creds, self.tempfile)
 157       self.assertEquals(self.DOCS_TOKEN, mocked_creds.docs_auth_token)
 158       self.assertEquals(self.TRACKER_TOKEN, mocked_creds.tracker_auth_token)
 159       self.assertFalse(mocked_creds.token_dirty)
 160
 161       # Clear auth_tokens before loading from just-created file.
 162       mocked_creds.docs_auth_token = None
 163       mocked_creds.tracker_auth_token = None
 164       mocked_creds.user = None
 165
 166       gdata_lib.Creds.LoadAuthToken(mocked_creds, self.tempfile)
 167       self.assertEquals(self.DOCS_TOKEN, mocked_creds.docs_auth_token)
 168       self.assertEquals(self.TRACKER_TOKEN, mocked_creds.tracker_auth_token)
 169       self.assertFalse(mocked_creds.token_dirty)
 170       self.assertEquals(self.USER, mocked_creds.user)
 171
 172     self.mox.VerifyAll()
 173
 174   def testSetCreds(self):
 175     # This is the replay script for the test.
 176     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 177     self.mox.ReplayAll()
 178
 179     # This is the test verification.
 180     gdata_lib.Creds.SetCreds(mocked_creds, self.USER,
 181                              password=self.PASSWORD)
 182     self.mox.VerifyAll()
 183     self.assertEquals(self.USER, mocked_creds.user)
 184     self.assertEquals(self.PASSWORD, mocked_creds.password)
 185     self.assertTrue(mocked_creds.creds_dirty)
 186
 187   def testSetCredsNoPassword(self):
 188     # Add test-specific mocks/stubs
 189     self.mox.StubOutWithMock(getpass, 'getpass')
 190
 191     # This is the replay script for the test.
 192     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 193     getpass.getpass(mox.IgnoreArg()).AndReturn(self.PASSWORD)
 194     self.mox.ReplayAll()
 195
 196     # This is the test verification.
 197     gdata_lib.Creds.SetCreds(mocked_creds, self.USER)
 198     self.mox.VerifyAll()
 199     self.assertEquals(self.USER, mocked_creds.user)
 200     self.assertEquals(self.PASSWORD, mocked_creds.password)
 201     self.assertTrue(mocked_creds.creds_dirty)
 202
 203   def testSetDocsToken(self):
 204     # This is the replay script for the test.
 205     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 206     self.mox.ReplayAll()
 207
 208     # This is the test verification.
 209     gdata_lib.Creds.SetDocsAuthToken(mocked_creds, self.DOCS_TOKEN)
 210     self.mox.VerifyAll()
 211     self.assertEquals(self.DOCS_TOKEN, mocked_creds.docs_auth_token)
 212     self.assertTrue(mocked_creds.token_dirty)
 213
 214   def testSetTrackerToken(self):
 215     # This is the replay script for the test.
 216     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 217     self.mox.ReplayAll()
 218
 219     # This is the test verification.
 220     gdata_lib.Creds.SetTrackerAuthToken(mocked_creds, self.TRACKER_TOKEN)
 221     self.mox.VerifyAll()
 222     self.assertEquals(self.TRACKER_TOKEN, mocked_creds.tracker_auth_token)
 223     self.assertTrue(mocked_creds.token_dirty)
 224
 225
 226 class SpreadsheetRowTest(cros_test_lib.OutputTestCase):
 227   """Tests related to spreadsheet row interaction."""
 228
 229   SS_ROW_OBJ = 'SSRowObj'
 230   SS_ROW_NUM = 5
 231
 232   def testEmpty(self):
 233     row = gdata_lib.SpreadsheetRow(self.SS_ROW_OBJ, self.SS_ROW_NUM)
 234
 235     self.assertEquals(0, len(row))
 236     self.assertEquals(self.SS_ROW_OBJ, row.ss_row_obj)
 237     self.assertEquals(self.SS_ROW_NUM, row.ss_row_num)
 238
 239     self.assertRaises(TypeError, row, '__setitem__', 'abc', 'xyz')
 240     self.assertEquals(0, len(row))
 241     self.assertFalse('abc' in row)
 242
 243   def testInit(self):
 244     starting_vals = {'abc': 'xyz', 'foo': 'bar'}
 245     row = gdata_lib.SpreadsheetRow(self.SS_ROW_OBJ, self.SS_ROW_NUM,
 246                                    starting_vals)
 247
 248     self.assertEquals(len(starting_vals), len(row))
 249     self.assertEquals(starting_vals, row)
 250     self.assertEquals(row['abc'], 'xyz')
 251     self.assertTrue('abc' in row)
 252     self.assertEquals(row['foo'], 'bar')
 253     self.assertTrue('foo' in row)
 254
 255     self.assertEquals(self.SS_ROW_OBJ, row.ss_row_obj)
 256     self.assertEquals(self.SS_ROW_NUM, row.ss_row_num)
 257
 258     self.assertRaises(TypeError, row, '__delitem__', 'abc')
 259     self.assertEquals(len(starting_vals), len(row))
 260     self.assertTrue('abc' in row)
 261
 262
 263 class SpreadsheetCommTest(cros_test_lib.MoxOutputTestCase):
 264   """Test Speadsheet communication."""
 265
 266   SS_KEY = 'TheSSKey'
 267   WS_NAME = 'TheWSName'
 268   WS_KEY = 'TheWSKey'
 269
 270   USER = 'dude'
 271   PASSWORD = 'shhh'
 272   TOKEN = 'authtoken'
 273
 274   COLUMNS = ('greeting', 'name', 'title')
 275   ROWS = (
 276       { 'greeting': 'Hi', 'name': 'George', 'title': 'Mr.' },
 277       { 'greeting': 'Howdy', 'name': 'Billy Bob', 'title': 'Mr.' },
 278       { 'greeting': 'Yo', 'name': 'Adriane', 'title': 'Ms.' },
 279       )
 280
 281   def MockScomm(self, connect=True):
 282     """Return a mocked SpreadsheetComm"""
 283     mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm)
 284
 285     mocked_scomm._columns = None
 286     mocked_scomm._rows = None
 287
 288     if connect:
 289       mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService)
 290       mocked_scomm.gd_client = mocked_gdclient
 291       mocked_scomm.ss_key = self.SS_KEY
 292       mocked_scomm.ws_name = self.WS_NAME
 293       mocked_scomm.ws_key = self.WS_KEY
 294     else:
 295       mocked_scomm.gd_client = None
 296       mocked_scomm.ss_key = None
 297       mocked_scomm.ws_name = None
 298       mocked_scomm.ws_key = None
 299
 300     return mocked_scomm
 301
 302   def NewScomm(self, gd_client=None, connect=True):
 303     """Return a non-mocked SpreadsheetComm."""
 304     scomm = gdata_lib.SpreadsheetComm()
 305     scomm.gd_client = gd_client
 306
 307     if connect:
 308       scomm.ss_key = self.SS_KEY
 309       scomm.ws_name = self.WS_NAME
 310       scomm.ws_key = self.WS_KEY
 311     else:
 312       scomm.ss_key = None
 313       scomm.ws_name = None
 314       scomm.ws_key = None
 315
 316     return scomm
 317
 318   def GenerateCreds(self, skip_user=False, skip_token=False):
 319     creds = gdata_lib.Creds()
 320     if not skip_user:
 321       creds.user = self.USER
 322       creds.password = self.PASSWORD
 323
 324     if not skip_token:
 325       creds.docs_auth_token = self.TOKEN
 326
 327     return creds
 328
 329   def testConnect(self):
 330     mocked_scomm = self.MockScomm(connect=False)
 331     creds = self.GenerateCreds()
 332
 333     # This is the replay script for the test.
 334     mocked_scomm._Login(creds, 'chromiumos')
 335     mocked_scomm.SetCurrentWorksheet(self.WS_NAME, ss_key=self.SS_KEY)
 336     self.mox.ReplayAll()
 337
 338     # This is the test verification.
 339     gdata_lib.SpreadsheetComm.Connect(mocked_scomm, creds,
 340                                       self.SS_KEY, self.WS_NAME)
 341     self.mox.VerifyAll()
 342
 343   def testColumns(self):
 344     """Test the .columns property.  Testing a property gets ugly."""
 345     self.mox.StubOutWithMock(gdata.spreadsheet.service, 'CellQuery')
 346     mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService)
 347     scomm = self.NewScomm(gd_client=mocked_gdclient, connect=True)
 348
 349     query = {'max-row': '1'}
 350
 351     # Simulate a Cells feed from spreadsheet for the column row.
 352     cols = [c[0].upper() + c[1:] for c in self.COLUMNS]
 353     entry = [cros_test_lib.EasyAttr(
 354         content=cros_test_lib.EasyAttr(text=c)) for c in cols]
 355     feed = cros_test_lib.EasyAttr(entry=entry)
 356
 357     # This is the replay script for the test.
 358     gdata.spreadsheet.service.CellQuery().AndReturn(query)
 359     mocked_gdclient.GetCellsFeed(self.SS_KEY, self.WS_KEY, query=query
 360                                  ).AndReturn(feed)
 361     self.mox.ReplayAll()
 362
 363     # This is the test verification.
 364     result = scomm.columns
 365     del scomm # Force deletion now before VerifyAll.
 366     self.mox.VerifyAll()
 367
 368     expected_result = self.COLUMNS
 369     self.assertEquals(expected_result, result)
 370
 371   def testRows(self):
 372     """Test the .rows property.  Testing a property gets ugly."""
 373     mocked_gdclient = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService)
 374     scomm = self.NewScomm(gd_client=mocked_gdclient, connect=True)
 375
 376     # Simulate a List feed from spreadsheet for all rows.
 377     rows = [{'col_name': 'Joe', 'col_age': '12', 'col_zip': '12345'},
 378             {'col_name': 'Bob', 'col_age': '15', 'col_zip': '54321'},
 379             ]
 380     entry = []
 381     for row in rows:
 382       custom = dict((k, cros_test_lib.EasyAttr(text=v))
 383                     for (k, v) in row.iteritems())
 384       entry.append(cros_test_lib.EasyAttr(custom=custom))
 385     feed = cros_test_lib.EasyAttr(entry=entry)
 386
 387     # This is the replay script for the test.
 388     mocked_gdclient.GetListFeed(self.SS_KEY, self.WS_KEY).AndReturn(feed)
 389     self.mox.ReplayAll()
 390
 391     # This is the test verification.
 392     result = scomm.rows
 393     del scomm # Force deletion now before VerifyAll.
 394     self.mox.VerifyAll()
 395     self.assertEquals(tuple(rows), result)
 396
 397     # Result tuple should have spreadsheet row num as attribute on each row.
 398     self.assertEquals(2, result[0].ss_row_num)
 399     self.assertEquals(3, result[1].ss_row_num)
 400
 401     # Result tuple should have spreadsheet row obj as attribute on each row.
 402     self.assertEquals(entry[0], result[0].ss_row_obj)
 403     self.assertEquals(entry[1], result[1].ss_row_obj)
 404
 405   def testSetCurrentWorksheetStart(self):
 406     mocked_scomm = self.MockScomm(connect=True)
 407
 408     # Undo worksheet settings.
 409     mocked_scomm.ss_key = None
 410     mocked_scomm.ws_name = None
 411     mocked_scomm.ws_key = None
 412
 413     # This is the replay script for the test.
 414     mocked_scomm._ClearCache()
 415     mocked_scomm._GetWorksheetKey(self.SS_KEY, self.WS_NAME
 416                                   ).AndReturn(self.WS_KEY)
 417     mocked_scomm._ClearCache()
 418     self.mox.ReplayAll()
 419
 420     # This is the test verification.
 421     gdata_lib.SpreadsheetComm.SetCurrentWorksheet(mocked_scomm, self.WS_NAME,
 422                                                   ss_key=self.SS_KEY)
 423     self.mox.VerifyAll()
 424
 425     self.assertEquals(self.SS_KEY, mocked_scomm.ss_key)
 426     self.assertEquals(self.WS_KEY, mocked_scomm.ws_key)
 427     self.assertEquals(self.WS_NAME, mocked_scomm.ws_name)
 428
 429   def testSetCurrentWorksheetRestart(self):
 430     mocked_scomm = self.MockScomm(connect=True)
 431
 432     other_ws_name = 'OtherWSName'
 433     other_ws_key = 'OtherWSKey'
 434
 435     # This is the replay script for the test.
 436     mocked_scomm._GetWorksheetKey(self.SS_KEY, other_ws_name
 437                                   ).AndReturn(other_ws_key)
 438     mocked_scomm._ClearCache()
 439     self.mox.ReplayAll()
 440
 441     # This is the test verification.
 442     gdata_lib.SpreadsheetComm.SetCurrentWorksheet(mocked_scomm, other_ws_name)
 443     self.mox.VerifyAll()
 444
 445     self.assertEquals(self.SS_KEY, mocked_scomm.ss_key)
 446     self.assertEquals(other_ws_key, mocked_scomm.ws_key)
 447     self.assertEquals(other_ws_name, mocked_scomm.ws_name)
 448
 449   def testClearCache(self):
 450     rows = 'SomeRows'
 451     cols = 'SomeColumns'
 452
 453     scomm = self.NewScomm()
 454     scomm._rows = rows
 455     scomm._columns = cols
 456
 457     scomm._ClearCache(keep_columns=True)
 458     self.assertTrue(scomm._rows is None)
 459     self.assertEquals(cols, scomm._columns)
 460
 461     scomm._rows = rows
 462     scomm._columns = cols
 463
 464     scomm._ClearCache(keep_columns=False)
 465     self.assertTrue(scomm._rows is None)
 466     self.assertTrue(scomm._columns is None)
 467
 468     scomm._rows = rows
 469     scomm._columns = cols
 470
 471     scomm._ClearCache()
 472     self.assertTrue(scomm._rows is None)
 473     self.assertTrue(scomm._columns is None)
 474
 475   def testLoginWithUserPassword(self):
 476     mocked_scomm = self.MockScomm(connect=False)
 477     creds = self.GenerateCreds(skip_token=True)
 478
 479     self.mox.StubOutClassWithMocks(gdata_lib, 'RetrySpreadsheetsService')
 480
 481     source = 'SomeSource'
 482
 483     # This is the replay script for the test.
 484     mocked_gdclient = gdata_lib.RetrySpreadsheetsService()
 485     mocked_gdclient.ProgrammaticLogin()
 486     mocked_gdclient.GetClientLoginToken().AndReturn(self.TOKEN)
 487     self.mox.ReplayAll()
 488
 489     # This is the test verification.
 490     with self.OutputCapturer():
 491       gdata_lib.SpreadsheetComm._Login(mocked_scomm, creds, source)
 492     self.mox.VerifyAll()
 493     self.assertEquals(self.USER, mocked_gdclient.email)
 494     self.assertEquals(self.PASSWORD, mocked_gdclient.password)
 495     self.assertEquals(self.TOKEN, creds.docs_auth_token)
 496     self.assertEquals(source, mocked_gdclient.source)
 497     self.assertEquals(mocked_gdclient, mocked_scomm.gd_client)
 498
 499   def testLoginWithToken(self):
 500     mocked_scomm = self.MockScomm(connect=False)
 501     creds = self.GenerateCreds(skip_user=True)
 502
 503     self.mox.StubOutClassWithMocks(gdata_lib, 'RetrySpreadsheetsService')
 504
 505     source = 'SomeSource'
 506
 507     # This is the replay script for the test.
 508     mocked_gdclient = gdata_lib.RetrySpreadsheetsService()
 509     mocked_gdclient.SetClientLoginToken(creds.docs_auth_token)
 510     self.mox.ReplayAll()
 511
 512     # This is the test verification.
 513     with self.OutputCapturer():
 514       gdata_lib.SpreadsheetComm._Login(mocked_scomm, creds, source)
 515     self.mox.VerifyAll()
 516     self.assertFalse(hasattr(mocked_gdclient, 'email'))
 517     self.assertFalse(hasattr(mocked_gdclient, 'password'))
 518     self.assertEquals(source, mocked_gdclient.source)
 519     self.assertEquals(mocked_gdclient, mocked_scomm.gd_client)
 520
 521   def testGetWorksheetKey(self):
 522     mocked_scomm = self.MockScomm()
 523
 524     entrylist = [
 525       cros_test_lib.EasyAttr(
 526           title=cros_test_lib.EasyAttr(text='Foo'), id='NotImportant'),
 527       cros_test_lib.EasyAttr(
 528           title=cros_test_lib.EasyAttr(text=self.WS_NAME),
 529           id=cros_test_lib.EasyAttr(text='/some/path/%s' % self.WS_KEY)),
 530       cros_test_lib.EasyAttr(
 531           title=cros_test_lib.EasyAttr(text='Bar'), id='NotImportant'),
 532       ]
 533     feed = cros_test_lib.EasyAttr(entry=entrylist)
 534
 535     # This is the replay script for the test.
 536     mocked_scomm.gd_client.GetWorksheetsFeed(self.SS_KEY).AndReturn(feed)
 537     self.mox.ReplayAll()
 538
 539     # This is the test verification.
 540     gdata_lib.SpreadsheetComm._GetWorksheetKey(mocked_scomm,
 541                                                self.SS_KEY, self.WS_NAME)
 542     self.mox.VerifyAll()
 543
 544   def testGetColumns(self):
 545     mocked_scomm = self.MockScomm()
 546     mocked_scomm.columns = 'SomeColumns'
 547
 548     # Replay script
 549     self.mox.ReplayAll()
 550
 551     # This is the test verification.
 552     result = gdata_lib.SpreadsheetComm.GetColumns(mocked_scomm)
 553     self.mox.VerifyAll()
 554     self.assertEquals('SomeColumns', result)
 555
 556   def testGetColumnIndex(self):
 557     # Note that spreadsheet column indices start at 1.
 558     mocked_scomm = self.MockScomm()
 559     mocked_scomm.columns = ['these', 'are', 'column', 'names']
 560
 561     # This is the replay script for the test.
 562     self.mox.ReplayAll()
 563
 564     # This is the test verification.
 565     result = gdata_lib.SpreadsheetComm.GetColumnIndex(mocked_scomm, 'are')
 566     self.mox.VerifyAll()
 567     self.assertEquals(2, result)
 568
 569   def testGetRows(self):
 570     mocked_scomm = self.MockScomm()
 571     rows = []
 572     for row_ix, row_dict in enumerate(self.ROWS):
 573       rows.append(gdata_lib.SpreadsheetRow('SSRowObj%d' % (row_ix + 2),
 574                                            (row_ix + 2), row_dict))
 575     mocked_scomm.rows = tuple(rows)
 576
 577     # This is the replay script for the test.
 578     self.mox.ReplayAll()
 579
 580     # This is the test verification.
 581     result = gdata_lib.SpreadsheetComm.GetRows(mocked_scomm)
 582     self.mox.VerifyAll()
 583     self.assertEquals(self.ROWS, result)
 584     for row_ix in xrange(len(self.ROWS)):
 585       self.assertEquals(row_ix + 2, result[row_ix].ss_row_num)
 586       self.assertEquals('SSRowObj%d' % (row_ix + 2), result[row_ix].ss_row_obj)
 587
 588   def testGetRowCacheByCol(self):
 589     mocked_scomm = self.MockScomm()
 590
 591     # This is the replay script for the test.
 592     mocked_scomm.GetRows().AndReturn(self.ROWS)
 593     self.mox.ReplayAll()
 594
 595     # This is the test verification.
 596     result = gdata_lib.SpreadsheetComm.GetRowCacheByCol(mocked_scomm, 'name')
 597     self.mox.VerifyAll()
 598
 599     # Result is a dict of rows by the 'name' column.
 600     for row in self.ROWS:
 601       name = row['name']
 602       self.assertEquals(row, result[name])
 603
 604   def testGetRowCacheByColDuplicates(self):
 605     mocked_scomm = self.MockScomm()
 606
 607     # Create new row list with duplicates by name column.
 608     rows = []
 609     for row in self.ROWS:
 610       new_row = dict(row)
 611       new_row['greeting'] = row['greeting'] + ' there'
 612       rows.append(new_row)
 613
 614     rows.extend(self.ROWS)
 615
 616     # This is the replay script for the test.
 617     mocked_scomm.GetRows().AndReturn(tuple(rows))
 618     self.mox.ReplayAll()
 619
 620     # This is the test verification.
 621     result = gdata_lib.SpreadsheetComm.GetRowCacheByCol(mocked_scomm, 'name')
 622     self.mox.VerifyAll()
 623
 624     # Result is a dict of rows by the 'name' column.  In this
 625     # test each result should be a list of the rows with the same
 626     # value in the 'name' column.
 627     num_rows = len(rows)
 628     for ix in xrange(num_rows / 2):
 629       row1 = rows[ix]
 630       row2 = rows[ix + (num_rows / 2)]
 631
 632       name = row1['name']
 633       self.assertEquals(name, row2['name'])
 634
 635       expected_rows = [row1, row2]
 636       self.assertEquals(expected_rows, result[name])
 637
 638   def testInsertRow(self):
 639     mocked_scomm = self.MockScomm()
 640
 641     row = 'TheRow'
 642
 643     # Replay script
 644     mocked_scomm.gd_client.InsertRow(row, mocked_scomm.ss_key,
 645                                      mocked_scomm.ws_key)
 646     mocked_scomm._ClearCache(keep_columns=True)
 647     self.mox.ReplayAll()
 648
 649     # This is the test verification.
 650     gdata_lib.SpreadsheetComm.InsertRow(mocked_scomm, row)
 651     self.mox.VerifyAll()
 652
 653   def testUpdateRowCellByCell(self):
 654     mocked_scomm = self.MockScomm()
 655
 656     rowIx = 5
 657     row = {'a': 123, 'b': 234, 'c': 345}
 658     colIndices = {'a': 1, 'b': None, 'c': 4}
 659
 660     # Replay script
 661     for colName in row:
 662       colIx = colIndices[colName]
 663       mocked_scomm.GetColumnIndex(colName).AndReturn(colIx)
 664       if colIx is not None:
 665         mocked_scomm.ReplaceCellValue(rowIx, colIx, row[colName])
 666     mocked_scomm._ClearCache(keep_columns=True)
 667     self.mox.ReplayAll()
 668
 669     # This is the test verification.
 670     gdata_lib.SpreadsheetComm.UpdateRowCellByCell(mocked_scomm, rowIx, row)
 671     self.mox.VerifyAll()
 672
 673   def testDeleteRow(self):
 674     mocked_scomm = self.MockScomm()
 675
 676     ss_row = 'TheRow'
 677
 678     # Replay script
 679     mocked_scomm.gd_client.DeleteRow(ss_row)
 680     mocked_scomm._ClearCache(keep_columns=True)
 681     self.mox.ReplayAll()
 682
 683     # This is the test verification.
 684     gdata_lib.SpreadsheetComm.DeleteRow(mocked_scomm, ss_row)
 685     self.mox.VerifyAll()
 686
 687   def testReplaceCellValue(self):
 688     mocked_scomm = self.MockScomm()
 689
 690     rowIx = 14
 691     colIx = 4
 692     val = 'TheValue'
 693
 694     # Replay script
 695     mocked_scomm.gd_client.UpdateCell(rowIx, colIx, val,
 696                                       mocked_scomm.ss_key, mocked_scomm.ws_key)
 697     mocked_scomm._ClearCache(keep_columns=True)
 698     self.mox.ReplayAll()
 699
 700     # Verify
 701     gdata_lib.SpreadsheetComm.ReplaceCellValue(mocked_scomm, rowIx, colIx, val)
 702     self.mox.VerifyAll()
 703
 704   def testClearCellValue(self):
 705     mocked_scomm = self.MockScomm()
 706
 707     rowIx = 14
 708     colIx = 4
 709
 710     # Replay script
 711     mocked_scomm.ReplaceCellValue(rowIx, colIx, None)
 712     self.mox.ReplayAll()
 713
 714     # Verify
 715     gdata_lib.SpreadsheetComm.ClearCellValue(mocked_scomm, rowIx, colIx)
 716     self.mox.VerifyAll()
 717
 718
 719 class IssueCommentTest(cros_test_lib.TestCase):
 720   """Test creating comments."""
 721
 722   def testInit(self):
 723     title = 'Greetings, Earthlings'
 724     text = 'I come in peace.'
 725     ic = gdata_lib.IssueComment(title, text)
 726
 727     self.assertEquals(title, ic.title)
 728     self.assertEquals(text, ic.text)
 729
 730 def createTrackerIssue(tid, labels, owner, status, content, title):
 731   tissue = cros_test_lib.EasyAttr()
 732   tissue.id = cros_test_lib.EasyAttr(
 733       text='http://www/some/path/%d' % tid)
 734   tissue.label = [cros_test_lib.EasyAttr(text=l) for l in labels]
 735   tissue.owner = cros_test_lib.EasyAttr(
 736       username=cros_test_lib.EasyAttr(text=owner))
 737   tissue.status = cros_test_lib.EasyAttr(text=status)
 738   tissue.content = cros_test_lib.EasyAttr(text=content)
 739   tissue.title = cros_test_lib.EasyAttr(text=title)
 740   return tissue
 741
 742 class IssueTest(cros_test_lib.MoxTestCase):
 743   """Test creating a bug."""
 744
 745   def testInitOverride(self):
 746     owner = 'somedude@chromium.org'
 747     status = 'Assigned'
 748     issue = gdata_lib.Issue(owner=owner, status=status)
 749
 750     self.assertEquals(owner, issue.owner)
 751     self.assertEquals(status, issue.status)
 752
 753   def testInitInvalidOverride(self):
 754     self.assertRaises(ValueError, gdata_lib.Issue,
 755                       foobar='NotARealAttr')
 756
 757   def testInitFromTracker(self):
 758     # Need to create a dummy Tracker Issue object.
 759     tissue_id = 123
 760     tissue_labels = ['Iteration-10', 'Effort-2']
 761     tissue_owner = 'thedude@chromium.org'
 762     tissue_status = 'Available'
 763     tissue_content = 'The summary message'
 764     tissue_title = 'The Big Title'
 765
 766     tissue = createTrackerIssue(tid=tissue_id, labels=tissue_labels,
 767                                 owner=tissue_owner, status=tissue_status,
 768                                 content=tissue_content, title=tissue_title)
 769
 770     mocked_issue = self.mox.CreateMock(gdata_lib.Issue)
 771
 772     # Replay script
 773     mocked_issue.GetTrackerIssueComments(tissue_id, 'TheProject').AndReturn([])
 774     self.mox.ReplayAll()
 775
 776     # Verify
 777     gdata_lib.Issue.InitFromTracker(mocked_issue, tissue, 'TheProject')
 778     self.mox.VerifyAll()
 779     self.assertEquals(tissue_id, mocked_issue.id)
 780     self.assertEquals(tissue_labels, mocked_issue.labels)
 781     self.assertEquals(tissue_owner, mocked_issue.owner)
 782     self.assertEquals(tissue_status, mocked_issue.status)
 783     self.assertEquals(tissue_content, mocked_issue.summary)
 784     self.assertEquals(tissue_title, mocked_issue.title)
 785     self.assertEquals([], mocked_issue.comments)
 786
 787
 788 class TrackerCommTest(cros_test_lib.MoxOutputTestCase):
 789   """Test bug tracker communication."""
 790
 791   def testConnectEmail(self):
 792     source = 'TheSource'
 793     token = 'TheToken'
 794     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 795     mocked_creds.user = 'dude'
 796     mocked_creds.password = 'shhh'
 797     mocked_creds.tracker_auth_token = None
 798     self.mox.StubOutClassWithMocks(gd_ph_client, 'ProjectHostingClient')
 799     mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm)
 800
 801     def set_token(*_args, **_kwargs):
 802       mocked_itclient.auth_token = cros_test_lib.EasyAttr(token_string=token)
 803
 804     # Replay script
 805     mocked_itclient = gd_ph_client.ProjectHostingClient()
 806     mocked_itclient.ClientLogin(mocked_creds.user, mocked_creds.password,
 807                                 source=source, service='code',
 808                                 account_type='GOOGLE'
 809                                 ).WithSideEffects(set_token)
 810     mocked_creds.SetTrackerAuthToken(token)
 811     self.mox.ReplayAll()
 812
 813     # Verify
 814     with self.OutputCapturer():
 815       gdata_lib.TrackerComm.Connect(mocked_tcomm, mocked_creds, 'TheProject',
 816                                     source=source)
 817     self.mox.VerifyAll()
 818     self.assertEquals(mocked_tcomm.it_client, mocked_itclient)
 819
 820   def testConnectToken(self):
 821     source = 'TheSource'
 822     token = 'TheToken'
 823     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 824     mocked_creds.user = 'dude'
 825     mocked_creds.password = 'shhh'
 826     mocked_creds.tracker_auth_token = token
 827     mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm)
 828
 829     self.mox.StubOutClassWithMocks(gd_ph_client, 'ProjectHostingClient')
 830     self.mox.StubOutClassWithMocks(gdata.gauth, 'ClientLoginToken')
 831
 832     # Replay script
 833     mocked_itclient = gd_ph_client.ProjectHostingClient()
 834     mocked_token = gdata.gauth.ClientLoginToken(token)
 835     self.mox.ReplayAll()
 836
 837     # Verify
 838     with self.OutputCapturer():
 839       gdata_lib.TrackerComm.Connect(mocked_tcomm, mocked_creds, 'TheProject',
 840                                     source=source)
 841     self.mox.VerifyAll()
 842     self.assertEquals(mocked_tcomm.it_client, mocked_itclient)
 843     self.assertEquals(mocked_itclient.auth_token, mocked_token)
 844
 845   def testGetTrackerIssueById(self):
 846     mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient)
 847     tcomm = gdata_lib.TrackerComm()
 848     tcomm.it_client = mocked_itclient
 849     tcomm.project_name = 'TheProject'
 850
 851     self.mox.StubOutClassWithMocks(gd_ph_client, 'Query')
 852     self.mox.StubOutClassWithMocks(gdata_lib, 'Issue')
 853     self.mox.StubOutWithMock(gdata_lib.Issue, 'InitFromTracker')
 854
 855     issue_id = 12345
 856     feed = cros_test_lib.EasyAttr(entry=['hi', 'there'])
 857
 858     # Replay script
 859     mocked_query = gd_ph_client.Query(issue_id=str(issue_id))
 860     mocked_itclient.get_issues('TheProject', query=mocked_query
 861                                ).AndReturn(feed)
 862     mocked_issue = gdata_lib.Issue()
 863     mocked_issue.InitFromTracker(feed.entry[0], 'TheProject')
 864     self.mox.ReplayAll()
 865
 866     # Verify
 867     issue = tcomm.GetTrackerIssueById(issue_id)
 868     self.mox.VerifyAll()
 869     self.assertEquals(mocked_issue, issue)
 870
 871   def testGetTrackerIssuesByText(self):
 872     author = 'TheAuthor'
 873     project = 'TheProject'
 874     text = "find me"
 875
 876     # Set up the fake tracker issue.
 877     tissue_id = 1
 878     tissue_labels = ['auto-filed']
 879     tissue_owner = 'someone@chromium.org'
 880     tissue_status = 'Available'
 881     tissue_content = 'find me in body'
 882     tissue_title = 'find me in title'
 883
 884     tissue = createTrackerIssue(tid=tissue_id, labels=tissue_labels,
 885                                 owner=tissue_owner, status=tissue_status,
 886                                 content=tissue_content, title=tissue_title)
 887
 888     issue = gdata_lib.Issue(id=tissue_id, labels=tissue_labels,
 889                             owner=tissue_owner, status=tissue_status,
 890                             title=tissue_title, summary=tissue_content)
 891
 892     # This will get called as part of Issue.InitFromTracker.
 893     self.mox.StubOutWithMock(gdata_lib.Issue, 'GetTrackerIssueComments')
 894
 895     mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient)
 896
 897     tcomm = gdata_lib.TrackerComm()
 898     tcomm.author = author
 899     tcomm.it_client = mocked_itclient
 900     tcomm.project_name = project
 901
 902     # We expect a Query instance to be passed into get_issues.
 903     # pylint: disable=E1120
 904     self.mox.StubOutClassWithMocks(gd_ph_client, 'Query')
 905
 906     mocked_query = gd_ph_client.Query(text_query='%s is:open' % text)
 907     feed = cros_test_lib.EasyAttr(entry=[tissue])
 908     mocked_itclient.get_issues(project, query=mocked_query).AndReturn(feed)
 909     gdata_lib.Issue.GetTrackerIssueComments(1, project).AndReturn([])
 910
 911     self.mox.ReplayAll()
 912
 913     issues = tcomm.GetTrackerIssuesByText(text)
 914     self.assertEquals(issues, [issue])
 915
 916   def testCreateTrackerIssue(self):
 917     author = 'TheAuthor'
 918     mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient)
 919     mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm)
 920     mocked_tcomm.author = author
 921     mocked_tcomm.it_client = mocked_itclient
 922     mocked_tcomm.project_name = 'TheProject'
 923
 924     issue = cros_test_lib.EasyAttr(title='TheTitle',
 925                                    summary='TheSummary',
 926                                    status='TheStatus',
 927                                    owner='TheOwner',
 928                                    labels='TheLabels',
 929                                    ccs=[])
 930
 931     # Replay script
 932     issue_id = cros_test_lib.EasyAttr(
 933         id=cros_test_lib.EasyAttr(text='foo/bar/123'))
 934     mocked_itclient.add_issue(project_name='TheProject',
 935                               title=issue.title,
 936                               content=issue.summary,
 937                               author=author,
 938                               status=issue.status,
 939                               owner=issue.owner,
 940                               labels=issue.labels,
 941                               ccs=issue.ccs,
 942                               ).AndReturn(issue_id)
 943     self.mox.ReplayAll()
 944
 945     # Verify
 946     result = gdata_lib.TrackerComm.CreateTrackerIssue(mocked_tcomm, issue)
 947     self.mox.VerifyAll()
 948     self.assertEquals(123, result)
 949
 950   def testAppendTrackerIssueById(self):
 951     author = 'TheAuthor'
 952     project_name = 'TheProject'
 953     mocked_itclient = self.mox.CreateMock(gd_ph_client.ProjectHostingClient)
 954     mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm)
 955     mocked_tcomm.author = author
 956     mocked_tcomm.it_client = mocked_itclient
 957     mocked_tcomm.project_name = project_name
 958
 959     issue_id = 54321
 960     comment = 'TheComment'
 961
 962     # Replay script
 963     mocked_itclient.update_issue(project_name=project_name,
 964                                  issue_id=issue_id,
 965                                  author=author,
 966                                  comment=comment,
 967                                  owner=None)
 968     self.mox.ReplayAll()
 969
 970     # Verify
 971     result = gdata_lib.TrackerComm.AppendTrackerIssueById(mocked_tcomm,
 972                                                           issue_id, comment)
 973     self.mox.VerifyAll()
 974     self.assertEquals(issue_id, result)
 975
 976
 977 class RetrySpreadsheetsServiceTest(cros_test_lib.MoxOutputTestCase):
 978   """Test Spreadsheet server retry helper."""
 979
 980   def testRequest(self):
 981     """Test that calling request method invokes _RetryRequest wrapper."""
 982     # pylint: disable=W0212
 983
 984     self.mox.StubOutWithMock(gdata_lib.RetrySpreadsheetsService,
 985                              '_RetryRequest')
 986
 987     # Use a real RetrySpreadsheetsService object rather than a mocked
 988     # one, because the .request method only exists if __init__ is run.
 989     # Also split up __new__ and __init__ in order to grab the original
 990     # rss.request method (inherited from base class at that point).
 991     rss = gdata_lib.RetrySpreadsheetsService.__new__(
 992         gdata_lib.RetrySpreadsheetsService)
 993     orig_request = rss.request
 994     rss.__init__()
 995
 996     args = ('GET', 'http://foo.bar')
 997
 998     # This is the replay script for the test.
 999     gdata_lib.RetrySpreadsheetsService._RetryRequest(orig_request, *args
1000                                                      ).AndReturn('wrapped')
1001     self.mox.ReplayAll()
1002
1003     # This is the test verification.
1004     retval = rss.request(*args)
1005     self.mox.VerifyAll()
1006     self.assertEquals('wrapped', retval)
1007
1008   def _TestHttpClientRetryRequest(self, statuses):
1009     """Test retry logic in http_client request during ProgrammaticLogin.
1010
1011     |statuses| is list of http codes to simulate, where 200 means success.
1012     """
1013     expect_success = statuses[-1] == 200
1014
1015     self.mox.StubOutWithMock(atom.http.ProxiedHttpClient, 'request')
1016     rss = gdata_lib.RetrySpreadsheetsService()
1017
1018     args = ('POST', 'https://www.google.com/accounts/ClientLogin')
1019     def _read():
1020       return 'Some response text'
1021
1022     # This is the replay script for the test.
1023     # Simulate the return codes in statuses.
1024     for status in statuses:
1025       retstatus = cros_test_lib.EasyAttr(status=status, read=_read)
1026       atom.http.ProxiedHttpClient.request(*args,
1027                                           data=mox.IgnoreArg(),
1028                                           headers=mox.IgnoreArg()
1029                                           ).AndReturn(retstatus)
1030     self.mox.ReplayAll()
1031
1032     # This is the test verification.
1033     with self.OutputCapturer():
1034       if expect_success:
1035         rss.ProgrammaticLogin()
1036       else:
1037         self.assertRaises(gdata.service.Error, rss.ProgrammaticLogin)
1038       self.mox.VerifyAll()
1039
1040     if not expect_success:
1041       # Retries did not help, request still failed.
1042       regexp = re.compile(r'^Giving up on HTTP request')
1043       self.AssertOutputContainsWarning(regexp=regexp)
1044     elif len(statuses) > 1:
1045       # Warning expected if retries were needed.
1046       self.AssertOutputContainsWarning()
1047     else:
1048       # First try worked, expect no warnings.
1049       self.AssertOutputContainsWarning(invert=True)
1050
1051   def testHttpClientRetryRequest(self):
1052     self._TestHttpClientRetryRequest([200])
1053
1054   def testHttpClientRetryRequest403(self):
1055     self._TestHttpClientRetryRequest([403, 200])
1056
1057   def testHttpClientRetryRequest403x2(self):
1058     self._TestHttpClientRetryRequest([403, 403, 200])
1059
1060   def testHttpClientRetryRequest403x3(self):
1061     self._TestHttpClientRetryRequest([403, 403, 403, 200])
1062
1063   def testHttpClientRetryRequest403x4(self):
1064     self._TestHttpClientRetryRequest([403, 403, 403, 403, 200])
1065
1066   def testHttpClientRetryRequest403x5(self):
1067     # This one should exhaust the retries.
1068     self._TestHttpClientRetryRequest([403, 403, 403, 403, 403])
1069
1070   def _TestRetryRequest(self, statuses):
1071     """Test retry logic for request method.
1072
1073     |statuses| is list of http codes to simulate, where 200 means success.
1074     """
1075     expect_success = statuses[-1] == 200
1076     expected_status_index = len(statuses) - 1 if expect_success else 0
1077
1078     mocked_ss = self.mox.CreateMock(gdata_lib.RetrySpreadsheetsService)
1079     args = ('GET', 'http://foo.bar')
1080
1081     # This is the replay script for the test.
1082     for ix, status in enumerate(statuses):
1083       # Add index of status to track which status the request function is
1084       # returning.  It is expected to return the last return status if
1085       # successful (retries or not), but first return status if failed.
1086       retval = cros_test_lib.EasyAttr(status=status, index=ix)
1087       mocked_ss.request(*args).AndReturn(retval)
1088
1089     self.mox.ReplayAll()
1090
1091     # This is the test verification.
1092     with self.OutputCapturer():
1093       # pylint: disable=W0212
1094       rval = gdata_lib.RetrySpreadsheetsService._RetryRequest(mocked_ss,
1095                                                               mocked_ss.request,
1096                                                               *args)
1097       self.mox.VerifyAll()
1098       self.assertEquals(statuses[expected_status_index], rval.status)
1099       self.assertEquals(expected_status_index, rval.index)
1100
1101     if not expect_success:
1102       # Retries did not help, request still failed.
1103       regexp = re.compile(r'^Giving up on HTTP request')
1104       self.AssertOutputContainsWarning(regexp=regexp)
1105     elif expected_status_index > 0:
1106       # Warning expected if retries were needed.
1107       self.AssertOutputContainsWarning()
1108     else:
1109       # First try worked, expect no warnings.
1110       self.AssertOutputContainsWarning(invert=True)
1111
1112   def testRetryRequest(self):
1113     self._TestRetryRequest([200])
1114
1115   def testRetryRequest403(self):
1116     self._TestRetryRequest([403, 200])
1117
1118   def testRetryRequest403x2(self):
1119     self._TestRetryRequest([403, 403, 200])
1120
1121   def testRetryRequest403x3(self):
1122     self._TestRetryRequest([403, 403, 403, 200])
1123
1124   def testRetryRequest403x4(self):
1125     self._TestRetryRequest([403, 403, 403, 403, 200])
1126
1127   def testRetryRequest403x5(self):
1128     # This one should exhaust the retries.
1129     self._TestRetryRequest([403, 403, 403, 403, 403])
1130
1131
1132 if __name__ == '__main__':
1133   cros_test_lib.main()