kiwi.command

  1# Copyright (c) 2015 SUSE Linux GmbH.  All rights reserved.
  2#
  3# This file is part of kiwi.
  4#
  5# kiwi is free software: you can redistribute it and/or modify
  6# it under the terms of the GNU General Public License as published by
  7# the Free Software Foundation, either version 3 of the License, or
  8# (at your option) any later version.
  9#
 10# kiwi is distributed in the hope that it will be useful,
 11# but WITHOUT ANY WARRANTY; without even the implied warranty of
 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 13# GNU General Public License for more details.
 14#
 15# You should have received a copy of the GNU General Public License
 16# along with kiwi.  If not, see <http://www.gnu.org/licenses/>
 17#
 18from typing import IO, Callable, Literal, List, MutableMapping, NamedTuple, Optional, overload
 19import logging
 20import os
 21import select
 22import subprocess
 23
 24# project
 25from kiwi.utils.codec import Codec
 26
 27from kiwi.exceptions import (
 28    KiwiCommandError,
 29    KiwiCommandNotFound
 30)
 31
 32log = logging.getLogger('kiwi')
 33
 34
 35class CommandT(NamedTuple):
 36    output: str
 37    error: str
 38    returncode: int
 39
 40
 41class CommandCallT(NamedTuple):
 42    output: IO[bytes]
 43    output_available: Callable[[], bool]
 44    error: IO[bytes]
 45    error_available: Callable[[], bool]
 46    process: subprocess.Popen
 47
 48
 49class Command:
 50    """
 51    **Implements command invocation**
 52
 53    An instance of Command provides methods to invoke external
 54    commands in blocking and non blocking mode. Control of
 55    stdout and stderr is given to the caller
 56    """
 57
 58    @overload
 59    @staticmethod
 60    def run(
 61        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
 62        raise_on_error: bool = True, stderr_to_stdout: bool = False,
 63        raise_on_command_not_found: Literal[False] = False
 64    ) -> CommandT:
 65        ...  # pragma: no cover
 66
 67    @overload
 68    @staticmethod
 69    def run(
 70        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
 71        raise_on_error: bool = True, stderr_to_stdout: bool = False,
 72        raise_on_command_not_found: bool = True
 73    ) -> Optional[CommandT]:
 74        ...  # pragma: no cover
 75
 76    @staticmethod
 77    def run(
 78        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
 79        raise_on_error: bool = True, stderr_to_stdout: bool = False,
 80        raise_on_command_not_found: bool = True
 81    ) -> Optional[CommandT]:
 82        """
 83        Execute a program and block the caller. The return value
 84        is a CommandT namedtuple containing the stdout, stderr
 85        and return code information. Unless raise_on_error is
 86        set to `False` an exception is thrown if the command
 87        exits with an error code not equal to zero. If
 88        raise_on_command_not_found is `False` and the command is
 89        not found, then `None` is returned.
 90
 91        Example:
 92
 93        .. code:: python
 94
 95            result = Command.run(['ls', '-l'])
 96
 97        :param list command: command and arguments
 98        :param dict custom_env: custom os.environ
 99        :param bool raise_on_error: control error behaviour
100        :param bool stderr_to_stdout: redirects stderr to stdout
101
102        :return:
103            Contains call results in command type
104
105            .. code:: python
106
107                CommandT(output='string', error='string', returncode=int)
108
109        :rtype: CommandT
110        """
111        from .path import Path
112        environment = custom_env or os.environ
113        cmd_abspath: Optional[str]
114        if command[0].startswith("/"):
115            cmd_abspath = command[0]
116            if not os.path.exists(cmd_abspath):
117                cmd_abspath = None
118        else:
119            cmd_abspath = Path.which(
120                command[0], custom_env=environment, access_mode=os.X_OK
121            )
122
123        if not cmd_abspath:
124            message = f'Command "{command[0]}" not found in the environment'
125            if raise_on_command_not_found:
126                raise KiwiCommandNotFound(message)
127            log.debug('EXEC: %s', message)
128            return None
129        stderr = subprocess.STDOUT if stderr_to_stdout else subprocess.PIPE
130        log.debug('EXEC: [%s]', ' '.join(command))
131        try:
132            process = subprocess.Popen(
133                [cmd_abspath] + command[1:],
134                stdout=subprocess.PIPE,
135                stderr=stderr,
136                env=environment
137            )
138        except (OSError, subprocess.SubprocessError) as e:
139            raise KiwiCommandError(
140                f'{command[0]}: {type(e).__name__}: {format(e)}'
141            ) from e
142
143        output, error = process.communicate()
144        if process.returncode != 0 and raise_on_error:
145            if not error:
146                error = bytes(b'(no output on stderr)')
147            if not output:
148                output = bytes(b'(no output on stdout)')
149            log.debug(
150                'EXEC: Failed with stderr: {0}, stdout: {1}'.format(
151                    Codec.decode(error), Codec.decode(output)
152                )
153            )
154            raise KiwiCommandError(
155                '{0}: stderr: {1}, stdout: {2}'.format(
156                    command[0], Codec.decode(error), Codec.decode(output)
157                )
158            )
159        return CommandT(
160            output=Codec.decode(output),
161            error=Codec.decode(error),
162            returncode=process.returncode
163        )
164
165    @staticmethod
166    def call(
167            command: List[str],
168            custom_env: Optional[MutableMapping[str, str]] = None
169    ) -> CommandCallT:
170        """
171        Execute a program and return an io file handle pair back.
172        stdout and stderr are both on different channels. The caller
173        must read from the output file handles in order to actually
174        run the command. This can be done using the CommandIterator
175        from command_process
176
177        Example:
178
179        .. code:: python
180
181            process = Command.call(['ls', '-l'])
182
183        :param list command: command and arguments
184        :param list custom_env: custom os.environ
185
186        :return:
187            Contains process results in command type
188
189            .. code:: python
190
191                command(
192                    output='string', output_available=bool,
193                    error='string', error_available=bool,
194                    process=subprocess
195                )
196
197        :rtype: namedtuple
198        """
199        from .path import Path
200        log.debug('EXEC: [%s]', ' '.join(command))
201        environment = custom_env or os.environ
202        if not Path.which(
203            command[0], custom_env=environment, access_mode=os.X_OK
204        ):
205            raise KiwiCommandNotFound(
206                f'Command "{command[0]}" not found in the environment'
207            )
208        try:
209            process = subprocess.Popen(
210                command,
211                stdout=subprocess.PIPE,
212                stderr=subprocess.PIPE,
213                env=environment
214            )
215        except Exception as e:
216            raise KiwiCommandError(
217                f'{type(e).__name__}: {format(e)}'
218            ) from e
219
220        # guaranteed to be true as stdout & stderr equal subprocess.PIPE
221        assert process.stdout and process.stderr
222
223        def output_available() -> Callable[[], bool]:
224            def _select():
225                readable, _, exceptional = select.select(
226                    [process.stdout], [], [process.stdout], 1e-4
227                )
228                if readable and not exceptional:
229                    return True
230                return False
231            return _select
232
233        def error_available() -> Callable[[], bool]:
234            def _select():
235                readable, _, exceptional = select.select(
236                    [process.stderr], [], [process.stderr], 1e-4
237                )
238                if readable and not exceptional:
239                    return True
240                return False
241            return _select
242
243        return CommandCallT(
244            output=process.stdout,
245            output_available=output_available(),
246            error=process.stderr,
247            error_available=error_available(),
248            process=process
249        )