xbuddy: Fix linter errors.
Also replace mox
BUG=chromium:403086
TEST=unittests, cros flash
Change-Id: I6052be6b45f1c480732925bf490ff5ecf38e39ca
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/dev-util/+/1822077
Tested-by: Achuith Bhandarkar <achuith@chromium.org>
Reviewed-by: Amin Hassani <ahassani@chromium.org>
Reviewed-by: Achuith Bhandarkar <achuith@chromium.org>
Commit-Queue: Achuith Bhandarkar <achuith@chromium.org>
diff --git a/xbuddy.py b/xbuddy.py
index d575d9a..7747ef7 100644
--- a/xbuddy.py
+++ b/xbuddy.py
@@ -7,10 +7,8 @@
from __future__ import print_function
-import cherrypy
-import ConfigParser
import datetime
-import distutils.version
+import distutils.version # pylint: disable=import-error,no-name-in-module
import operator
import os
import re
@@ -18,6 +16,9 @@
import time
import threading
+import cherrypy # pylint: disable=import-error
+from six.moves import configparser
+
import artifact_info
import build_artifact
import build_util
@@ -60,10 +61,10 @@
FACTORY_SHIM = 'factory_shim'
# Local build constants
-ANY = "ANY"
-LATEST = "latest"
-LOCAL = "local"
-REMOTE = "remote"
+ANY = 'ANY'
+LATEST = 'latest'
+LOCAL = 'local'
+REMOTE = 'remote'
# TODO(sosa): Fix a lot of assumptions about these aliases. There is too much
# implicit logic here that's unnecessary. What should be done:
@@ -133,9 +134,9 @@
GS_ALIAS_TO_FILENAME = dict(zip(GS_ALIASES, GS_FILE_NAMES))
GS_ALIAS_TO_ARTIFACT = dict(zip(GS_ALIASES, ARTIFACTS))
-LATEST_OFFICIAL = "latest-official"
+LATEST_OFFICIAL = 'latest-official'
-RELEASE = "-release"
+RELEASE = '-release'
class XBuddyException(Exception):
@@ -143,8 +144,6 @@
pass
-# no __init__ method
-#pylint: disable=W0232
class Timestamp(object):
"""Class to translate build path strings and timestamp filenames."""
@@ -163,12 +162,11 @@
def UpdateTimestamp(timestamp_dir, build_id):
"""Update timestamp file of build with build_id."""
common_util.MkDirP(timestamp_dir)
- _Log("Updating timestamp for %s", build_id)
+ _Log('Updating timestamp for %s', build_id)
time_file = os.path.join(timestamp_dir,
Timestamp.BuildToTimestamp(build_id))
- with file(time_file, 'a'):
+ with open(time_file, 'a'):
os.utime(time_file, None)
-#pylint: enable=W0232
class XBuddy(build_util.BuildObject):
@@ -245,7 +243,7 @@
Raises:
XBuddyException if the config file is missing.
"""
- xbuddy_config = ConfigParser.ConfigParser()
+ xbuddy_config = configparser.ConfigParser()
config_file = os.path.join(self.devserver_dir, CONFIG_FILE)
if os.path.exists(config_file):
xbuddy_config.read(config_file)
@@ -263,7 +261,7 @@
_Log('Using shadow config file stored at %s', shadow_config_file)
if os.path.exists(shadow_config_file):
- shadow_xbuddy_config = ConfigParser.ConfigParser()
+ shadow_xbuddy_config = configparser.ConfigParser()
shadow_xbuddy_config.read(shadow_config_file)
# Merge shadow config in.
@@ -282,14 +280,14 @@
"""Checks if xBuddy is managing local builds using the current config."""
try:
return self.ParseBoolean(self.config.get(GENERAL, 'manage_builds'))
- except ConfigParser.Error:
+ except configparser.Error:
return False
def _Capacity(self):
"""Gets the xbuddy capacity from the current config."""
try:
return int(self.config.get(GENERAL, 'capacity'))
- except ConfigParser.Error:
+ except configparser.Error:
return 5
def LookupAlias(self, alias, board=None, version=None):
@@ -312,12 +310,12 @@
"""
try:
suffix = self.config.get(LOCATION_SUFFIXES, alias)
- except ConfigParser.Error:
+ except configparser.Error:
suffix = RELEASE
try:
val = self.config.get(PATH_REWRITES, alias)
- except ConfigParser.Error:
+ except configparser.Error:
# No alias lookup found. Return original path.
val = None
@@ -326,12 +324,12 @@
else:
# The found value is not an empty string.
# Fill in the board and version.
- val = val.replace("BOARD", "%(board)s")
- val = val.replace("VERSION", "%(version)s")
+ val = val.replace('BOARD', '%(board)s')
+ val = val.replace('VERSION', '%(version)s')
val = val % {'board': board or self._board,
'version': version or self._version or LATEST}
- _Log("Path is %s, location suffix is %s", val, suffix)
+ _Log('Path is %s, location suffix is %s', val, suffix)
return val, suffix
@staticmethod
@@ -351,7 +349,7 @@
def _LookupOfficial(self, board, suffix, image_dir=None):
"""Check LATEST-master for the version number of interest."""
- _Log("Checking gs for latest %s-%s image", board, suffix)
+ _Log('Checking gs for latest %s-%s image', board, suffix)
image_dir = XBuddy._ResolveImageDir(image_dir)
latest_addr = (devserver_constants.GS_LATEST_MASTER %
{'image_dir': image_dir,
@@ -398,9 +396,9 @@
list_result = self._LS(path, list_subdirectory=list_subdirectory)
dir_names = [os.path.basename(p.rstrip('/')) for p in list_result]
try:
- filter_re = re.compile(devserver_constants.VERSION_RE if with_release
- else devserver_constants.VERSION)
- versions = filter(filter_re.match, dir_names)
+ versions_re = re.compile(devserver_constants.VERSION_RE if with_release
+ else devserver_constants.VERSION)
+ versions = [d for d in dir_names if versions_re.match(d)]
latest_version = max(versions, key=distutils.version.LooseVersion)
except ValueError:
raise gs.GSContextException(
@@ -523,8 +521,8 @@
if re.match(devserver_constants.VERSION_RE, version):
return self._RemoteBuildId(board, suffix, version), None
elif re.match(devserver_constants.VERSION, version):
- raise XBuddyException('\'%s\' is not valid. Should provide the fully '
- 'qualified version with a version prefix \'RX-\' '
+ raise XBuddyException('"%s" is not valid. Should provide the fully '
+ 'qualified version with a version prefix "RX-" '
'due to crbug.com/585914' % version)
elif version == LATEST_OFFICIAL:
# latest-official --> LATEST build in board-release
@@ -551,7 +549,7 @@
@staticmethod
def _Symlink(link, target):
"""Symlinks link to target, and removes whatever link was there before."""
- _Log("Linking to %s from %s", link, target)
+ _Log('Linking to %s from %s', link, target)
if os.path.lexists(link):
os.unlink(link)
os.symlink(target, link)
@@ -609,7 +607,7 @@
Raises:
XBuddyException: if the path can't be resolved into valid components
"""
- path_list = filter(None, path.split('/'))
+ path_list = [p for p in path.split('/') if p]
# Do the stuff that is well known first. We know that if paths have a
# image_type, it must be one of the GS/LOCAL aliases and it must be at the
@@ -647,8 +645,8 @@
version = path_list.pop(0)
if path_list:
- raise XBuddyException("Path isn't valid. Could not figure out how to "
- "parse remaining components: %s." % path_list)
+ raise XBuddyException('Path is not valid. Could not figure out how to '
+ 'parse remaining components: %s.' % path_list)
_Log("Get artifact '%s' with board %s and version %s'. Locally? %s",
image_type, board, version, is_local)
@@ -703,7 +701,7 @@
build_id = Timestamp.TimestampToBuild(f)
stale_time = datetime.timedelta(seconds=(time.time() - last_accessed))
build_dict[build_id] = stale_time
- return_tup = sorted(build_dict.iteritems(), key=operator.itemgetter(1))
+ return_tup = sorted(build_dict.items(), key=operator.itemgetter(1))
return return_tup
def _Download(self, gs_url, artifacts, build_id):
@@ -716,7 +714,7 @@
with XBuddy._staging_thread_count_lock:
XBuddy._staging_thread_count += 1
try:
- _Log("Downloading %s from %s", artifacts, gs_url)
+ _Log('Downloading %s from %s', artifacts, gs_url)
dl = downloader.GoogleStorageDownloader(self.static_dir, gs_url, build_id)
factory = build_artifact.ChromeOSArtifactFactory(
dl.GetBuildDir(), artifacts, [], dl.GetBuild())
diff --git a/xbuddy_unittest.py b/xbuddy_unittest.py
index bcee666..bd39abb 100755
--- a/xbuddy_unittest.py
+++ b/xbuddy_unittest.py
@@ -8,34 +8,33 @@
from __future__ import print_function
-import ConfigParser
import os
import shutil
import tempfile
import time
import unittest
-import mox
+import mock
+from six.moves import configparser
import xbuddy
# Make sure that chromite is available to import.
-import setup_chromite # pylint: disable=unused-import
+import setup_chromite # pylint: disable=unused-import
try:
from chromite.lib import gs
except ImportError as e:
gs = None
-#pylint: disable=W0212
-#pylint: disable=no-value-for-parameter
+# pylint: disable=protected-access
+# pylint: disable=no-value-for-parameter
GS_ALTERNATE_DIR = 'gs://chromeos-alternate-archive/'
-class xBuddyTest(mox.MoxTestBase):
+class xBuddyTest(unittest.TestCase):
"""Regression tests for xbuddy."""
def setUp(self):
- mox.MoxTestBase.setUp(self)
self.static_image_dir = tempfile.mkdtemp('xbuddy_unittest_static')
@@ -61,7 +60,6 @@
def testGetLatestVersionFromGsDir(self):
"""Test that we can get the most recent version from gsutil calls."""
- self.mox.StubOutWithMock(self.mock_xb, '_LS')
mock_data1 = """gs://chromeos-releases/stable-channel/parrot/3701.96.0/
gs://chromeos-releases/stable-channel/parrot/3701.98.0/
gs://chromeos-releases/stable-channel/parrot/3912.100.0/
@@ -73,134 +71,116 @@
gs://chromeos-image-archive/parrot-release/R27-3912.101.0
gs://chromeos-image-archive/parrot-release/R28-3912.101.0"""
- self.mock_xb._LS(mox.IgnoreArg(), list_subdirectory=False).AndReturn(
- mock_data1.splitlines())
- self.mock_xb._LS(mox.IgnoreArg(), list_subdirectory=True).AndReturn(
- mock_data2.splitlines())
-
- self.mox.ReplayAll()
- url = ''
- self.assertEqual(
- self.mock_xb._GetLatestVersionFromGsDir(url, with_release=False),
- '3912.101.0')
- self.assertEqual(
- self.mock_xb._GetLatestVersionFromGsDir(url, list_subdirectory=True,
- with_release=True),
- 'R28-3912.101.0')
- self.mox.VerifyAll()
+ with mock.patch.object(
+ self.mock_xb, '_LS',
+ side_effect=[mock_data1.splitlines(),
+ mock_data2.splitlines()]) as ls_mock:
+ self.assertEqual(
+ self.mock_xb._GetLatestVersionFromGsDir('url1', with_release=False),
+ '3912.101.0')
+ ls_mock.assert_called_with('url1', list_subdirectory=False)
+ self.assertEqual(
+ self.mock_xb._GetLatestVersionFromGsDir(
+ 'url2', list_subdirectory=True, with_release=True),
+ 'R28-3912.101.0')
+ ls_mock.assert_called_with('url2', list_subdirectory=True)
def testLookupOfficial(self):
"""Basic test of _LookupOfficial. Checks that a given suffix is handled."""
- self.mox.StubOutWithMock(gs.GSContext, 'Cat')
- gs.GSContext.Cat(mox.IgnoreArg()).AndReturn('v')
- expected = 'b-s/v'
- self.mox.ReplayAll()
- self.assertEqual(self.mock_xb._LookupOfficial('b', suffix='-s'), expected)
- self.mox.VerifyAll()
+ with mock.patch.object(gs.GSContext, 'Cat', return_value='v') as cat_mock:
+ self.assertEqual(self.mock_xb._LookupOfficial('b', suffix='-s'), 'b-s/v')
+ cat_mock.assert_called_with(
+ 'gs://chromeos-image-archive/b-s/LATEST-master')
- def testLookupChannel(self):
+ @mock.patch('xbuddy.XBuddy._GetLatestVersionFromGsDir',
+ side_effect=['4100.68.0', 'R28-4100.68.0'])
+ def testLookupChannel(self, version_mock):
"""Basic test of _LookupChannel. Checks that a given suffix is handled."""
- self.mox.StubOutWithMock(self.mock_xb, '_GetLatestVersionFromGsDir')
- mock_data1 = '4100.68.0'
- self.mock_xb._GetLatestVersionFromGsDir(
- mox.IgnoreArg(), with_release=False).AndReturn(mock_data1)
- mock_data2 = 'R28-4100.68.0'
- self.mock_xb._GetLatestVersionFromGsDir(
- mox.IgnoreArg(), list_subdirectory=True).AndReturn(mock_data2)
- self.mox.ReplayAll()
- expected = 'b-release/R28-4100.68.0'
self.assertEqual(self.mock_xb._LookupChannel('b', '-release'),
- expected)
- self.mox.VerifyAll()
+ 'b-release/R28-4100.68.0')
+ version_mock.assert_called_with(
+ 'gs://chromeos-image-archive/b-release/R*4100.68.0',
+ list_subdirectory=True)
def testLookupAliasPathRewrite(self):
"""Tests LookupAlias of path rewrite, including keyword substitution."""
- alias = 'foobar'
path = 'remote/BOARD/VERSION/test'
- self.mox.StubOutWithMock(self.mock_xb.config, 'get')
- self.mock_xb.config.get('LOCATION_SUFFIXES', alias).AndRaise(
- ConfigParser.Error())
- self.mock_xb.config.get('PATH_REWRITES', alias).AndReturn(path)
- self.mox.ReplayAll()
- self.assertEqual(('remote/parrot/1.2.3/test', '-release'),
- self.mock_xb.LookupAlias(alias, board='parrot',
- version='1.2.3'))
+ with mock.patch.object(
+ self.mock_xb.config, 'get',
+ side_effect=[configparser.Error, path]) as get_mock:
+ self.assertEqual(('remote/parrot/1.2.3/test', '-release'),
+ self.mock_xb.LookupAlias('foobar', board='parrot',
+ version='1.2.3'))
+ get_mock.assert_called_with('PATH_REWRITES', 'foobar')
def testLookupAliasSuffix(self):
"""Tests LookupAlias of location suffix."""
- alias = 'foobar'
- suffix = '-random'
- self.mox.StubOutWithMock(self.mock_xb.config, 'get')
- self.mock_xb.config.get('LOCATION_SUFFIXES', alias).AndReturn(suffix)
- self.mock_xb.config.get('PATH_REWRITES', alias).AndRaise(
- ConfigParser.Error())
- self.mox.ReplayAll()
- self.assertEqual((alias, suffix),
- self.mock_xb.LookupAlias(alias, board='parrot',
- version='1.2.3'))
+ with mock.patch.object(
+ self.mock_xb.config, 'get',
+ side_effect=['-random', configparser.Error]) as get_mock:
+ self.assertEqual(('foobar', '-random'),
+ self.mock_xb.LookupAlias('foobar', board='parrot',
+ version='1.2.3'))
+ get_mock.assert_called_with('PATH_REWRITES', 'foobar')
def testLookupAliasPathRewriteAndSuffix(self):
"""Tests LookupAlias with both path rewrite and suffix."""
- alias = 'foobar'
path = 'remote/BOARD/VERSION/test'
- suffix = '-random'
- self.mox.StubOutWithMock(self.mock_xb.config, 'get')
- self.mock_xb.config.get('LOCATION_SUFFIXES', alias).AndReturn(suffix)
- self.mock_xb.config.get('PATH_REWRITES', alias).AndReturn(path)
- self.mox.ReplayAll()
- self.assertEqual(('remote/parrot/1.2.3/test', suffix),
- self.mock_xb.LookupAlias(alias, board='parrot',
- version='1.2.3'))
+ with mock.patch.object(
+ self.mock_xb.config, 'get',
+ side_effect=['-random', path]) as get_mock:
+ self.assertEqual(('remote/parrot/1.2.3/test', '-random'),
+ self.mock_xb.LookupAlias('foobar', board='parrot',
+ version='1.2.3'))
+ get_mock.assert_called_with('PATH_REWRITES', 'foobar')
- def testResolveVersionToBuildIdAndChannel_Official(self):
+ @mock.patch('xbuddy.XBuddy._LookupOfficial')
+ def testResolveVersionToBuildIdAndChannel_Official(self, lookup_mock):
"""Check _ResolveVersionToBuildIdAndChannel support for official build."""
- board = 'b'
+ board = 'chell'
suffix = '-s'
- # aliases that should be redirected to LookupOfficial
-
- self.mox.StubOutWithMock(self.mock_xb, '_LookupOfficial')
- self.mock_xb._LookupOfficial(board, suffix, image_dir=None)
- self.mock_xb._LookupOfficial(board, suffix,
- image_dir=GS_ALTERNATE_DIR)
- self.mock_xb._LookupOfficial(board, 'paladin', image_dir=None)
- self.mock_xb._LookupOfficial(board, 'paladin',
- image_dir=GS_ALTERNATE_DIR)
-
- self.mox.ReplayAll()
version = 'latest-official'
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version)
+ lookup_mock.assert_called_with('chell', '-s', image_dir=None)
+
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version,
image_dir=GS_ALTERNATE_DIR)
+ lookup_mock.assert_called_with('chell', '-s',
+ image_dir='gs://chromeos-alternate-archive/')
+
version = 'latest-official-paladin'
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version)
+ lookup_mock.assert_called_with('chell', 'paladin', image_dir=None)
+
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version,
image_dir=GS_ALTERNATE_DIR)
- self.mox.VerifyAll()
+ lookup_mock.assert_called_with('chell', 'paladin',
+ image_dir='gs://chromeos-alternate-archive/')
- def testResolveVersionToBuildIdAndChannel_Channel(self):
+ @mock.patch('xbuddy.XBuddy._LookupChannel')
+ def testResolveVersionToBuildIdAndChannel_Channel(self, lookup_mock):
"""Check _ResolveVersionToBuildIdAndChannel support for channels."""
- board = 'b'
+ board = 'chell'
suffix = '-s'
- # aliases that should be redirected to LookupChannel
- self.mox.StubOutWithMock(self.mock_xb, '_LookupChannel')
- self.mock_xb._LookupChannel(board, suffix, image_dir=None)
- self.mock_xb._LookupChannel(board, suffix, image_dir=GS_ALTERNATE_DIR)
- self.mock_xb._LookupChannel(board, suffix, channel='dev', image_dir=None)
- self.mock_xb._LookupChannel(board, suffix, channel='dev',
- image_dir=GS_ALTERNATE_DIR)
-
- self.mox.ReplayAll()
version = 'latest'
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version)
+ lookup_mock.assert_called_with('chell', '-s', image_dir=None)
+
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version,
image_dir=GS_ALTERNATE_DIR)
+ lookup_mock.assert_called_with('chell', '-s',
+ image_dir='gs://chromeos-alternate-archive/')
+
version = 'latest-dev'
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version)
+ lookup_mock.assert_called_with('chell', '-s', channel='dev', image_dir=None)
+
self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version,
image_dir=GS_ALTERNATE_DIR)
- self.mox.VerifyAll()
+ lookup_mock.assert_called_with('chell', '-s', channel='dev',
+ image_dir='gs://chromeos-alternate-archive/')
# TODO(dgarrett): Re-enable when crbug.com/585914 is fixed.
# def testResolveVersionToBuildId_BaseVersion(self):
@@ -329,56 +309,57 @@
result = self.mock_xb._ListBuildTimes()
self.assertEqual(len(result), 4)
- ############### Public Methods
def testXBuddyCaching(self):
"""Caching & replacement of timestamp files."""
- path_a = ('remote', 'a', 'R0', 'test')
- path_b = ('remote', 'b', 'R0', 'test')
- self.mox.StubOutWithMock(gs.GSContext, 'LS')
- self.mox.StubOutWithMock(self.mock_xb, '_Download')
- for _ in range(8):
- self.mock_xb._Download(mox.IsA(str), mox.In(mox.IsA(str)), mox.IsA(str))
- # All non-release urls are invalid so we can meet expectations.
- gs.GSContext.LS(
- mox.Not(mox.StrContains('-release'))).MultipleTimes().AndRaise(
- gs.GSContextException('bad url'))
- gs.GSContext.LS(mox.StrContains('-release')).MultipleTimes()
+ def _ReleaseOnly(name):
+ # All non-release URLs are invalid so we can meet expectations.
+ if name.find('-release') == -1:
+ raise gs.GSContextException('bad URL')
+ return name
- self.mox.ReplayAll()
+ with mock.patch.object(
+ gs.GSContext, 'LS', side_effect=_ReleaseOnly) as ls_mock:
+ with mock.patch.object(
+ self.mock_xb, '_Download') as download_mock:
+ version = '%s-release/R0'
- # requires default capacity
- self.assertEqual(self.mock_xb.Capacity(), '5')
+ def _Download(image):
+ gs_image = 'gs://chromeos-image-archive/' + version
+ self.mock_xb.Get(('remote', image, 'R0', 'test'))
+ ls_mock.assert_called_with(gs_image % image)
+ download_mock.assert_called_with(gs_image % image, ['test_image'],
+ version % image)
+ time.sleep(0.05)
- # Get 6 different images: a,b,c,d,e,f
- images = ['a', 'b', 'c', 'd', 'e', 'f']
- for c in images:
- self.mock_xb.Get(('remote', c, 'R0', 'test'))
- time.sleep(0.05)
+ # Requires default capacity.
+ self.assertEqual(self.mock_xb.Capacity(), '5')
- # check that b,c,d,e,f are still stored
- result = self.mock_xb._ListBuildTimes()
- self.assertEqual(len(result), 5)
+ # Get 6 different images: a,b,c,d,e,f.
+ images = ['a', 'b', 'c', 'd', 'e', 'f']
+ for image in images:
+ _Download(image)
- # Flip the list to get reverse chronological order
- images.reverse()
- for i in range(5):
- self.assertEqual(result[i][0], '%s-release/R0' % images[i])
+ # Check that b,c,d,e,f are still stored.
+ result = self.mock_xb._ListBuildTimes()
+ self.assertEqual(len(result), 5)
- # Get b,a
- self.mock_xb.Get(path_b)
- time.sleep(0.05)
- self.mock_xb.Get(path_a)
- time.sleep(0.05)
+ # Flip the list to get reverse chronological order
+ images.reverse()
+ for i in range(5):
+ self.assertEqual(result[i][0], version % images[i])
- # check that d,e,f,b,a are still stored
- result = self.mock_xb._ListBuildTimes()
- self.assertEqual(len(result), 5)
- images_expected = ['a', 'b', 'f', 'e', 'd']
- for i in range(5):
- self.assertEqual(result[i][0], '%s-release/R0' % images_expected[i])
+ # Get b,a.
+ images = ['b', 'a']
+ for image in images:
+ _Download(image)
- self.mox.VerifyAll()
+ # Check that d,e,f,b,a are still stored.
+ result = self.mock_xb._ListBuildTimes()
+ self.assertEqual(len(result), 5)
+ images_expected = ['a', 'b', 'f', 'e', 'd']
+ for i in range(5):
+ self.assertEqual(result[i][0], '%s-release/R0' % images_expected[i])
if __name__ == '__main__':
