src/third_party/chromite/scripts/upload_package_status_unittest.py

   1 #!/usr/bin/python
   2
   3 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
   4 # Use of this source code is governed by a BSD-style license that can be
   5 # found in the LICENSE file.
   6
   7 """Unit tests for cros_portage_upgrade.py."""
   8
   9 import exceptions
  10
  11 import mox
  12
  13 from chromite.lib import cros_test_lib
  14 from chromite.lib import gdata_lib
  15 from chromite.lib import osutils
  16 from chromite.lib import table as tablelib
  17 from chromite.scripts import merge_package_status as mps
  18 from chromite.scripts import upload_package_status as ups
  19
  20 # pylint: disable=W0212,R0904,E1120,E1101
  21
  22
  23 class SSEntry(object):
  24   """Class to simulate one spreadsheet entry."""
  25   def __init__(self, text):
  26     self.text = text
  27
  28
  29 class SSRow(object):
  30   """Class for simulating spreadsheet row."""
  31   def __init__(self, row, cols=None):
  32     self.custom = {}
  33
  34     if not cols:
  35       # If columns not specified, then column order doesn't matter.
  36       cols = row.keys()
  37     for col in cols:
  38       ss_col = gdata_lib.PrepColNameForSS(col)
  39       val = row[col]
  40       ss_val = gdata_lib.PrepValForSS(val)
  41       self.custom[ss_col] = SSEntry(ss_val)
  42
  43
  44 class SSFeed(object):
  45   """Class for simulating spreadsheet list feed."""
  46   def __init__(self, rows, cols=None):
  47     self.entry = []
  48     for row in rows:
  49       self.entry.append(SSRow(row, cols))
  50
  51
  52 class UploaderTest(cros_test_lib.MoxOutputTestCase):
  53   """Test the functionality of upload_package_status.Uploader class."""
  54
  55   COL_PKG = 'Package'
  56   COL_SLOT = 'Slot'
  57   COL_OVERLAY = 'Overlay'
  58   COL_STATUS = 'Status'
  59   COL_VER = 'Current Version'
  60   COL_STABLE_UP = 'Stable Upstream Version'
  61   COL_LATEST_UP = 'Latest Upstream Version'
  62   COL_TARGET = 'Chrome OS Root Target'
  63
  64   SS_COL_PKG = gdata_lib.PrepColNameForSS(COL_PKG)
  65   SS_COL_SLOT = gdata_lib.PrepColNameForSS(COL_SLOT)
  66   SS_COL_OVERLAY = gdata_lib.PrepColNameForSS(COL_OVERLAY)
  67   SS_COL_STATUS = gdata_lib.PrepColNameForSS(COL_STATUS)
  68   SS_COL_VER = gdata_lib.PrepColNameForSS(COL_VER)
  69   SS_COL_STABLE_UP = gdata_lib.PrepColNameForSS(COL_STABLE_UP)
  70   SS_COL_LATEST_UP = gdata_lib.PrepColNameForSS(COL_LATEST_UP)
  71   SS_COL_TARGET = gdata_lib.PrepColNameForSS(COL_TARGET)
  72
  73   COLS = [COL_PKG,
  74           COL_SLOT,
  75           COL_OVERLAY,
  76           COL_STATUS,
  77           COL_VER,
  78           COL_STABLE_UP,
  79           COL_LATEST_UP,
  80           COL_TARGET,
  81           ]
  82
  83   ROW0 = {COL_PKG: 'lib/foo',
  84           COL_SLOT: '0',
  85           COL_OVERLAY: 'portage',
  86           COL_STATUS: 'needs upgrade',
  87           COL_VER: '3.0.2',
  88           COL_STABLE_UP: '3.0.9',
  89           COL_LATEST_UP: '3.0.11',
  90           COL_TARGET: 'virtual/target-os',
  91           }
  92   ROW1 = {COL_PKG: 'sys-dev/bar',
  93           COL_SLOT: '0',
  94           COL_OVERLAY: 'chromiumos-overlay',
  95           COL_STATUS: 'needs upgrade',
  96           COL_VER: '1.2.3-r1',
  97           COL_STABLE_UP: '1.2.3-r2',
  98           COL_LATEST_UP: '1.2.4',
  99           COL_TARGET: 'virtual/target-os-dev',
 100           }
 101   ROW2 = {COL_PKG: 'sys-dev/raster',
 102           COL_SLOT: '1',
 103           COL_OVERLAY: 'chromiumos-overlay',
 104           COL_STATUS: 'current',
 105           COL_VER: '1.2.3',
 106           COL_STABLE_UP: '1.2.3',
 107           COL_LATEST_UP: '1.2.4',
 108           COL_TARGET: 'chromeos-test',
 109           }
 110
 111   SS_ROW0 = dict([(gdata_lib.PrepColNameForSS(c), v) for c, v in ROW0.items()])
 112   SS_ROW1 = dict([(gdata_lib.PrepColNameForSS(c), v) for c, v in ROW1.items()])
 113   SS_ROW2 = dict([(gdata_lib.PrepColNameForSS(c), v) for c, v in ROW2.items()])
 114
 115   EMAIL = 'knights@ni.com'
 116   PASSWORD = 'the'
 117
 118   def _MockUploader(self, table=None):
 119     """Set up a mocked Uploader object."""
 120     uploader = self.mox.CreateMock(ups.Uploader)
 121
 122     if not table:
 123       # Use default table
 124       table = self._CreateDefaultTable()
 125
 126     for slot in ups.Uploader.__slots__:
 127       uploader.__setattr__(slot, None)
 128
 129     uploader._csv_table = table
 130     uploader._scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm)
 131     uploader._creds = cros_test_lib.EasyAttr(user=self.EMAIL,
 132                                              password=self.PASSWORD)
 133     uploader._ss_row_cache = self._CreateRowCache(table)
 134
 135     return uploader
 136
 137   def _CreateRowCache(self, table):
 138     """Recreate the expected row cache (by pkg) from |table|."""
 139     if not table:
 140       return None
 141
 142     row_cache = {}
 143     for rowIx, row in enumerate(table):
 144       pkg = row[self.COL_PKG]
 145
 146       # Translate column names now.
 147       ss_row_dict = {}
 148       for col in row:
 149         ss_row_dict[gdata_lib.PrepColNameForSS(col)] = row[col]
 150
 151       ss_row = gdata_lib.SpreadsheetRow('OrigRow%d' % (rowIx + 2),
 152                                         rowIx + 2, ss_row_dict)
 153       entry = row_cache.get(pkg)
 154       if not entry:
 155         row_cache[pkg] = ss_row
 156       elif type(entry) == list:
 157         row_cache[pkg] = entry + [ss_row]
 158       else:
 159         row_cache[pkg] = [entry, ss_row]
 160     return row_cache
 161
 162   def _CreateDefaultTable(self):
 163     return self._CreateTableWithRows(self.COLS,
 164                                      [self.ROW0, self.ROW1])
 165
 166   def _CreateTableWithRows(self, cols, rows):
 167     mytable = tablelib.Table(list(cols))
 168     if rows:
 169       for row in rows:
 170         mytable.AppendRow(dict(row))
 171     return mytable
 172
 173   def testLoadTable(self):
 174     # Note that this test is not actually for method of Uploader class.
 175
 176     self.mox.StubOutWithMock(tablelib.Table, 'LoadFromCSV')
 177     csv = 'any.csv'
 178
 179     # Replay script
 180     tablelib.Table.LoadFromCSV(csv).AndReturn('loaded_table')
 181     self.mox.ReplayAll()
 182
 183     # Verification steps.
 184     with self.OutputCapturer():
 185       loaded_table = ups.LoadTable(csv)
 186       self.assertEquals(loaded_table, 'loaded_table')
 187
 188   def testGetSSRowForPackage(self):
 189     mocked_uploader = self._MockUploader()
 190
 191     # No replay script.
 192     self.mox.ReplayAll()
 193
 194     # Verification steps.
 195     result = ups.Uploader._GetSSRowForPackage(mocked_uploader,
 196                                               self.ROW0[self.COL_PKG])
 197     self.assertEquals(result, self.SS_ROW0)
 198     self.assertEquals(2, result.ss_row_num)
 199     result = ups.Uploader._GetSSRowForPackage(mocked_uploader,
 200                                               self.ROW1[self.COL_PKG])
 201     self.assertEquals(result, self.SS_ROW1)
 202     self.assertEquals(3, result.ss_row_num)
 203     result = ups.Uploader._GetSSRowForPackage(mocked_uploader,
 204                                               self.ROW2[self.COL_PKG])
 205     self.assertEquals(result, None)
 206
 207     self.mox.VerifyAll()
 208
 209   def testUploadFirstWorksheet(self):
 210     mocked_uploader = self._MockUploader()
 211
 212     # Clear ._scomm attribute to simulate uploading first worksheet.
 213     mocked_scomm = mocked_uploader._scomm
 214     mocked_uploader._scomm = None
 215
 216     self.mox.StubOutWithMock(gdata_lib.SpreadsheetComm, '__new__')
 217
 218     ss_key = 'Some ss_key'
 219     ws_name = 'Some ws_name'
 220
 221     # Replay script
 222     gdata_lib.SpreadsheetComm.__new__(gdata_lib.SpreadsheetComm
 223                                       ).AndReturn(mocked_scomm)
 224     mocked_scomm.Connect(mocked_uploader._creds, ss_key, ws_name,
 225                          source='Upload Package Status')
 226     mocked_scomm.GetRowCacheByCol(self.SS_COL_PKG).AndReturn('RowCache')
 227     mocked_uploader._UploadChangedRows().AndReturn(tuple([0, 1, 2]))
 228     mocked_uploader._DeleteOldRows().AndReturn(tuple([3, 4]))
 229     self.mox.ReplayAll()
 230
 231     # Verify
 232     with self.OutputCapturer():
 233       ups.Uploader.Upload(mocked_uploader, ss_key, ws_name)
 234       self.mox.VerifyAll()
 235
 236   def testUploadSecondWorksheet(self):
 237     mocked_uploader = self._MockUploader()
 238
 239     ss_key = 'Some ss_key'
 240     ws_name = 'Some ws_name'
 241
 242     # Replay script
 243     mocked_uploader._scomm.SetCurrentWorksheet(ws_name)
 244     mocked_uploader._scomm.GetRowCacheByCol(self.SS_COL_PKG).AndReturn('RCache')
 245     mocked_uploader._UploadChangedRows().AndReturn(tuple([0, 1, 2]))
 246     mocked_uploader._DeleteOldRows().AndReturn(tuple([3, 4]))
 247     self.mox.ReplayAll()
 248
 249     # Verify
 250     with self.OutputCapturer():
 251       ups.Uploader.Upload(mocked_uploader, ss_key, ws_name)
 252       self.mox.VerifyAll()
 253
 254   def testUploadChangedRows(self):
 255     table = self._CreateTableWithRows(self.COLS,
 256                                       [self.ROW0, self.ROW1, self.ROW2])
 257     mocked_uploader = self._MockUploader(table=table)
 258
 259     def RowVerifier(row_delta, golden_col_set, golden_row):
 260       if golden_col_set != set(row_delta.keys()):
 261         return False
 262
 263       for col in row_delta:
 264         val = row_delta[col]
 265         if val != golden_row[col]:
 266           return False
 267
 268       return True
 269
 270     # First Row.
 271     # Pretend first row does not exist already in online spreadsheet
 272     # by returning (None, None) from _GetSSRowForPackage.
 273     #
 274     row0_pkg = self.ROW0[self.COL_PKG]
 275     mocked_uploader._GetSSRowForPackage(row0_pkg).AndReturn(None)
 276     mocked_uploader._scomm.InsertRow(mox.IgnoreArg())
 277
 278     # Second Row.
 279     # Pretend second row does already exist in online spreadsheet, and
 280     # pretend that it has a different value that needs to be changed
 281     # by an upload.
 282     row1_pkg = self.ROW1[self.COL_PKG]
 283     row1_reverse_delta = { self.SS_COL_VER: '1.2.3' }
 284     ss_row1 = dict(self.SS_ROW1)
 285     for col in row1_reverse_delta:
 286       ss_row1[col] = row1_reverse_delta[col]
 287     ss_row1 = gdata_lib.SpreadsheetRow('OrigRow1', 3, ss_row1)
 288     mocked_uploader._GetSSRowForPackage(row1_pkg).AndReturn(ss_row1)
 289     # Prepare verfication for row.
 290     g_col_set1 = set(row1_reverse_delta.keys())
 291     g_row1 = gdata_lib.PrepRowForSS(self.SS_ROW1)
 292     row1_verifier = lambda rdelta : RowVerifier(rdelta, g_col_set1, g_row1)
 293     mocked_uploader._scomm.UpdateRowCellByCell(3, mox.Func(row1_verifier))
 294
 295     # Third Row.
 296     # Pretend third row does already exist in online spreadsheet, and
 297     # pretend that several values need to be changed by an upload.
 298     row2_pkg = self.ROW2[self.COL_PKG]
 299     row2_reverse_delta = { self.SS_COL_STATUS: 'needs upgrade',
 300                            self.SS_COL_VER: '0.5',
 301                            self.SS_COL_TARGET: 'chromeos-foo',
 302                            }
 303     ss_row2 = dict(self.SS_ROW2)
 304     for col in row2_reverse_delta:
 305       ss_row2[col] = row2_reverse_delta[col]
 306     ss_row2 = gdata_lib.SpreadsheetRow('OrigRow2', 4, ss_row2)
 307     mocked_uploader._GetSSRowForPackage(row2_pkg).AndReturn(ss_row2)
 308     # Prepare verification for row.
 309     g_col_set2 = set(row2_reverse_delta.keys())
 310     g_row2 = gdata_lib.PrepRowForSS(self.SS_ROW2)
 311     row2_verifier = lambda rdelta : RowVerifier(rdelta, g_col_set2, g_row2)
 312     mocked_uploader._scomm.UpdateRowCellByCell(4, mox.Func(row2_verifier))
 313
 314     self.mox.ReplayAll()
 315
 316     # Verify
 317     with self.OutputCapturer():
 318       ups.Uploader._UploadChangedRows(mocked_uploader)
 319     self.mox.VerifyAll()
 320
 321   def testDeleteOldRows(self):
 322     mocked_uploader = self._MockUploader()
 323
 324     # Pretend spreadsheet has 2 rows, one in table and one not.
 325     ss_row1 = gdata_lib.SpreadsheetRow('OrigRow1', 2, self.SS_ROW1)
 326     ss_row2 = gdata_lib.SpreadsheetRow('OrigRow2', 3, self.SS_ROW2)
 327     ss_rows = (ss_row1, ss_row2)
 328
 329     mocked_uploader._scomm.GetRows().AndReturn(ss_rows)
 330     # We expect ROW2 in spreadsheet to be deleted.
 331     mocked_uploader._scomm.DeleteRow('OrigRow2')
 332     self.mox.ReplayAll()
 333
 334     # Verify
 335     with self.OutputCapturer():
 336       ups.Uploader._DeleteOldRows(mocked_uploader)
 337     self.mox.VerifyAll()
 338
 339
 340 class MainTest(cros_test_lib.MoxOutputTestCase):
 341   """Test argument handling at the main method level."""
 342
 343   def testHelp(self):
 344     """Test that --help is functioning"""
 345     with self.OutputCapturer() as output:
 346       # Running with --help should exit with code==0
 347       try:
 348         ups.main(['--help'])
 349       except exceptions.SystemExit as e:
 350         self.assertEquals(e.args[0], 0)
 351
 352     # Verify that a message beginning with "Usage: " was printed
 353     stdout = output.GetStdout()
 354     self.assertTrue(stdout.startswith('Usage: '))
 355
 356   def testMissingCSV(self):
 357     """Test that running without a csv file argument exits with an error."""
 358     with self.OutputCapturer():
 359       # Running without a package should exit with code!=0
 360       try:
 361         ups.main([])
 362       except exceptions.SystemExit as e:
 363         self.assertNotEquals(e.args[0], 0)
 364
 365     self.AssertOutputEndsInError(check_stdout=True)
 366
 367   def testPrepareCredsEmailPassword(self):
 368     email = 'foo@g.com'
 369     password = 'shh'
 370     creds_file = 'bogus'
 371     token_file = 'boguser'
 372
 373     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 374     self.mox.StubOutWithMock(gdata_lib.Creds, '__new__')
 375
 376     gdata_lib.Creds.__new__(gdata_lib.Creds).AndReturn(mocked_creds)
 377     mocked_creds.SetCreds(email, password)
 378     self.mox.ReplayAll()
 379
 380     ups.PrepareCreds(creds_file, token_file, email, password)
 381     self.mox.VerifyAll()
 382
 383   def testMainEmailPassword(self):
 384     """Verify that running main with email/password follows flow."""
 385     csv = 'any.csv'
 386     email = 'foo@g.com'
 387     password = '123'
 388
 389     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 390     creds_file = 'non-existing-file'
 391
 392     self.mox.StubOutWithMock(ups, 'PrepareCreds')
 393     self.mox.StubOutWithMock(ups, 'LoadTable')
 394     self.mox.StubOutWithMock(mps, 'FinalizeTable')
 395     self.mox.StubOutWithMock(ups.Uploader, 'Upload')
 396
 397     ups.PrepareCreds(creds_file, None, email, password).AndReturn(mocked_creds)
 398     ups.LoadTable(csv).AndReturn('csv_table')
 399     mps.FinalizeTable('csv_table')
 400     ups.Uploader.Upload(mox.IgnoreArg(), ws_name='Packages')
 401     ups.Uploader.Upload(mox.IgnoreArg(), ws_name='Dependencies')
 402     mocked_creds.StoreCredsIfNeeded(creds_file)
 403     self.mox.ReplayAll()
 404
 405     ups.main(['--email=%s' % email,
 406               '--password=%s' % password,
 407               '--cred-file=%s' % creds_file,
 408                csv])
 409
 410     self.mox.VerifyAll()
 411
 412   @osutils.TempFileDecorator
 413   def testMainCredsFile(self):
 414     """Verify that running main with creds file follows flow."""
 415     csv = 'any.csv'
 416     creds_file = self.tempfile
 417     token_file = 'non-existing-file'
 418
 419     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 420     mocked_creds.auth_token_loaded = False
 421
 422     self.mox.StubOutWithMock(ups, 'PrepareCreds')
 423     self.mox.StubOutWithMock(ups, 'LoadTable')
 424     self.mox.StubOutWithMock(mps, 'FinalizeTable')
 425     self.mox.StubOutWithMock(ups.Uploader, 'Upload')
 426
 427     ups.PrepareCreds(creds_file, token_file, None, None).AndReturn(mocked_creds)
 428     ups.LoadTable(csv).AndReturn('csv_table')
 429     mps.FinalizeTable('csv_table')
 430     ups.Uploader.Upload(mox.IgnoreArg(), ws_name=ups.PKGS_WS_NAME)
 431     ups.Uploader.Upload(mox.IgnoreArg(), ws_name=ups.DEPS_WS_NAME)
 432     mocked_creds.StoreCredsIfNeeded(creds_file)
 433     mocked_creds.StoreAuthTokenIfNeeded(token_file)
 434     self.mox.ReplayAll()
 435
 436     ups.main(['--cred-file=%s' % creds_file,
 437               '--auth-token-file=%s' % token_file,
 438               csv])
 439
 440     self.mox.VerifyAll()
 441
 442
 443   @osutils.TempFileDecorator
 444   def testMainTokenFile(self):
 445     """Verify that running main with token file follows flow."""
 446     csv = 'any.csv'
 447     token_file = self.tempfile
 448     creds_file = 'non-existing-file'
 449
 450     mocked_creds = self.mox.CreateMock(gdata_lib.Creds)
 451     mocked_creds.auth_token_loaded = True
 452
 453     self.mox.StubOutWithMock(ups, 'PrepareCreds')
 454     self.mox.StubOutWithMock(ups, 'LoadTable')
 455     self.mox.StubOutWithMock(mps, 'FinalizeTable')
 456     self.mox.StubOutWithMock(ups.Uploader, 'Upload')
 457
 458     ups.PrepareCreds(creds_file, token_file, None, None).AndReturn(mocked_creds)
 459     ups.LoadTable(csv).AndReturn('csv_table')
 460     mps.FinalizeTable('csv_table')
 461     ups.Uploader.Upload(mox.IgnoreArg(), ws_name=ups.PKGS_WS_NAME)
 462     ups.Uploader.Upload(mox.IgnoreArg(), ws_name=ups.DEPS_WS_NAME)
 463     mocked_creds.StoreCredsIfNeeded(creds_file)
 464     mocked_creds.StoreAuthTokenIfNeeded(token_file)
 465     self.mox.ReplayAll()
 466
 467     ups.main(['--cred-file=%s' % creds_file,
 468               '--auth-token-file=%s' % token_file,
 469               csv])
 470
 471     self.mox.VerifyAll()
 472
 473
 474 if __name__ == '__main__':
 475   cros_test_lib.main()