blob: 8a3bd1b60f914c877f6b2ce97478e75f6b67936c [file] [log] [blame]
# Copyright (c) 2011-2012 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.
"""Unittests for commands. Needs to be run inside of chroot for mox."""
import itertools
import logging
import mox
import os
import sys
import copy
import shutil
import time
import unittest
import constants
sys.path.insert(0, constants.SOURCE_ROOT)
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.buildbot import patch as cros_patch
from chromite.buildbot import gerrit_helper
_GetNumber = iter(itertools.count()).next
"project":"tacos/chromite", "branch":"master",
"currentPatchSet": {
"number":"2", "ref":"refs/changes/72/5172/1",
"subject":"chromite commit",
"owner":{"name":"Chromite Master", "email":""},
"open": True,
# Change-ID of a known open change in public gerrit.
class _PatchSuppression(object):
"""Mixin to suppress certain behaviour in patch objects for testing."""
def ProjectDir(self, buildroot):
# Short circuit this to not require manifest parsing.
return buildroot
def _GetUpstreamBranch(self, buildroot, force_default=False):
# Tests are built around the assumption of this upstream value.
return 'refs/remotes/origin/master'
class TestGitRepoPatch(cros_test_lib.TempDirMixin, unittest.TestCase):
# No pymox bits are to be used in this class's tests.
# This needs to actually validate git output, and git behaviour, rather
# than test our assumptions about git's behaviour/output.
class patch_kls(_PatchSuppression, cros_patch.GitRepoPatch):
"""commit abcdefgh
Author: Fake person
Date: Tue Oct 99
I am the first commit.
# Boolean controlling whether the target class natively knows its
# ChangeId; only GerritPatches do.
has_native_change_id = False
def _CreateSourceRepo(self, path):
"""Generate a new repo with a single commit."""
tmp_path = '%s-tmp' % path
self._run(['git', 'init', '--separate-git-dir', path], cwd=tmp_path)
# Add an initial commit then wipe the working tree.
self._run(['git', 'commit', '--allow-empty', '-m', 'initial commit'],
def setUp(self):
# Create an empty repo to work from.
self.source = os.path.join(self.tempdir, 'source.git')
self.default_cwd = os.path.join(self.tempdir, 'unwritable')
self.original_cwd = os.getcwd()
# Disallow write so as to smoke out any invalid writes to
# cwd.
os.chmod(self.default_cwd, 0500)
def tearDown(self):
# shutil.rmtree won't reset perms on an unwritable directory; do it
# ourselves.
os.chmod(self.default_cwd, 0700)
def _MkPatch(self, source, sha1, ref='refs/heads/master', **kwds):
return self.patch_kls(source, 'chromiumos/chromite', ref,
'origin/master', sha1=sha1, **kwds)
def _run(self, cmd, cwd=None):
# Note that cwd is intentionally set to a location the user can't write
# to; this flushes out any bad usage in the tests that would work by
# fluke of being invoked from w/in a git repo.
if cwd is None:
cwd = self.default_cwd
return cros_build_lib.RunCommandCaptureOutput(
cmd, cwd=cwd, print_cmd=False).output.strip()
def _GetSha1(self, cwd, refspec):
return self._run(['git', 'rev-list', '-n1', refspec], cwd=cwd)
def _MakeRepo(self, name, clone, branch='master', alternates=True):
path = os.path.join(self.tempdir, name)
cmd = ['git', 'clone', clone, path]
if alternates:
cmd += ['--reference', clone]
return path
def _MakeCommit(self, repo, commit=None):
if commit is None:
commit = "commit at %s" % (time.time(),)
self._run(['git', 'commit', '-a', '-m', commit], repo)
return self._GetSha1(repo, 'HEAD')
def CommitFile(self, repo, filename, content, commit=None, **kwds):
osutils.WriteFile(os.path.join(repo, filename), content)
self._run(['git', 'add', filename], repo)
sha1 = self._MakeCommit(repo, commit=commit)
if not self.has_native_change_id:
kwds.pop('ChangeId', None)
patch = self._MkPatch(repo, sha1, **kwds)
self.assertEqual(patch.sha1, sha1)
return patch
def _CommonGitSetup(self):
git1 = self._MakeRepo('git1', self.source)
git2 = self._MakeRepo('git2', self.source)
patch = self.CommitFile(git1, 'monkeys', 'foon')
return git1, git2, patch
def testFetch(self):
git1, git2, patch = self._CommonGitSetup()
self.assertEqual(patch.sha1, self._GetSha1(git2, 'FETCH_HEAD'))
# Verify reuse; specifically that Fetch doesn't actually run since
# the rev is already available locally via alternates.
patch.project_url = '/dev/null'
git3 = self._MakeRepo('git3', git2)
self.assertEqual(patch.sha1, self._GetSha1(git3, patch.sha1))
def testAlreadyApplied(self):
git1 = self._MakeRepo('git1', self.source)
patch = self.CommitFile(git1, 'monkeys', 'rule')
# Note that apply switches to a separate branch; thus the
# double apply. The first lands the change, the second
# verifies the machinery doesn't scream when we try
# landing it a second time.
def testCleanlyApply(self):
git1, git2, patch = self._CommonGitSetup()
# Clone git3 before we modify git2; else we'll just wind up
# cloning it's master.
git3 = self._MakeRepo('git3', git2)
self.assertEqual(patch.sha1, self._GetSha1(git2, 'HEAD'))
# Verify reuse; specifically that Fetch doesn't actually run since
# the object is available in alternates. testFetch partially
# validates this; the Apply usage here fully validates it via
# ensuring that the attempted Apply goes boom if it can't get the
# required sha1.
self.assertEqual(patch.sha1, self._GetSha1(git3, 'HEAD'))
def testFailsApply(self):
git1, git2, patch1 = self._CommonGitSetup()
patch2 = self.CommitFile(git2, 'monkeys', 'not foon')
# Note that Apply creates it's own branch, resetting to master
# thus we have to re-apply (even if it looks stupid, it's right).
except cros_patch.ApplyPatchException, e:
raise AssertionError("patch1.Apply didn't throw a failing "
def MakeChangeId(self, how_many=1):
l = [cros_patch.MakeChangeId() for _ in xrange(how_many)]
if how_many == 1:
return l[0]
return l
def CommitChangeIdFile(self, repo, changeid=None, extra=None,
filename='monkeys', content='flinging',
template = self.COMMIT_TEMPLATE
if changeid is None:
changeid = self.MakeChangeId()
if raw_changeid_text is None:
raw_changeid_text = 'Change-Id: %s' % (changeid,)
if extra is None:
extra = ''
commit = template % {'change-id':raw_changeid_text, 'extra':extra}
return self.CommitFile(repo, filename, content, commit=commit,
def testGerritDependencies(self):
git1 = self._MakeRepo('git1', self.source)
# Check that we handle the edge case of the first commit in a
# repo...
patch = self._MkPatch(git1, self._GetSha1(git1, 'HEAD'))
self.assertEqual(patch.GerritDependencies(git1), [])
cid1, cid2, cid3 = self.MakeChangeId(3)
patch = self.CommitChangeIdFile(git1, cid1)
# Since it's parent is ToT, there are no deps.
self.assertEqual(patch.GerritDependencies(git1), [])
patch = self.CommitChangeIdFile(git1, cid2, content='poo')
self.assertEqual(patch.GerritDependencies(git1), [cid1])
# Check the behaviour for missing ChangeId in a parent next.
patch = self.CommitChangeIdFile(git1, cid1, content='thus',
# Note the ordering; leftmost needs to be the nearest child of the commit.
self.assertEqual(patch.GerritDependencies(git1), [cid2, cid1])
patch = self.CommitChangeIdFile(git1, cid3, content='the glass walls.')
patch.GerritDependencies, git1)
def _CheckPaladin(self, repo, master_id, ids, extra):
patch = self.CommitChangeIdFile(
repo, master_id, extra=extra,
filename='paladincheck', content=str(_GetNumber()))
deps = patch.PaladinDependencies(repo)
# Assert that are parsing unique'ifies the results.
self.assertEqual(len(deps), len(set(deps)))
deps = set(deps)
ids = set(ids)
self.assertEqual(ids, deps)
set(cros_patch.FormatChangeId(x) for x in deps if x[0] in 'Ii'),
set(x for x in ids if x[0] in 'Ii'))
return patch
def testPaladinDependencies(self):
git1 = self._MakeRepo('git1', self.source)
cid1, cid2, cid3, cid4 = self.MakeChangeId(4)
# Verify it handles nonexistant CQ-DEPEND.
self._CheckPaladin(git1, cid1, [], '')
# Single key, single value.
self._CheckPaladin(git1, cid1, [cid2],
'CQ-DEPEND=%s' % cid2)
# Single key, multiple values
self._CheckPaladin(git1, cid1, [cid2, cid3],
'CQ-DEPEND=%s %s' % (cid2, cid3))
# Dumb comma behaviour
self._CheckPaladin(git1, cid1, [cid2, cid3],
'CQ-DEPEND=%s, %s,' % (cid2, cid3))
# Multiple keys.
self._CheckPaladin(git1, cid1, [cid2, cid3, cid4],
'CQ-DEPEND=%s, %s\nCQ-DEPEND=%s' % (cid2, cid3, cid4))
# Ensure it goes boom on invalid data.
self.assertRaises(cros_patch.BrokenCQDepends, self._CheckPaladin,
git1, cid1, [], 'CQ-DEPEND=monkeys')
self.assertRaises(cros_patch.BrokenCQDepends, self._CheckPaladin,
git1, cid1, [], 'CQ-DEPEND=%s monkeys' % (cid2,))
# Validate numeric is allowed.
self._CheckPaladin(git1, cid1, [cid2, '1'], 'CQ-DEPEND=1 %s' % cid2)
# Validate that it unique'ifies the results.
self._CheckPaladin(git1, cid1, ['1'], 'CQ-DEPEND=1 1')
class TestLocalPatchGit(TestGitRepoPatch):
class patch_kls(_PatchSuppression, cros_patch.LocalPatch):
def setUp(self):
self.sourceroot = os.path.join(self.tempdir, 'sourceroot')
def _MkPatch(self, source, sha1, ref='refs/heads/master', **kwds):
return self.patch_kls(source, 'chromiumos/chromite', ref,
'origin/master', sha1,
def testUpload(self):
def ProjectDirMock(sourceroot):
return git1
git1, git2, patch = self._CommonGitSetup()
git2_sha1 = self._GetSha1(git2, 'HEAD')
patch.ProjectDir = ProjectDirMock
# First suppress carbon copy behaviour so we verify pushing
# plain works.
sha1 = patch.sha1
patch._GetCarbonCopy = lambda: sha1
patch.Upload(git2, 'refs/testing/test1')
self.assertEqual(self._GetSha1(git2, 'refs/testing/test1'),
# Enable CarbonCopy behaviour; verify it lands a different
# sha1. Additionally verify it didn't corrupt the patch's sha1 locally.
del patch._GetCarbonCopy
patch.Upload(git2, 'refs/testing/test2')
self.assertNotEqual(self._GetSha1(git2, 'refs/testing/test2'),
self.assertEqual(patch.sha1, sha1)
# Ensure the carbon creation didn't damage the target repo.
self.assertEqual(self._GetSha1(git1, 'HEAD'), sha1)
# Ensure we didn't damage the target repo's state at all.
self.assertEqual(git2_sha1, self._GetSha1(git2, 'HEAD'))
# Ensure the content is the same.
base = ['git', 'show']
self._run(base + ['refs/testing/test1:monkeys'], git2),
self._run(base + ['refs/testing/test2:monkeys'], git2))
base = ['git', 'log', '--format=%B', '-n1']
self._run(base + ['refs/testing/test1'], git2),
self._run(base + ['refs/testing/test2'], git2))
class TestUploadedLocalPatch(TestGitRepoPatch):
PROJECT = 'chromiumos/chromite'
ORIGINAL_BRANCH = 'original_branch'
ORIGINAL_SHA1 = 'ffffffff'
class patch_kls(_PatchSuppression, cros_patch.UploadedLocalPatch):
def _MkPatch(self, source, sha1, ref='refs/heads/master', **kwds):
return self.patch_kls(source, self.PROJECT, ref,
'origin/master', self.ORIGINAL_BRANCH,
self.ORIGINAL_SHA1, carbon_copy_sha1=sha1, **kwds)
def testStringRepresentation(self):
git1, git2, patch = self._CommonGitSetup()
str_rep = str(patch).split(':')
for element in [self.PROJECT, self.ORIGINAL_BRANCH, self.ORIGINAL_SHA1]:
self.assertTrue(element in str_rep)
class TestGerritPatch(TestGitRepoPatch):
has_native_change_id = True
class patch_kls(_PatchSuppression, cros_patch.GerritPatch):
# Suppress the behaviour pointing the project url at actual gerrit,
# instead slaving it back to a local repo for tests.
def _GetProjectUrl(self, project, internal):
assert hasattr(self, 'patch_dict')
return self.patch_dict['_unittest_url_bypass']
def test_GetProjectUrl(self):
# We test this since we explicitly override the behaviour
# for all other usage.
kls = cros_patch.GerritPatch
kls._GetProjectUrl('monkeys', False),
os.path.join(kls._PUBLIC_URL, 'monkeys'))
kls._GetProjectUrl('monkeys', True),
os.path.join(constants.GERRIT_INT_SSH_URL, 'monkeys'))
def test_json(self):
return copy.deepcopy(FAKE_PATCH_JSON)
def _MkPatch(self, source, sha1, ref='refs/heads/master', **kwds):
json = self.test_json
internal = kwds.pop('internal', False)
suppress_branch = kwds.pop('suppress_branch', False)
change_id = kwds.pop('ChangeId', None)
if change_id is None:
change_id = self.MakeChangeId()
change_num, patch_num = _GetNumber(), _GetNumber()
# Note we intentionally use a gerrit like refspec here; we want to
# ensure that none of our common code pathways puke on a non head/tag.
refspec = 'refs/changes/%i/%i/%i' % (
change_num % 100, change_num + 1000, patch_num)
dict(number=patch_num, ref=refspec, revision=sha1))
json['branch'] = os.path.basename(ref)
json['_unittest_url_bypass'] = source
json['id'] = change_id
obj = self.patch_kls(json.copy(), internal)
self.assertEqual(obj.patch_dict, json)
self.assertEqual(obj.internal, internal)
self.assertEqual(obj.project, json['project'])
self.assertEqual(obj.ref, refspec)
self.assertEqual(, change_id)
# Now make the fetching actually work, if desired.
if not suppress_branch:
# Note that a push is needed here, rather than a branch; branch
# will just make it under refs/heads, we want it literally in
# refs/changes/
self._run(['git', 'push', source, '%s:%s' % (sha1, refspec)], source)
return obj
def testIsAlreadyMerged(self):
# Note that these are magic constants- they're known to be
# merged (and the other abandoned) in public gerrit.
# If old changes are ever flushed, or something 'special' occurs,
# then this will break. That it's an acceptable risk.
# Note we should be checking a known open one; seems rather likely
# that'll get closed inadvertantly thus breaking the tests (not
# an acceptable risk in the authors opinion).
merged, abandoned, still_open = gerrit_helper.GetGerritPatchInfo(
def test_json(self):
return copy.deepcopy(FAKE_PATCH_JSON)
class PrepareLocalPatchesTests(mox.MoxTestBase):
def setUp(self):
self.patches = ['my/project:mybranch']
self.mox.StubOutWithMock(cros_build_lib, 'GetProjectDir')
self.mox.StubOutWithMock(cros_build_lib, 'GetCurrentBranch')
self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')
self.manifest = self.mox.CreateMock(cros_build_lib.ManifestCheckout)
def VerifyPatchInfo(self, patch_info, project, branch, tracking_branch):
"""Check the returned GitRepoPatchInfo against golden values."""
self.assertEquals(patch_info.project, project)
self.assertEquals(patch_info.ref, branch)
self.assertEquals(patch_info.tracking_branch, tracking_branch)
def testBranchSpecifiedSuccessRun(self):
"""Test success with branch specified by user."""
output_obj = self.mox.CreateMock(cros_build_lib.CommandResult)
output_obj.output= '12345'
self.manifest.GetProjectPath('my/project', True).AndReturn('mydir')
patch_info = cros_patch.PrepareLocalPatches(self.manifest, self.patches)
self.VerifyPatchInfo(patch_info[0], 'my/project', 'mybranch', 'kernel')
def testBranchSpecifiedNoChanges(self):
"""Test when no changes on the branch specified by user."""
output_obj = self.mox.CreateMock(cros_build_lib.CommandResult)
self.manifest.GetProjectPath('my/project', True).AndReturn('mydir')
self.manifest, self.patches)
class ApplyLocalPatchesTests(mox.MoxTestBase):
def testWrongTrackingBranch(self):
"""When the original patch branch does not track buildroot's branch."""
patch = cros_patch.GitRepoPatch('/path/to/my/project.git',
'my/project', 'mybranch',
self.assertRaises(cros_patch.PatchException, patch.ApplyIntoGitRepo,
'/tmp/notadirectory', 'origin/R19')
if __name__ == '__main__':
logging_format = '%(asctime)s - %(filename)s - %(levelname)-8s: %(message)s'
date_format = constants.LOGGER_DATE_FMT
logging.basicConfig(level=logging.DEBUG, format=logging_format,