blob: b458364daff684bcbc8c0cfd152e671050c970cc [file] [log] [blame]
# Copyright 2020 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Stage the Chromium checkout to update CTS test version."""
import contextlib
import json
import operator
import os
import re
import sys
import tempfile
import threading
try:
# Workaround for py2/3 compatibility.
# TODO(pbirk): remove once py2 support is no longer needed.
import urllib.request as urllib_request
except ImportError:
import urllib as urllib_request
import zipfile
sys.path.append(
os.path.join(
os.path.dirname(__file__), os.pardir, os.pardir, 'third_party',
'catapult', 'devil'))
# pylint: disable=wrong-import-position,import-error
from devil.utils import cmd_helper
sys.path.append(
os.path.join(
os.path.dirname(__file__), os.pardir, os.pardir, 'third_party',
'catapult', 'common', 'py_utils'))
# pylint: disable=wrong-import-position,import-error
from py_utils import tempfile_ext
SRC_DIR = os.path.abspath(
os.path.join(os.path.dirname(__file__), os.pardir, os.pardir))
TOOLS_DIR = os.path.join('android_webview', 'tools')
CONFIG_FILE = os.path.join('cts_config', 'webview_cts_gcs_path.json')
CONFIG_PATH = os.path.join(SRC_DIR, TOOLS_DIR, CONFIG_FILE)
CIPD_FILE = os.path.join('cts_archive', 'cipd.yaml')
CIPD_PATH = os.path.join(SRC_DIR, TOOLS_DIR, CIPD_FILE)
DEPS_FILE = 'DEPS'
TEST_SUITES_FILE = os.path.join('testing', 'buildbot', 'test_suites.pyl')
CTS_DEP_NAME = 'src/android_webview/tools/cts_archive'
CTS_DEP_PACKAGE = 'chromium/android_webview/tools/cts_archive'
CIPD_REFERRERS = [DEPS_FILE, TEST_SUITES_FILE]
_GENERATE_BUILDBOT_JSON = os.path.join('testing', 'buildbot',
'generate_buildbot_json.py')
_ENSURE_FORMAT = """$ParanoidMode CheckIntegrity
@Subdir cipd
{} {}"""
_ENSURE_SUBDIR = 'cipd'
_RE_COMMENT_OR_BLANK = re.compile(r'^ *(#.*)?$')
class CTSConfig:
"""Represents a CTS config file."""
def __init__(self, file_path=CONFIG_PATH):
"""Constructs a representation of the CTS config file.
Only read operations are provided by this object. Users should edit the
file manually for any modifications.
Args:
file_path: Path to file.
"""
self._path = os.path.abspath(file_path)
with open(self._path) as f:
self._config = json.load(f)
def save(self):
with open(self._path, 'w') as file:
json.dump(self._config, file, indent=2)
file.write("\n")
def get_platforms(self):
return sorted(self._config.keys())
def get_archs(self, platform):
return sorted(self._config[platform]['arch'].keys())
def get_git_tag_prefix(self, platform):
return self._config[platform]['git']['tag_prefix']
def iter_platforms(self):
for p in self.get_platforms():
yield p
def iter_platform_archs(self):
for p in self.get_platforms():
for a in self.get_archs(p):
yield p, a
def get_cipd_zip(self, platform, arch):
return self._config[platform]['arch'][arch]['filename']
def get_origin(self, platform, arch):
return self._config[platform]['arch'][arch]['_origin']
def get_origin_zip(self, platform, arch):
return os.path.basename(self.get_origin(platform, arch))
def get_apks(self, platform):
return sorted([r['apk'] for r in self._config[platform]['test_runs']])
def get_additional_apks(self, platform):
return [
apk['apk'] for r in self._config[platform]['test_runs']
for apk in r.get('additional_apks', [])
]
def set_release_version(self, platform, arch, release):
pattern = re.compile(r'(?<=_r)\d*')
def update_release_version(field):
return pattern.sub(str(release),
self._config[platform]['arch'][arch][field])
self._config[platform]['arch'][arch] = {
'filename': update_release_version('filename'),
'_origin': update_release_version('_origin'),
'unzip_dir': update_release_version('unzip_dir'),
}
class CTSCIPDYaml:
"""Represents a CTS CIPD yaml file."""
RE_PACKAGE = r'^package:\s*(\S+)\s*$'
RE_DESC = r'^description:\s*(.+)$'
RE_DATA = r'^data:\s*$'
RE_FILE = r'^\s+-\s+file:\s*(.+)$'
# TODO(crbug.com/1049432): Replace with yaml parser
@classmethod
def parse(cls, lines):
result = {}
for line in lines:
if len(line) == 0 or line[0] == '#':
continue
package_match = re.match(cls.RE_PACKAGE, line)
if package_match:
result['package'] = package_match.group(1)
continue
desc_match = re.match(cls.RE_DESC, line)
if desc_match:
result['description'] = desc_match.group(1)
continue
if re.match(cls.RE_DATA, line):
result['data'] = []
if 'data' in result:
file_match = re.match(cls.RE_FILE, line)
if file_match:
result['data'].append({'file': file_match.group(1)})
return result
def __init__(self, file_path=CIPD_PATH):
"""Constructs a representation of CTS CIPD yaml file.
Note the file won't be modified unless write is called
with its path.
Args:
file_path: Path to file.
"""
self._path = os.path.abspath(file_path)
self._header = []
# Read header comments
with open(self._path) as f:
for l in f.readlines():
if re.match(_RE_COMMENT_OR_BLANK, l):
self._header.append(l)
else:
break
# Read yaml data
with open(self._path) as f:
self._yaml = CTSCIPDYaml.parse(f.readlines())
def get_file_path(self):
"""Get full file path of yaml file that this was constructed from."""
return self._path
def get_file_basename(self):
"""Get base file name that this was constructed from."""
return os.path.basename(self._path)
def get_package(self):
"""Get package name."""
return self._yaml['package']
def clear_files(self):
"""Clears all files in file (only in local memory, does not modify file)."""
self._yaml['data'] = []
def append_file(self, file_name):
"""Add file_name to list of files."""
self._yaml['data'].append({'file': str(file_name)})
def remove_file(self, file_name):
"""Remove file_name from list of files."""
old_file_names = self.get_files()
new_file_names = [name for name in old_file_names if name != file_name]
self._yaml['data'] = [{'file': name} for name in new_file_names]
def get_files(self):
"""Get list of files in yaml file."""
return [e['file'] for e in self._yaml['data']]
def write(self, file_path):
"""(Over)write file_path with the cipd.yaml representation."""
dir_name = os.path.dirname(file_path)
if not os.path.isdir(dir_name):
os.makedirs(dir_name)
with open(file_path, 'w') as f:
f.writelines(self._get_yamls())
def _get_yamls(self):
"""Return the cipd.yaml file contents of this object."""
output = []
output += self._header
output.append('package: {}\n'.format(self._yaml['package']))
output.append('description: {}\n'.format(self._yaml['description']))
output.append('data:\n')
for d in sorted(self._yaml['data'], key=operator.itemgetter('file')):
output.append(' - file: {}\n'.format(d.get('file')))
return output
def cipd_ensure(package, version, root_dir):
"""Ensures CIPD package is installed at root_dir.
Args:
package: CIPD name of package
version: Package version
root_dir: Directory to install package into
"""
def _createEnsureFile(package, version, file_path):
with open(file_path, 'w') as f:
f.write(_ENSURE_FORMAT.format(package, version))
def _ensure(root, ensure_file):
ret = cmd_helper.RunCmd(
['cipd', 'ensure', '-root', root, '-ensure-file', ensure_file])
if ret:
raise IOError('Error while running cipd ensure: ' + ret)
with tempfile.NamedTemporaryFile() as f:
_createEnsureFile(package, version, f.name)
_ensure(root_dir, f.name)
def cipd_download(cipd, version, download_dir):
"""Downloads CIPD package files.
This is different from cipd ensure in that actual files will exist at
download_dir instead of symlinks.
Args:
cipd: CTSCIPDYaml object
version: Version of package
download_dir: Destination directory
"""
package = cipd.get_package()
download_dir_abs = os.path.abspath(download_dir)
if not os.path.isdir(download_dir_abs):
os.makedirs(download_dir_abs)
with tempfile_ext.NamedTemporaryDirectory() as workDir, chdir(workDir):
cipd_ensure(package, version, '.')
for file_name in cipd.get_files():
src_path = os.path.join(_ENSURE_SUBDIR, file_name)
dest_path = os.path.join(download_dir_abs, file_name)
dest_dir = os.path.dirname(dest_path)
if not os.path.isdir(dest_dir):
os.makedirs(dest_dir)
ret = cmd_helper.RunCmd(['cp', '--reflink=never', src_path, dest_path])
if ret:
raise IOError('Error file copy from ' + file_name + ' to ' + dest_path)
def filter_cts_file(cts_config, cts_zip_file, dest_dir):
"""Filters out non-webview test apks from downloaded CTS zip file.
Args:
cts_config: CTSConfig object
cts_zip_file: Path to downloaded CTS zip, retaining the original filename
dest_dir: Destination directory to filter to, filename will be unchanged
"""
for p in cts_config.get_platforms():
for a in cts_config.get_archs(p):
o = cts_config.get_origin(p, a)
base_name = os.path.basename(o)
if base_name == os.path.basename(cts_zip_file):
filterzip(cts_zip_file,
cts_config.get_apks(p) + cts_config.get_additional_apks(p),
os.path.join(dest_dir, base_name))
return
raise ValueError('Could not find platform and arch for: ' + cts_zip_file)
class ChromiumRepoHelper:
"""Performs operations on Chromium checkout."""
def __init__(self, root_dir=SRC_DIR):
self._root_dir = os.path.abspath(root_dir)
self._cipd_referrers = [
os.path.join(self._root_dir, p) for p in CIPD_REFERRERS
]
@property
def cipd_referrers(self):
return self._cipd_referrers
@property
def cts_cipd_package(self):
return CTS_DEP_PACKAGE
def get_cipd_dependency_rev(self):
"""Return CTS CIPD revision in the checkout's DEPS file."""
deps_file = os.path.join(self._root_dir, DEPS_FILE)
# Use the gclient command instead of gclient_eval since the latter is not
# intended for direct use outside of depot_tools. The .bat file extension
# must be explicitly specified when shell=False.
gclient = 'gclient.bat' if os.name == 'nt' else 'gclient'
cmd = [
gclient, 'getdep', '--revision',
'%s:%s' % (CTS_DEP_NAME, CTS_DEP_PACKAGE), '--deps-file', deps_file
]
env = os.environ
# Disable auto-update of depot tools since update_depot_tools may not be
# available (for example, on the presubmit bot), and it's probably best not
# to perform surprise updates anyways.
env.update({'DEPOT_TOOLS_UPDATE': '0'})
status, output, err = cmd_helper.GetCmdStatusOutputAndError(cmd, env=env)
if status != 0:
raise Exception('Command "%s" failed: %s' % (' '.join(cmd), err))
return output.strip()
def update_cts_cipd_rev(self, new_version):
"""Update references to CTS CIPD revision in checkout.
Args:
new_version: New version to use
"""
old_version = self.get_cipd_dependency_rev()
for path in self.cipd_referrers:
replace_cipd_revision(path, old_version, new_version)
def git_status(self, path):
"""Returns canonical git status of file.
Args:
path: Path to file.
Returns:
Output of git status --porcelain.
"""
with chdir(self._root_dir):
output = cmd_helper.GetCmdOutput(['git', 'status', '--porcelain', path])
return output
def update_testing_json(self):
"""Performs generate_buildbot_json.py.
Raises:
IOError: If generation failed.
"""
with chdir(self._root_dir):
ret = cmd_helper.RunCmd(['vpython3', _GENERATE_BUILDBOT_JSON])
if ret:
raise IOError('Error while generating_buildbot_json.py')
def rebase(self, *rel_path_parts):
"""Construct absolute path from parts relative to root_dir.
Args:
rel_path_parts: Parts of the root relative path.
Returns:
The absolute path.
"""
return os.path.join(self._root_dir, *rel_path_parts)
def replace_cipd_revision(file_path, old_revision, new_revision):
"""Replaces cipd revision strings in file.
Args:
file_path: Path to file.
old_revision: Old cipd revision to be replaced.
new_revision: New cipd revision to use as replacement.
Returns:
Number of replaced occurrences.
Raises:
IOError: If no occurrences were found.
"""
with open(file_path) as f:
contents = f.read()
num = contents.count(old_revision)
if not num:
raise IOError('Did not find old CIPD revision {} in {}'.format(
old_revision, file_path))
newcontents = contents.replace(old_revision, new_revision)
with open(file_path, 'w') as f:
f.write(newcontents)
return num
@contextlib.contextmanager
def chdir(dirPath):
"""Context manager that changes working directory."""
cwd = os.getcwd()
os.chdir(dirPath)
try:
yield
finally:
os.chdir(cwd)
def filterzip(inputPath, pathList, outputPath):
"""Copy a subset of files from input archive into output archive.
Args:
inputPath: Input archive path
pathList: List of file names from input archive to copy
outputPath: Output archive path
"""
with zipfile.ZipFile(os.path.abspath(inputPath), 'r') as inputZip,\
zipfile.ZipFile(os.path.abspath(outputPath), 'w') as outputZip,\
tempfile_ext.NamedTemporaryDirectory() as workDir,\
chdir(workDir):
for p in pathList:
inputZip.extract(p)
outputZip.write(p)
def download(url, destination):
"""Asynchronously download url to path specified by destination.
Args:
url: Url location of file.
destination: Path where file should be saved to.
If destination parent directories do not exist, they will be created.
Returns the download thread which can then be joined by the caller to
wait for download completion.
"""
dest_dir = os.path.dirname(destination)
if not os.path.isdir(dest_dir):
os.makedirs(dest_dir)
t = threading.Thread(target=urllib_request.urlretrieve,
args=(url, destination))
t.start()
return t
def update_cipd_package(cipd_yaml_path):
"""Updates the CIPD package specified by cipd_yaml_path.
Args:
cipd_yaml_path: Path of cipd yaml specification file
"""
cipd_yaml_path_abs = os.path.abspath(cipd_yaml_path)
with chdir(os.path.dirname(cipd_yaml_path_abs)),\
tempfile.NamedTemporaryFile() as jsonOut:
ret = cmd_helper.RunCmd([
'cipd', 'create', '-pkg-def', cipd_yaml_path_abs, '-json-output',
jsonOut.name
])
if ret:
raise IOError('Error during cipd create.')
return json.load(jsonOut)['result']['instance_id']