blob: 3d4a0a3b233ddfe7634420532c75781b2beb20ac [file] [log] [blame]
# Copyright (c) 2011 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.
"""This implements calling of scripts and other utilities/tools.
We support running inside and outside the chroot. To do this, we permit
a ## prefix which resolves to the chroot. Within the chroot this will be
/ but outside it will be the full path to the chroot.
So we can use filenames like this:
import doctest
import optparse
import os
import re
import shutil
import struct
import sys
import tempfile
import unittest
from chromite.lib import cros_build_lib
from chromite.lib import git
import cros_output
# Attributes defined outside __init__
#pylint: disable=W0201
class CmdError(Exception):
"""An error in the execution of a command."""
class Tools:
"""A class to encapsulate the external tools we want to run.
This provides convenient functions for running tools inside/outside the
Public properties:
outdir: The output directory to write output files to.
search_paths: The list of directories to search for files we are asked
to read.
The tools class also provides common paths:
chroot_path: chroot directory
src_path: source directory
script_path: scripts directory (src/scripts)
overlay_path: overlays directory (src/overlays)
priv_overlay_path: private overlays directory (src/private-overlays)
board_path: build directory (/build in chroot)
third_party_path: third_parth directory (src/third_party)
cros_overlay_path: Chromium OS overlay (src/chromiumos-overlay)
def __init__(self, output):
"""Set up the tools system.
output: cros_output object to use for output.
IOError: Unable to find .repo directory
# Detect whether we're inside a chroot or not
self.in_chroot = cros_build_lib.IsInsideChroot()
self._out = output
self._root = None
self.chroot_path = None
root_dir = None
if self.in_chroot:
root_dir = os.getenv('CROS_WORKON_SRCROOT')
repo = git.FindRepoDir('.')
if repo:
root_dir = os.path.dirname(repo)
if root_dir:
self._out.Info("Chroot is at '%s'" % self.chroot_path)
self._out.Info('Running outside chroot')
self._tools = {
'make_bmp_image': '##/usr/share/vboot/bitmaps/',
'bct_dump': '##/usr/bin/bct_dump',
'tegrarcm': '##/usr/bin/tegrarcm',
'gbb_utility': '##/usr/bin/gbb_utility',
'cbfstool': '##/usr/bin/cbfstool',
'fdisk': '##/sbin/fdisk',
self.outdir = None # We have no output directory yet
self.search_paths = []
def __enter__(self):
return self
def __exit__(self, the_type, value, traceback):
return False
def _SetRoot(self, root_dir):
"""Sets the root directory for the build envionrment.
The root directory is the one containing .repo, chroot and src.
This should be called once the root is known. All other parts are
calculated from this.
root_dir: The path to the root directory.
self._root = os.path.normpath(root_dir)
# Set up the path to prepend to get to the chroot
if self.in_chroot:
self.chroot_path = '/'
self.chroot_path = os.path.join(self._root, 'chroot')
self.src_path = os.path.join(self._root, 'src')
self.script_path = os.path.join(self.src_path, 'scripts')
self.overlay_path = os.path.join(self.src_path, 'overlays')
self.priv_overlay_path = os.path.join(self.src_path,
self.board_path = os.path.join(self.chroot_path, 'build')
self.third_party_path = os.path.join(self.src_path, 'third_party')
self.cros_overlay_path = os.path.join(self.third_party_path,
def Filename(self, fname):
"""Resolve a file path to an absolute path.
If fname starts with ##/ and chroot is available, ##/ gets replaced with
the chroot path. If chroot is not available, this file name can not be
resolved, `None' is returned.
If fname is not prepended with the above prefix, and is not an existing
file, the actual file name is retrieved from the passed in string and the
search_paths directories (if any) are searched to for the file. If found -
the path to the found file is returned, `None' is returned otherwise.
fname: a string, the path to resolve.
Absolute path to the file or None if not found.
if fname.startswith('##/'):
if self.chroot_path:
fname = os.path.join(self.chroot_path, fname[3:])
return None
# Search for a pathname that exists, and return it if found
if fname and not os.path.exists(fname):
for path in self.search_paths:
pathname = os.path.join(path, os.path.basename(fname))
if os.path.exists(pathname):
return pathname
# If not found, just return the standard, unchanged path
return fname
def Run(self, tool, args, cwd=None, sudo=False):
"""Run a tool with given arguments.
The tool name may be used unchanged or substituted with a full path if
The tool and arguments can use ##/ to signify the chroot (at the beginning
of the tool/argument).
tool: Name of tool to run.
args: List of arguments to pass to tool.
cwd: Directory to change into before running tool (None if none).
sudo: True to run the tool with sudo
Output of tool (stdout).
CmdError: If running the tool, or the tool itself creates an error.
if tool in self._tools:
tool = self._tools[tool]
tool = self.Filename(tool)
args = [self.Filename(arg) for arg in args]
cmd = [tool] + args
if sudo and os.getuid():
cmd.insert(0, 'sudo')
result = cros_build_lib.RunCommandCaptureOutput(
cmd, cwd=cwd, print_cmd=self._out.verbose > 3,
combine_stdout_stderr=True, error_code_ok=True)
except cros_build_lib.RunCommandError as ex:
raise CmdError(str(ex))
stdout = result.output
if result.returncode:
raise CmdError('Command failed: %s\n%s' % (' '.join(cmd), stdout))
return stdout
def ReadFile(self, fname):
"""Read and return the contents of a file.
fname: path to filename to read, where ## signifiies the chroot.
data read from file, as a string.
fd = open(self.Filename(fname), 'rb')
data =
self._out.Info("Read file '%s' size %d (%#0x)" %
(fname, len(data), len(data)))
return data
def WriteFile(self, fname, data):
"""Write data into a file.
fname: path to filename to write, where ## signifiies the chroot.
data: data to write to file, as a string.
self._out.Info("Write file '%s' size %d (%#0x)" %
(fname, len(data), len(data)))
fd = open(self.Filename(fname), 'wb')
def ReadFileAndConcat(self, filenames, compress=None, with_index=False):
"""Read several files and concat them.
filenames: a list containing name of the files to read.
with_index: If true, an index structure is prepended to the data.
A tuple of a string and two list. The string is the concated data read
from file, in the same order as in filenames, aligned to 4-byte. The
first list contains the offset of each file in the data string and
the second one contains the actual (non-padded) length of each file,
both in the same order.
The optional index structure is a 32 bit integer set to the number of
entries in the index, followed by that many pairs of integers which
describe the offset and length of each chunk.
data = ''
offsets = []
lengths = []
for fname in filenames:
content = self.ReadFile(fname)
pad_len = ((len(content) + 3) & ~3) - len(content)
data += content + chr(0xff) * pad_len
if with_index:
index_size = 4 + len(filenames) * 8
index = struct.pack("<I", len(filenames))
offsets = tuple(offset + index_size for offset in offsets)
for _, offset, length in zip(filenames, offsets, lengths):
index += struct.pack("<II", offset, length)
data = index + data
if compress:
if compress == 'lzo':
# Would be nice to just pipe here. but we don't have RunPipe().
fname = self.GetOutputFilename('data.tmp')
outname = self.GetOutputFilename('data.tmp.lzo')
if os.path.exists(outname):
self.WriteFile(fname, data)
args = ['-9', fname]
self.Run('lzop', args)
data = self.ReadFile(outname)
raise ValueError("Unknown compression method '%s'" % compress)
return data, offsets, lengths
def GetChromeosVersion(self):
"""Returns the ChromeOS version string.
This works by finding and executing the version script:
Version string in the form '0.14.726.2011_07_07_1635'
CmdError: If the version script cannot be found, or is found but cannot
be executed.
version_script = os.path.join(self.cros_overlay_path, 'chromeos', 'config',
if os.path.exists(version_script):
result = self.Run('sh', ['-c', '. %s >/dev/null; '
% version_script])
return result.strip()
raise CmdError("Cannot find version script ''")
def CheckTool(self, name, ebuild=None):
"""Check that the specified tool exists.
If it does not exist in the PATH, then generate a useful error message
indicating how to install the ebuild that contains the required tool.
name: filename of tool to look for on path.
ebuild: name of ebuild which should be emerged to install this tool,
or None if it is the same as the filename.
CmdError(msg) if the tool is not found.
filename = name
if filename in self._tools:
filename = self._tools[filename]
filename = self.Filename(filename)
self.Run('which', [filename])
except CmdError:
raise CmdError("The '%s' utility was not found in your path. "
"Run the following command in \nyour chroot to install "
"it: sudo -E emerge %s" % (filename, ebuild or name))
def OutputSize(self, label, filename, level=cros_output.NOTICE):
"""Display the filename and size of an object.
label: Label for this file.
filename: Filename to output.
level: Verbosity level to attach to this message
filename = self.Filename(filename)
size = os.stat(filename).st_size
self._out.DoOutput(level, '%s: %s; size: %d / %#x' %
(label, filename, size, size))
def PrepareOutputDir(self, outdir, preserve=False):
"""Select an output directory, ensuring it exists.
This either creates a temporary directory or checks that the one supplied
by the user is valid. For a temporary directory, it makes a note to
remove it later if required.
outdir: a string, name of the output directory to use to store
intermediate and output files. If is None - create a temporary
preserve: a Boolean. If outdir above is None and preserve is False, the
created temporary directory will be destroyed on exit.
OSError: If it cannot create the output directory.
self.preserve_outdir = outdir or preserve
if outdir:
self.outdir = outdir
if not os.path.isdir(self.outdir):
except OSError as err:
raise CmdError("Cannot make output directory '%s': '%s'" %
(self.outdir, err.strerror))
self.outdir = tempfile.mkdtemp(prefix='cros-dev.')
self._out.Debug("Using temporary directory '%s'" % self.outdir)
def FinalizeOutputDir(self):
"""Tidy up: delete output directory if temporary and not preserved."""
if self.outdir and not self.preserve_outdir:
self._out.Debug("Deleted temporary directory '%s'" %
self.outdir = None
def GetOutputFilename(self, fname):
"""Return a filename within the output directory.
fname: Filename to use for new file
The full path of the filename, within the output directory
return os.path.join(self.outdir, fname)
# pylint: disable=W0212
class ToolsTests(unittest.TestCase):
"""Unit tests for this module."""
def setUp(self):
self.out = cros_output.Output(False) = Tools(self.out)
def MakeOutsideChroot(self, base):
tools = Tools(self.out)
tools.in_chroot = False
return tools
def testPaths(self):
tools =
self.assertTrue(os.path.isdir(os.path.join(tools._root, '.repo')))
def _testToolsPaths(self, base, tools):
"""Common paths tests to run inside and outside chroot.
These tests are the same inside and outside the choot, so we put them in a
separate function.
base: Base directory to use for testing (contains the 'src' directory).
tools: Tools object to use.
self.assertEqual(tools._root, base[:-1])
self.assertEqual(tools.src_path, base + 'src')
self.assertEqual(tools.script_path, base + 'src/scripts')
self.assertEqual(tools.overlay_path, base + 'src/overlays')
self.assertEqual(tools.priv_overlay_path, base + 'src/private-overlays')
self.assertEqual(tools.third_party_path, base + 'src/third_party')
self.assertEqual(tools.cros_overlay_path, base +
def testSetRootInsideChroot(self):
"""Inside the chroot, paths are slightly different from outside."""
tools = Tools(self.out)
tools.in_chroot = True
# Force our own path.
base = '/air/bridge/'
# We should get a full path from that without the trailing '/'.
self.assertEqual(tools.chroot_path, '/')
self.assertEqual(tools.board_path, '/build')
self._testToolsPaths(base, tools)
def testSetRootOutsideChroot(self):
"""Pretend to be outside the chroot, and check that paths are correct."""
# Force our own path, outside the chroot.
base = '/spotty/light/'
tools = self.MakeOutsideChroot(base)
# We should get a full path from that without the trailing '/'.
self.assertEqual(tools.chroot_path, base + 'chroot')
self.assertEqual(tools.board_path, tools.chroot_path + '/build')
self._testToolsPaths(base, tools)
def _testToolsFilenames(self, tools):
"""Common filename tests to run inside and outside chroot.
These tests are the same inside and outside the choot, so we put them in a
separate function.
tools: Tools object to use.
# Try search paths in /bin and /ls.
tools.search_paths = ['/bin', '/lib']
file_in_bin = os.listdir('/bin')[0]
self.assertEqual(tools.Filename(file_in_bin), '/bin/%s' % file_in_bin)
file_in_lib = os.listdir('/lib')[0]
self.assertEqual(tools.Filename(file_in_lib), '/lib/%s' % file_in_lib)
self.assertEqual(tools.Filename('i-am-not-here'), 'i-am-not-here')
# Don't search for an empty file.
self.assertEqual(tools.Filename(''), '')
def testFilenameInsideChroot(self):
"""Test that we can specify search paths and they work correctly.
Test search patches inside the chroot.
tools = Tools(self.out)
tools.in_chroot = True
# Force our own path.
base = '/air/bridge/'
self.assertEqual(tools.Filename('##/fred'), '/fred')
def testFilenameOutsideChroot(self):
"""Test that we can specify search paths and they work correctly.
Test search patches outside the chroot.
base = '/home/'
tools = self.MakeOutsideChroot(base)
self.assertEqual(tools.Filename('##/fred'), base + 'chroot/fred')
base + 'chroot/just/a/short/dir/')
def testReadWriteFile(self):
"""Test our read/write utility functions."""
tools = Tools(self.out)
data = 'some context here' * 2
fname = tools.GetOutputFilename('bang')
tools.WriteFile(fname, data)
# Check that the file looks correct.
compare = tools.ReadFile(fname)
self.assertEqual(data, compare)
def testReadFileAndConcat(self):
"""Test 'cat' of several files."""
tools = Tools(self.out)
file_list = ['one', 'empty', 'two', 'three', 'four']
out_list = [tools.GetOutputFilename(fname) for fname in file_list]
file_list[1] = '' # Empty the 'empty' file.
for upto in range(len(file_list)):
tools.WriteFile(out_list[upto], file_list[upto])
data, offset, length = tools.ReadFileAndConcat(out_list)
self.assertEqual(len(data), 20)
self.assertEqual(offset, [0, 4, 4, 8, 16])
self.assertEqual(length, [3, 0, 3, 5, 4])
def testGetChromeosVersion(self):
"""Test for GetChromeosVersion() inside and outside chroot.
This function returns a string like '2893.0.2012_09_16_2219'.
tools =
re_version_pattern = '^\d{4}.\d+.(?:\d{4}_\d{2}_\d{2}_\d+|\d+)$'
re_version = re.compile(re_version_pattern)
reported_version = tools.GetChromeosVersion()
msg='%s !~= %s' % (reported_version, re_version_pattern))
tools = Tools(self.out)
# Force our own path, outside the chroot. This should fail.
base = 'invalid-dir'
tools = self.MakeOutsideChroot(base)
tools.in_chroot = False
self.assertRaises(CmdError, tools.GetChromeosVersion)
def testCheckTool(self):
"""Test for the CheckTool() method."""
tools =
self.assertRaises(CmdError, tools.CheckTool, 'non-existent-tool')
self.assertRaises(CmdError, tools.CheckTool, '/usr/bin/fdisk')
def testRun(self):
"""Test for the Run() method."""
tools =
# Ask fdisk for its version - this utility must be in the chroot.
re_fdisk = re.compile('fdisk \(util-linux .*\)')
self.assertTrue(re_fdisk.match(tools.Run('fdisk', ['-v'])))
# We need sudo for looking at disks.
self.assertEqual(tools.Run('fdisk', ['-l', '/dev/sda']),
'Cannot open /dev/sda\n')
out = tools.Run('fdisk', ['-l', '/dev/sda'], sudo=True)
# Don't look at the specific output, but it will have > 5 lines.
self.assertTrue(len(out.splitlines()) > 5)
self.assertEqual(tools.Run('pwd', [], cwd='/tmp'), '/tmp\n')
def testOutputDir(self):
"""Test output directory creation and deletion."""
tools =
# First check basic operation, creating and deleting a tmpdir.
fname = tools.GetOutputFilename('fred')
tools.WriteFile(fname, 'You are old, Father William, the young man said')
dirname = tools.outdir
# Try preserving it.
tools.PrepareOutputDir(None, True)
fname = tools.GetOutputFilename('fred')
tools.WriteFile(fname, 'and your hair has become very white')
dirname = tools.outdir
# Use our own directory, which is always preserved.
testdir = '/tmp/tools-test.test'
fname = tools.GetOutputFilename('fred')
tools.WriteFile(fname, 'and yet you incessantly stand on your head')
dirname = tools.outdir
# Try creating an invalid directory.
testdir = '/sys/cannot/do/this/here'
self.assertRaises(CmdError, tools.PrepareOutputDir, testdir)
fname = tools.GetOutputFilename('fred')
self.assertRaises(IOError, tools.WriteFile, fname,
'do you think at your age it is right?')
dirname = tools.outdir
def _OutputMock(self, level, msg, color=None):
self._level = level
self._msg = msg
self._color = color
def testOutputSize(self):
"""Test for OutputSize() function."""
tools =
# Rather than mocks, use a special Output object.
out = tools._out
out._Output = self._OutputMock
fname = tools.GetOutputFilename('fred')
text_string = 'test of output size'
tools.WriteFile(fname, text_string)
re_fname = re.compile('fred')
re_size = re.compile('.*size: (\d*)')
tools.OutputSize('first', fname, level=cros_output.ERROR)
self.assertEqual(self._level, cros_output.ERROR)
self.assertEqual(self._color, None)
# Check the default level, and that the filename length is given.
tools.OutputSize('second', fname)
self.assertEqual(self._level, cros_output.NOTICE)
self.assertEqual(self._color, None)
m = re_size.match(self._msg)
self.assertEqual(, str(len(text_string)))
def _Test(argv):
"""Run any built-in tests."""
assert doctest.testmod().failed == 0
def main():
"""Main function for tools.
We provide a way to call a few of our useful functions.
TODO(sjg) Move into the Chromite libraries when these are ready.
parser = optparse.OptionParser()
parser.add_option('-v', '--verbosity', dest='verbosity', default=1,
help='Control verbosity: 0=silent, 1=progress, 3=full, '
help_str = '%s [options] cmd [args]\n\nAvailable commands:\n' % sys.argv[0]
help_str += '\tchromeos-version\tDisplay Chrome OS version'
parser.usage = help_str
(options, args) = parser.parse_args(sys.argv)
args = args[1:]
out = cros_output.Output(options.verbosity)
tools = Tools(out)
if not args:
parser.error('No command provided')
elif args[0] == 'chromeos-version':
print tools.GetChromeosVersion()
parser.error("Unknown command '%s'" % args[0])
if __name__ == '__main__':
if sys.argv[1:2] == ['--test']:
_Test([sys.argv[0]] + sys.argv[2:])