tests/commit_queue_test.py

#!/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()