blob: 37ade864d497edfe82420b1a39126ca716b1085a [file] [log] [blame]
# 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.
"""Common mocks."""
import copy
import os
import sys
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(ROOT_DIR, '..'))
import find_depot_tools # pylint: disable=W0611
import breakpad
import patch
import async_push
import context
import pending_manager
from testing_support import auto_stub
'Index: chrome/\n'
'--- chrome/\t(revision 74690)\n'
'+++ chrome/\t(working copy)\n'
'@@ -80,10 +80,10 @@\n'
' // Foo\n'
' // Bar\n'
' void foo() {\n'
'- return bar;\n'
'+ return foo;\n'
' }\n'
' \n'
' \n')
class RietveldMock(auto_stub.SimpleMock):
url = 'http://nowhere'
email = 'fake_email'
password = 'fake_password'
def __init__(self, unit_test):
super(RietveldMock, self).__init__(unit_test)
self.issues = {
31337: {
"description": u"foo",
"created": "2010-12-27 03:23:31.149045",
"cc": ["",],
"reviewers": [""],
"owner_email": "",
"patchsets": [1],
"modified": "2011-01-10 20:52:39.127231",
"private": False,
"base_url": "svn://fake/repo",
"closed": False,
"owner": "Author",
"issue": 31337,
"subject": 'foo',
"messages": [
"date": "2010-12-27 03:23:32.489999",
"text": u"hi!",
"sender": "",
"recipients": ["", ""],
"approval": False,
"date": "2010-12-27 03:23:32.489999",
"text": u"hi!",
"sender": "",
"recipients": ["", ""],
"approval": True,
"commit": True,
self.patchsets = []
# Key is (issue, patchset)
self.patchsets_properties = {}
def get_pending_issues(self):
return self.issues.keys()
def get_issue_properties(self, issue_id, _):
return copy.deepcopy(self.issues[issue_id])
def get_patchset_properties(self, issue_id, patchset):
key = (issue_id, patchset)
key in self.patchsets_properties, (key, self.patchsets_properties))
return copy.deepcopy(self.patchsets_properties[key])
def close_issue(self, *args, **kwargs):
self._register_call(*args, **kwargs)
def update_description(self, *args, **kwargs):
self._register_call(*args, **kwargs)
def set_flag(self, *args, **kwargs):
self._register_call(*args, **kwargs)
return True
def add_comment(self, *args, **kwargs):
self._register_call(*args, **kwargs)
def get_patch(self, _issue, _patchset):
patch.FilePatchDiff('chrome/', SVN_PATCH, []),
patch.FilePatchDelete('other/place/foo', True),
patch.FilePatchBinary('foo', 'data', [], True),
return self.patchsets[-1]
def _send(self, *args, **kwargs):
self._register_call(*args, **kwargs)
def trigger_try_jobs(self, *args, **kwargs):
self._register_call(*args, **kwargs)
def xsrf_token():
return 'XSRF-free'
def post(self, *args, **kwargs):
self._register_call(*args, **kwargs)
class SvnCheckoutMock(auto_stub.SimpleMock):
def __init__(self, *args):
super(SvnCheckoutMock, self).__init__(*args)
self.project_path = os.getcwd()
self.project_name = os.path.basename(self.project_path)
self.post_processors = []
def prepare(self, revision):
# prepare() should always return a valid revision.
return revision or 124
def apply_patch(self, *args):
def commit(self, *args):
return 125
def get_settings(_key):
return None
def revisions(_revision1, _revision2):
"""Number of revisions between 2 revisions."""
return 3
class AsyncPushMock(auto_stub.SimpleMock, async_push.AsyncPushNoop):
def __init__(self, *args):
auto_stub.SimpleMock.__init__(self, *args)
self.queue = []
def send(self, packet, pending):
self.queue.append(self._package(packet, pending))
def pop_packets(self):
packets = self.queue
self.queue = []
return packets
def check_packets(self, expected):
self.assertEqual(expected, self.pop_packets())
def check_names(self, expected):
self.assertEqual(expected, [i['verification'] for i in self.pop_packets()])
class BuildbotBuilder(object):
def __init__(self):
self.builds = {}
class Data(object):
def __init__(self): = {}
self.pending_builds = Data()
class BuildbotBuild(object):
def __init__(self, revision, key, completed, parent_key):
self.properties_as_dict = {
'try_job_key': key,
'parent_try_job_key': parent_key,
self.revision = revision
self.start_time = 1
self.steps = []
self.completed = completed
self.duration = 180
class BuildbotBuildStep(object):
def __init__(self, name, result): = name
self.simplified_result = result
class BuildbotBuilders(object):
def __init__(self, initial):
self.values = initial.copy()
def __getitem__(self, key):
return self.values[key]
def keys(self):
return self.values.keys()
def setdefault(self, key, value):
return self.values.setdefault(key, value)
class BuildbotMock(auto_stub.SimpleMock):
"""Stateful try server mock."""
def __init__(self, unit_test):
super(BuildbotMock, self).__init__(unit_test) = BuildbotBuilders({
u'mac': BuildbotBuilder(),
u'linux': BuildbotBuilder(),
def discard(self):
def add_build(self, builder, number, revision, key, completed, parent_key):
"""Add a build to a builder."""
builder =, BuildbotBuilder())
builder.builds[number] = BuildbotBuild(revision, key, completed, parent_key)
return builder.builds[number]
class TestCase(auto_stub.TestCase):
def setUp(self):
super(TestCase, self).setUp()
self.mock(breakpad, 'SendStack', self._send_stack_mock)
self.context = context.Context(
RietveldMock(self), SvnCheckoutMock(self), AsyncPushMock(self), False)
self.pending = pending_manager.PendingCommit(
def tearDown(self):
if not self.has_failed():
super(TestCase, self).tearDown()
def _send_stack_mock(self, last_tb, stack, *_args, **_kwargs):
"""Fails a test that calls SendStack.
In practice it doesn't happen when a test pass but will when a test fails so
hook it here so breakpad doesn't send too many stack traces to maintainers.
"""'%s, %s' % (last_tb, stack))