diff --git a/.github/workflows/requirements/style/requirements.txt b/.github/workflows/requirements/style/requirements.txt index 25a9cdc05d5..c4a7f11aa1d 100644 --- a/.github/workflows/requirements/style/requirements.txt +++ b/.github/workflows/requirements/style/requirements.txt @@ -2,6 +2,6 @@ black==24.10.0 clingo==5.7.1 flake8==7.1.1 isort==5.13.2 -mypy==1.8.0 +mypy==1.11.2 types-six==1.17.0.20241205 vermin==1.6.0 diff --git a/.github/workflows/valid-style.yml b/.github/workflows/valid-style.yml index e1dc18d863d..6b85f5d0c67 100644 --- a/.github/workflows/valid-style.yml +++ b/.github/workflows/valid-style.yml @@ -20,7 +20,7 @@ jobs: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b with: - python-version: '3.11' + python-version: '3.13' cache: 'pip' - name: Install Python Packages run: | @@ -39,7 +39,7 @@ jobs: fetch-depth: 0 - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b with: - python-version: '3.11' + python-version: '3.13' cache: 'pip' - name: Install Python packages run: | @@ -58,7 +58,7 @@ jobs: secrets: inherit with: with_coverage: ${{ inputs.with_coverage }} - python_version: '3.11' + python_version: '3.13' # Check that spack can bootstrap the development environment on Python 3.6 - RHEL8 bootstrap-dev-rhel8: runs-on: ubuntu-latest diff --git a/lib/spack/spack/util/executable.py b/lib/spack/spack/util/executable.py index 72ce80edea8..529ed458622 100644 --- a/lib/spack/spack/util/executable.py +++ b/lib/spack/spack/util/executable.py @@ -7,11 +7,12 @@ import subprocess import sys from pathlib import Path, PurePath +from typing import Callable, Dict, Optional, Sequence, TextIO, Type, Union, overload import llnl.util.tty as tty import spack.error -import spack.util.environment +from spack.util.environment import EnvironmentModifications __all__ = ["Executable", "which", "which_string", "ProcessError"] @@ -19,33 +20,29 @@ class Executable: """Class representing a program that can be run on the command line.""" - def __init__(self, name): + def __init__(self, name: str) -> None: file_path = str(Path(name)) if sys.platform != "win32" and name.startswith("."): # pathlib strips the ./ from relative paths so it must be added back file_path = os.path.join(".", file_path) + self.exe = [file_path] - - self.default_env = {} - - self.default_envmod = spack.util.environment.EnvironmentModifications() - self.returncode = None + self.default_env: Dict[str, str] = {} + self.default_envmod = EnvironmentModifications() + self.returncode = 0 self.ignore_quotes = False - if not self.exe: - raise ProcessError("Cannot construct executable for '%s'" % name) - - def add_default_arg(self, *args): + def add_default_arg(self, *args: str) -> None: """Add default argument(s) to the command.""" self.exe.extend(args) - def with_default_args(self, *args): + def with_default_args(self, *args: str) -> "Executable": """Same as add_default_arg, but returns a copy of the executable.""" new = self.copy() new.add_default_arg(*args) return new - def copy(self): + def copy(self) -> "Executable": """Return a copy of this Executable.""" new = Executable(self.exe[0]) new.exe[:] = self.exe @@ -53,7 +50,7 @@ def copy(self): new.default_envmod.extend(self.default_envmod) return new - def add_default_env(self, key, value): + def add_default_env(self, key: str, value: str) -> None: """Set an environment variable when the command is run. Parameters: @@ -62,68 +59,109 @@ def add_default_env(self, key, value): """ self.default_env[key] = value - def add_default_envmod(self, envmod): + def add_default_envmod(self, envmod: EnvironmentModifications) -> None: """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 - """ + def command(self) -> str: + """Returns the entire command-line string""" return " ".join(self.exe) @property - def name(self): - """The executable name. - - Returns: - str: The basename of the executable - """ + def name(self) -> str: + """Returns the executable name""" return PurePath(self.path).name @property - def path(self): - """The path to the executable. - - Returns: - str: The path to the executable - """ + def path(self) -> str: + """Returns the executable path""" return str(PurePath(self.exe[0])) - def __call__(self, *args, **kwargs): - """Run this executable in a subprocess. + @overload + def __call__( + self, + *args: str, + fail_on_error: bool = True, + ignore_errors: Union[int, Sequence[int]] = (), + ignore_quotes: Optional[bool] = None, + timeout: Optional[int] = None, + env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + input: Optional[TextIO] = None, + output: Union[Optional[TextIO], str] = None, + error: Union[Optional[TextIO], str] = None, + _dump_env: Optional[Dict[str, str]] = None, + ) -> None: ... + + @overload + def __call__( + self, + *args: str, + fail_on_error: bool = True, + ignore_errors: Union[int, Sequence[int]] = (), + ignore_quotes: Optional[bool] = None, + timeout: Optional[int] = None, + env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + input: Optional[TextIO] = None, + output: Union[Type[str], Callable] = ..., + error: Union[Optional[TextIO], str, Type[str], Callable] = None, + _dump_env: Optional[Dict[str, str]] = None, + ) -> str: ... + + @overload + def __call__( + self, + *args: str, + fail_on_error: bool = True, + ignore_errors: Union[int, Sequence[int]] = (), + ignore_quotes: Optional[bool] = None, + timeout: Optional[int] = None, + env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + input: Optional[TextIO] = None, + output: Union[Optional[TextIO], str, Type[str], Callable] = None, + error: Union[Type[str], Callable] = ..., + _dump_env: Optional[Dict[str, str]] = None, + ) -> str: ... + + def __call__( + self, + *args: str, + fail_on_error: bool = True, + ignore_errors: Union[int, Sequence[int]] = (), + ignore_quotes: Optional[bool] = None, + timeout: Optional[int] = None, + env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None, + input: Optional[TextIO] = None, + output: Union[Optional[TextIO], str, Type[str], Callable] = None, + error: Union[Optional[TextIO], str, Type[str], Callable] = None, + _dump_env: Optional[Dict[str, str]] = None, + ) -> Optional[str]: + """Runs 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. - timeout (int or float): The number of seconds to wait before killing - the child process - input: Where to read stdin from - output: Where to send stdout - error: Where to send stderr + *args: command-line arguments to the executable to run + fail_on_error: if True, raises an exception if the subprocess returns an error + The return code is available as ``self.returncode`` + ignore_errors: a sequence 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: if False, warn users that quotes are not needed, as Spack does not + use a shell. If None, use ``self.ignore_quotes``. + timeout: the number of seconds to wait before killing the child process + env: the environment with which to run the executable + extra_env: extra items to add to the environment (neither requires nor precludes env) + input: where to read stdin from + output: where to send stdout + error: where to send stderr + _dump_env: dict to be set to the environment actually used (envisaged for + testing purposes only) 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 @@ -133,8 +171,11 @@ def __call__(self, *args, **kwargs): 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. + For output and error it's also accepted: + * filenames, which will be automatically opened for writing + + By default, the subprocess inherits the parent's file descriptors. """ def process_cmd_output(out, err): @@ -159,44 +200,34 @@ def process_cmd_output(out, err): sys.stderr.write(errstr) return result - # 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) + current_environment = os.environ.copy() if env is None else {} + self.default_envmod.apply_modifications(current_environment) + current_environment.update(self.default_env) # Apply env argument - if isinstance(env_arg, spack.util.environment.EnvironmentModifications): - env_arg.apply_modifications(env) - elif env_arg: - env.update(env_arg) + if isinstance(env, EnvironmentModifications): + env.apply_modifications(current_environment) + elif env: + current_environment.update(env) # Apply extra env - extra_env = kwargs.get("extra_env", {}) - if isinstance(extra_env, spack.util.environment.EnvironmentModifications): - extra_env.apply_modifications(env) - else: - env.update(extra_env) + if isinstance(extra_env, EnvironmentModifications): + extra_env.apply_modifications(current_environment) + elif extra_env is not None: + current_environment.update(extra_env) - if "_dump_env" in kwargs: - kwargs["_dump_env"].clear() - kwargs["_dump_env"].update(env) + if _dump_env is not None: + _dump_env.clear() + _dump_env.update(current_environment) - fail_on_error = kwargs.pop("fail_on_error", True) - ignore_errors = kwargs.pop("ignore_errors", ()) - ignore_quotes = kwargs.pop("ignore_quotes", self.ignore_quotes) - timeout = kwargs.pop("timeout", None) + if ignore_quotes is None: + ignore_quotes = self.ignore_quotes # 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.") @@ -230,9 +261,15 @@ def streamify(arg, mode): cmd_line_string = " ".join(escaped_cmd) tty.debug(cmd_line_string) + result = None try: proc = subprocess.Popen( - cmd, stdin=istream, stderr=estream, stdout=ostream, env=env, close_fds=False + cmd, + stdin=istream, + stderr=estream, + stdout=ostream, + env=current_environment, + close_fds=False, ) out, err = proc.communicate(timeout=timeout) @@ -248,9 +285,6 @@ def streamify(arg, mode): long_msg += "\n" + result raise ProcessError("Command exited with status %d:" % proc.returncode, long_msg) - - return result - except OSError as e: message = "Command: " + cmd_line_string if " " in self.exe[0]: @@ -286,6 +320,8 @@ def streamify(arg, mode): if close_istream: istream.close() + return result + def __eq__(self, other): return hasattr(other, "exe") and self.exe == other.exe