blob: d1082448867622ff9a99af5a2395adec3d856f84 [file] [log] [blame]
# Copyright (c) 2012 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.
"""Purpose of this module is to hold common script/commandline functionality.
This ranges from optparse, to a basic script wrapper setup (much like
what is used for chromite.bin.*).
from __future__ import print_function
import argparse
import collections
import datetime
import functools
import os
import optparse
import signal
import sys
import urlparse
# TODO(build): sort the cbuildbot.constants/lib.constants issue;
# lib shouldn't have to import from buildbot like this.
from chromite.cbuildbot import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import gs
from chromite.lib import osutils
from chromite.lib import path_util
from chromite.lib import terminal
class ChrootRequiredError(Exception):
"""Raised when a command must be run in the chroot
This exception is intended to be caught by code which will restart execution
in the chroot. Throwing this exception allows contexts to be exited and
general cleanup to happen before we exec an external binary.
The command to run inside the chroot, and (optionally) special cros_sdk
arguments are attached to the exception. Any adjustments to the arguments
should be done before raising the exception.
def __init__(self, cmd, chroot_args=None, extra_env=None):
"""Constructor for ChrootRequiredError.
cmd: Command line to run inside the chroot as a list of strings.
chroot_args: Arguments to pass directly to cros_sdk.
extra_env: Environmental variables to set in the chroot.
super(ChrootRequiredError, self).__init__()
self.cmd = cmd
self.chroot_args = chroot_args
self.extra_env = extra_env
class ExecRequiredError(Exception):
"""Raised when a command needs to exec, after cleanup.
This exception is intended to be caught by code which will exec another
command. Throwing this exception allows contexts to be exited and general
cleanup to happen before we exec an external binary.
The command to run is attached to the exception. Any adjustments to the
arguments should be done before raising the exception.
def __init__(self, cmd):
"""Constructor for ExecRequiredError.
cmd: Command line to run inside the chroot as a list of strings.
super(ExecRequiredError, self).__init__()
self.cmd = cmd
def AbsolutePath(_option, _opt, value):
"""Expand paths and make them absolute."""
return osutils.ExpandPath(value)
def NormalizeGSPath(value):
"""Normalize GS paths."""
url = gs.CanonicalizeURL(value, strict=True)
return '%s%s' % (gs.BASE_GS_URL, os.path.normpath(url[len(gs.BASE_GS_URL):]))
def NormalizeLocalOrGSPath(value):
"""Normalize a local or GS path."""
ptype = 'gs_path' if gs.PathIsGs(value) else 'path'
return VALID_TYPES[ptype](value)
def ParseBool(value):
"""Parse bool argument into a bool value.
For the existing type=bool functionality, the parser uses the built-in bool(x)
function to determine the value. This function will only return false if x
is False or omitted. Even with this type specified, however, arguments that
are generated from a command line initially get parsed as a string, and for
any string value passed in to bool(x), it will always return True.
value: String representing a boolean value.
True or False.
return cros_build_lib.BooleanShellValue(value, False)
def ParseDate(value):
"""Parse date argument into a object.
value: String representing a single date in "YYYY-MM-DD" format.
A object.
return datetime.datetime.strptime(value, '%Y-%m-%d').date()
except ValueError:
# Give a helpful error message about the format expected. Putting this
# message in the exception is useless because argparse ignores the
# exception message and just says the value is invalid.
logging.error('Date is expected to be in format YYYY-MM-DD.')
def NormalizeUri(value):
"""Normalize a local path or URI."""
o = urlparse.urlparse(value)
if o.scheme == 'file':
# Trim off the file:// prefix.
return VALID_TYPES['path'](value[7:])
elif o.scheme not in ('', 'gs'):
o = list(o)
o[2] = os.path.normpath(o[2])
return urlparse.urlunparse(o)
return NormalizeLocalOrGSPath(value)
# A Device object holds information parsed from the command line input:
# username: String SSH username or None.
# hostname: String SSH hostname or None.
# port: Int SSH port or None.
# path: String USB/file path or None.
# raw: String raw input from the command line.
# For now this is a superset of all information for USB, SSH, or file devices.
# If functionality diverges based on type, it may be useful to split this into
# separate device classes instead.
Device = cros_build_lib.Collection(
'Device', scheme=None, username=None, hostname=None, port=None, path=None,
class DeviceParser(object):
"""Parses devices as an argparse argument type.
In addition to parsing user input, this class will also ensure that only
supported device schemes are accepted by the parser. For example,
`cros deploy` only makes sense with an SSH device, but `cros flash` can use
SSH, USB, or file device schemes.
If the device input is malformed or the scheme is wrong, an error message will
be printed and the program will exit.
Valid device inputs are:
- [ssh://][username@]hostname[:port].
- usb://[path].
- file://path or /absolute_path.
- [ssh://]:vm:.
The last item above is an alias for ssh'ing into a virtual machine on a
localhost. It gets translated into 'localhost:9222'.
parser = argparse.ArgumentParser()
def __init__(self, schemes):
"""Initializes the parser.
See the class comments for usage examples.
schemes: A scheme or list of schemes to accept.
self.schemes = [schemes] if isinstance(schemes, basestring) else schemes
# Provide __name__ for argparse to print on failure, or else it will use
# repr() which creates a confusing error message.
self.__name__ = type(self).__name__
def __call__(self, value):
"""Parses a device input and enforces constraints.
DeviceParser is an object so that a set of valid schemes can be specified,
but argparse expects a parsing function, so we overload __call__() for
argparse to use.
value: String representing a device target. See class comments for
valid device input formats.
A Device object.
ValueError: |value| is not a valid device specifier or doesn't
match the supported list of schemes.
device = self._ParseDevice(value)
self._EnforceConstraints(device, value)
return device
except ValueError as e:
# argparse ignores exception messages, so print the message manually.
except Exception as e:
logging.error('Internal error while parsing device input: %s', e)
def _EnforceConstraints(self, device, value):
"""Verifies that user-specified constraints are upheld.
Checks that the parsed device has a scheme that matches what the user
expects. Additional constraints can be added if needed.
device: Device object.
value: String representing a device target.
ValueError: |device| has the wrong scheme.
if device.scheme not in self.schemes:
raise ValueError('Unsupported scheme "%s" for device "%s"' %
(device.scheme, value))
def _ParseDevice(self, value):
"""Parse a device argument.
value: String representing a device target.
A Device object.
ValueError: |value| is not a valid device specifier.
# ':vm:' is an alias for ssh'ing into a virtual machihne on localhost;
# translate it appropriately.
if value.strip().lower() == ':vm:':
value = 'localhost:9222'
elif value.strip().lower() == 'ssh://:vm:':
value = 'ssh://localhost:9222'
parsed = urlparse.urlparse(value)
if not parsed.scheme:
# Default to a file scheme for absolute paths, SSH scheme otherwise.
if value and value[0] == '/':
# urlparse won't provide hostname/username/port unless a scheme is
# specified so we need to re-parse.
parsed = urlparse.urlparse('%s://%s' % (DEVICE_SCHEME_SSH, value))
scheme = parsed.scheme.lower()
if scheme == DEVICE_SCHEME_SSH:
hostname = parsed.hostname
port = parsed.port
if hostname == 'localhost' and not port:
# Use of localhost as the actual machine is uncommon enough relative to
# the use of KVM that we require users to specify localhost:22 if they
# actually want to connect to the localhost. Otherwise the expectation
# is that they intend to access the VM but forget or didn't know to use
# port 9222.
raise ValueError('To connect to localhost, use ssh://localhost:22 '
'explicitly, or use ssh://localhost:9222 for the local'
' VM.')
if not hostname:
raise ValueError('Hostname is required for device "%s"' % value)
return Device(scheme=scheme, username=parsed.username, hostname=hostname,
port=port, raw=value)
elif scheme == DEVICE_SCHEME_USB:
path = parsed.netloc + parsed.path
# Change path '' to None for consistency.
return Device(scheme=scheme, path=path if path else None, raw=value)
elif scheme == DEVICE_SCHEME_FILE:
path = parsed.netloc + parsed.path
if not path:
raise ValueError('Path is required for "%s"' % value)
return Device(scheme=scheme, path=path, raw=value)
raise ValueError('Unknown device scheme "%s" in "%s"' % (scheme, value))
'bool': ParseBool,
'date': ParseDate,
'path': osutils.ExpandPath,
'gs_path': NormalizeGSPath,
'local_or_gs_path': NormalizeLocalOrGSPath,
'path_or_uri': NormalizeUri,
def OptparseWrapCheck(desc, check_f, _option, opt, value):
"""Optparse adapter for type checking functionality."""
return check_f(value)
except ValueError:
raise optparse.OptionValueError(
'Invalid %s given: --%s=%s' % (desc, opt, value))
class Option(optparse.Option):
"""Subclass to implement path evaluation & other useful types."""
_EXTRA_TYPES = ('path', 'gs_path')
TYPES = optparse.Option.TYPES + _EXTRA_TYPES
TYPE_CHECKER = optparse.Option.TYPE_CHECKER.copy()
for t in _EXTRA_TYPES:
TYPE_CHECKER[t] = functools.partial(OptparseWrapCheck, t, VALID_TYPES[t])
class FilteringOption(Option):
"""Subclass that supports Option filtering for FilteringOptionParser"""
def take_action(self, action, dest, opt, value, values, parser):
if action in FilteringOption.ACTIONS:
Option.take_action(self, action, dest, opt, value, values, parser)
if value is None:
value = []
elif not self.nargs or self.nargs <= 1:
value = [value]
parser.AddParsedArg(self, opt, [str(v) for v in value])
# TODO: logging.Formatter is not a subclass of object in python
# 2.6. Make ColoredFormatter explicitly inherit from object so that
# functions such as super() will not fail. This should be removed
# after python is upgraded to 2.7 on master2 (
class ColoredFormatter(logging.Formatter, object):
"""A logging formatter that can color the messages."""
'WARNING': terminal.Color.YELLOW,
'ERROR': terminal.Color.RED,
def __init__(self, *args, **kwargs):
"""Initializes the formatter.
args: See logging.Formatter for specifics.
kwargs: See logging.Formatter for specifics.
enable_color: Whether to enable colored logging. Defaults
to None, where terminal.Color will set to a sane default.
self.color = terminal.Color(enabled=kwargs.pop('enable_color', None))
super(ColoredFormatter, self).__init__(*args, **kwargs)
def format(self, record, **kwargs):
"""Formats |record| with color."""
msg = super(ColoredFormatter, self).format(record, **kwargs)
color = self._COLOR_MAPPING.get(record.levelname)
return msg if not color else self.color.Color(color, msg)
class ChromiteStreamHandler(logging.StreamHandler):
"""A stream handler for logging."""
class BaseParser(object):
"""Base parser class that includes the logic to add logging controls."""
DEFAULT_LOG_LEVELS = ('fatal', 'critical', 'error', 'warning', 'notice',
'info', 'debug')
def __init__(self, **kwargs):
"""Initialize this parser instance.
logging: Defaults to ALLOW_LOGGING from the class; if given,
add --log-level.
default_log_level: If logging is enabled, override the default logging
level. Defaults to the class's DEFAULT_LOG_LEVEL value.
log_levels: If logging is enabled, this overrides the enumeration of
allowed logging levels. If not given, defaults to the classes
manual_debug: If logging is enabled and this is True, suppress addition
of a --debug alias. This option defaults to True unless 'debug' has
been exempted from the allowed logging level targets.
caching: If given, must be either a callable that discerns the cache
location if it wasn't specified (the prototype must be akin to
lambda parser, values:calculated_cache_dir_path; it may return None to
indicate that it handles setting the value on its own later in the
parsing including setting the env), or True; if True, the
machinery defaults to invoking the class's FindCacheDir method
(which can be overridden). FindCacheDir $CROS_CACHEDIR, falling
back to $REPO/.cache, finally falling back to $TMP.
Note that the cache_dir is not created, just discerned where it
should live.
If False, or caching is not given, then no --cache-dir option will be
self.debug_enabled = False
self.caching_group = None
self.debug_group = None
self.default_log_level = None
self.log_levels = None
self.logging_enabled = kwargs.get('logging', self.ALLOW_LOGGING)
self.default_log_level = kwargs.get('default_log_level',
self.log_levels = tuple(x.lower() for x in
kwargs.get('log_levels', self.DEFAULT_LOG_LEVELS))
self.debug_enabled = (not kwargs.get('manual_debug', False)
and 'debug' in self.log_levels)
self.caching = kwargs.get('caching', False)
def PopUsedArgs(kwarg_dict):
"""Removes keys used by the base parser from the kwarg namespace."""
parser_keys = ['logging', 'default_log_level', 'log_levels', 'manual_debug',
for key in parser_keys:
kwarg_dict.pop(key, None)
def SetupOptions(self):
"""Sets up special chromite options for an OptionParser."""
if self.logging_enabled:
self.debug_group = self.add_option_group('Debug options')
self.debug_group, '--log-level', choices=self.log_levels,
help='Set logging level to report at.')
self.debug_group, '--log_format', action='store',
help='Set logging format to use.')
if self.debug_enabled:
self.debug_group, '--debug', action='store_const', const='debug',
dest='log_level', help='Alias for `--log-level=debug`. '
'Useful for debugging bugs/failures.')
self.debug_group, '--nocolor', action='store_false', dest='color',
help='Do not use colorized output (or `export NOCOLOR=true`)')
if self.caching:
self.caching_group = self.add_option_group('Caching Options')
self.caching_group, '--cache-dir', default=None, type='path',
help='Override the calculated chromeos cache directory; '
"typically defaults to '$REPO/.cache' .")
def SetupLogging(self, opts):
"""Sets up logging based on |opts|."""
value = opts.log_level.upper()
logger = logging.getLogger()
logger.setLevel(getattr(logging, value))
formatter = ColoredFormatter(fmt=opts.log_format,
# Only set colored formatter for ChromiteStreamHandler instances,
# which could have been added by ScriptWrapperMain() below.
chromite_handlers = [x for x in logger.handlers if
isinstance(x, ChromiteStreamHandler)]
for handler in chromite_handlers:
return value
def DoPostParseSetup(self, opts, args):
"""Method called to handle post opts/args setup.
This can be anything from logging setup to positional arg count validation.
opts: optparse.Values or argparse.Namespace instance
args: position arguments unconsumed from parsing.
(opts, args), w/ whatever modification done.
if self.logging_enabled:
value = self.SetupLogging(opts)
if self.debug_enabled:
opts.debug = (value == 'DEBUG')
if self.caching:
path = os.environ.get(constants.SHARED_CACHE_ENVVAR)
if path is not None and opts.cache_dir is None:
opts.cache_dir = os.path.abspath(path)
opts.cache_dir_specified = opts.cache_dir is not None
if not opts.cache_dir_specified:
func = self.FindCacheDir if not callable(self.caching) else self.caching
opts.cache_dir = func(self, opts)
if opts.cache_dir is not None:
return opts, args
def ConfigureCacheDir(cache_dir):
if cache_dir is None:
os.environ.pop(constants.SHARED_CACHE_ENVVAR, None)
logging.debug('Removed cache_dir setting')
os.environ[constants.SHARED_CACHE_ENVVAR] = cache_dir
logging.debug('Configured cache_dir to %r', cache_dir)
def FindCacheDir(cls, _parser, _opts):
logging.debug('Cache dir lookup.')
return path_util.FindCacheDir()
def add_option_group(self, *args, **kwargs):
"""Returns a new option group see optparse.OptionParser.add_option_group."""
raise NotImplementedError('Subclass must override this method')
def add_option_to_group(group, *args, **kwargs):
"""Adds the given option defined by args and kwargs to group."""
group.add_option(*args, **kwargs)
class ArgumentNamespace(argparse.Namespace):
"""Class to mimic argparse.Namespace with value freezing support."""
__metaclass__ = cros_build_lib.FrozenAttributesClass
_FROZEN_ERR_MSG = 'Option values are frozen, cannot alter %s.'
# Note that because optparse.Values is not a new-style class this class
# must use the mixin FrozenAttributesMixin rather than the metaclass
# FrozenAttributesClass.
class OptionValues(cros_build_lib.FrozenAttributesMixin, optparse.Values):
"""Class to mimic optparse.Values with value freezing support."""
_FROZEN_ERR_MSG = 'Option values are frozen, cannot alter %s.'
def __init__(self, defaults, *args, **kwargs):
optparse.Values.__init__(self, defaults, *args, **kwargs)
# Used by FilteringParser.
self.parsed_args = None
PassedOption = collections.namedtuple(
'PassedOption', ['opt_inst', 'opt_str', 'value_str'])
class FilteringParser(optparse.OptionParser, BaseParser):
"""Custom option parser for filtering options.
Aside from adding a couple of types (path for absolute paths,
gs_path for google storage urls, and log_level for logging level control),
this additionally exposes logging control by default; if undesired,
either derive from this class setting ALLOW_LOGGING to False, or
pass in logging=False to the constructor.
def __init__(self, usage=None, **kwargs):
BaseParser.__init__(self, **kwargs)
kwargs.setdefault('option_class', self.DEFAULT_OPTION_CLASS)
optparse.OptionParser.__init__(self, usage=usage, **kwargs)
def parse_args(self, args=None, values=None):
# If no Values object is specified then use our custom OptionValues.
if values is None:
values = OptionValues(defaults=self.defaults)
values.parsed_args = []
opts, remaining = optparse.OptionParser.parse_args(
self, args=args, values=values)
return self.DoPostParseSetup(opts, remaining)
def AddParsedArg(self, opt_inst, opt_str, value_str):
"""Add a parsed argument with attributes.
opt_inst: An instance of a raw optparse.Option object that represents the
opt_str: The option string.
value_str: A list of string-ified values dentified by OptParse.
self.values.parsed_args.append(PassedOption(opt_inst, opt_str, value_str))
def FilterArgs(parsed_args, filter_fn):
"""Filter the argument by passing it through a function.
parsed_args: The list of parsed argument namedtuples to filter. Tuples
are of the form (opt_inst, opt_str, value_str).
filter_fn: A function with signature f(PassedOption), and returns True if
the argument is to be passed through. False if not.
A tuple containing two lists - one of accepted arguments and one of
removed arguments.
removed = []
accepted = []
for arg in parsed_args:
target = accepted if filter_fn(arg) else removed
return accepted, removed
class SharedParser(argparse.ArgumentParser):
"""A type of parser that may be used as a shared parent for subparsers."""
def __init__(self, **kwargs):
kwargs.setdefault('add_help', False)
argparse.ArgumentParser.__init__(self, **kwargs)
class ArgumentParser(BaseParser, argparse.ArgumentParser):
"""Custom argument parser for use by chromite.
This class additionally exposes logging control by default; if undesired,
either derive from this class setting ALLOW_LOGGING to False, or
pass in logging=False to the constructor.
def __init__(self, usage=None, **kwargs):
kwargs.setdefault('formatter_class', argparse.RawDescriptionHelpFormatter)
BaseParser.__init__(self, **kwargs)
argparse.ArgumentParser.__init__(self, usage=usage, **kwargs)
def _SetupTypes(self):
"""Register types with ArgumentParser."""
for t, check_f in VALID_TYPES.iteritems():
self.register('type', t, check_f)
def add_option_group(self, *args, **kwargs):
"""Return an argument group rather than an option group."""
return self.add_argument_group(*args, **kwargs)
def add_option_to_group(group, *args, **kwargs):
"""Adds an argument rather than an option to the given group."""
return group.add_argument(*args, **kwargs)
def parse_args(self, args=None, namespace=None):
"""Translates OptionParser call to equivalent ArgumentParser call."""
# If no Namespace object is specified then use our custom ArgumentNamespace.
if namespace is None:
namespace = ArgumentNamespace()
# Unlike OptionParser, ArgParser works only with a single namespace and no
# args. Re-use BaseParser DoPostParseSetup but only take the namespace.
namespace = argparse.ArgumentParser.parse_args(
self, args=args, namespace=namespace)
return self.DoPostParseSetup(namespace, None)[0]
class _ShutDownException(SystemExit):
"""Exception raised when user hits CTRL+C."""
def __init__(self, sig_num, message):
self.signal = sig_num
# Setup a usage message primarily for any code that may intercept it
# while this exception is crashing back up the stack to us.
SystemExit.__init__(self, message)
self.args = (sig_num, message)
def _DefaultHandler(signum, _frame):
# Don't double process sigterms; just trigger shutdown from the first
# exception.
signal.signal(signum, signal.SIG_IGN)
raise _ShutDownException(
signum, 'Received signal %i; shutting down' % (signum,))
def _RestartInChroot(cmd, chroot_args, extra_env):
"""Rerun inside the chroot.
cmd: Command line to run inside the chroot as a list of strings.
chroot_args: Arguments to pass directly to cros_sdk (or None).
extra_env: Dictionary of environmental variables to set inside the
chroot (or None).
return cros_build_lib.RunCommand(cmd, error_code_ok=True,
enter_chroot=True, chroot_args=chroot_args,
def RunInsideChroot(command, chroot_args=None):
"""Restart the current command inside the chroot.
This method is only valid for any code that is run via ScriptWrapperMain.
It allows proper cleanup of the local context by raising an exception handled
in ScriptWrapperMain.
command: An instance of CliCommand to be restarted inside the chroot.
chroot_args: List of command-line arguments to pass to cros_sdk, if invoked.
if cros_build_lib.IsInsideChroot():
# Produce the command line to execute inside the chroot.
argv = sys.argv[:]
argv[0] = path_util.ToChrootPath(argv[0])
# Set log-level of cros_sdk to be same as log-level of command entering the
# chroot.
if chroot_args is None:
chroot_args = []
chroot_args += ['--log-level', command.options.log_level]
raise ChrootRequiredError(argv, chroot_args)
def ReExec():
"""Restart the current command.
This method is only valid for any code that is run via ScriptWrapperMain.
It allows proper cleanup of the local context by raising an exception handled
in ScriptWrapperMain.
# The command to exec.
raise ExecRequiredError(sys.argv[:])
def ScriptWrapperMain(find_target_func, argv=None,
"""Function usable for chromite.script.* style wrapping.
Note that this function invokes sys.exit on the way out by default.
find_target_func: a function, which, when given the absolute
pathway the script was invoked via (for example,
/home/ferringb/cros/trunk/chromite/bin/cros_sdk; note that any
trailing .py from the path name will be removed),
will return the main function to invoke (that functor will take
a single arg- a list of arguments, and shall return either None
or an integer, to indicate the exit code).
argv: sys.argv, or an equivalent tuple for testing. If nothing is
given, sys.argv is defaulted to.
log_level: Default logging level to start at.
log_format: Default logging format to use.
if argv is None:
argv = sys.argv[:]
target = os.path.abspath(argv[0])
name = os.path.basename(target)
if target.endswith('.py'):
target = os.path.splitext(target)[0]
target = find_target_func(target)
if target is None:
print('Internal error detected- no main functor found in module %r.' %
(name,), file=sys.stderr)
# Set up basic logging information for all modules that use logging.
# Note a script target may setup default logging in its module namespace
# which will take precedence over this.
logger = logging.getLogger()
logger_handler = ChromiteStreamHandler()
logging.Formatter(fmt=log_format, datefmt=constants.LOGGER_DATE_FMT))
signal.signal(signal.SIGTERM, _DefaultHandler)
ret = 1
ret = target(argv[1:])
except _ShutDownException as e:
print('%s: Signaled to shutdown: caught %i signal.' % (name, e.signal),
except SystemExit as e:
# Right now, let this crash through- longer term, we'll update the scripts
# in question to not use sys.exit, and make this into a flagged error.
except ChrootRequiredError as e:
ret = _RestartInChroot(e.cmd, e.chroot_args, e.extra_env)
except ExecRequiredError as e:
# This does not return.
os.execv(e.cmd[0], e.cmd)
except Exception as e:
print('%s: Unhandled exception:' % (name,), file=sys.stderr)
if ret is None:
ret = 0