blob: 92dbcd6fb7e88d0f343e35946a8435f847e2c294 [file] [log] [blame]
# Copyright 2018 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import contextlib
import urllib
from recipe_engine import recipe_api
DEFAULT_CIPD_VERSION = 'git_revision:4520b945a32ae72645adf048ff86bfdc87ac9810'
class IsolatedApi(recipe_api.RecipeApi):
"""API for interacting with isolated.
The isolated client implements a tar-like scatter-gather mechanism for
archiving files. The tool's source lives at
This module will deploy the client to [CACHE]/isolated_client/; users should
add this path to the named cache for their builder.
def __init__(self, isolated_properties, *args, **kwargs):
super(IsolatedApi, self).__init__(*args, **kwargs)
self._server = isolated_properties.get('server', None)
self._namespace = isolated_properties.get('namespace', 'default-gzip')
self._version = isolated_properties.get('version', DEFAULT_CIPD_VERSION)
self._client_dir = None
self._client = None
def initialize(self):
if self._test_data.enabled:
self._server = ''
if self.m.runtime.is_experimental:
self._version = 'latest'
self._client_dir = self.m.path['cache'].join('isolated_client')
def _ensure_isolated(self):
"""Ensures that the isolated Go binary is installed."""
if self._client:
with self.m.step.nest('ensure isolated'):
with self.m.context(infra_steps=True):
pkgs = self.m.cipd.EnsureFile()
pkgs.add_package('infra/tools/luci/isolated/${platform}', self._version)
self.m.cipd.ensure(self._client_dir, pkgs)
self._client = self._client_dir.join('isolated')
def isolate_server(self):
"""Returns the associated isolate server."""
assert self._server
return self._server
def namespace(self):
"""Returns the associated namespace."""
assert self._namespace
return self._namespace
def _run(self, name, cmd, step_test_data=None):
"""Return an isolated command step.
name: (str): name of the step.
cmd (list(str|Path)): isolated client subcommand to run.
return self.m.step(name,
[self._client] + list(cmd),
def on_path(self):
"""This context manager ensures the go isolated client is available on
with api.isolated.on_path():
# do your steps which require the isolated binary on path
with self.m.context(env_prefixes={'PATH': [self._client_dir]}):
def isolated(self, root_dir):
"""Returns an Isolated object that can be used to archive a set of files
and directories, relative to a given root directory.
root_dir (Path): directory relative to which files and directory will be
return Isolated(self.m, root_dir)
def download(self, step_name, isolated_hash, output_dir, isolate_server=None):
"""Downloads an isolated tree from an isolate server.
step_name (str): name of the step.
isolated_hash (str): the hash of an isolated tree.
output_dir (Path): Path to an output directory. If a non-existent
directory, it will be created; else if already existent,
conflicting files will be overwritten and non-conflicting files
already in the directory will be ignored.
isolate_server (str|None): an isolate server to donwload from; if None,
the module's default server will be used instead.
isolate_server = isolate_server or self.isolate_server
cmd = [
'-isolate-server', isolate_server,
'-isolated', isolated_hash,
'-output-dir', output_dir,
return self._run(step_name, cmd)
class Isolated(object):
"""Used to gather a list of files and directories to an isolated, relative to
a provided root directory."""
def __init__(self, api, root_dir):
assert root_dir
self._api = api
self._root_dir = root_dir
self._files = []
self._dirs = []
def _isolated_path_format(self, path):
"""Returns the path format consumed by the isolated CLI."""
if self._root_dir.is_parent_of(path):
return '%s:%s' % (
assert path == self._root_dir, \
"isolated path must be equal to or within %s" % self._root_dir
return '%s:.' % self._root_dir
def add_file(self, path):
"""Stages a single file to be added to the isolated.
path (Path): absolute path to a file.
assert self._root_dir.is_parent_of(path)
def add_files(self, paths):
"""Stages a list of files to be added to the isolated.
paths list(Path): list of absolute paths to files.
for path in paths:
def add_dir(self, path):
"""Stages a single directory to be added to the isolated.
path (Path): absolute path to a directory.
assert self._root_dir == path or self._root_dir.is_parent_of(path)
def archive(self, step_name, isolate_server=None):
"""Step to archive all staged files and directories.
If no isolate_server is provided, the IsolatedApis's default server will be
used instead.
step_name (str): name of the step.
isolate_server (str): an isolate server to archive to.
The hash of the isolated tree.
isolate_server = isolate_server or self._api.isolated.isolate_server
cmd = [
'-isolate-server', isolate_server,
'-namespace', self._api.isolated.namespace,
'-dump-hash', self._api.raw_io.output_text(),
for f in self._files:
cmd.extend(['-files', self._isolated_path_format(f)])
for d in self._dirs:
cmd.extend(['-dirs', self._isolated_path_format(d)])
isolated_hash = self._api.isolated._run(
q = {
'hash': isolated_hash,
'namespace': self._api.isolated.namespace,
self._api.step.active_result.presentation.links['isolated UI'] = (
'%s/browse?%s' % (isolate_server, urllib.urlencode(q))
return isolated_hash