Source code for spack.util.executable

# Copyright 2013-2022 Lawrence Livermore National Security, LLC and other
# Spack Project Developers. See the top-level COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)

import os
import re
import shlex
import subprocess
import sys

from six import string_types, text_type

import llnl.util.tty as tty

import spack.error
from spack.util.path import Path, format_os_path, path_to_os_path, system_path_filter

__all__ = ['Executable', 'which', 'ProcessError']


[docs]class Executable(object): """Class representing a program that can be run on the command line.""" def __init__(self, name): # necesary here for the shlex call to succeed name = format_os_path(name, mode=Path.unix) self.exe = shlex.split(str(name)) # filter back to platform dependent path self.exe = path_to_os_path(*self.exe) self.default_env = {} from spack.util.environment import EnvironmentModifications # no cycle self.default_envmod = EnvironmentModifications() self.returncode = None if not self.exe: raise ProcessError("Cannot construct executable for '%s'" % name)
[docs] @system_path_filter def add_default_arg(self, arg): """Add a default argument to the command.""" self.exe.append(arg)
[docs] @system_path_filter def add_default_env(self, key, value): """Set an environment variable when the command is run. Parameters: key: The environment variable to set value: The value to set it to """ self.default_env[key] = value
[docs] def add_default_envmod(self, envmod): """Set an EnvironmentModifications to use when the command is run.""" self.default_envmod.extend(envmod)
@property def command(self): """The command-line string. Returns: str: The executable and default arguments """ return ' '.join(self.exe) @property def name(self): """The executable name. Returns: str: The basename of the executable """ return os.path.basename(self.path) @property def path(self): """The path to the executable. Returns: str: The path to the executable """ return self.exe[0] def __call__(self, *args, **kwargs): """Run this executable in a subprocess. Parameters: *args (str): Command-line arguments to the executable to run Keyword Arguments: _dump_env (dict): Dict to be set to the environment actually used (envisaged for testing purposes only) env (dict or EnvironmentModifications): The environment with which to run the executable extra_env (dict or EnvironmentModifications): Extra items to add to the environment (neither requires nor precludes env) fail_on_error (bool): Raise an exception if the subprocess returns an error. Default is True. The return code is available as ``exe.returncode`` ignore_errors (int or list): A list of error codes to ignore. If these codes are returned, this process will not raise an exception even if ``fail_on_error`` is set to ``True`` ignore_quotes (bool): If False, warn users that quotes are not needed as Spack does not use a shell. Defaults to False. input: Where to read stdin from output: Where to send stdout error: Where to send stderr Accepted values for input, output, and error: * python streams, e.g. open Python file objects, or ``os.devnull`` * filenames, which will be automatically opened for writing * ``str``, as in the Python string type. If you set these to ``str``, output and error will be written to pipes and returned as a string. If both ``output`` and ``error`` are set to ``str``, then one string is returned containing output concatenated with error. Not valid for ``input`` * ``str.split``, as in the ``split`` method of the Python string type. Behaves the same as ``str``, except that value is also written to ``stdout`` or ``stderr``. By default, the subprocess inherits the parent's file descriptors. """ # Environment env_arg = kwargs.get('env', None) # Setup default environment env = os.environ.copy() if env_arg is None else {} self.default_envmod.apply_modifications(env) env.update(self.default_env) from spack.util.environment import EnvironmentModifications # no cycle # Apply env argument if isinstance(env_arg, EnvironmentModifications): env_arg.apply_modifications(env) elif env_arg: env.update(env_arg) # Apply extra env extra_env = kwargs.get('extra_env', {}) if isinstance(extra_env, EnvironmentModifications): extra_env.apply_modifications(env) else: env.update(extra_env) if '_dump_env' in kwargs: kwargs['_dump_env'].clear() kwargs['_dump_env'].update(env) fail_on_error = kwargs.pop('fail_on_error', True) ignore_errors = kwargs.pop('ignore_errors', ()) ignore_quotes = kwargs.pop('ignore_quotes', False) # If they just want to ignore one error code, make it a tuple. if isinstance(ignore_errors, int): ignore_errors = (ignore_errors, ) input = kwargs.pop('input', None) output = kwargs.pop('output', None) error = kwargs.pop('error', None) if input is str: raise ValueError('Cannot use `str` as input stream.') def streamify(arg, mode): if isinstance(arg, string_types): return open(arg, mode), True elif arg in (str, str.split): return subprocess.PIPE, False else: return arg, False ostream, close_ostream = streamify(output, 'w') estream, close_estream = streamify(error, 'w') istream, close_istream = streamify(input, 'r') if not ignore_quotes: quoted_args = [arg for arg in args if re.search(r'^".*"$|^\'.*\'$', arg)] if quoted_args: tty.warn( "Quotes in command arguments can confuse scripts like" " configure.", "The following arguments may cause problems when executed:", str("\n".join([" " + arg for arg in quoted_args])), "Quotes aren't needed because spack doesn't use a shell. " "Consider removing them.", "If multiple levels of quotation are required, use " "`ignore_quotes=True`.") cmd = self.exe + list(args) escaped_cmd = ["'%s'" % arg.replace("'", "'\"'\"'") for arg in cmd] cmd_line_string = " ".join(escaped_cmd) tty.debug(cmd_line_string) try: proc = subprocess.Popen( cmd, stdin=istream, stderr=estream, stdout=ostream, env=env, close_fds=False,) out, err = proc.communicate() result = None if output in (str, str.split) or error in (str, str.split): result = '' if output in (str, str.split): if sys.platform == 'win32': outstr = text_type(out.decode('ISO-8859-1')) else: outstr = text_type(out.decode('utf-8')) result += outstr if output is str.split: sys.stdout.write(outstr) if error in (str, str.split): if sys.platform == 'win32': errstr = text_type(err.decode('ISO-8859-1')) else: errstr = text_type(err.decode('utf-8')) result += errstr if error is str.split: sys.stderr.write(errstr) rc = self.returncode = proc.returncode if fail_on_error and rc != 0 and (rc not in ignore_errors): long_msg = cmd_line_string if result: # If the output is not captured in the result, it will have # been stored either in the specified files (e.g. if # 'output' specifies a file) or written to the parent's # stdout/stderr (e.g. if 'output' is not specified) long_msg += '\n' + result raise ProcessError('Command exited with status %d:' % proc.returncode, long_msg) return result except OSError as e: raise ProcessError( '%s: %s' % (self.exe[0], e.strerror), 'Command: ' + cmd_line_string) except subprocess.CalledProcessError as e: if fail_on_error: raise ProcessError( str(e), '\nExit status %d when invoking command: %s' % (proc.returncode, cmd_line_string)) finally: if close_ostream: ostream.close() if close_estream: estream.close() if close_istream: istream.close() def __eq__(self, other): return hasattr(other, 'exe') and self.exe == other.exe def __neq__(self, other): return not (self == other) def __hash__(self): return hash((type(self), ) + tuple(self.exe)) def __repr__(self): return '<exe: %s>' % self.exe def __str__(self): return ' '.join(self.exe)
@system_path_filter def which_string(*args, **kwargs): """Like ``which()``, but return a string instead of an ``Executable``.""" path = kwargs.get('path', os.environ.get('PATH', '')) required = kwargs.get('required', False) if isinstance(path, string_types): path = path.split(os.pathsep) for name in args: win_candidates = [] if sys.platform == "win32" and (not name.endswith(".exe") and not name.endswith(".bat")): win_candidates = [name + ext for ext in ['.exe', '.bat']] candidate_names = [name] if not win_candidates else win_candidates for candidate_name in candidate_names: if os.path.sep in candidate_name: exe = os.path.abspath(candidate_name) if os.path.isfile(exe) and os.access(exe, os.X_OK): return exe else: for directory in path: directory = path_to_os_path(directory).pop() exe = os.path.join(directory, candidate_name) if os.path.isfile(exe) and os.access(exe, os.X_OK): return exe if required: raise CommandNotFoundError( "spack requires '%s'. Make sure it is in your path." % args[0]) return None
[docs]def which(*args, **kwargs): """Finds an executable in the path like command-line which. If given multiple executables, returns the first one that is found. If no executables are found, returns None. Parameters: *args (str): One or more executables to search for Keyword Arguments: path (list or str): The path to search. Defaults to ``PATH`` required (bool): If set to True, raise an error if executable not found Returns: Executable: The first executable that is found in the path """ exe = which_string(*args, **kwargs) return Executable(exe) if exe else None
[docs]class ProcessError(spack.error.SpackError): """ProcessErrors are raised when Executables exit with an error code."""
class CommandNotFoundError(spack.error.SpackError): """Raised when ``which()`` can't find a required executable."""