blob: 7e45c10732c0661e6c442883c5537f1b1ed6b813 [file] [log] [blame]
# Copyright 2014 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.
from recipe_engine import recipe_api
class ZipApi(recipe_api.RecipeApi):
"""Provides steps to zip and unzip files."""
def make_package(self, root, output):
"""Returns ZipPackage object that can be used to compress a set of files.
Usage:
pkg = api.zip.make_package(root, output)
pkg.add_file(root.join('file'))
pkg.add_directory(root.join('directory'))
yield pkg.zip('zipping step')
Args:
root: a directory that would become root of a package, all files added to
an archive will have archive paths relative to this directory.
output: path to a zip file to create.
Returns:
ZipPackage object.
"""
return ZipPackage(self.m, root, output)
def directory(self, step_name, directory, output):
"""Step to compress a single directory.
Args:
step_name: display name of the step.
directory: path to a directory to compress, it would become the root of
an archive, i.e. |directory|/file.txt would be named 'file.txt' in
the archive.
output: path to a zip file to create.
"""
pkg = self.make_package(directory, output)
pkg.add_directory(directory)
pkg.zip(step_name)
def namelist(self, step_name, zip_file):
"""Step to get the name list of |zip_file|.
Args:
step_name: display name of a step.
zip_file: path to a zip file to get its namelist, should exist.
"""
script_input = {
'zip_file': str(zip_file),
}
names_step = self.m.step(
step_name,
['python3', self.resource('namelist.py')],
stdin=self.m.json.input(script_input),
stdout=self.m.json.output(),
)
return names_step.stdout or []
def unzip(self, step_name, zip_file, output, quiet=False):
"""Step to uncompress |zip_file| into |output| directory.
Zip package will be unpacked to |output| so that root of an archive is in
|output|, i.e. archive.zip/file.txt will become |output|/file.txt.
Step will FAIL if |output| already exists.
Args:
step_name: display name of a step.
zip_file: path to a zip file to uncompress, should exist.
output: path to a directory to unpack to, it should NOT exist.
quiet (bool): If True, print terse output instead of the name
of each unzipped file.
"""
# TODO(vadimsh): Use 7zip on Windows if available?
script_input = {
'output': str(output),
'zip_file': str(zip_file),
'quiet': quiet,
}
self.m.step(
step_name, ['python3', self.resource('unzip.py')],
stdin=self.m.json.input(script_input)
)
class ZipPackage(object):
"""Used to gather a list of files to zip."""
def __init__(self, api, root, output):
self._api = api
self._root = root
self._output = output
self._entries = []
@property
def root(self):
return self._root
@property
def output(self):
return self._output
def add_file(self, path, archive_name=None):
"""Stages single file to be added to the package.
Args:
path: absolute path to a file, should be in |root| subdirectory.
archive_name: name of the file in the archive, if non-None
"""
assert self._root.is_parent_of(path), path
self._entries.append({
'type': 'file', 'path': str(path), 'archive_name': archive_name
})
def add_directory(self, path):
"""Stages a directory with all its content to be added to the package.
Args:
path: absolute path to a directory, should be in |root| subdirectory.
"""
# TODO(vadimsh): Implement 'exclude' filter.
assert self._root.is_parent_of(path) or path == self._root, path
self._entries.append({
'type': 'dir',
'path': str(path),
})
def zip(self, step_name):
"""Step to zip all staged files."""
script_input = {
'entries': self._entries,
'output': str(self._output),
'root': str(self._root),
}
step_result = self._api.step(
step_name, ['python3', self._api.zip.resource('zip.py')],
stdin=self._api.json.input(script_input)
)
self._api.path.mock_add_paths(self._output)
return step_result