Refactor Command class

Command.run() currently has a bit of a confusing behavior: if raise_on_error is
False and the executable is not found, then a weird CommandT is returned (return
code is -1 and stdout+stderr is None). This makes it possible to hanlde command
not found errors separately, but it makes that needlessly verbose. So instead,
let's just return None in *this* special case.

That in turn uncovered, that in most cases when we set `raise_on_error=True`, we
actually want an error if the command is not present but no error if the command
fails to execute (e.g. because it returns -1 if you run `$cmd --version`). Hence we
introduce the flag `raise_on_command_not_found`, which causes an exception to
be raised if the command is not found. This makes it independent of the
`raise_on_error` flag.

Additionally, we add a small optimization: if command starts with /, then we
assume it's a full path and we omit the call to which (and just check whether it
exists).

Co-authored-by: Marcus Schäfer <[email protected]>

.github/workflows/ci-units-types.yml

@@ -12,7 +12,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"]
 
     steps:
       - uses: actions/checkout@v3

kiwi/builder/install.py

@@ -280,9 +280,9 @@ def create_install_pxe_archive(self) -> None:
             source_filename=self.diskname,
             keep_source_on_compress=True
         )
-        compress.xz(self.xz_options)
+        xz_archive = compress.xz(self.xz_options)
         Command.run(
-            ['mv', compress.compressed_filename, pxe_image_filename]
+            ['mv', xz_archive, pxe_image_filename]
         )
 
         # the system image transfer is checked against a checksum

kiwi/command.py

@@ -15,11 +15,17 @@
 # You should have received a copy of the GNU General Public License
 # along with kiwi.  If not, see <http://www.gnu.org/licenses/>
 #
-from typing import NamedTuple
-import select
-import os
+from typing import IO, Callable, List, MutableMapping, NamedTuple, Optional, overload
 import logging
+import os
+import select
 import subprocess
+import sys
+
+if sys.version_info >= (3, 8):
+    from typing import Literal  # pragma: no cover
+else:  # pragma: no cover
+    from typing_extensions import Literal  # pragma: no cover
 
 # project
 from kiwi.utils.codec import Codec
@@ -31,23 +37,19 @@
 
 log = logging.getLogger('kiwi')
 
-command_type = NamedTuple(
-    'command_type', [
-        ('output', str),
-        ('error', str),
-        ('returncode', int)
-    ]
-)
 
-command_call_type = NamedTuple(
-    'command_call_type', [
-        ('output', str),
-        ('output_available', bool),
-        ('error', str),
-        ('error_available', bool),
-        ('process', subprocess.Popen)
-    ]
-)
+class CommandT(NamedTuple):
+    output: str
+    error: str
+    returncode: int
+
+
+class CommandCallT(NamedTuple):
+    output: IO[bytes]
+    output_available: Callable[[], bool]
+    error: IO[bytes]
+    error_available: Callable[[], bool]
+    process: subprocess.Popen
 
 
 class Command:
@@ -58,16 +60,39 @@ class Command:
     commands in blocking and non blocking mode. Control of
     stdout and stderr is given to the caller
     """
+
+    @overload
+    @staticmethod
+    def run(
+        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
+        raise_on_error: bool = True, stderr_to_stdout: bool = False,
+        raise_on_command_not_found: Literal[False] = False
+    ) -> CommandT:
+        ...  # pragma: no cover
+
+    @overload
+    @staticmethod
+    def run(
+        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
+        raise_on_error: bool = True, stderr_to_stdout: bool = False,
+        raise_on_command_not_found: bool = True
+    ) -> Optional[CommandT]:
+        ...  # pragma: no cover
+
     @staticmethod
     def run(
-        command, custom_env=None, raise_on_error=True, stderr_to_stdout=False
-    ):
+        command: List[str], custom_env: Optional[MutableMapping[str, str]] = None,
+        raise_on_error: bool = True, stderr_to_stdout: bool = False,
+        raise_on_command_not_found: bool = True
+    ) -> Optional[CommandT]:
         """
         Execute a program and block the caller. The return value
-        is a hash containing the stdout, stderr and return code
-        information. Unless raise_on_error is set to false an
-        exception is thrown if the command exits with an error
-        code not equal to zero
+        is a CommandT namedtuple containing the stdout, stderr
+        and return code information. Unless raise_on_error is
+        set to `False` an exception is thrown if the command
+        exits with an error code not equal to zero. If
+        raise_on_command_not_found is `False` and the command is
+        not found, then `None` is returned.
 
         Example:
 
@@ -76,7 +101,7 @@ def run(
             result = Command.run(['ls', '-l'])
 
         :param list command: command and arguments
-        :param list custom_env: custom os.environ
+        :param dict custom_env: custom os.environ
         :param bool raise_on_error: control error behaviour
         :param bool stderr_to_stdout: redirects stderr to stdout
 
@@ -85,40 +110,42 @@ def run(
 
             .. code:: python
 
-                command(output='string', error='string', returncode=int)
+                CommandT(output='string', error='string', returncode=int)
 
-        :rtype: namedtuple
+        :rtype: CommandT
         """
         from .path import Path
         log.debug('EXEC: [%s]', ' '.join(command))
-        environment = os.environ
-        if custom_env:
-            environment = custom_env
-        if not Path.which(
-            command[0], custom_env=environment, access_mode=os.X_OK
-        ):
+        environment = custom_env or os.environ
+        cmd_abspath: Optional[str]
+        if command[0].startswith("/"):
+            cmd_abspath = command[0]
+            if not os.path.exists(cmd_abspath):
+                cmd_abspath = None
+        else:
+            cmd_abspath = Path.which(
+                command[0], custom_env=environment, access_mode=os.X_OK
+            )
+
+        if not cmd_abspath:
             message = 'Command "%s" not found in the environment' % command[0]
-            if not raise_on_error:
-                log.debug('EXEC: %s', message)
-                return command_type(
-                    output=None,
-                    error=None,
-                    returncode=-1
-                )
-            else:
+            if raise_on_command_not_found:
                 raise KiwiCommandNotFound(message)
+            log.debug('EXEC: %s', message)
+            return None
         stderr = subprocess.STDOUT if stderr_to_stdout else subprocess.PIPE
         try:
             process = subprocess.Popen(
-                command,
+                [cmd_abspath] + command[1:],
                 stdout=subprocess.PIPE,
                 stderr=stderr,
                 env=environment
             )
-        except Exception as e:
+        except (OSError, subprocess.SubprocessError) as e:
             raise KiwiCommandError(
                 '%s: %s: %s' % (command[0], type(e).__name__, format(e))
-            )
+            ) from e
+
         output, error = process.communicate()
         if process.returncode != 0 and raise_on_error:
             if not error:
@@ -135,14 +162,17 @@ def run(
                     command[0], Codec.decode(error), Codec.decode(output)
                 )
             )
-        return command_type(
+        return CommandT(
             output=Codec.decode(output),
             error=Codec.decode(error),
             returncode=process.returncode
         )
 
     @staticmethod
-    def call(command, custom_env=None):
+    def call(
+            command: List[str],
+            custom_env: Optional[MutableMapping[str, str]] = None
+    ) -> CommandCallT:
         """
         Execute a program and return an io file handle pair back.
         stdout and stderr are both on different channels. The caller
@@ -174,9 +204,7 @@ def call(command, custom_env=None):
         """
         from .path import Path
         log.debug('EXEC: [%s]', ' '.join(command))
-        environment = os.environ
-        if custom_env:
-            environment = custom_env
+        environment = custom_env or os.environ
         if not Path.which(
             command[0], custom_env=environment, access_mode=os.X_OK
         ):
@@ -193,31 +221,32 @@ def call(command, custom_env=None):
         except Exception as e:
             raise KiwiCommandError(
                 '%s: %s' % (type(e).__name__, format(e))
-            )
+            ) from e
+
+        # guaranteed to be true as stdout & stderr equal subprocess.PIPE
+        assert process.stdout and process.stderr
 
-        def output_available():
+        def output_available() -> Callable[[], bool]:
             def _select():
-                descriptor_lists = select.select(
+                readable, _, exceptional = select.select(
                     [process.stdout], [], [process.stdout], 1e-4
                 )
-                readable = descriptor_lists[0]
-                exceptional = descriptor_lists[2]
                 if readable and not exceptional:
                     return True
+                return False
             return _select
 
-        def error_available():
+        def error_available() -> Callable[[], bool]:
             def _select():
-                descriptor_lists = select.select(
+                readable, _, exceptional = select.select(
                     [process.stderr], [], [process.stderr], 1e-4
                 )
-                readable = descriptor_lists[0]
-                exceptional = descriptor_lists[2]
                 if readable and not exceptional:
                     return True
+                return False
             return _select
 
-        return command_call_type(
+        return CommandCallT(
             output=process.stdout,
             output_available=output_available(),
             error=process.stderr,

kiwi/command_process.py

@@ -17,6 +17,7 @@
 #
 import logging
 from collections import namedtuple
+from kiwi.command import CommandCallT
 
 # project
 from kiwi.utils.codec import Codec
@@ -37,7 +38,7 @@ class CommandProcess:
     :param subprocess command: instance of subprocess
     :param string log_topic: topic string for logging
     """
-    def __init__(self, command, log_topic='system'):
+    def __init__(self, command: CommandCallT, log_topic='system') -> None:
         self.command = CommandIterator(command)
         self.log_topic = log_topic
         self.items_processed = 0
@@ -154,7 +155,7 @@ class CommandIterator:
 
     :param subprocess command: instance of subprocess
     """
-    def __init__(self, command):
+    def __init__(self, command: CommandCallT) -> None:
         self.command = command
         self.command_error_output = bytes(b'')
         self.command_output_line = bytes(b'')
@@ -196,7 +197,7 @@ def get_error_output(self):
         """
         return Codec.decode(self.command_error_output)
 
-    def get_error_code(self):
+    def get_error_code(self) -> int:
         """
         Provide return value from processed command
 
@@ -206,7 +207,7 @@ def get_error_code(self):
         """
         return self.command.process.returncode
 
-    def get_pid(self):
+    def get_pid(self) -> int:
         """
         Provide process ID of command while running
 
@@ -216,7 +217,7 @@ def get_pid(self):
         """
         return self.command.process.pid
 
-    def kill(self):
+    def kill(self) -> None:
         """
         Send kill signal SIGTERM to command process
         """

kiwi/mount_manager.py

@@ -216,7 +216,4 @@ def is_mounted(self) -> bool:
             command=['mountpoint', '-q', self.mountpoint],
             raise_on_error=False
         )
-        if mountpoint_call.returncode == 0:
-            return True
-        else:
-            return False
+        return mountpoint_call.returncode == 0

kiwi/package_manager/apt.py

@@ -24,7 +24,7 @@
 )
 
 # project
-from kiwi.command import command_call_type
+from kiwi.command import CommandCallT
 from kiwi.command import Command
 from kiwi.path import Path
 from kiwi.package_manager.base import PackageManagerBase
@@ -128,7 +128,7 @@ def setup_repository_modules(
 
     def process_install_requests_bootstrap(
         self, root_bind: RootBind = None, bootstrap_package: str = None
-    ) -> command_call_type:
+    ) -> CommandCallT:
         """
         Process package install requests for bootstrap phase (no chroot)
         Either debootstrap or a prebuilt bootstrap package can be used
@@ -155,7 +155,7 @@ def process_install_requests_bootstrap(
 
     def _process_install_requests_bootstrap_prebuild_root(
         self, bootstrap_package: str
-    ) -> command_call_type:
+    ) -> CommandCallT:
         """
         Process bootstrap phase (no chroot) using a prebuilt bootstrap
         package. The package has to provide a tarball below the
@@ -207,7 +207,7 @@ def _process_install_requests_bootstrap_prebuild_root(
 
     def _process_install_requests_bootstrap_debootstrap(
         self, root_bind: RootBind = None
-    ) -> command_call_type:
+    ) -> CommandCallT:
         """
         Process package install requests for bootstrap phase (no chroot)
         The debootstrap program is used to bootstrap a new system
@@ -316,7 +316,7 @@ def post_process_install_requests_bootstrap(
         if root_bind:
             root_bind.mount_kernel_file_systems(delta_root)
 
-    def process_install_requests(self) -> command_call_type:
+    def process_install_requests(self) -> CommandCallT:
         """
         Process package install requests for image phase (chroot)
 
@@ -344,7 +344,7 @@ def process_install_requests(self) -> command_call_type:
             apt_get_command, self.command_env
         )
 
-    def process_delete_requests(self, force: bool = False) -> command_call_type:
+    def process_delete_requests(self, force: bool = False) -> CommandCallT:
         """
         Process package delete requests (chroot)
 
@@ -452,7 +452,7 @@ def post_process_delete_requests(
             ]
         )
 
-    def update(self) -> command_call_type:
+    def update(self) -> CommandCallT:
         """
         Process package update requests (chroot)