blob: b7cdea594c7654259e2beb71b5c27dae0990c750 [file] [log] [blame]
# Copyright (c) 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.
"""Unit tests for cros_portage_upgrade.py."""
from __future__ import print_function
import exceptions
import filecmp
import mox
import os
import re
import shutil
import tempfile
import unittest
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import upgrade_table as utable
from chromite.scripts import cros_portage_upgrade as cpu
from chromite.scripts import parallel_emerge
from portage.package.ebuild import config as portcfg # pylint: disable=F0401
# This no longer gets installed by portage. Stub it as None to avoid
# `cros lint` errors.
#from portage.tests.resolver import ResolverPlayground as respgnd
respgnd = None
# Enable color invariably. Since we rely on color for error/warn message
# recognition, leaving this to be decided based on stdout being a tty
# will make the tests fail/succeed based on how they are run.
# pylint: disable=W0102,W0212,E1120,E1101
cpu.oper._color._enabled = True
DEFAULT_PORTDIR = '/usr/portage'
# Configuration for generating a temporary valid ebuild hierarchy.
# ResolverPlayground sets up a default profile with ARCH=x86, so
# other architectures are irrelevant for now.
DEFAULT_ARCH = 'x86'
EBUILDS = {
'dev-libs/A-1': {'RDEPEND': 'dev-libs/B'},
'dev-libs/A-2': {'RDEPEND': 'dev-libs/B'},
'dev-libs/B-1': {'RDEPEND': 'dev-libs/C'},
'dev-libs/B-2': {'RDEPEND': 'dev-libs/C'},
'dev-libs/C-1': {},
'dev-libs/C-2': {},
'dev-libs/D-1': {'RDEPEND': '!dev-libs/E'},
'dev-libs/D-2': {},
'dev-libs/D-3': {},
'dev-libs/E-2': {'RDEPEND': '!dev-libs/D'},
'dev-libs/E-3': {},
'dev-libs/F-1': {'SLOT': '1'},
'dev-libs/F-2': {'SLOT': '2'},
'dev-libs/F-2-r1': {
'SLOT': '2',
'KEYWORDS': '~amd64 ~x86 ~arm',
},
'dev-apps/X-1': {
'EAPI': '3',
'SLOT': '0',
'KEYWORDS': 'amd64 arm x86',
'RDEPEND': '=dev-libs/C-1',
},
'dev-apps/Y-2': {
'EAPI': '3',
'SLOT': '0',
'KEYWORDS': 'amd64 arm x86',
'RDEPEND': '=dev-libs/C-2',
},
'chromeos-base/flimflam-0.0.1-r228': {
'EAPI': '2',
'SLOT': '0',
'KEYWORDS': 'amd64 x86 arm',
'RDEPEND': '>=dev-libs/D-2',
},
'chromeos-base/flimflam-0.0.2-r123': {
'EAPI': '2',
'SLOT': '0',
'KEYWORDS': '~amd64 ~x86 ~arm',
'RDEPEND': '>=dev-libs/D-3',
},
'chromeos-base/libchrome-57098-r4': {
'EAPI': '2',
'SLOT': '0',
'KEYWORDS': 'amd64 x86 arm',
'RDEPEND': '>=dev-libs/E-2',
},
'chromeos-base/libcros-1': {
'EAPI': '2',
'SLOT': '0',
'KEYWORDS': 'amd64 x86 arm',
'RDEPEND': 'dev-libs/B dev-libs/C chromeos-base/flimflam',
'DEPEND': ('dev-libs/B dev-libs/C chromeos-base/flimflam '
'chromeos-base/libchrome'),
},
'virtual/libusb-0': {
'EAPI': '2',
'SLOT': '0',
'RDEPEND': (
'|| ( >=dev-libs/libusb-0.1.12-r1:0 dev-libs/libusb-compat '
'>=sys-freebsd/freebsd-lib-8.0[usb] )'
),
},
'virtual/libusb-1': {
'EAPI':'2', 'SLOT': '1',
'RDEPEND': '>=dev-libs/libusb-1.0.4:1',
},
'dev-libs/libusb-0.1.13': {},
'dev-libs/libusb-1.0.5': {'SLOT':'1'},
'dev-libs/libusb-compat-1': {},
'sys-freebsd/freebsd-lib-8': {'IUSE': '+usb'},
'sys-fs/udev-164': {'EAPI': '1', 'RDEPEND': 'virtual/libusb:0'},
'virtual/jre-1.5.0': {
'SLOT': '1.5',
'RDEPEND': '|| ( =dev-java/sun-jre-bin-1.5.0* =virtual/jdk-1.5.0* )',
},
'virtual/jre-1.5.0-r1': {
'SLOT': '1.5',
'RDEPEND': '|| ( =dev-java/sun-jre-bin-1.5.0* =virtual/jdk-1.5.0* )',
},
'virtual/jre-1.6.0': {
'SLOT': '1.6',
'RDEPEND': '|| ( =dev-java/sun-jre-bin-1.6.0* =virtual/jdk-1.6.0* )',
},
'virtual/jre-1.6.0-r1': {
'SLOT': '1.6',
'RDEPEND': '|| ( =dev-java/sun-jre-bin-1.6.0* =virtual/jdk-1.6.0* )',
},
'virtual/jdk-1.5.0': {
'SLOT': '1.5',
'RDEPEND': '|| ( =dev-java/sun-jdk-1.5.0* dev-java/gcj-jdk )',
},
'virtual/jdk-1.5.0-r1': {
'SLOT': '1.5',
'RDEPEND': '|| ( =dev-java/sun-jdk-1.5.0* dev-java/gcj-jdk )',
},
'virtual/jdk-1.6.0': {
'SLOT': '1.6',
'RDEPEND': '|| ( =dev-java/icedtea-6* =dev-java/sun-jdk-1.6.0* )',
},
'virtual/jdk-1.6.0-r1': {
'SLOT': '1.6',
'RDEPEND': '|| ( =dev-java/icedtea-6* =dev-java/sun-jdk-1.6.0* )',
},
'dev-java/gcj-jdk-4.5': {},
'dev-java/gcj-jdk-4.5-r1': {},
'dev-java/icedtea-6.1': {},
'dev-java/icedtea-6.1-r1': {},
'dev-java/sun-jdk-1.5': {'SLOT': '1.5'},
'dev-java/sun-jdk-1.6': {'SLOT': '1.6'},
'dev-java/sun-jre-bin-1.5': {'SLOT': '1.5'},
'dev-java/sun-jre-bin-1.6': {'SLOT': '1.6'},
'dev-java/ant-core-1.8': {'DEPEND': '>=virtual/jdk-1.4'},
'dev-db/hsqldb-1.8': {'RDEPEND': '>=virtual/jre-1.6'},
}
WORLD = [
'dev-libs/A',
'dev-libs/D',
'virtual/jre',
]
INSTALLED = {
'dev-libs/A-1': {},
'dev-libs/B-1': {},
'dev-libs/C-1': {},
'dev-libs/D-1': {},
'virtual/jre-1.5.0': {
'SLOT': '1.5',
'RDEPEND': '|| ( =virtual/jdk-1.5.0* =dev-java/sun-jre-bin-1.5.0* )',
},
'virtual/jre-1.6.0': {
'SLOT': '1.6',
'RDEPEND': '|| ( =virtual/jdk-1.6.0* =dev-java/sun-jre-bin-1.6.0* )',
},
'virtual/jdk-1.5.0': {
'SLOT': '1.5',
'RDEPEND': '|| ( =dev-java/sun-jdk-1.5.0* dev-java/gcj-jdk )',
},
'virtual/jdk-1.6.0': {
'SLOT': '1.6',
'RDEPEND': '|| ( =dev-java/icedtea-6* =dev-java/sun-jdk-1.6.0* )',
},
'dev-java/gcj-jdk-4.5': {},
'dev-java/icedtea-6.1': {},
'virtual/libusb-0': {
'EAPI': '2',
'SLOT': '0',
'RDEPEND': (
'|| ( >=dev-libs/libusb-0.1.12-r1:0 dev-libs/libusb-compat '
'>=sys-freebsd/freebsd-lib-8.0[usb] )'
)
},
}
# For verifying dependency graph results
GOLDEN_DEP_GRAPHS = {
'dev-libs/A-2': {
'needs': {'dev-libs/B-2': 'runtime'},
'action': 'merge',
},
'dev-libs/B-2': {'needs': {'dev-libs/C-2': 'runtime'}},
'dev-libs/C-2': {'needs': {}},
'dev-libs/D-3': {'needs': {}},
# TODO(mtennant): Bug in parallel_emerge deps graph makes blocker show up
# for E-3, rather than in just E-2 where it belongs. See crosbug.com/22190.
# To repeat bug, swap the commented status of next two lines.
#'dev-libs/E-3': {'needs': {}},
'dev-libs/E-3': {'needs': {'dev-libs/D-3': 'blocker'}},
'chromeos-base/libcros-1': {
'needs': {
'dev-libs/B-2': 'runtime/buildtime',
'dev-libs/C-2': 'runtime/buildtime',
'chromeos-base/libchrome-57098-r4': 'buildtime',
'chromeos-base/flimflam-0.0.1-r228': 'runtime/buildtime'
}
},
'chromeos-base/flimflam-0.0.1-r228': {
'needs': {'dev-libs/D-3': 'runtime'},
},
'chromeos-base/libchrome-57098-r4': {
'needs': {'dev-libs/E-3': 'runtime'},
},
}
# For verifying dependency set results
GOLDEN_DEP_SETS = {
'dev-libs/A': set(['dev-libs/A-2', 'dev-libs/B-2', 'dev-libs/C-2']),
'dev-libs/B': set(['dev-libs/B-2', 'dev-libs/C-2']),
'dev-libs/C': set(['dev-libs/C-2']),
'dev-libs/D': set(['dev-libs/D-3']),
'virtual/libusb': set(['virtual/libusb-1', 'dev-libs/libusb-1.0.5']),
'chromeos-base/libcros': set(['chromeos-base/libcros-1',
'dev-libs/B-2',
'chromeos-base/libchrome-57098-r4',
'dev-libs/E-3',
'chromeos-base/flimflam-0.0.1-r228',
'dev-libs/D-3',
'dev-libs/C-2',])
}
def _GetGoldenDepsSet(pkg):
"""Retrieve the golden dependency set for |pkg| from GOLDEN_DEP_SETS."""
return GOLDEN_DEP_SETS.get(pkg, None)
def _VerifyDepsGraph(deps_graph, pkgs):
for pkg in pkgs:
if not _VerifyDepsGraphOnePkg(deps_graph, pkg):
return False
return True
def _VerifyDepsGraphOnePkg(deps_graph, pkg):
"""Verfication function for Mox to validate deps graph for |pkg|."""
if deps_graph is None:
print('Error: no dependency graph passed into _GetPreOrderDepGraph')
return False
if type(deps_graph) != dict:
print('Error: dependency graph is expected to be a dict. Instead:\n%r' %
deps_graph)
return False
validated = True
golden_deps_set = _GetGoldenDepsSet(pkg)
if golden_deps_set == None:
print('Error: golden dependency list not configured for %s package' % pkg)
validated = False
# Verify dependencies, by comparing them to GOLDEN_DEP_GRAPHS
for p in deps_graph:
golden_pkg_info = None
try:
golden_pkg_info = GOLDEN_DEP_GRAPHS[p]
except KeyError:
print('Error: golden dependency graph not configured for %s package' % p)
validated = False
continue
pkg_info = deps_graph[p]
for key in golden_pkg_info:
golden_value = golden_pkg_info[key]
value = pkg_info[key]
if not value == golden_value:
print('Error: while verifying "%s" value for %s package,'
' expected:\n%r\nBut instead found:\n%r'
% (key, p, golden_value, value))
validated = False
if not validated:
print('Error: dependency graph for %s is not as expected. Instead:\n%r' %
(pkg, deps_graph))
return validated
def _GenDepsGraphVerifier(pkgs):
"""Generate a graph verification function for the given package."""
return lambda deps_graph: _VerifyDepsGraph(deps_graph, pkgs)
class ManifestLine(object):
"""Class to represent a Manifest line."""
__slots__ = (
'type', # DIST, EBUILD, etc.
'file',
'size',
'RMD160',
'SHA1',
'SHA256',
)
__attrlist__ = __slots__
def __init__(self, line=None, **kwargs):
"""Parse |line| from manifest file."""
if line:
tokens = line.split()
self.type = tokens[0]
self.file = tokens[1]
self.size = tokens[2]
self.RMD160 = tokens[4]
self.SHA1 = tokens[6]
self.SHA256 = tokens[8]
assert tokens[3] == 'RMD160'
assert tokens[5] == 'SHA1'
assert tokens[7] == 'SHA256'
# Entries in kwargs are overwrites.
for attr in self.__attrlist__:
if attr in kwargs or not hasattr(self, attr):
setattr(self, attr, kwargs.get(attr))
def __str__(self):
return ('%s %s %s RMD160 %s SHA1 %s SHA256 %s' %
(self.type, self.file, self.size,
self.RMD160, self.SHA1, self.SHA256))
def __eq__(self, other):
"""Equality support."""
if type(self) != type(other):
return False
no_attr = object()
for attr in self.__attrlist__:
if getattr(self, attr, no_attr) != getattr(other, attr, no_attr):
return False
return True
def __ne__(self, other):
"""Inequality for completeness."""
return not self == other
class RunCommandResult(object):
"""Class to simulate result of cros_build_lib.RunCommand."""
__slots__ = ['returncode', 'output']
def __init__(self, returncode, output):
self.returncode = returncode
self.output = output
class PInfoTest(cros_test_lib.OutputTestCase):
"""Tests for the PInfo class."""
def testInit(self):
pinfo = cpu.PInfo(category='SomeCat', user_arg='SomeArg')
self.assertEquals('SomeCat', pinfo.category)
self.assertEquals('SomeArg', pinfo.user_arg)
self.assertEquals(None, pinfo.cpv)
self.assertEquals(None, pinfo.overlay)
self.assertRaises(AttributeError, getattr, pinfo, 'foobar')
def testEqAndNe(self):
pinfo1 = cpu.PInfo(category='SomeCat', user_arg='SomeArg')
self.assertEquals(pinfo1, pinfo1)
self.assertTrue(pinfo1 == pinfo1)
self.assertFalse(pinfo1 != pinfo1)
pinfo2 = cpu.PInfo(category='SomeCat', user_arg='SomeArg')
self.assertEquals(pinfo1, pinfo2)
self.assertTrue(pinfo1 == pinfo2)
self.assertFalse(pinfo1 != pinfo2)
pinfo3 = cpu.PInfo(category='SomeCat', user_arg='SomeOtherArg')
self.assertNotEquals(pinfo1, pinfo3)
self.assertFalse(pinfo1 == pinfo3)
self.assertTrue(pinfo1 != pinfo3)
pinfo4 = cpu.PInfo(category='SomeCat', slot='SomeSlot')
self.assertNotEquals(pinfo1, pinfo4)
self.assertFalse(pinfo1 == pinfo4)
self.assertTrue(pinfo1 != pinfo4)
class CpuTestBase(cros_test_lib.MoxTempDirTestOutputCase):
"""Base class for all test classes in this file."""
__slots__ = [
'playground',
'playground_envvars',
]
def setUp(self):
self.playground = None
self.playground_envvars = None
def tearDown(self):
self._TearDownPlayground()
def _SetUpPlayground(self, ebuilds=EBUILDS, installed=INSTALLED, world=WORLD,
active=True):
"""Prepare the temporary ebuild playground (ResolverPlayground).
This leverages test code in existing Portage modules to create an ebuild
hierarchy. This can be a little slow.
Args:
ebuilds: A list of hashes representing ebuild files in a portdir.
installed: A list of hashes representing ebuilds files already installed.
world: A list of lines to simulate in the world file.
active: True means that os.environ variables should be set
to point to the created playground, such that Portage tools
(such as emerge) can be run now using the playground as the active
PORTDIR. Also saves the playground as self._playground. If |active|
is False, then no os.environ variables are set and playground is
not saved (only returned).
Returns:
Tuple (playground, envvars).
"""
# TODO(mtennant): Support multiple overlays? This essentially
# creates just a default overlay.
# Also note that ResolverPlayground assumes ARCH=x86 for the
# default profile it creates.
playground = respgnd.ResolverPlayground(ebuilds=ebuilds,
installed=installed,
world=world)
# Set all envvars needed by parallel_emerge, since parallel_emerge
# normally does that when --board is given.
eroot = self._GetPlaygroundRoot(playground)
portdir = self._GetPlaygroundPortdir(playground)
envvars = {
'PORTAGE_CONFIGROOT': eroot,
'ROOT': eroot,
'PORTDIR': portdir,
# See _GenPortageEnvvars for more info on this setting.
'PORTDIR_OVERLAY': portdir,
}
if active:
for envvar in envvars:
os.environ[envvar] = envvars[envvar]
self.playground = playground
self.playground_envvars = envvars
return (playground, envvars)
def _GetPlaygroundRoot(self, playground=None):
"""Get the temp dir playground is using as ROOT."""
if playground is None:
playground = self.playground
eroot = playground.eroot
if eroot[-1:] == '/':
eroot = eroot[:-1]
return eroot
def _GetPlaygroundPortdir(self, playground=None):
"""Get the temp portdir without playground."""
if playground is None:
playground = self.playground
eroot = self._GetPlaygroundRoot(playground)
portdir = '%s%s' % (eroot, DEFAULT_PORTDIR)
return portdir
def _TearDownPlayground(self):
"""Delete the temporary ebuild playground files."""
if self.playground:
self.playground.cleanup()
self.playground = None
self.playground_envvars = None
def _MockUpgrader(self, cmdargs=None, **kwargs):
"""Set up a mocked Upgrader object with the given args."""
upgrader_slot_defaults = {
'_curr_arch': DEFAULT_ARCH,
'_curr_board': 'some_board',
'_unstable_ok': False,
'_verbose': False,
}
upgrader = self.mox.CreateMock(cpu.Upgrader)
# Initialize each attribute with default value.
for slot in cpu.Upgrader.__slots__:
value = upgrader_slot_defaults.get(slot, None)
upgrader.__setattr__(slot, value)
# Initialize with command line if given.
if cmdargs is not None:
parser = cpu._CreateParser()
options = parser.parse_args(cmdargs)
cpu.Upgrader.__init__(upgrader, options)
# Override Upgrader attributes if requested.
for slot in cpu.Upgrader.__slots__:
value = None
if slot in kwargs:
upgrader.__setattr__(slot, kwargs[slot])
return upgrader
@unittest.skip('relies on portage module not currently available')
class CopyUpstreamTest(CpuTestBase):
"""Test Upgrader._CopyUpstreamPackage, _CopyUpstreamEclass, _CreateManifest"""
# This is a hack until crosbug.com/21965 is completed and upstreamed
# to Portage. Insert eclass simulation into tree.
def _AddEclassToPlayground(self, eclass, lines=None,
ebuilds=None, missing=False):
"""Hack to insert an eclass into the playground source.
Args:
eclass: Name of eclass to create (without .eclass suffix). Will be
created as an empty file unless |lines| is specified.
lines: Lines of text to put into created eclass, if given.
ebuilds: List of ebuilds to put inherit line into. Should be path
to ebuild from playground portdir.
missing: If True, do not actually create the eclass file. Only makes
sense if |ebuilds| is non-empty, presumably to test inherit failure.
Returns:
Full path to the eclass file, whether it was created or not.
"""
portdir = self._GetPlaygroundPortdir()
eclass_path = os.path.join(portdir, 'eclass', '%s.eclass' % eclass)
# Create the eclass file
osutils.WriteFile(eclass_path, '\n'.join(lines if lines else []))
# Insert the inherit line into the ebuild file, if requested.
if ebuilds:
for ebuild in ebuilds:
ebuild_path = os.path.join(portdir, ebuild)
text = osutils.ReadFile(ebuild_path)
def repl(match):
return match.group(1) + '\ninherit ' + eclass
text = re.sub(r'(EAPI.*)', repl, text)
osutils.WriteFile(ebuild_path, text)
# Remove the Manifest file
os.remove(os.path.join(os.path.dirname(ebuild_path), 'Manifest'))
# Recreate the Manifests using the ebuild utility.
cmd = ['ebuild', ebuild_path, 'manifest']
cros_build_lib.RunCommand(cmd, print_cmd=False, redirect_stdout=True,
combine_stdout_stderr=True)
# If requested, remove the eclass.
if missing:
os.remove(eclass_path)
return eclass_path
#
# _IdentifyNeededEclass
#
def _TestIdentifyNeededEclass(self, cpv, ebuild, eclass, create_eclass):
"""Test Upgrader._IdentifyNeededEclass"""
self._SetUpPlayground()
portdir = self._GetPlaygroundPortdir()
mocked_upgrader = self._MockUpgrader(cmdargs=[],
_stable_repo=portdir)
self._AddEclassToPlayground(eclass,
ebuilds=[ebuild],
missing=not create_eclass)
# Add test-specific mocks/stubs
# Replay script
envvars = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
mocked_upgrader._curr_arch,
unstable_ok=True)
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok=True,
).AndReturn(envvars)
mocked_upgrader._GetBoardCmd('equery').AndReturn('equery')
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
result = cpu.Upgrader._IdentifyNeededEclass(mocked_upgrader, cpv)
self.mox.VerifyAll()
return result
def testIdentifyNeededEclassMissing(self):
result = self._TestIdentifyNeededEclass('dev-libs/A-2',
'dev-libs/A/A-2.ebuild',
'inheritme',
False)
self.assertEquals('inheritme.eclass', result)
def testIdentifyNeededEclassOK(self):
result = self._TestIdentifyNeededEclass('dev-libs/A-2',
'dev-libs/A/A-2.ebuild',
'inheritme',
True)
self.assertTrue(result is None)
#
# _CopyUpstreamEclass
#
def _TestCopyUpstreamEclass(self, eclass, do_copy,
local_copy_identical=None, error=None):
"""Test Upgrader._CopyUpstreamEclass"""
self._SetUpPlayground()
upstream_portdir = self._GetPlaygroundPortdir()
portage_stable = self.tempdir
mocked_upgrader = self._MockUpgrader(_curr_board=None,
_upstream=upstream_portdir,
_stable_repo=portage_stable)
eclass_subpath = os.path.join('eclass', eclass + '.eclass')
eclass_path = os.path.join(portage_stable, eclass_subpath)
upstream_eclass_path = None
if do_copy or local_copy_identical:
lines = ['#Dummy eclass', '#Hi']
upstream_eclass_path = self._AddEclassToPlayground(eclass,
lines=lines)
if local_copy_identical:
# Make it look like identical eclass already exists in portage-stable.
os.makedirs(os.path.dirname(eclass_path))
shutil.copy2(upstream_eclass_path, eclass_path)
elif local_copy_identical is not None:
# Make local copy some other gibberish.
os.makedirs(os.path.dirname(eclass_path))
osutils.WriteFile(eclass_path, 'garblety gook')
# Add test-specific mocks/stubs
# Replay script
if do_copy:
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['add', eclass_subpath])
self.mox.ReplayAll()
# Verify
result = None
with self.OutputCapturer():
if error:
self.assertRaises(error, cpu.Upgrader._CopyUpstreamEclass,
mocked_upgrader, eclass + '.eclass')
else:
result = cpu.Upgrader._CopyUpstreamEclass(mocked_upgrader,
eclass + '.eclass')
self.mox.VerifyAll()
if do_copy:
self.assertTrue(result)
# Verify that eclass has been copied into portage-stable.
self.assertTrue(os.path.exists(eclass_path))
# Verify that eclass contents are correct.
self.assertTrue(filecmp.cmp(upstream_eclass_path, eclass_path))
else:
self.assertFalse(result)
def testCopyUpstreamEclassCopyBecauseMissing(self):
self._TestCopyUpstreamEclass('inheritme',
do_copy=True)
def testCopyUpstreamEclassCopyBecauseDifferent(self):
self._TestCopyUpstreamEclass('inheritme',
do_copy=True,
local_copy_identical=False)
def testCopyUpstreamEclassNoCopyBecauseIdentical(self):
self._TestCopyUpstreamEclass('inheritme',
do_copy=False,
local_copy_identical=True)
def testCopyUpstreamEclassNoCopyBecauseUpstreamMissing(self):
self._TestCopyUpstreamEclass('inheritme',
do_copy=False,
error=RuntimeError)
#
# _CopyUpstreamPackage
#
def _TestCopyUpstreamPackage(self, catpkg, verrev, success,
existing_files, extra_upstream_files,
error=None):
"""Test Upgrader._CopyUpstreamPackage"""
upstream_cpv = '%s-%s' % (catpkg, verrev)
ebuild = '%s-%s.ebuild' % (catpkg.split('/')[-1], verrev)
self._SetUpPlayground()
upstream_portdir = self._GetPlaygroundPortdir()
# Simulate extra files in upsteam package dir.
if extra_upstream_files:
pkg_dir = os.path.join(upstream_portdir, catpkg)
if os.path.exists(pkg_dir):
for extra_file in extra_upstream_files:
open(os.path.join(pkg_dir, extra_file), 'w')
# Prepare dummy portage-stable dir, with extra previously
# existing files simulated if requested.
portage_stable = self.tempdir
if existing_files:
pkg_dir = os.path.join(portage_stable, catpkg)
os.makedirs(pkg_dir)
for existing_file in existing_files:
open(os.path.join(pkg_dir, existing_file), 'w')
mocked_upgrader = self._MockUpgrader(_curr_board=None,
_upstream=upstream_portdir,
_stable_repo=portage_stable,
)
# Replay script
if success:
def git_rm(*args, **_kwargs):
# Identify file that psuedo-git is to remove, then remove it.
# As with real "git rm", if the dir is then empty remove that.
# File to remove is last argument in git command (arg 1)
dirpath = args[0]
for f in args[1][2:]:
os.remove(os.path.join(dirpath, f))
try:
os.rmdir(os.path.dirname(dirpath))
except OSError:
pass
pkgdir = os.path.join(mocked_upgrader._stable_repo, catpkg)
if existing_files:
rm_list = [os.path.join(catpkg, f) for f in existing_files]
# Accept files to remove in any order.
def rm_cmd_verifier(cmd):
cmd_args = cmd[2:] # Peel off 'rm -rf'.
return sorted(cmd_args) == sorted(rm_list)
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
mox.Func(rm_cmd_verifier),
redirect_stdout=True
).WithSideEffects(git_rm)
mocked_upgrader._CreateManifest(os.path.join(upstream_portdir, catpkg),
pkgdir, ebuild)
mocked_upgrader._IdentifyNeededEclass(upstream_cpv).AndReturn(None)
self.mox.ReplayAll()
# Verify
result = None
with self.OutputCapturer():
if error:
self.assertRaises(error, cpu.Upgrader._CopyUpstreamPackage,
mocked_upgrader, upstream_cpv)
else:
result = cpu.Upgrader._CopyUpstreamPackage(mocked_upgrader,
upstream_cpv)
self.mox.VerifyAll()
if success:
self.assertEquals(result, upstream_cpv)
# Verify that ebuild has been copied into portage-stable.
ebuild_path = os.path.join(portage_stable, catpkg, ebuild)
self.assertTrue(os.path.exists(ebuild_path),
msg='Missing expected ebuild after copy from upstream')
# Verify that any extra files upstream are also copied.
for extra_file in extra_upstream_files:
file_path = os.path.join(portage_stable, catpkg, extra_file)
msg = ('Missing expected extra file %s after copy from upstream' %
extra_file)
self.assertTrue(os.path.exists(file_path), msg=msg)
else:
self.assertTrue(result is None)
def testCopyUpstreamPackageEmptyStable(self):
existing_files = []
extra_upstream_files = []
self._TestCopyUpstreamPackage('dev-libs/D', '2', True,
existing_files,
extra_upstream_files)
def testCopyUpstreamPackageClutteredStable(self):
existing_files = ['foo', 'bar', 'foobar.ebuild', 'D-1.ebuild']
extra_upstream_files = []
self._TestCopyUpstreamPackage('dev-libs/D', '2', True,
existing_files,
extra_upstream_files)
def testCopyUpstreamPackageVersionNotAvailable(self):
"""Should fail, dev-libs/D version 5 does not exist 'upstream'"""
existing_files = []
extra_upstream_files = []
self._TestCopyUpstreamPackage('dev-libs/D', '5', False,
existing_files,
extra_upstream_files,
error=RuntimeError)
def testCopyUpstreamPackagePackageNotAvailable(self):
"""Should fail, a-b-c/D does not exist 'upstream' in any version"""
existing_files = []
extra_upstream_files = []
self._TestCopyUpstreamPackage('a-b-c/D', '5', False,
existing_files,
extra_upstream_files,
error=RuntimeError)
def testCopyUpstreamPackageExtraUpstreamFiles(self):
existing_files = ['foo', 'bar']
extra_upstream_files = ['keepme', 'andme']
self._TestCopyUpstreamPackage('dev-libs/F', '2-r1', True,
existing_files,
extra_upstream_files)
def _SetupManifestTest(self, ebuild,
upstream_mlines, current_mlines):
upstream_dir = tempfile.mkdtemp(dir=self.tempdir)
current_dir = tempfile.mkdtemp(dir=self.tempdir)
upstream_manifest = os.path.join(upstream_dir, 'Manifest')
current_manifest = os.path.join(current_dir, 'Manifest')
if upstream_mlines:
osutils.WriteFile(upstream_manifest,
'\n'.join(str(x) for x in upstream_mlines) + '\n')
if current_mlines:
osutils.WriteFile(current_manifest,
'\n'.join(str(x) for x in current_mlines) + '\n')
ebuild_path = os.path.join(current_dir, ebuild)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')
# Prepare test replay script.
run_result = RunCommandResult(returncode=0, output='')
cros_build_lib.RunCommand(['ebuild', ebuild_path, 'manifest'],
error_code_ok=True, print_cmd=False,
redirect_stdout=True, combine_stdout_stderr=True
).AndReturn(run_result)
self.mox.ReplayAll()
return (upstream_dir, current_dir)
def _AssertManifestContents(self, manifest_path, expected_manifest_lines):
manifest_lines = []
with open(manifest_path, 'r') as f:
for line in f:
manifest_lines.append(ManifestLine(line))
msg = ('Manifest contents not as expected. Expected:\n%s\n'
'\nBut got:\n%s\n' %
('\n'.join([str(ml) for ml in expected_manifest_lines]),
'\n'.join([str(ml) for ml in manifest_lines])))
self.assertTrue(manifest_lines == expected_manifest_lines, msg=msg)
self.assertFalse(manifest_lines != expected_manifest_lines, msg=msg)
def testCreateManifestNew(self):
"""Test case with upstream but no current Manifest."""
mocked_upgrader = self._MockUpgrader()
ebuild = 'some-pkg.ebuild'
upst_mlines = [ManifestLine(type='DIST',
file='fileA',
size='100',
RMD160='abc',
SHA1='123',
SHA256='abc123'),
ManifestLine(type='EBUILD',
file=ebuild,
size='254',
RMD160='def',
SHA1='456',
SHA256='def456'),]
upstream_dir, current_dir = self._SetupManifestTest(ebuild,
upst_mlines, None)
upstream_manifest = os.path.join(upstream_dir, 'Manifest')
current_manifest = os.path.join(current_dir, 'Manifest')
# Run test verification.
self.assertFalse(os.path.exists(current_manifest))
cpu.Upgrader._CreateManifest(mocked_upgrader,
upstream_dir, current_dir, ebuild)
self.mox.VerifyAll()
self.assertTrue(filecmp.cmp(upstream_manifest, current_manifest))
def testCreateManifestMerge(self):
"""Test case with upstream but no current Manifest."""
mocked_upgrader = self._MockUpgrader()
ebuild = 'some-pkg.ebuild'
curr_mlines = [ManifestLine(type='DIST',
file='fileA',
size='101',
RMD160='abc',
SHA1='123',
SHA256='abc123'),
ManifestLine(type='DIST',
file='fileC',
size='321',
RMD160='cde',
SHA1='345',
SHA256='cde345'),
ManifestLine(type='EBUILD',
file=ebuild,
size='254',
RMD160='def',
SHA1='789',
SHA256='def789'),]
upst_mlines = [ManifestLine(type='DIST',
file='fileA',
size='100',
RMD160='abc',
SHA1='123',
SHA256='abc123'),
# This file is different from current manifest.
# It should be picked up by _CreateManifest.
ManifestLine(type='DIST',
file='fileB',
size='345',
RMD160='bcd',
SHA1='234',
SHA256='bcd234'),
ManifestLine(type='EBUILD',
file=ebuild,
size='254',
RMD160='def',
SHA1='789',
SHA256='def789'),]
upstream_dir, current_dir = self._SetupManifestTest(ebuild,
upst_mlines,
curr_mlines)
current_manifest = os.path.join(current_dir, 'Manifest')
# Run test verification.
self.assertTrue(os.path.exists(current_manifest))
cpu.Upgrader._CreateManifest(mocked_upgrader,
upstream_dir, current_dir, ebuild)
self.mox.VerifyAll()
expected_mlines = curr_mlines + upst_mlines[1:2]
self._AssertManifestContents(current_manifest, expected_mlines)
class GetPackageUpgradeStateTest(CpuTestBase):
"""Test Upgrader._GetPackageUpgradeState"""
def _TestGetPackageUpgradeState(self, pinfo,
exists_upstream):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._FindUpstreamCPV(pinfo.cpv, unstable_ok=True,
).AndReturn(exists_upstream)
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GetPackageUpgradeState(mocked_upgrader, pinfo)
self.mox.VerifyAll()
return result
def testGetPackageUpgradeStateLocalOnly(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='chromiumos-overlay',
cpv_cmp_upstream=None,
latest_upstream_cpv=None)
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result, utable.UpgradeTable.STATE_LOCAL_ONLY)
def testGetPackageUpgradeStateUnknown(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='portage',
cpv_cmp_upstream=None,
latest_upstream_cpv=None)
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result, utable.UpgradeTable.STATE_UNKNOWN)
def testGetPackageUpgradeStateUpgradeAndDuplicated(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='chromiumos-overlay',
cpv_cmp_upstream=1, # outdated
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=True)
self.assertEquals(result,
utable.UpgradeTable.STATE_NEEDS_UPGRADE_AND_DUPLICATED)
def testGetPackageUpgradeStateUpgradeAndPatched(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='chromiumos-overlay',
cpv_cmp_upstream=1, # outdated
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result,
utable.UpgradeTable.STATE_NEEDS_UPGRADE_AND_PATCHED)
def testGetPackageUpgradeStateUpgrade(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='portage-stable',
cpv_cmp_upstream=1, # outdated
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result, utable.UpgradeTable.STATE_NEEDS_UPGRADE)
def testGetPackageUpgradeStateDuplicated(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='chromiumos-overlay',
cpv_cmp_upstream=0, # current
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=True)
self.assertEquals(result, utable.UpgradeTable.STATE_DUPLICATED)
def testGetPackageUpgradeStatePatched(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='chromiumos-overlay',
cpv_cmp_upstream=0, # current
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result, utable.UpgradeTable.STATE_PATCHED)
def testGetPackageUpgradeStateCurrent(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
overlay='portage-stable',
cpv_cmp_upstream=0, # current
latest_upstream_cpv='not important')
result = self._TestGetPackageUpgradeState(pinfo, exists_upstream=False)
self.assertEquals(result, utable.UpgradeTable.STATE_CURRENT)
@unittest.skip('relies on portage module not currently available')
class EmergeableTest(CpuTestBase):
"""Test Upgrader._AreEmergeable."""
def _TestAreEmergeable(self, cpvlist, expect,
debug=False, world=WORLD):
"""Test the Upgrader._AreEmergeable method.
|cpvlist| is passed to _AreEmergeable.
|expect| is boolean, expected return value of _AreEmergeable
|debug| requests that emerge output in _AreEmergeable be shown.
|world| is list of lines to override default world contents.
"""
cmdargs = ['--upgrade'] + cpvlist
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
self._SetUpPlayground(world=world)
# Add test-specific mocks/stubs
# Replay script
envvars = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
mocked_upgrader._curr_arch,
unstable_ok=False)
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok=False).AndReturn(envvars)
mocked_upgrader._GetBoardCmd('emerge').AndReturn('emerge')
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._AreEmergeable(mocked_upgrader, cpvlist)
self.mox.VerifyAll()
(code, _cmd, output) = result
if debug or code != expect:
print('\nTest ended with success==%r (expected==%r)' % (code, expect))
print('Emerge output:\n%s' % output)
self.assertEquals(code, expect)
def testAreEmergeableOnePkg(self):
"""Should pass, one cpv target."""
cpvlist = ['dev-libs/A-1']
return self._TestAreEmergeable(cpvlist, True)
def testAreEmergeableTwoPkgs(self):
"""Should pass, two cpv targets."""
cpvlist = ['dev-libs/A-1', 'dev-libs/B-1']
return self._TestAreEmergeable(cpvlist, True)
def testAreEmergeableOnePkgTwoVersions(self):
"""Should fail, targets two versions of same package."""
cpvlist = ['dev-libs/A-1', 'dev-libs/A-2']
return self._TestAreEmergeable(cpvlist, False)
def testAreEmergeableStableFlimFlam(self):
"""Should pass, target stable version of pkg."""
cpvlist = ['chromeos-base/flimflam-0.0.1-r228']
return self._TestAreEmergeable(cpvlist, True)
def testAreEmergeableUnstableFlimFlam(self):
"""Should fail, target unstable version of pkg."""
cpvlist = ['chromeos-base/flimflam-0.0.2-r123']
return self._TestAreEmergeable(cpvlist, False)
def testAreEmergeableBlockedPackages(self):
"""Should fail, targets have blocking deps on each other."""
cpvlist = ['dev-libs/D-1', 'dev-libs/E-2']
return self._TestAreEmergeable(cpvlist, False)
def testAreEmergeableBlockedByInstalledPkg(self):
"""Should fail because of installed D-1 pkg."""
cpvlist = ['dev-libs/E-2']
return self._TestAreEmergeable(cpvlist, False)
def testAreEmergeableNotBlockedByInstalledPkgNotInWorld(self):
"""Should pass because installed D-1 pkg not in world."""
cpvlist = ['dev-libs/E-2']
return self._TestAreEmergeable(cpvlist, True, world=[])
def testAreEmergeableSamePkgDiffSlots(self):
"""Should pass, same package but different slots."""
cpvlist = ['dev-libs/F-1', 'dev-libs/F-2']
return self._TestAreEmergeable(cpvlist, True)
def testAreEmergeableTwoPackagesIncompatibleDeps(self):
"""Should fail, targets depend on two versions of same pkg."""
cpvlist = ['dev-apps/X-1', 'dev-apps/Y-2']
return self._TestAreEmergeable(cpvlist, False)
class CPVUtilTest(CpuTestBase):
"""Test various CPV utilities in Upgrader"""
def _TestCmpCpv(self, cpv1, cpv2):
"""Test Upgrader._CmpCpv"""
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._CmpCpv(cpv1, cpv2)
self.mox.VerifyAll()
return result
def testCmpCpv(self):
# cpvs to compare
equal = [('foo/bar-1', 'foo/bar-1'),
('a-b-c/x-y-z-1.2.3-r1', 'a-b-c/x-y-z-1.2.3-r1'),
('foo/bar-1', 'foo/bar-1-r0'),
(None, None)]
for (cpv1, cpv2) in equal:
self.assertEqual(0, self._TestCmpCpv(cpv1, cpv2))
lessthan = [(None, 'foo/bar-1'),
('foo/bar-1', 'foo/bar-2'),
('foo/bar-1', 'foo/bar-1-r1'),
('foo/bar-1-r1', 'foo/bar-1-r2'),
('foo/bar-1.2.3', 'foo/bar-1.2.4'),
('foo/bar-5a', 'foo/bar-5b')]
for (cpv1, cpv2) in lessthan:
self.assertTrue(self._TestCmpCpv(cpv1, cpv2) < 0)
self.assertTrue(self._TestCmpCpv(cpv2, cpv1) > 0)
not_comparable = [('foo/bar-1', 'bar/foo-1')]
for (cpv1, cpv2) in not_comparable:
self.assertEquals(None, self._TestCmpCpv(cpv1, cpv2))
def _TestGetCatPkgFromCpv(self, cpv):
"""Test Upgrader._GetCatPkgFromCpv"""
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GetCatPkgFromCpv(cpv)
self.mox.VerifyAll()
return result
def testGetCatPkgFromCpv(self):
# (input, output) tuples
data = [('foo/bar-1', 'foo/bar'),
('a-b-c/x-y-z-1', 'a-b-c/x-y-z'),
('a-b-c/x-y-z-1.2.3-r3', 'a-b-c/x-y-z'),
('bar-1', 'bar'),
('bar', None)]
for (cpv, catpn) in data:
result = self._TestGetCatPkgFromCpv(cpv)
self.assertEquals(catpn, result)
def _TestGetVerRevFromCpv(self, cpv):
"""Test Upgrader._GetVerRevFromCpv"""
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GetVerRevFromCpv(cpv)
self.mox.VerifyAll()
return result
def testGetVerRevFromCpv(self):
# (input, output) tuples
data = [('foo/bar-1', '1'),
('a-b-c/x-y-z-1', '1'),
('a-b-c/x-y-z-1.2.3-r3', '1.2.3-r3'),
('foo/bar-3.222-r0', '3.222'),
('bar-1', '1'),
('bar', None)]
for (cpv, verrev) in data:
result = self._TestGetVerRevFromCpv(cpv)
self.assertEquals(verrev, result)
def _TestGetEbuildPathFromCpv(self, cpv):
"""Test Upgrader._GetEbuildPathFromCpv"""
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GetEbuildPathFromCpv(cpv)
self.mox.VerifyAll()
return result
def testGetEbuildPathFromCpv(self):
# (input, output) tuples
data = [('foo/bar-1', 'foo/bar/bar-1.ebuild'),
('a-b-c/x-y-z-1', 'a-b-c/x-y-z/x-y-z-1.ebuild'),
('a-b-c/x-y-z-1.2.3-r3', 'a-b-c/x-y-z/x-y-z-1.2.3-r3.ebuild'),
('foo/bar-3.222-r0', 'foo/bar/bar-3.222-r0.ebuild'),]
for (cpv, verrev) in data:
result = self._TestGetEbuildPathFromCpv(cpv)
self.assertEquals(verrev, result)
class PortageStableTest(CpuTestBase):
"""Test Upgrader methods _SaveStatusOnStableRepo, _CheckStableRepoOnBranch"""
STATUS_MIX = {'path1/file1': 'M',
'path1/path2/file2': 'A',
'a/b/.x/y~': 'D',
'foo/bar': 'C',
'/bar/foo': 'U',
'unknown/file': '??',}
STATUS_UNKNOWN = {'foo/bar': '??',
'a/b/c': '??',}
STATUS_EMPTY = {}
#
# _CheckStableRepoOnBranch
#
def _TestCheckStableRepoOnBranch(self, run_result, expect_err):
"""Test Upgrader._CheckStableRepoOnBranch"""
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['branch'], redirect_stdout=True,
).AndReturn(run_result)
self.mox.ReplayAll()
# Verify
try:
cpu.Upgrader._CheckStableRepoOnBranch(mocked_upgrader)
self.assertFalse(expect_err, 'Expected RuntimeError, but none raised.')
except RuntimeError as ex:
self.assertTrue(expect_err, 'Unexpected RuntimeError: %s' % str(ex))
self.mox.VerifyAll()
def testCheckStableRepoOnBranchNoBranch(self):
"""Should fail due to 'git branch' saying 'no branch'"""
output = '* (no branch)\n somebranch\n otherbranch\n'
run_result = RunCommandResult(returncode=0, output=output)
self._TestCheckStableRepoOnBranch(run_result, True)
def testCheckStableRepoOnBranchOK1(self):
"""Should pass as 'git branch' indicates a branch"""
output = '* somebranch\n otherbranch\n'
run_result = RunCommandResult(returncode=0, output=output)
self._TestCheckStableRepoOnBranch(run_result, False)
def testCheckStableRepoOnBranchOK2(self):
"""Should pass as 'git branch' indicates a branch"""
output = ' somebranch\n* otherbranch\n'
run_result = RunCommandResult(returncode=0, output=output)
self._TestCheckStableRepoOnBranch(run_result, False)
def testCheckStableRepoOnBranchFail(self):
"""Should fail as 'git branch' failed"""
output = 'does not matter'
run_result = RunCommandResult(returncode=1, output=output)
self._TestCheckStableRepoOnBranch(run_result, True)
#
# _SaveStatusOnStableRepo
#
def _TestSaveStatusOnStableRepo(self, run_result):
"""Test Upgrader._SaveStatusOnStableRepo"""
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['status', '-s'], redirect_stdout=True,
).AndReturn(run_result)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._SaveStatusOnStableRepo(mocked_upgrader)
self.mox.VerifyAll()
self.assertFalse(mocked_upgrader._stable_repo_stashed)
return mocked_upgrader._stable_repo_status
def testSaveStatusOnStableRepoFailed(self):
"""Test case where 'git status -s' fails, should raise RuntimeError"""
run_result = RunCommandResult(returncode=1,
output=None)
self.assertRaises(RuntimeError,
self._TestSaveStatusOnStableRepo,
run_result)
def testSaveStatusOnStableRepoAllKinds(self):
"""Test where 'git status -s' returns all status kinds"""
status_lines = ['%2s %s' % (v, k) for (k, v) in self.STATUS_MIX.items()]
status_output = '\n'.join(status_lines)
run_result = RunCommandResult(returncode=0,
output=status_output)
status = self._TestSaveStatusOnStableRepo(run_result)
self.assertEqual(status, self.STATUS_MIX)
def testSaveStatusOnStableRepoRename(self):
"""Test where 'git status -s' shows a file rename"""
old = 'path/foo-1'
new = 'path/foo-2'
status_lines = [' R %s --> %s' % (old, new)]
status_output = '\n'.join(status_lines)
run_result = RunCommandResult(returncode=0,
output=status_output)
status = self._TestSaveStatusOnStableRepo(run_result)
self.assertEqual(status, {old: 'D', new: 'A'})
def testSaveStatusOnStableRepoEmpty(self):
"""Test empty response from 'git status -s'"""
run_result = RunCommandResult(returncode=0,
output='')
status = self._TestSaveStatusOnStableRepo(run_result)
self.assertEqual(status, {})
#
# _AnyChangesStaged
#
def _TestAnyChangesStaged(self, status_dict):
"""Test Upgrader._AnyChangesStaged"""
mocked_upgrader = self._MockUpgrader(_stable_repo_status=status_dict)
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._AnyChangesStaged(mocked_upgrader)
self.mox.VerifyAll()
return result
def testAnyChangesStagedMix(self):
"""Should return True"""
self.assertTrue(self._TestAnyChangesStaged(self.STATUS_MIX),
'Failed to notice files with changed status.')
def testAnyChangesStagedUnknown(self):
"""Should return False, only files with '??' status"""
self.assertFalse(self._TestAnyChangesStaged(self.STATUS_UNKNOWN),
'Should not consider files with "??" status.')
def testAnyChangesStagedEmpty(self):
"""Should return False, no file statuses"""
self.assertFalse(self._TestAnyChangesStaged(self.STATUS_EMPTY),
'No files should mean no changes staged.')
#
# _StashChanges
#
def testStashChanges(self):
"""Test Upgrader._StashChanges"""
mocked_upgrader = self._MockUpgrader(cmdargs=[],
_stable_repo_stashed=False)
self.assertFalse(mocked_upgrader._stable_repo_stashed)
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['stash', 'save'],
redirect_stdout=True,
combine_stdout_stderr=True)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._StashChanges(mocked_upgrader)
self.mox.VerifyAll()
self.assertTrue(mocked_upgrader._stable_repo_stashed)
#
# _UnstashAnyChanges
#
def _TestUnstashAnyChanges(self, stashed):
"""Test Upgrader._UnstashAnyChanges"""
mocked_upgrader = self._MockUpgrader(cmdargs=[],
_stable_repo_stashed=stashed)
self.assertEqual(stashed, mocked_upgrader._stable_repo_stashed)
# Add test-specific mocks/stubs
# Replay script
if stashed:
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['stash', 'pop', '--index'],
redirect_stdout=True,
combine_stdout_stderr=True)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._UnstashAnyChanges(mocked_upgrader)
self.mox.VerifyAll()
self.assertFalse(mocked_upgrader._stable_repo_stashed)
def testUnstashAnyChanges(self):
self._TestUnstashAnyChanges(True)
self._TestUnstashAnyChanges(False)
#
# _DropAnyStashedChanges
#
def _TestDropAnyStashedChanges(self, stashed):
"""Test Upgrader._DropAnyStashedChanges"""
mocked_upgrader = self._MockUpgrader(cmdargs=[],
_stable_repo_stashed=stashed)
self.assertEqual(stashed, mocked_upgrader._stable_repo_stashed)
# Add test-specific mocks/stubs
# Replay script
if stashed:
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['stash', 'drop'],
redirect_stdout=True,
combine_stdout_stderr=True)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._DropAnyStashedChanges(mocked_upgrader)
self.mox.VerifyAll()
self.assertFalse(mocked_upgrader._stable_repo_stashed)
def testDropAnyStashedChanges(self):
self._TestDropAnyStashedChanges(True)
self._TestDropAnyStashedChanges(False)
class UtilityTest(CpuTestBase):
"""Test several Upgrader methods.
Test these Upgrader methods: _SplitEBuildPath, _GenPortageEnvvars
"""
#
# _IsInUpgradeMode
#
def _TestIsInUpgradeMode(self, cmdargs):
"""Test Upgrader._IsInUpgradeMode. Pretty simple."""
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._IsInUpgradeMode(mocked_upgrader)
self.mox.VerifyAll()
return result
def testIsInUpgradeModeNoOpts(self):
"""Should not be in upgrade mode with no options."""
result = self._TestIsInUpgradeMode([])
self.assertFalse(result)
def testIsInUpgradeModeUpgrade(self):
"""Should be in upgrade mode with --upgrade."""
result = self._TestIsInUpgradeMode(['--upgrade'])
self.assertTrue(result)
def testIsInUpgradeModeUpgradeDeep(self):
"""Should be in upgrade mode with --upgrade-deep."""
result = self._TestIsInUpgradeMode(['--upgrade-deep'])
self.assertTrue(result)
#
# _GetBoardCmd
#
def _TestGetBoardCmd(self, cmd, board):
"""Test Upgrader._GetBoardCmd."""
mocked_upgrader = self._MockUpgrader(_curr_board=board)
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GetBoardCmd(mocked_upgrader, cmd)
self.mox.VerifyAll()
return result
def testGetBoardCmdKnownCmds(self):
board = 'x86-alex'
for cmd in ['emerge', 'equery', 'portageq']:
result = self._TestGetBoardCmd(cmd, cpu.Upgrader.HOST_BOARD)
self.assertEquals(result, cmd)
result = self._TestGetBoardCmd(cmd, board)
self.assertEquals(result, '%s-%s' % (cmd, board))
def testGetBoardCmdUnknownCmd(self):
board = 'x86-alex'
cmd = 'foo'
result = self._TestGetBoardCmd(cmd, cpu.Upgrader.HOST_BOARD)
self.assertEquals(result, cmd)
result = self._TestGetBoardCmd(cmd, board)
self.assertEquals(result, cmd)
#
# _GenPortageEnvvars testing
#
def _TestGenPortageEnvvars(self, arch, unstable_ok,
portdir=None, portage_configroot=None):
"""Testing the behavior of the Upgrader._GenPortageEnvvars method."""
mocked_upgrader = self._MockUpgrader()
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
arch, unstable_ok,
portdir, portage_configroot)
self.mox.VerifyAll()
keyw = arch
if unstable_ok:
keyw = arch + ' ~' + arch
self.assertEquals(result['ACCEPT_KEYWORDS'], keyw)
if portdir is None:
self.assertFalse('PORTDIR' in result)
else:
self.assertEquals(result['PORTDIR'], portdir)
if portage_configroot is None:
self.assertFalse('PORTAGE_CONFIGROOT' in result)
else:
self.assertEquals(result['PORTAGE_CONFIGROOT'], portage_configroot)
def testGenPortageEnvvars1(self):
self._TestGenPortageEnvvars('arm', False)
def testGenPortageEnvvars2(self):
self._TestGenPortageEnvvars('x86', True)
def testGenPortageEnvvars3(self):
self._TestGenPortageEnvvars('x86', True,
portdir='/foo/bar',
portage_configroot='/bar/foo')
#
# _SplitEBuildPath testing
#
def _TestSplitEBuildPath(self, ebuild_path, golden_result):
"""Test the behavior of the Upgrader._SplitEBuildPath method."""
mocked_upgrader = self._MockUpgrader()
# Replay script
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._SplitEBuildPath(mocked_upgrader,
ebuild_path)
self.assertEquals(result, golden_result)
self.mox.VerifyAll()
def testSplitEBuildPath1(self):
self._TestSplitEBuildPath('/foo/bar/portage/dev-libs/A/A-2.ebuild',
('portage', 'dev-libs', 'A', 'A-2'))
def testSplitEBuildPath2(self):
self._TestSplitEBuildPath('/foo/ooo/ccc/ppp/ppp-1.2.3-r123.ebuild',
('ooo', 'ccc', 'ppp', 'ppp-1.2.3-r123'))
@unittest.skip('relies on portage module not currently available')
class TreeInspectTest(CpuTestBase):
"""Test Upgrader methods: _FindCurrentCPV, _FindUpstreamCPV"""
def _GenerateTestInput(self, category, pkg_name, ver_rev,
path_prefix=DEFAULT_PORTDIR):
"""Return tuple (ebuild_path, cpv, cp)."""
ebuild_path = None
cpv = None
if ver_rev:
ebuild_path = '%s/%s/%s/%s-%s.ebuild' % (path_prefix,
category, pkg_name,
pkg_name, ver_rev)
cpv = '%s/%s-%s' % (category, pkg_name, ver_rev)
cp = '%s/%s' % (category, pkg_name)
return (ebuild_path, cpv, cp)
#
# _FindUpstreamCPV testing
#
def _TestFindUpstreamCPV(self, pkg_arg, ebuild_expect, unstable_ok=False):
"""Test Upgrader._FindUpstreamCPV
This points _FindUpstreamCPV at the ResolverPlayground as if it is
the upstream tree.
"""
self._SetUpPlayground()
eroot = self._GetPlaygroundRoot()
mocked_upgrader = self._MockUpgrader(_curr_board=None,
_upstream=eroot)
# Add test-specific mocks/stubs
# Replay script
envvars = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
mocked_upgrader._curr_arch,
unstable_ok,
portdir=eroot,
portage_configroot=eroot)
portage_configroot = mocked_upgrader._emptydir
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok,
portdir=mocked_upgrader._upstream,
portage_configroot=portage_configroot,
).AndReturn(envvars)
if ebuild_expect:
ebuild_path = eroot + ebuild_expect
split_path = cpu.Upgrader._SplitEBuildPath(mocked_upgrader, ebuild_path)
mocked_upgrader._SplitEBuildPath(ebuild_path).AndReturn(split_path)
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._FindUpstreamCPV(mocked_upgrader, pkg_arg,
unstable_ok)
self.mox.VerifyAll()
self.assertTrue(bool(ebuild_expect) == bool(result))
return result
def testFindUpstreamA2(self):
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-libs',
pkg_name='A',
ver_rev='2')
result = self._TestFindUpstreamCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindUpstreamAAA(self):
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-apps',
pkg_name='AAA',
ver_rev=None)
result = self._TestFindUpstreamCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindUpstreamF(self):
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-libs',
pkg_name='F',
ver_rev='2')
result = self._TestFindUpstreamCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindUpstreamFlimflam(self):
"""Should find 0.0.1-r228 because more recent flimflam unstable."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='chromeos-base',
pkg_name='flimflam',
ver_rev='0.0.1-r228')
result = self._TestFindUpstreamCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindUpstreamFlimflamUnstable(self):
"""Should find 0.0.2-r123 because of unstable_ok."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='chromeos-base',
pkg_name='flimflam',
ver_rev='0.0.2-r123')
result = self._TestFindUpstreamCPV(cp, ebuild, unstable_ok=True)
self.assertEquals(result, cpv)
#
# _FindCurrentCPV testing
#
def _TestFindCurrentCPV(self, pkg_arg, ebuild_expect):
"""Test Upgrader._FindCurrentCPV
This test points Upgrader._FindCurrentCPV to the ResolverPlayground
tree as if it is the local source.
"""
mocked_upgrader = self._MockUpgrader(_curr_board=None)
self._SetUpPlayground()
eroot = self._GetPlaygroundRoot()
# Add test-specific mocks/stubs
# Replay script
envvars = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
mocked_upgrader._curr_arch,
unstable_ok=False)
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok=False).AndReturn(envvars)
mocked_upgrader._GetBoardCmd('equery').AndReturn('equery')
if ebuild_expect:
ebuild_path = eroot + ebuild_expect
split_path = cpu.Upgrader._SplitEBuildPath(mocked_upgrader, ebuild_path)
mocked_upgrader._SplitEBuildPath(ebuild_path).AndReturn(split_path)
self.mox.ReplayAll()
# Verify
result = cpu.Upgrader._FindCurrentCPV(mocked_upgrader, pkg_arg)
self.mox.VerifyAll()
return result
def testFindCurrentA(self):
"""Should find dev-libs/A-2."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-libs',
pkg_name='A',
ver_rev='2')
result = self._TestFindCurrentCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindCurrentAAA(self):
"""Should find None, because dev-libs/AAA does not exist in tree."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-libs',
pkg_name='AAA',
ver_rev=None)
result = self._TestFindCurrentCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindCurrentF(self):
"""Should find dev-libs/F-2."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='dev-libs',
pkg_name='F',
ver_rev='2')
result = self._TestFindCurrentCPV(cp, ebuild)
self.assertEquals(result, cpv)
def testFindCurrentFlimflam(self):
"""Should find 0.0.1-r228 because more recent flimflam unstable."""
(ebuild, cpv, cp) = self._GenerateTestInput(category='chromeos-base',
pkg_name='flimflam',
ver_rev='0.0.1-r228')
result = self._TestFindCurrentCPV(cp, ebuild)
self.assertEquals(result, cpv)
class RunBoardTest(CpuTestBase):
"""Test Upgrader.RunBoard,PrepareToRun,RunCompleted."""
def testRunCompletedSpecified(self):
cmdargs = ['--upstream=/some/dir']
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_emptydir='empty-dir',
_curr_board=None)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(osutils, 'RmDir')
# Replay script
osutils.RmDir('empty-dir', ignore_missing=True)
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
cpu.Upgrader.RunCompleted(mocked_upgrader)
self.mox.VerifyAll()
def testRunCompletedRemoveCache(self):
cmdargs = ['--no-upstream-cache']
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_emptydir='empty-dir',
_curr_board=None)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(osutils, 'RmDir')
self.mox.StubOutWithMock(osutils, 'SafeUnlink')
# Replay script
osutils.RmDir(mocked_upgrader._upstream, ignore_missing=True)
osutils.SafeUnlink('%s-README' % mocked_upgrader._upstream)
osutils.RmDir('empty-dir', ignore_missing=True)
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
cpu.Upgrader.RunCompleted(mocked_upgrader)
self.mox.VerifyAll()
def testRunCompletedKeepCache(self):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_emptydir='empty-dir',
_curr_board=None)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(osutils, 'RmDir')
# Replay script
osutils.RmDir('empty-dir', ignore_missing=True)
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
cpu.Upgrader.RunCompleted(mocked_upgrader)
self.mox.VerifyAll()
def testPrepareToRunUpstreamRepoExists(self):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_curr_board=None)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(os.path, 'exists')
self.mox.StubOutWithMock(osutils, 'RmDir')
# Replay script
os.path.exists('/tmp/portage/.git/shallow').AndReturn(False)
osutils.RmDir('/tmp/portage', ignore_missing=True)
os.path.exists('/tmp/portage').AndReturn(True)
mocked_upgrader._RunGit(
'/tmp/portage', ['remote', 'set-url', 'origin',
cpu.Upgrader.PORTAGE_GIT_URL])
mocked_upgrader._RunGit(
'/tmp/portage', ['remote', 'update'])
mocked_upgrader._RunGit(
'/tmp/portage', ['checkout', '-f', 'origin/gentoo'],
combine_stdout_stderr=True, redirect_stdout=True)
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
cpu.Upgrader.PrepareToRun(mocked_upgrader)
self.mox.VerifyAll()
def testPrepareToRunUpstreamRepoNew(self):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_upstream=self.tempdir,
_curr_board=None)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(os.path, 'dirname')
self.mox.StubOutWithMock(os.path, 'basename')
self.mox.StubOutWithMock(tempfile, 'mkdtemp')
# Replay script
tempfile.mkdtemp()
root = os.path.dirname(mocked_upgrader._upstream).AndReturn('root')
name = os.path.basename(mocked_upgrader._upstream).AndReturn('name')
os.path.basename('origin/gentoo').AndReturn('gentoo')
mocked_upgrader._RunGit(root,
['clone', '--branch', 'gentoo', '--depth', '1',
cpu.Upgrader.PORTAGE_GIT_URL, name])
self.mox.ReplayAll()
# Verify
try:
with self.OutputCapturer():
cpu.Upgrader.PrepareToRun(mocked_upgrader)
self.mox.VerifyAll()
finally:
self.mox.UnsetStubs()
readme_path = self.tempdir + '-README'
self.assertTrue(os.path.exists(readme_path))
os.remove(readme_path)
def _TestRunBoard(self, pinfolist, upgrade=False, staged_changes=False):
"""Test Upgrader.RunBoard."""
targetlist = [pinfo.user_arg for pinfo in pinfolist]
upstream_only_pinfolist = [pinfo for pinfo in pinfolist if not pinfo.cpv]
cmdargs = targetlist
if upgrade:
cmdargs = ['--upgrade'] + cmdargs
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_curr_board=None)
board = 'runboard_testboard'
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cpu.Upgrader, '_FindBoardArch')
# Replay script
mocked_upgrader._SaveStatusOnStableRepo()
mocked_upgrader._LoadStableRepoCategories()
cpu.Upgrader._FindBoardArch(board).AndReturn('x86')
upgrade_mode = cpu.Upgrader._IsInUpgradeMode(mocked_upgrader)
mocked_upgrader._IsInUpgradeMode().AndReturn(upgrade_mode)
mocked_upgrader._AnyChangesStaged().AndReturn(staged_changes)
if staged_changes:
mocked_upgrader._StashChanges()
mocked_upgrader._ResolveAndVerifyArgs(targetlist,
upgrade_mode).AndReturn(pinfolist)
if upgrade:
mocked_upgrader._FinalizeUpstreamPInfolist(pinfolist).AndReturn([])
else:
mocked_upgrader._GetCurrentVersions(pinfolist).AndReturn(pinfolist)
mocked_upgrader._FinalizeLocalPInfolist(pinfolist).AndReturn([])
if upgrade_mode:
mocked_upgrader._FinalizeUpstreamPInfolist(
upstream_only_pinfolist).AndReturn([])
mocked_upgrader._UnstashAnyChanges()
mocked_upgrader._UpgradePackages([])
mocked_upgrader._DropAnyStashedChanges()
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
cpu.Upgrader.RunBoard(mocked_upgrader, board)
self.mox.VerifyAll()
def testRunBoard1(self):
target_pinfolist = [cpu.PInfo(user_arg='dev-libs/A',
cpv='dev-libs/A-1',
upstream_cpv='dev-libs/A-2')]
return self._TestRunBoard(target_pinfolist)
def testRunBoard2(self):
target_pinfolist = [cpu.PInfo(user_arg='dev-libs/A',
cpv='dev-libs/A-1',
upstream_cpv='dev-libs/A-2')]
return self._TestRunBoard(target_pinfolist, upgrade=True)
def testRunBoard3(self):
target_pinfolist = [cpu.PInfo(user_arg='dev-libs/A',
cpv='dev-libs/A-1',
upstream_cpv='dev-libs/A-2')]
return self._TestRunBoard(target_pinfolist, upgrade=True,
staged_changes=True)
def testRunBoardUpstreamOnlyStatusMode(self):
"""Status mode with package that is only upstream should error."""
pinfolist = [cpu.PInfo(user_arg='dev-libs/M',
cpv=None,
upstream_cpv='dev-libs/M-2'),]
targetlist = [pinfo.user_arg for pinfo in pinfolist]
mocked_upgrader = self._MockUpgrader(cmdargs=['dev-libs/M'],
_curr_board=None)
board = 'runboard_testboard'
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cpu.Upgrader, '_FindBoardArch')
# Replay script
mocked_upgrader._SaveStatusOnStableRepo()
mocked_upgrader._LoadStableRepoCategories()
cpu.Upgrader._FindBoardArch(board).AndReturn('x86')
upgrade_mode = cpu.Upgrader._IsInUpgradeMode(mocked_upgrader)
mocked_upgrader._IsInUpgradeMode().AndReturn(upgrade_mode)
mocked_upgrader._AnyChangesStaged().AndReturn(False)
mocked_upgrader._ResolveAndVerifyArgs(targetlist,
upgrade_mode).AndReturn(pinfolist)
mocked_upgrader._DropAnyStashedChanges()
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
self.assertRaises(RuntimeError,
cpu.Upgrader.RunBoard,
mocked_upgrader, board)
self.mox.VerifyAll()
class GiveEmergeResultsTest(CpuTestBase):
"""Test Upgrader._GiveEmergeResults"""
def _TestGiveEmergeResultsOK(self, pinfolist, ok, error=None):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._AreEmergeable(mox.IgnoreArg(),
).AndReturn((ok, None, None))
self.mox.ReplayAll()
# Verify
result = None
with self.OutputCapturer():
if error:
self.assertRaises(error, cpu.Upgrader._GiveEmergeResults,
mocked_upgrader, pinfolist)
else:
result = cpu.Upgrader._GiveEmergeResults(mocked_upgrader, pinfolist)
self.mox.VerifyAll()
return result
def testGiveEmergeResultsUnmaskedOK(self):
pinfolist = [cpu.PInfo(upgraded_cpv='abc/def-4', upgraded_unmasked=True),
cpu.PInfo(upgraded_cpv='bcd/efg-8', upgraded_unmasked=True)]
self._TestGiveEmergeResultsOK(pinfolist, True)
def testGiveEmergeResultsUnmaskedNotOK(self):
pinfolist = [cpu.PInfo(upgraded_cpv='abc/def-4', upgraded_unmasked=True),
cpu.PInfo(upgraded_cpv='bcd/efg-8', upgraded_unmasked=True)]
self._TestGiveEmergeResultsOK(pinfolist, False, error=RuntimeError)
def _TestGiveEmergeResultsMasked(self, pinfolist, ok, masked_cpvs,
error=None):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
# Replay script
emergeable_tuple = (ok, 'some-cmd', 'some-output')
mocked_upgrader._AreEmergeable(mox.IgnoreArg(),
).AndReturn(emergeable_tuple)
if not ok:
for cpv in masked_cpvs:
mocked_upgrader._GiveMaskedError(cpv, 'some-output').InAnyOrder()
self.mox.ReplayAll()
# Verify
result = None
with self.OutputCapturer():
if error:
self.assertRaises(error, cpu.Upgrader._GiveEmergeResults,
mocked_upgrader, pinfolist)
else:
result = cpu.Upgrader._GiveEmergeResults(mocked_upgrader, pinfolist)
self.mox.VerifyAll()
return result
def testGiveEmergeResultsMaskedOK(self):
pinfolist = [cpu.PInfo(upgraded_cpv='abc/def-4', upgraded_unmasked=False),
cpu.PInfo(upgraded_cpv='bcd/efg-8', upgraded_unmasked=False)]
masked_cpvs = ['abc/def-4', 'bcd/efg-8']
self._TestGiveEmergeResultsMasked(pinfolist, True, masked_cpvs,
error=RuntimeError)
def testGiveEmergeResultsMaskedNotOK(self):
pinfolist = [cpu.PInfo(upgraded_cpv='abc/def-4', upgraded_unmasked=False),
cpu.PInfo(upgraded_cpv='bcd/efg-8', upgraded_unmasked=False)]
masked_cpvs = ['abc/def-4', 'bcd/efg-8']
self._TestGiveEmergeResultsMasked(pinfolist, False, masked_cpvs,
error=RuntimeError)
class CheckStagedUpgradesTest(CpuTestBase):
"""Test Upgrader._CheckStagedUpgrades"""
def testCheckStagedUpgradesTwoStaged(self):
cmdargs = []
ebuild1 = 'a/b/foo/bar/bar-1.ebuild'
ebuild2 = 'x/y/bar/foo/foo-3.ebuild'
repo_status = {ebuild1: 'A',
'a/b/foo/garbage': 'A',
ebuild2: 'A',}
pinfolist = [cpu.PInfo(package='foo/bar'),
cpu.PInfo(package='bar/foo')]
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_stable_repo_status=repo_status)
# Add test-specific mocks/stubs
# Replay script
for ebuild in [ebuild1, ebuild2]:
split = cpu.Upgrader._SplitEBuildPath(mocked_upgrader, ebuild)
mocked_upgrader._SplitEBuildPath(ebuild).InAnyOrder().AndReturn(split)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._CheckStagedUpgrades(mocked_upgrader, pinfolist)
self.mox.VerifyAll()
def testCheckStagedUpgradesTwoStagedOneUnexpected(self):
cmdargs = []
ebuild1 = 'a/b/foo/bar/bar-1.ebuild'
ebuild2 = 'x/y/bar/foo/foo-3.ebuild'
repo_status = {ebuild1: 'A',
'a/b/foo/garbage': 'A',
ebuild2: 'A',}
# Without foo/bar in the pinfolist it should complain about that
# package having staged changes.
pinfolist = [cpu.PInfo(package='bar/foo')]
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_stable_repo_status=repo_status)
# Add test-specific mocks/stubs
# Replay script
for ebuild in [ebuild1, ebuild2]:
split = cpu.Upgrader._SplitEBuildPath(mocked_upgrader, ebuild)
mocked_upgrader._SplitEBuildPath(ebuild).InAnyOrder().AndReturn(split)
self.mox.ReplayAll()
# Verify
self.assertRaises(RuntimeError, cpu.Upgrader._CheckStagedUpgrades,
mocked_upgrader, pinfolist)
self.mox.VerifyAll()
def testCheckStagedUpgradesNoneStaged(self):
cmdargs = []
pinfolist = [cpu.PInfo(package='foo/bar-1'),
cpu.PInfo(package='bar/foo-3')]
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_stable_repo_status=None)
# Add test-specific mocks/stubs
# Replay script
self.mox.ReplayAll()
# Verify
cpu.Upgrader._CheckStagedUpgrades(mocked_upgrader, pinfolist)
self.mox.VerifyAll()
class UpgradePackagesTest(CpuTestBase):
"""Test Upgrader._UpgradePackages"""
def _TestUpgradePackages(self, pinfolist, upgrade):
cmdargs = []
if upgrade:
cmdargs.append('--upgrade')
table = utable.UpgradeTable('some-arch')
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_curr_table=table)
# Add test-specific mocks/stubs
# Replay script
upgrades_this_run = False
for pinfo in pinfolist:
pkg_result = bool(pinfo.upgraded_cpv)
mocked_upgrader._UpgradePackage(pinfo).InAnyOrder(
'up').AndReturn(pkg_result)
if pkg_result:
upgrades_this_run = True
for pinfo in pinfolist:
if pinfo.upgraded_cpv:
mocked_upgrader._VerifyPackageUpgrade(pinfo).InAnyOrder('ver')
mocked_upgrader._PackageReport(pinfo).InAnyOrder('ver')
if upgrades_this_run:
mocked_upgrader._GiveEmergeResults(pinfolist)
upgrade_mode = cpu.Upgrader._IsInUpgradeMode(mocked_upgrader)
mocked_upgrader._IsInUpgradeMode().AndReturn(upgrade_mode)
if upgrade_mode:
mocked_upgrader._CheckStagedUpgrades(pinfolist)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._UpgradePackages(mocked_upgrader, pinfolist)
self.mox.VerifyAll()
def testUpgradePackagesUpgradeModeWithUpgrades(self):
pinfolist = [cpu.PInfo(upgraded_cpv='abc/def-4'),
cpu.PInfo(upgraded_cpv='bcd/efg-8'),
cpu.PInfo(upgraded_cpv=None),
cpu.PInfo(upgraded_cpv=None)]
self._TestUpgradePackages(pinfolist, True)
def testUpgradePackagesUpgradeModeNoUpgrades(self):
pinfolist = [cpu.PInfo(upgraded_cpv=None),
cpu.PInfo(upgraded_cpv=None)]
self._TestUpgradePackages(pinfolist, True)
def testUpgradePackagesStatusModeNoUpgrades(self):
pinfolist = [cpu.PInfo(upgraded_cpv=None),
cpu.PInfo(upgraded_cpv=None)]
self._TestUpgradePackages(pinfolist, False)
class CategoriesRoundtripTest(cros_test_lib.MoxTempDirTestOutputCase):
"""Tests for full "round trip" runs."""
def _TestCategoriesRoundtrip(self, categories):
stable_repo = self.tempdir
cat_file = cpu.Upgrader.CATEGORIES_FILE
profiles_dir = os.path.join(stable_repo, os.path.dirname(cat_file))
self.mox.StubOutWithMock(cpu.Upgrader, '_RunGit')
# Prepare replay script.
cpu.Upgrader._RunGit(stable_repo, ['add', cat_file])
self.mox.ReplayAll()
options = cros_test_lib.EasyAttr(srcroot='foobar', upstream=None,
packages='')
upgrader = cpu.Upgrader(options=options)
upgrader._stable_repo = stable_repo
os.makedirs(profiles_dir)
# Verification phase. Write then load categories.
upgrader._stable_repo_categories = set(categories)
upgrader._WriteStableRepoCategories()
upgrader._stable_repo_categories = None
upgrader._LoadStableRepoCategories()
self.mox.VerifyAll()
self.assertEquals(sorted(categories),
sorted(upgrader._stable_repo_categories))
def test1(self):
categories = ['alpha-omega', 'omega-beta', 'beta-chi']
self._TestCategoriesRoundtrip(categories)
def test2(self):
categories = []
self._TestCategoriesRoundtrip(categories)
def test3(self):
categories = ['virtual', 'happy-days', 'virtually-there']
self._TestCategoriesRoundtrip(categories)
class UpgradePackageTest(CpuTestBase):
"""Test Upgrader._UpgradePackage"""
def _TestUpgradePackage(self, pinfo, upstream_cpv, upstream_cmp,
stable_up, latest_up,
upgrade_requested, upgrade_staged,
unstable_ok, force):
cmdargs = []
if unstable_ok:
cmdargs.append('--unstable-ok')
if force:
cmdargs.append('--force')
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')
# Replay script
mocked_upgrader._FindUpstreamCPV(pinfo.package).AndReturn(stable_up)
mocked_upgrader._FindUpstreamCPV(pinfo.package,
unstable_ok=True).AndReturn(latest_up)
if upstream_cpv:
mocked_upgrader._PkgUpgradeRequested(pinfo).AndReturn(upgrade_requested)
if upgrade_requested:
mocked_upgrader._PkgUpgradeStaged(
upstream_cpv).AndReturn(upgrade_staged)
if (not upgrade_staged and
(upstream_cmp > 0 or (upstream_cmp == 0 and force))):
mocked_upgrader._CopyUpstreamPackage(
upstream_cpv).AndReturn(upstream_cpv)
upgrade_staged = True
if upgrade_staged:
mocked_upgrader._SetUpgradedMaskBits(pinfo)
ebuild_path = cpu.Upgrader._GetEbuildPathFromCpv(upstream_cpv)
ebuild_path = os.path.join(mocked_upgrader._stable_repo,
ebuild_path)
mocked_upgrader._StabilizeEbuild(ebuild_path)
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['add', pinfo.package])
mocked_upgrader._UpdateCategories(pinfo)
cache_files = 'metadata/md5-cache/%s-[0-9]*' % pinfo.package
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['rm', '--ignore-unmatch', '-q', '-f',
cache_files])
cmd = ['egencache', '--update', '--repo=portage-stable',
pinfo.package]
run_result = RunCommandResult(returncode=0, output=None)
cros_build_lib.RunCommand(cmd, print_cmd=False,
redirect_stdout=True,
combine_stdout_stderr=True).AndReturn(
run_result)
mocked_upgrader._RunGit(mocked_upgrader._stable_repo,
['add', cache_files])
self.mox.ReplayAll()
# Verify
with self.OutputCapturer():
result = cpu.Upgrader._UpgradePackage(mocked_upgrader, pinfo)
self.mox.VerifyAll()
if upstream_cpv:
self.assertEquals(upstream_cpv, pinfo.upstream_cpv)
if upgrade_requested and (upstream_cpv != pinfo.cpv or force):
self.assertEquals(upstream_cpv, pinfo.upgraded_cpv)
else:
self.assertTrue(pinfo.upgraded_cpv is None)
else:
self.assertTrue(pinfo.upstream_cpv is None)
self.assertTrue(pinfo.upgraded_cpv is None)
self.assertEquals(stable_up, pinfo.stable_upstream_cpv)
self.assertEquals(latest_up, pinfo.latest_upstream_cpv)
return result
# Dimensions to vary:
# 1) Upgrade for this package requested or not
# 2) Upgrade can be stable or not
# 3) Specific version to upgrade to specified
# 4) Upgrade already staged or not
# 5) Upgrade needed or not (current)
def testUpgradePackageOutdatedRequestedStable(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv=None)
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-3',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=False,
unstable_ok=False,
force=False)
self.assertTrue(result)
def testUpgradePackageOutdatedRequestedUnstable(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv=None)
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-5',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=False,
unstable_ok=True,
force=False)
self.assertTrue(result)
def testUpgradePackageOutdatedRequestedStableSpecified(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv='foo/bar-4')
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-4',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=False,
unstable_ok=False, # not important
force=False)
self.assertTrue(result)
def testUpgradePackageCurrentRequestedStable(self):
pinfo = cpu.PInfo(cpv='foo/bar-3',
package='foo/bar',
upstream_cpv=None)
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-3',
upstream_cmp=0, # current
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=False,
unstable_ok=False,
force=False)
self.assertFalse(result)
def testUpgradePackageCurrentRequestedStableForce(self):
pinfo = cpu.PInfo(cpv='foo/bar-3',
package='foo/bar',
upstream_cpv='foo/bar-3')
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-3',
upstream_cmp=0, # current
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=False,
unstable_ok=False,
force=True)
self.assertTrue(result)
def testUpgradePackageOutdatedStable(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv=None)
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-3',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=False,
upgrade_staged=False,
unstable_ok=False,
force=False)
self.assertFalse(result)
def testUpgradePackageOutdatedRequestedStableStaged(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv=None)
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-3',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=True,
unstable_ok=False,
force=False)
self.assertTrue(result)
def testUpgradePackageOutdatedRequestedUnstableStaged(self):
pinfo = cpu.PInfo(cpv='foo/bar-2',
package='foo/bar',
upstream_cpv='foo/bar-5')
result = self._TestUpgradePackage(pinfo,
upstream_cpv='foo/bar-5',
upstream_cmp=1, # outdated
stable_up='foo/bar-3',
latest_up='foo/bar-5',
upgrade_requested=True,
upgrade_staged=True,
unstable_ok=True,
force=False)
self.assertTrue(result)
class VerifyPackageTest(CpuTestBase):
"""Tests for _VerifyPackageUpgrade()."""
def _TestVerifyPackageUpgrade(self, pinfo):
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs)
was_overwrite = pinfo.cpv_cmp_upstream == 0
# Add test-specific mocks/stubs
# Replay script
mocked_upgrader._VerifyEbuildOverlay(pinfo.upgraded_cpv,
'portage-stable',
was_overwrite)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._VerifyPackageUpgrade(mocked_upgrader, pinfo)
self.mox.VerifyAll()
def testVerifyPackageUpgrade(self):
pinfo = cpu.PInfo(upgraded_cpv='foo/bar-3')
for cpv_cmp_upstream in (0, 1):
pinfo.cpv_cmp_upstream = cpv_cmp_upstream
self._TestVerifyPackageUpgrade(pinfo)
def _TestVerifyEbuildOverlay(self, cpv, overlay, ebuild_path, was_overwrite):
"""Test Upgrader._VerifyEbuildOverlay"""
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_curr_arch=DEFAULT_ARCH)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')
# Replay script
envvars = cpu.Upgrader._GenPortageEnvvars(mocked_upgrader,
mocked_upgrader._curr_arch,
unstable_ok=False)
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok=False).AndReturn(envvars)
mocked_upgrader._GetBoardCmd('equery').AndReturn('equery')
run_result = RunCommandResult(returncode=0,
output=ebuild_path)
cros_build_lib.RunCommand(['equery', '-C', 'which', '--include-masked',
cpv], error_code_ok=True,
extra_env=envvars, print_cmd=False,
redirect_stdout=True, combine_stdout_stderr=True,
).AndReturn(run_result)
split_ebuild = cpu.Upgrader._SplitEBuildPath(mocked_upgrader, ebuild_path)
mocked_upgrader._SplitEBuildPath(ebuild_path).AndReturn(split_ebuild)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._VerifyEbuildOverlay(mocked_upgrader, cpv,
overlay, was_overwrite)
self.mox.VerifyAll()
def testVerifyEbuildOverlayGood(self):
cpv = 'foo/bar-2'
overlay = 'some-overlay'
good_path = '/some/path/%s/foo/bar/bar-2.ebuild' % overlay
self._TestVerifyEbuildOverlay(cpv, overlay, good_path, False)
def testVerifyEbuildOverlayEvilNonOverwrite(self):
cpv = 'foo/bar-2'
overlay = 'some-overlay'
evil_path = '/some/path/spam/foo/bar/bar-2.ebuild'
self.assertRaises(RuntimeError,
self._TestVerifyEbuildOverlay,
cpv, overlay, evil_path, False)
def testVerifyEbuildOverlayEvilOverwrite(self):
cpv = 'foo/bar-2'
overlay = 'some-overlay'
evil_path = '/some/path/spam/foo/bar/bar-2.ebuild'
self.assertRaises(RuntimeError,
self._TestVerifyEbuildOverlay,
cpv, overlay, evil_path, True)
def _TestSetUpgradedMaskBits(self, pinfo, output):
cpv = pinfo.upgraded_cpv
cmdargs = []
mocked_upgrader = self._MockUpgrader(cmdargs=cmdargs,
_curr_arch=DEFAULT_ARCH)
# Add test-specific mocks/stubs
self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')
# Replay script
mocked_upgrader._GenPortageEnvvars(mocked_upgrader._curr_arch,
unstable_ok=False).AndReturn('envvars')
mocked_upgrader._GetBoardCmd('equery').AndReturn('equery')
run_result = RunCommandResult(returncode=0,
output=output)
cros_build_lib.RunCommand(['equery', '-qCN', 'list', '-F',
'$mask|$cpv:$slot', '-op', cpv],
error_code_ok=True,
extra_env='envvars', print_cmd=False,
redirect_stdout=True, combine_stdout_stderr=True,
).AndReturn(run_result)
self.mox.ReplayAll()
# Verify
cpu.Upgrader._SetUpgradedMaskBits(mocked_upgrader, pinfo)
self.mox.VerifyAll()
def testGetMaskBitsUnmaskedStable(self):
output = ' |foo/bar-2.7.0:0'
pinfo = cpu.PInfo(upgraded_cpv='foo/bar-2.7.0')
self._TestSetUpgradedMaskBits(pinfo, output)
self.assertTrue(pinfo.upgraded_unmasked)
def testGetMaskBitsUnmaskedUnstable(self):
output = ' ~|foo/bar-2.7.3:0'
pinfo = cpu.PInfo(upgraded_cpv='foo/bar-2.7.3')
self._TestSetUpgradedMaskBits(pinfo, output)
self.assertTrue(pinfo.upgraded_unmasked)
def testGetMaskBitsMaskedStable(self):
output = 'M |foo/bar-2.7.4:0'
pinfo = cpu.PInfo(upgraded_cpv='foo/bar-2.7.4')
self._TestSetUpgradedMaskBits(pinfo, output)
self.assertFalse(pinfo.upgraded_unmasked)
def testGetMaskBitsMaskedUnstable(self):
output = 'M~|foo/bar-2.7.4-r1:0'
pinfo = cpu.PInfo(upgraded_cpv='foo/bar-2.7.4-r1')
self._TestSetUpgradedMaskBits(pinfo, output)
self.assertFalse(pinfo.upgraded_unmasked)
class CommitTest(CpuTestBase):
"""Test various commit-related Upgrader methods"""
#
# _ExtractUpgradedPkgs