Add a new shell_out with support for hooks to create async functionality.

benchkit/communication/docker.py

@@ -105,6 +105,7 @@ def shell(
             timeout=timeout,
             output_is_log=output_is_log,
             ignore_ret_codes=ignore_ret_codes,
+            ignore_any_error_code=ignore_any_error_code,
         )
 
         return output

benchkit/shell/command_execution/command_process.py

@@ -0,0 +1,106 @@
+# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations  # Otherwise Queue comlains about typing
+
+import itertools
+from multiprocessing import Queue
+from subprocess import CalledProcessError, Popen, TimeoutExpired
+from threading import Thread
+from typing import Iterable, Optional, Tuple
+
+from benchkit.shell.command_execution.io.output import Output
+
+
+class CommandProcess:
+    """Encaptulation of the Popen process with functions to use it in an asyncronous way"""
+    def __init__(
+        self,
+        popen_object: Popen[bytes],
+        output: Output,
+        timeout: Optional[int],
+        success_code: int = 0,
+        ignore_exit_codes: Optional[Iterable[int]] = None,
+    ):
+
+        self.__popen_object: Popen[bytes] = popen_object
+        self.__output: Output = output
+        self.__timeout: Optional[int] = timeout
+        self.__retcode_queue: Queue[Tuple[int, Optional[Exception]]] = Queue()
+        self.success_code: int = success_code
+        self.ignore_exit_codes: Iterable[int] = (
+            (success_code,) if ignore_exit_codes is None else ignore_exit_codes
+        )
+
+        # add the success_code to the return codes to ignore
+        if self.success_code not in self.ignore_exit_codes:
+            self.ignore_exit_codes = itertools.chain([success_code], self.ignore_exit_codes)
+
+        self.retcode: Optional[int] = None
+        self.error: Optional[Exception] = None
+        self.process: Thread = self.__wait_async()
+
+    @staticmethod
+    def __wait_func(
+        subprocess: Popen[bytes],
+        queue: Queue[Tuple[int, Optional[Exception]]],
+        timeout: Optional[int],
+        ignore_exit_codes: Iterable[int],
+    ) -> None:
+        try:
+            retcode = subprocess.wait(timeout)
+            if retcode not in ignore_exit_codes:
+                queue.put(
+                    (
+                        retcode,
+                        CalledProcessError(
+                            retcode,
+                            subprocess.args,
+                        ),
+                    )
+                )
+            else:
+                queue.put((retcode, None))
+        except TimeoutExpired as exc:
+            # TODO: we can add some form of logging here to warn the user if something went wrong
+            subprocess.terminate()
+            subprocess.wait(1)
+            queue.put((-1, exc))
+
+    def __wait_async(self) -> Thread:
+        waiting_thread = Thread(
+            target=self.__wait_func,
+            args=(
+                self.__popen_object,
+                self.__retcode_queue,
+                self.__timeout,
+                self.ignore_exit_codes,
+            ),
+        )
+        waiting_thread.start()
+        return waiting_thread
+
+    def get_output(self) -> Output:
+        """get the Output object related to this process
+           can be used as input for other processes"""
+        return self.__output
+
+    def get_return_code(self) -> int:
+        """halt until the process has a return code
+           if the return code is not ignored
+           or the waittime was exceded throw an error instead"""
+        if self.error is not None:
+            raise self.error
+        if self.retcode:
+            return self.retcode
+        self.process.join()
+        self.retcode, self.error = self.__retcode_queue.get()
+        if self.error is not None:
+            raise self.error
+        return self.retcode
+
+    # TODO: check how this interacts with ssh
+    # THIS DOES NOT SEND IT TO THE RIGHT ONE -> move abstraction higher
+    def signal(self, signalcode: int) -> None:
+        self.__popen_object.send_signal(signalcode)
+        self.__popen_object.wait(1)

benchkit/shell/command_execution/execute.py

@@ -0,0 +1,104 @@
+# Copyright (C) 2024 Vrije Universiteit Brussel. All rights reserved.
+# SPDX-License-Identifier: MIT
+
+# Otherwise os.PathLike[Any] complains
+from __future__ import annotations
+
+import pathlib
+import subprocess
+from typing import Dict, Iterable, List, Optional
+
+from benchkit.shell.command_execution.command_process import CommandProcess
+from benchkit.shell.command_execution.io.hooks.hook import (
+    IOHook,
+    IOWriterHook,
+    OutputHook,
+)
+from benchkit.shell.command_execution.io.stream import (
+    EmptyIOStream,
+    ReadableIOStream,
+    WritableIOStream,
+)
+from benchkit.shell.command_execution.io.output import popen_get_output
+
+
+def execute_command(
+    # needed for starting the command
+    command: List[str],
+    # This dir can only be a path on the local machine
+    current_dir: Optional[pathlib.Path] = None,
+    # Do we want to add os.environ to this?
+    environment: Optional[Dict[str, str]] = None,
+    # needed for construction and evaluation of output
+    timeout: Optional[int] = None,
+    ignore_ret_codes: Optional[Iterable[int]] = None,
+    success_value: int = 0,
+    # working with the IOStreams of the command
+    std_input: Optional[ReadableIOStream] = None,
+    ordered_input_hooks: Optional[List[IOHook]] = None,
+    ordered_output_hooks: Optional[List[OutputHook]] = None,
+) -> CommandProcess:
+
+    if environment is None:
+        environment = {}
+
+    process = subprocess.Popen(
+        command,
+        cwd=current_dir,
+        env=environment,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        stdin=subprocess.PIPE,
+    )
+    try:
+        # 1) manipulate the input stream using the ordered input hooks
+        if ordered_input_hooks is not None:
+            if std_input is None:
+                std_input = EmptyIOStream()
+            for inhook in ordered_input_hooks:
+                inhook.start_hook_function(std_input)
+                std_input = inhook.get_outgoing_io_stream()
+
+        # 2) Write the input to the command
+        # hookfunction to write the ReadableIOStream given as input to stdin
+        def pasalong(input_stream: ReadableIOStream, _: WritableIOStream) -> None:
+            if process.stdin is not None:
+                outline = input_stream.read(1)
+                while outline:
+                    process.stdin.write(outline)
+                    process.stdin.flush()
+                    outline = input_stream.read(1)
+                process.stdin.close()
+
+        if std_input is not None:
+            hook = IOWriterHook(pasalong)
+            hook.start_hook_function(std_input)
+        if process.stdin is not None:
+            process.stdin.close()
+
+        # 3) manipulate teh output stream using the orderd output hooks
+        command_output = popen_get_output(process.stdout, process.stderr)
+
+        if ordered_output_hooks is not None:
+            for outhook in ordered_output_hooks:
+                command_output = outhook.attatch(command_output)
+
+        # close all the main thread file descriptors
+        if process.stdout is not None:
+            process.stdout.close()
+        if process.stderr is not None:
+            process.stderr.close()
+        if process.stdin is not None:
+            process.stdin.close()
+
+        # 4) construct the object we can use to monitor the process
+        return CommandProcess(
+            process, command_output, timeout, success_value, ignore_ret_codes
+        )
+
+    except Exception:
+        # make sure the process is terminated for cleanup
+        # TODO: this needs some test cases
+        process.terminate()
+        process.wait()
+        raise

benchkit/shell/command_execution/io/hooks/basic_hooks.py

@@ -0,0 +1,107 @@
+# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations  # Otherwise Queue comlains about typing
+
+from multiprocessing import Queue
+from typing import Any
+from pathlib import Path
+
+from benchkit.shell.command_execution.io.hooks.hook import (
+    IOReaderHook,
+    IOResultHook,
+    IOWriterHook,
+    OutputHook,
+)
+from benchkit.shell.command_execution.io.stream import (
+    ReadableIOStream,
+    StringIOStream,
+    WritableIOStream,
+    try_converting_bystring_to_readable_characters,
+)
+
+
+def create_voiding_result_hook() -> IOResultHook:
+    def hook_function(
+        input_object: ReadableIOStream, _: WritableIOStream, result_queue: Queue[Any]
+    ):
+        # we do not write to the out stream thus this is "voiding"
+        outlines: bytes = b""
+        outline = input_object.read(10)
+        while outline:
+            outlines += outline
+            outline = input_object.read(10)
+        result_queue.put(outlines)
+
+    return IOResultHook(hook_function)
+
+def stream_prepend_hook(stream:StringIOStream):
+    def hook_function(
+        input_object: ReadableIOStream, output_object: WritableIOStream,
+    ):
+        outline = stream.read(10)
+        while outline:
+            output_object.write(outline)
+            outline = input_object.read(10)
+        outline = input_object.read(10)
+        while outline:
+            output_object.write(outline)
+            outline = input_object.read(10)
+
+    return IOWriterHook(hook_function)
+
+def write_to_file_hook(path:Path,mode:str="a"):
+    def hook_function(
+        input_object: ReadableIOStream,
+    ):
+        with path.open(mode=f'{mode}b', buffering=0) as file:
+            outline = input_object.read(10)
+            while outline:
+                file.write(outline)
+                outline = input_object.read(10)
+            file.flush()
+
+    return IOReaderHook(hook_function)
+
+
+
+def create_stream_line_logger_hook(formating_string: str) -> IOReaderHook:
+    def hook_function_line(input_object: ReadableIOStream):
+        byt = input_object.read_line()
+        while byt:
+            print(
+                formating_string.format(f"{try_converting_bystring_to_readable_characters(byt)}"),
+                end="",
+            )
+            byt = input_object.read_line()
+
+    return IOReaderHook(hook_function_line)
+
+
+# TODO: Voiding can be done be done better but this will do for now
+# problem: if there are hooks on the output they will wait for input still
+# can be resolved by making use of EmptyIOStream
+# Needs to be done on a higher level than hooks
+def void_input(input_object: ReadableIOStream, _: WritableIOStream):
+    outline = input_object.read(10)
+    while outline:
+        outline = input_object.read(10)
+
+
+def logger_line_hook(outformat: str, errformat: str):
+    return OutputHook(
+        create_stream_line_logger_hook(outformat),
+        create_stream_line_logger_hook(errformat),
+    )
+
+
+def void_hook():
+    return OutputHook(IOWriterHook(void_input), IOWriterHook(void_input))
+
+
+def std_out_result_void_err():
+    output_hook_object = create_voiding_result_hook()
+
+    voiding_result_hook = OutputHook(output_hook_object, IOWriterHook(void_input))
+
+    return (output_hook_object, voiding_result_hook)