| #!/usr/bin/python |
| |
| # 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 cros_portage_upgrade.py.""" |
| |
| import os |
| import sys |
| |
| sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), |
| '..', '..')) |
| from chromite.lib import cros_test_lib |
| from chromite.lib import gdata_lib |
| from chromite.lib import upgrade_table as utable |
| from chromite.scripts import sync_package_status as sps |
| |
| # pylint: disable=W0212,R0904 |
| |
| |
| class SyncerTest(cros_test_lib.MoxOutputTestCase): |
| |
| col_amd64 = utable.UpgradeTable.GetColumnName(utable.UpgradeTable.COL_STATE, |
| 'amd64') |
| col_amd64 = gdata_lib.PrepColNameForSS(col_amd64) |
| col_arm = utable.UpgradeTable.GetColumnName(utable.UpgradeTable.COL_STATE, |
| 'arm') |
| col_arm = gdata_lib.PrepColNameForSS(col_arm) |
| col_x86 = utable.UpgradeTable.GetColumnName(utable.UpgradeTable.COL_STATE, |
| 'x86') |
| col_x86 = gdata_lib.PrepColNameForSS(col_x86) |
| |
| def testInit(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| tcomm, scomm = 'TComm', 'SComm' |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| sps.Syncer.__init__(mocked_syncer, tcomm, scomm) |
| self.mox.VerifyAll() |
| self.assertEquals(scomm, mocked_syncer.scomm) |
| self.assertEquals(tcomm, mocked_syncer.tcomm) |
| self.assertEquals(None, mocked_syncer.teams) |
| self.assertEquals(None, mocked_syncer.owners) |
| self.assertEquals(False, mocked_syncer.pretend) |
| self.assertEquals(False, mocked_syncer.verbose) |
| |
| def testReduceTeamName(self): |
| syncer = sps.Syncer('tcomm_obj', 'scomm_obj') |
| |
| tests = { |
| 'build/bdavirro': 'build', |
| 'build/rtc': 'build', |
| 'build': 'build', |
| 'UI/zel': 'ui', |
| 'UI': 'ui', |
| 'Build': 'build', |
| None: None, |
| } |
| |
| # Verify |
| for key in tests: |
| result = syncer._ReduceTeamName(key) |
| self.assertEquals(tests[key], result) |
| |
| def testReduceOwnerName(self): |
| syncer = sps.Syncer('tcomm_obj', 'scomm_obj') |
| |
| tests = { |
| 'joe': 'joe', |
| 'Joe': 'joe', |
| 'joe@chromium.org': 'joe', |
| 'Joe@chromium.org': 'joe', |
| 'Joe.Bob@chromium.org': 'joe.bob', |
| None: None, |
| } |
| |
| # Verify |
| for key in tests: |
| result = syncer._ReduceOwnerName(key) |
| self.assertEquals(tests[key], result) |
| |
| def testSetTeamFilterOK(self): |
| syncer = sps.Syncer('tcomm_obj', 'scomm_obj') |
| |
| tests = { |
| 'build:system:ui': set(['build', 'system', 'ui']), |
| 'Build:system:UI': set(['build', 'system', 'ui']), |
| 'kernel': set(['kernel']), |
| 'KERNEL': set(['kernel']), |
| None: None, |
| '': None, |
| } |
| |
| # Verify |
| for test in tests: |
| syncer.SetTeamFilter(test) |
| self.assertEquals(tests[test], syncer.teams) |
| |
| def testSetTeamFilterError(self): |
| syncer = sps.Syncer('tcomm_obj', 'scomm_obj') |
| |
| # "systems" is not valid (should be "system") |
| teamarg = 'build:systems' |
| |
| # Verify |
| with self.OutputCapturer(): |
| self.assertRaises(SystemExit, sps.Syncer.SetTeamFilter, |
| syncer, teamarg) |
| |
| def testSetOwnerFilter(self): |
| syncer = sps.Syncer('tcomm_obj', 'scomm_obj') |
| |
| tests = { |
| 'joe:bill:bob': set(['joe', 'bill', 'bob']), |
| 'Joe:Bill:BOB': set(['joe', 'bill', 'bob']), |
| 'joe@chromium.org:bill:bob': set(['joe', 'bill', 'bob']), |
| 'joe': set(['joe']), |
| 'joe@chromium.org': set(['joe']), |
| '': None, |
| None: None, |
| } |
| |
| # Verify |
| for test in tests: |
| syncer.SetOwnerFilter(test) |
| self.assertEquals(tests[test], syncer.owners) |
| |
| def testRowPassesFilters(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| row1 = { sps.COL_TEAM: 'build', sps.COL_OWNER: 'joe' } |
| row2 = { sps.COL_TEAM: 'build', sps.COL_OWNER: 'bob' } |
| row3 = { sps.COL_TEAM: 'build', sps.COL_OWNER: None } |
| row4 = { sps.COL_TEAM: None, sps.COL_OWNER: None } |
| |
| teams1 = set(['build']) |
| teams2 = set(['kernel']) |
| teams3 = set(['build', 'ui']) |
| |
| owners1 = set(['joe']) |
| owners2 = set(['bob']) |
| owners3 = set(['joe', 'dan']) |
| |
| tests = [ |
| { 'row': row1, 'teams': None, 'owners': None, 'result': True }, |
| { 'row': row1, 'teams': teams1, 'owners': None, 'result': True }, |
| { 'row': row1, 'teams': teams2, 'owners': None, 'result': False }, |
| { 'row': row1, 'teams': teams3, 'owners': None, 'result': True }, |
| { 'row': row1, 'teams': teams1, 'owners': owners1, 'result': True }, |
| { 'row': row1, 'teams': None, 'owners': owners2, 'result': False }, |
| { 'row': row1, 'teams': None, 'owners': owners3, 'result': True }, |
| |
| { 'row': row2, 'teams': None, 'owners': None, 'result': True }, |
| { 'row': row2, 'teams': teams1, 'owners': None, 'result': True }, |
| { 'row': row2, 'teams': teams2, 'owners': None, 'result': False }, |
| { 'row': row2, 'teams': teams3, 'owners': None, 'result': True }, |
| { 'row': row2, 'teams': teams1, 'owners': owners1, 'result': False }, |
| { 'row': row2, 'teams': None, 'owners': owners2, 'result': True }, |
| { 'row': row2, 'teams': None, 'owners': owners3, 'result': False }, |
| |
| { 'row': row3, 'teams': None, 'owners': None, 'result': True }, |
| { 'row': row3, 'teams': teams1, 'owners': None, 'result': True }, |
| { 'row': row3, 'teams': teams2, 'owners': None, 'result': False }, |
| { 'row': row3, 'teams': teams3, 'owners': None, 'result': True }, |
| { 'row': row3, 'teams': teams1, 'owners': owners1, 'result': False }, |
| { 'row': row3, 'teams': None, 'owners': owners2, 'result': False }, |
| { 'row': row3, 'teams': None, 'owners': owners3, 'result': False }, |
| |
| { 'row': row4, 'teams': None, 'owners': None, 'result': True }, |
| { 'row': row4, 'teams': teams1, 'owners': None, 'result': False }, |
| { 'row': row4, 'teams': teams1, 'owners': owners1, 'result': False }, |
| { 'row': row4, 'teams': None, 'owners': owners2, 'result': False }, |
| ] |
| |
| # Replay script |
| for test in tests: |
| done = False |
| |
| if test['teams']: |
| row_team = test['row'][sps.COL_TEAM] |
| mocked_syncer._ReduceTeamName(row_team).AndReturn(row_team) |
| done = row_team not in test['teams'] |
| |
| if not done and test['owners']: |
| row_owner = test['row'][sps.COL_OWNER] |
| mocked_syncer._ReduceOwnerName(row_owner).AndReturn(row_owner) |
| self.mox.ReplayAll() |
| |
| # Verify |
| for test in tests: |
| mocked_syncer.teams = test['teams'] |
| mocked_syncer.owners = test['owners'] |
| result = sps.Syncer._RowPassesFilters(mocked_syncer, test['row']) |
| |
| msg = ('Expected following row to %s filter, but it did not:\n%r' % |
| ('pass' if test['result'] else 'fail', test['row'])) |
| msg += '\n Using teams filter : %r' % mocked_syncer.teams |
| msg += '\n Using owners filter: %r' % mocked_syncer.owners |
| self.assertEquals(test['result'], result, msg) |
| self.mox.VerifyAll() |
| |
| def testSyncMissingTrackerColumn(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tcomm = mocked_tcomm |
| |
| # Replay script |
| mocked_scomm.GetColumnIndex(sps.COL_TRACKER).AndReturn(None) |
| self.mox.ReplayAll() |
| |
| # Verify |
| self.assertRaises(sps.SyncError, sps.Syncer.Sync, mocked_syncer) |
| self.mox.VerifyAll() |
| |
| def testSyncNewIssues(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tcomm = mocked_tcomm |
| |
| rows = [ |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: None }, |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: 'joe' }, |
| ] |
| |
| # Replay script |
| mocked_scomm.GetColumnIndex(sps.COL_TRACKER).AndReturn(1) # Any index ok. |
| mocked_scomm.GetRows().AndReturn(rows) |
| |
| for ix in xrange(len(rows)): |
| mocked_syncer._RowPassesFilters(rows[ix]).AndReturn(True) |
| mocked_syncer._GenIssueForRow(rows[ix]).AndReturn('NewIssue%d' % ix) |
| mocked_syncer._GetRowTrackerId(rows[ix]).AndReturn(None) |
| mocked_syncer._CreateRowIssue(ix + 2, rows[ix], 'NewIssue%d' % ix) |
| self.mox.ReplayAll() |
| |
| # Verify |
| sps.Syncer.Sync(mocked_syncer) |
| self.mox.VerifyAll() |
| |
| def testSyncClearIssues(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tcomm = mocked_tcomm |
| |
| rows = [ |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: None }, |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: 'joe' }, |
| ] |
| |
| # Replay script |
| mocked_scomm.GetColumnIndex(sps.COL_TRACKER).AndReturn(1) # Any index ok. |
| mocked_scomm.GetRows().AndReturn(rows) |
| |
| for ix in xrange(len(rows)): |
| mocked_syncer._RowPassesFilters(rows[ix]).AndReturn(True) |
| mocked_syncer._GenIssueForRow(rows[ix]).AndReturn(None) |
| mocked_syncer._GetRowTrackerId(rows[ix]).AndReturn(123 + ix) |
| mocked_syncer._ClearRowIssue(ix + 2, rows[ix]) |
| self.mox.ReplayAll() |
| |
| # Verify |
| sps.Syncer.Sync(mocked_syncer) |
| self.mox.VerifyAll() |
| |
| def testSyncFilteredOut(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tcomm = mocked_tcomm |
| |
| rows = [ |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: None }, |
| { sps.COL_PACKAGE: 'd/f', sps.COL_TEAM: 'build', sps.COL_OWNER: 'joe' }, |
| ] |
| |
| # Replay script |
| mocked_scomm.GetColumnIndex(sps.COL_TRACKER).AndReturn(1) # Any index ok. |
| mocked_scomm.GetRows().AndReturn(rows) |
| |
| for ix in xrange(len(rows)): |
| mocked_syncer._RowPassesFilters(rows[ix]).AndReturn(False) |
| self.mox.ReplayAll() |
| |
| # Verify |
| sps.Syncer.Sync(mocked_syncer) |
| self.mox.VerifyAll() |
| |
| def testGetRowValue(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| row = { |
| self.col_amd64: 'ABC', |
| self.col_arm: 'XYZ', |
| self.col_x86: 'FooBar', |
| sps.COL_TEAM: 'build', |
| } |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = sps.Syncer._GetRowValue(mocked_syncer, row, |
| 'stateonamd64', 'amd64') |
| self.assertEquals('ABC', result) |
| result = sps.Syncer._GetRowValue(mocked_syncer, row, |
| 'stateonarm', 'arm') |
| self.assertEquals('XYZ', result) |
| result = sps.Syncer._GetRowValue(mocked_syncer, row, |
| 'stateonamd64', 'amd64') |
| self.assertEquals('ABC', result) |
| result = sps.Syncer._GetRowValue(mocked_syncer, row, sps.COL_TEAM) |
| self.assertEquals('build', result) |
| self.mox.VerifyAll() |
| |
| def _TestGenIssueForRowNeedsUpgrade(self, row): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_syncer.default_owner = None |
| mocked_syncer.scomm = cros_test_lib.EasyAttr(ss_key='SomeSSKey') |
| |
| # Replay script |
| for arch in sps.ARCHES: |
| state = sps.Syncer._GetRowValue(mocked_syncer, row, |
| utable.UpgradeTable.COL_STATE, arch) |
| mocked_syncer._GetRowValue(row, utable.UpgradeTable.COL_STATE, |
| arch).AndReturn(state) |
| red_team = sps.Syncer._ReduceTeamName(mocked_syncer, row[sps.COL_TEAM]) |
| mocked_syncer._ReduceTeamName(row[sps.COL_TEAM]).AndReturn(red_team) |
| red_owner = sps.Syncer._ReduceOwnerName(mocked_syncer, row[sps.COL_OWNER]) |
| mocked_syncer._ReduceOwnerName(row[sps.COL_OWNER]).AndReturn(red_owner) |
| for arch in sps.ARCHES: |
| mocked_syncer._GetRowValue(row, utable.UpgradeTable.COL_CURRENT_VER, |
| arch).AndReturn('1') |
| mocked_syncer._GetRowValue(row, |
| utable.UpgradeTable.COL_STABLE_UPSTREAM_VER, |
| arch).AndReturn('2') |
| mocked_syncer._GetRowValue(row, |
| utable.UpgradeTable.COL_LATEST_UPSTREAM_VER, |
| arch).AndReturn('3') |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = sps.Syncer._GenIssueForRow(mocked_syncer, row) |
| self.mox.VerifyAll() |
| return result |
| |
| def testGenIssueForRowNeedsUpgrade1(self): |
| row = { |
| self.col_amd64: utable.UpgradeTable.STATE_NEEDS_UPGRADE, |
| self.col_arm: 'Not important', |
| self.col_x86: 'Not important', |
| sps.COL_TEAM: 'build', |
| sps.COL_OWNER: None, |
| sps.COL_PACKAGE: 'dev/foo', |
| } |
| |
| result = self._TestGenIssueForRowNeedsUpgrade(row) |
| self.assertEquals(None, result.owner) |
| self.assertEquals(0, result.id) |
| self.assertEquals('Untriaged', result.status) |
| |
| def testGenIssueForRowNeedsUpgrade2(self): |
| row = { |
| self.col_amd64: utable.UpgradeTable.STATE_NEEDS_UPGRADE, |
| self.col_arm: utable.UpgradeTable.STATE_NEEDS_UPGRADE_AND_PATCHED, |
| self.col_x86: 'Not important', |
| sps.COL_TEAM: 'build', |
| sps.COL_OWNER: 'joe', |
| sps.COL_PACKAGE: 'dev/foo', |
| } |
| |
| result = self._TestGenIssueForRowNeedsUpgrade(row) |
| self.assertEquals('joe@chromium.org', result.owner) |
| self.assertEquals(0, result.id) |
| self.assertEquals('Available', result.status) |
| |
| def testGenIssueForRowNeedsUpgrade3(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| row = { |
| self.col_amd64: utable.UpgradeTable.STATE_NEEDS_UPGRADE, |
| self.col_arm: utable.UpgradeTable.STATE_NEEDS_UPGRADE_AND_PATCHED, |
| self.col_x86: 'Not important', |
| sps.COL_TEAM: None, |
| sps.COL_OWNER: 'joe', |
| sps.COL_PACKAGE: 'dev/foo', |
| } |
| |
| # Replay script |
| for arch in sps.ARCHES: |
| state = sps.Syncer._GetRowValue(mocked_syncer, row, |
| utable.UpgradeTable.COL_STATE, arch) |
| mocked_syncer._GetRowValue(row, utable.UpgradeTable.COL_STATE, |
| arch).AndReturn(state) |
| reduced = sps.Syncer._ReduceTeamName(mocked_syncer, row[sps.COL_TEAM]) |
| mocked_syncer._ReduceTeamName(row[sps.COL_TEAM]).AndReturn(reduced) |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| self.assertRaises(RuntimeError, sps.Syncer._GenIssueForRow, |
| mocked_syncer, row) |
| self.mox.VerifyAll() |
| |
| def testGenIssueForRowNoUpgrade(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| row = { |
| self.col_amd64: 'Not important', |
| self.col_arm: 'Not important', |
| self.col_x86: 'Not important', |
| sps.COL_TEAM: None, |
| sps.COL_OWNER: 'joe', |
| sps.COL_PACKAGE: 'dev/foo', |
| } |
| |
| # Replay script |
| for arch in sps.ARCHES: |
| state = sps.Syncer._GetRowValue(mocked_syncer, row, |
| utable.UpgradeTable.COL_STATE, arch) |
| mocked_syncer._GetRowValue(row, utable.UpgradeTable.COL_STATE, |
| arch).AndReturn(state) |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = sps.Syncer._GenIssueForRow(mocked_syncer, row) |
| self.mox.VerifyAll() |
| self.assertEquals(None, result) |
| |
| def testGetRowTrackerId(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| row = { sps.COL_TRACKER: '321' } |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| result = sps.Syncer._GetRowTrackerId(mocked_syncer, row) |
| self.mox.VerifyAll() |
| self.assertEquals(321, result) |
| |
| def testCreateRowIssuePretend(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_syncer.pretend = True |
| |
| row = { sps.COL_PACKAGE: 'dev/foo' } |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| sps.Syncer._CreateRowIssue(mocked_syncer, 5, row, 'some_issue') |
| self.mox.VerifyAll() |
| |
| def testCreateRowIssue(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_tcomm = self.mox.CreateMock(gdata_lib.TrackerComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tcomm = mocked_tcomm |
| mocked_syncer.tracker_col_ix = 8 |
| mocked_syncer.pretend = False |
| |
| row_ix = 5 |
| row = { sps.COL_PACKAGE: 'dev/foo' } |
| issue = 'SomeIssue' |
| issue_id = 234 |
| ss_issue_val = 'Hyperlink%d' % issue_id |
| |
| # Replay script |
| mocked_scomm.ClearCellValue(row_ix, mocked_syncer.tracker_col_ix) |
| mocked_tcomm.CreateTrackerIssue(issue).AndReturn(issue_id) |
| mocked_syncer._GenSSLinkToIssue(issue_id).AndReturn(ss_issue_val) |
| mocked_scomm.ReplaceCellValue(row_ix, mocked_syncer.tracker_col_ix, |
| ss_issue_val) |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| sps.Syncer._CreateRowIssue(mocked_syncer, row_ix, row, issue) |
| self.mox.VerifyAll() |
| |
| def testGenSSLinkToIssue(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| |
| issue_id = 123 |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| result = sps.Syncer._GenSSLinkToIssue(mocked_syncer, issue_id) |
| self.mox.VerifyAll() |
| self.assertEquals('=hyperlink("crbug.com/123";"123")', result) |
| |
| def testClearRowIssue(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tracker_col_ix = 8 |
| mocked_syncer.pretend = False |
| |
| row_ix = 44 |
| row = { sps.COL_PACKAGE: 'dev/foo' } |
| |
| # Replay script |
| mocked_scomm.ClearCellValue(row_ix, mocked_syncer.tracker_col_ix) |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| sps.Syncer._ClearRowIssue(mocked_syncer, row_ix, row) |
| self.mox.VerifyAll() |
| |
| def testClearRowIssuePretend(self): |
| mocked_syncer = self.mox.CreateMock(sps.Syncer) |
| mocked_scomm = self.mox.CreateMock(gdata_lib.SpreadsheetComm) |
| mocked_syncer.scomm = mocked_scomm |
| mocked_syncer.tracker_col_ix = 8 |
| mocked_syncer.pretend = True |
| |
| row_ix = 44 |
| row = { sps.COL_PACKAGE: 'dev/foo' } |
| |
| # Replay script |
| self.mox.ReplayAll() |
| |
| # Verify |
| with self.OutputCapturer(): |
| sps.Syncer._ClearRowIssue(mocked_syncer, row_ix, row) |
| self.mox.VerifyAll() |
| |
| |
| if __name__ == '__main__': |
| cros_test_lib.main() |
