Index: commit-queue/tests/commit_queue_test.py |
=================================================================== |
--- commit-queue/tests/commit_queue_test.py (revision 249146) |
+++ commit-queue/tests/commit_queue_test.py (working copy) |
@@ -1,193 +0,0 @@ |
-#!/usr/bin/env python |
-# Copyright (c) 2012 The Chromium 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 commit_queue.py.""" |
- |
-import logging |
-import os |
-import StringIO |
-import sys |
-import time |
-import traceback |
-import unittest |
- |
-ROOT_DIR = os.path.dirname(os.path.abspath(__file__)) |
-sys.path.insert(0, os.path.join(ROOT_DIR, '..')) |
- |
-import commit_queue |
-import context |
-import creds |
- |
-from testing_support import auto_stub |
- |
-# From /tests |
-import mocks |
- |
- |
-class Stop(Exception): |
- pass |
- |
- |
-class PendingManagerMock(auto_stub.SimpleMock): |
- def __init__(self, unit_test): |
- super(PendingManagerMock, self).__init__(unit_test) |
- self.context = context.Context( |
- mocks.RietveldMock(unit_test), mocks.SvnCheckoutMock(unit_test), None) |
- self.count = 0 |
- |
- def load(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- def save(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- def close(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- def look_for_new_pending_commit(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- self.count += 1 |
- if self.count > 3: |
- raise Stop() |
- |
- def process_new_pending_commit(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- def update_status(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- def scan_results(self, *args, **kwargs): |
- self._register_call(*args, **kwargs) |
- |
- |
-class CredentialsMock(object): |
- @staticmethod |
- def get(user): |
- return '1%s1' % user |
- |
- |
-class CommitQueueTest(auto_stub.TestCase): |
- def setUp(self): |
- super(CommitQueueTest, self).setUp() |
- self.mock(sys, 'argv', ['commit_queue.py']) |
- self.mock(sys, 'stdout', StringIO.StringIO()) |
- self.mock(sys, 'stderr', StringIO.StringIO()) |
- self.mock(commit_queue.projects, 'load_project', None) |
- self.mock(commit_queue, 'SetupLogging', lambda _: None) |
- # Setup logging attached to the mocked sys.stderr, printing |
- # only the exception name to make tests less fragile. |
- handler = logging.StreamHandler() |
- formatter = logging.Formatter() |
- formatter.formatException = lambda _: traceback.format_exc(0) |
- handler.setFormatter(formatter) |
- logging.getLogger().handlers = [handler] |
- self._time = 1 |
- self.mock(time, 'time', self._get_time) |
- self.mock(creds, 'Credentials', self._get_cred) |
- |
- def tearDown(self): |
- try: |
- if not self.has_failed(): |
- self._check('stdout', '') |
- self._check('stderr', '') |
- finally: |
- super(CommitQueueTest, self).tearDown() |
- |
- def _check(self, pipe, expected): |
- self.assertEqual(expected, self._pop(pipe)) |
- |
- def _get_time(self): |
- self._time += 10 |
- return self._time |
- |
- @staticmethod |
- def _pop(pipe): |
- data = getattr(sys, pipe).getvalue() |
- setattr(sys, pipe, StringIO.StringIO()) |
- return data |
- |
- def _get_cred(self, pwd): |
- rootdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
- workdir = os.path.join(rootdir, 'workdir') |
- self.assertEqual(os.path.join(workdir, '.gaia_pwd'), pwd) |
- return CredentialsMock() |
- |
- def testHelp(self): |
- sys.argv.append('--help') |
- try: |
- commit_queue.main() |
- self.fail() |
- except SystemExit as e: |
- self.assertEqual(0, e.code) |
- output = self._pop('stdout') |
- # Cannot compare for the exact string since the formatting depends on the |
- # screen size. |
- self.assertIn('Minimum delay between each polling loop', output) |
- self.assertIn('Run for real instead of dry-run mode which', output) |
- self.assertLess(600, len(output), output) |
- |
- def testChromium(self): |
- sys.argv.extend(('--project', 'chromium')) |
- calls = [] |
- def load_project(*args): |
- calls.append(args) |
- return PendingManagerMock(self) |
- |
- self.mock(commit_queue.projects, 'load_project', load_project) |
- try: |
- commit_queue.main() |
- self.fail() |
- except Stop: |
- pass |
- self.assertEqual(1, len(calls)) |
- self.assertEqual('chromium', calls[0][0]) |
- self.assertEqual('commit-bot@chromium.org', calls[0][1]) |
- self.assertEqual( |
- os.path.join(os.path.dirname(ROOT_DIR), 'workdir'), calls[0][2]) |
- self.assertEqual(None, calls[0][4]) |
- self._check( |
- 'stdout', |
- 'Using read-only Rietveld\n' |
- 'Using read-only checkout\n' |
- 'Using read-only chromium-status interface\n') |
- self._check( |
- 'stderr', |
- 'CQ loop terminating\n' |
- 'Traceback (most recent call last):\n' |
- 'Stop\n\n' |
- 'Saving db...\ndb save successful.\n') |
- |
- def testDryRun(self): |
- sys.argv.extend(('--project', 'chromium')) |
- pc = PendingManagerMock(self) |
- self.mock( |
- commit_queue.projects, |
- 'load_project', |
- lambda *args: pc) |
- try: |
- commit_queue.main() |
- self.fail() |
- except Stop: |
- pass |
- self.assertEqual( |
- 'ReadOnlyCheckout', pc.context.checkout.__class__.__name__) |
- # Ugh. |
- self.assertEqual( |
- 'RietveldMock', pc.context.rietveld.__class__.__name__) |
- self._check( |
- 'stdout', |
- 'Using read-only Rietveld\n' |
- 'Using read-only checkout\n' |
- 'Using read-only chromium-status interface\n') |
- self._check( |
- 'stderr', |
- 'CQ loop terminating\n' |
- 'Traceback (most recent call last):\n' |
- 'Stop\n\n' |
- 'Saving db...\ndb save successful.\n') |
- |
- |
-if __name__ == '__main__': |
- unittest.main() |