diff --git a/podman/__init__.py b/podman/__init__.py index 8e342bd9..b39cc881 100644 --- a/podman/__init__.py +++ b/podman/__init__.py @@ -1,7 +1,8 @@ """Podman client module.""" from podman.client import PodmanClient, from_env +from podman.command import PodmanCommand from podman.version import __version__ # isort: unique-list -__all__ = ['PodmanClient', '__version__', 'from_env'] +__all__ = ['PodmanClient', 'PodmanCommand', '__version__', 'from_env'] diff --git a/podman/command/__init__.py b/podman/command/__init__.py new file mode 100644 index 00000000..608424bd --- /dev/null +++ b/podman/command/__init__.py @@ -0,0 +1,4 @@ +from .command import PodmanCommand +from .cli_runner import GlobalOptions + +__all__ = ["PodmanCommand", "GlobalOptions"] diff --git a/podman/command/cli_runner.py b/podman/command/cli_runner.py new file mode 100644 index 00000000..15154cd6 --- /dev/null +++ b/podman/command/cli_runner.py @@ -0,0 +1,268 @@ +import dataclasses +import logging +import os +import platform +import shlex +import shutil +import subprocess +from pathlib import Path +from typing import Optional, Union + +from .. import errors + +logger = logging.getLogger("podman.command.cli_runner") + + +@dataclasses.dataclass +class GlobalOptions: + """Global options for Podman commands. + + Attributes: + cdi_spec_dir (Union[str, Path, list[str], list[Path], None]): The CDI spec directory path (can be a list of paths). + cgroup_manager: CGroup manager to use. + config: Location of config file, mainly for Docker compatibility. + conmon: Path to the conmon binary. + connection: Connection to use for remote Podman. + events_backend: Backend to use for storing events. + hooks_dir: Directory for hooks (can be a list of directories). + identity: Path to SSH identity file. + imagestore: Path to the image store. + log_level: Logging level. + module: Load a containers.conf module. + network_cmd_path: Path to slirp4netns command. + network_config_dir: Path to network config directory. + remote: When true, access to the Podman service is remote. + root: Storage root dir in which data, including images, is stored + runroot: Storage state directory where all state information is stored + runtime: Name or path of the OCI runtime. + runtime_flag: Global flags for the container runtime + ssh: Change SSH mode. + storage_driver: Storage driver to use. + storage_opt: Storage options. + syslog: Output logging information to syslog as well as the console. + tmpdir: Path to the tmp directory, for libpod runtime content. + transient_store: Whether to use a transient store. + url: URL for Podman service. + volumepath: Volume directory where builtin volume information is stored + """ + + cdi_spec_dir: Union[str, Path, list[str], list[Path], None] = None + cgroup_manager: Union[str, None] = None + config: Union[str, Path, None] = None + conmon: Union[str, Path, None] = None + connection: Union[str, None] = None + events_backend: Union[str, None] = None + hooks_dir: Union[str, Path, list[str], list[Path], None] = None + identity: Union[str, Path, None] = None + imagestore: Union[str, None] = None + log_level: Union[str, None] = None + module: Union[str, None] = None + network_cmd_path: Union[str, Path, None] = None + network_config_dir: Union[str, Path, None] = None + remote: Union[bool, None] = None + root: Union[str, Path, None] = None + runroot: Union[str, Path, None] = None + runtime: Union[str, Path, None] = None + runtime_flag: Union[str, list[str], None] = None + ssh: Union[str, None] = None + storage_driver: Union[str, None] = None + storage_opt: Union[str, list[str], None] = None + syslog: Union[bool, None] = None + tmpdir: Union[str, Path, None] = None + transient_store: Union[bool, None] = False + url: Union[str, None] = None + volumepath: Union[str, Path, None] = None + + +def get_subprocess_startupinfo(): + if platform.system() == "Windows": + startupinfo = subprocess.STARTUPINFO() + startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW + return startupinfo + else: + return None + + +class Runner: + """Runner class to execute Podman commands. + + Attributes: + podman_path (Path): Path to the Podman executable. + privileged (bool): Whether to run commands with elevated privileges. + options (GlobalOptions): Global options for Podman commands. + env (dict): Environment variables for the subprocess. + """ + + def __init__( + self, + path: Path = None, + privileged: bool = False, + options: GlobalOptions = None, + env: dict = None, + ): + """Initialize the Runner. + + Args: + path (Path, optional): Path to the Podman executable. Defaults to the system path. + privileged (bool, optional): Whether to run commands with elevated privileges. Defaults to False. + options (GlobalOptions, optional): Global options for Podman commands. Defaults to None. + env (dict, optional): Environment variables for the subprocess. Defaults to None. + + Raises: + errors.PodmanNotInstalled: If Podman is not installed. + """ + if path is None: + path = shutil.which("podman") + if path is None: + raise errors.PodmanNotInstalled() + Path(path) + + self.podman_path = path + if privileged and platform.system() == "Windows": + raise errors.PodmanError("Cannot run privileged Podman command on Windows") + self.privileged = privileged + self.options = options + self.env = env + + def display(self, cmd): + """Display a list of command-line options as a single command invocation.""" + parts = [str(part) for part in cmd] + return shlex.join(parts) + + def format_cli_opts(self, *args, **kwargs) -> list[str]: + """Format Pythonic arguments into command-line options for the Podman command. + + Args: + *args: Positional arguments to format. + **kwargs: Keyword arguments to format. + + Returns: + list[str]: A list of formatted command-line options. + """ + cmd = [] + # Positional arguments (*args) are added as is, provided that they are + # defined. + for arg in args: + if arg is not None: + cmd.append(arg) + + for arg, value in kwargs.items(): + option_name = "--" + arg.replace("_", "-") + if value is True: + # Options like cli_flag=True get converted to ["--cli-flag"]. + cmd.append(option_name) + elif isinstance(value, list): + # Options like cli_flag=["foo", "bar"] get converted to + # ["--cli-flag", "foo", "--cli-flag", "bar"]. + for v in value: + cmd += [option_name, str(v)] + elif value is not None and value is not False: + # Options like cli_flag="foo" get converted to + # ["--cli-flag", "foo"]. + cmd += [option_name, str(value)] + return cmd + + def construct(self, *args, **kwargs) -> list[str]: + """Construct the full command to run. + + Construct the base Podman command, along with the global CLI options. + Then, format the Pythonic arguments for the Podman command + (*args/**kwargs) and append them to the final command. + + Args: + *args: Positional arguments for the command. + **kwargs: Keyword arguments for the command. + + Returns: + list[str]: The constructed command as a list of strings. + """ + cmd = [] + if self.privileged: + cmd.append("sudo") + + cmd.append(str(self.podman_path)) + + if self.options: + cmd += self.format_cli_opts(**dataclasses.asdict(self.options)) + + cmd += self.format_cli_opts(*args, **kwargs) + return cmd + + def run( + self, + cmd: list[str], + *, + check: bool = True, + capture_output=True, + wait=True, + **skwargs, + ) -> Union[str, subprocess.Popen]: + """Run the specified Podman command. + + Args: + cmd (list[str]): The command to run, as a list of strings. + check (bool, optional): Whether to check for errors. Defaults to True. + capture_output (bool, optional): Whether to capture output. Defaults to True. + wait (bool, optional): Whether to wait for the command to complete. Defaults to True. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + Optional[str]: The output of the command if captured, otherwise the + subprocess.Popen instance. + + Raises: + errors.CommandError: If the command fails. + """ + cmd = self.construct() + cmd + return self.run_raw(cmd, check=check, capture_output=capture_output, wait=wait, **skwargs) + + def run_raw( + self, + cmd: list[str], + *, + check: bool = True, + capture_output=True, + stdin=subprocess.DEVNULL, + wait=True, + **skwargs, + ) -> Union[str, subprocess.Popen]: + """Run the command without additional construction. Mostly for internal use. + + Args: + cmd (list[str]): The full command to run. + check (bool, optional): Whether to check for errors. Defaults to True. + capture_output (bool, optional): Whether to capture output. Defaults to True. + stdin: Control the process' stdin. Disabled by default, to avoid hanging commands. + wait (bool, optional): Whether to wait for the command to complete. Defaults to True. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + Optional[str]: The output of the command if captured, otherwise the + subprocess.Popen instance. + + Raises: + errors.CommandError: If the command fails. + """ + logger.debug(f"Running: {self.display(cmd)}") + if not wait: + return subprocess.Popen( + cmd, + env=self.env, + startupinfo=get_subprocess_startupinfo(), + **skwargs, + ) + + try: + ret = subprocess.run( + cmd, + env=self.env, + check=check, + capture_output=capture_output, + stdin=stdin, + startupinfo=get_subprocess_startupinfo(), + **skwargs, + ) + except subprocess.CalledProcessError as e: + raise errors.CommandError(e) from e + if capture_output: + return ret.stdout.decode().rstrip() diff --git a/podman/command/command.py b/podman/command/command.py new file mode 100644 index 00000000..ebd04d29 --- /dev/null +++ b/podman/command/command.py @@ -0,0 +1,206 @@ +import contextlib +import platform +import subprocess +import time +from pathlib import Path +from typing import Optional, Union + +from . import cli_runner, machine_manager +from .. import client, errors + + +class PodmanCommand: + """Main class for executing Podman commands. + + Attributes: + runner (cli_runner.Runner): The runner instance to execute commands. + machine (machine_manager.MachineManager): Manager for machine operations. + """ + + GlobalOptions = cli_runner.GlobalOptions + + def __init__( + self, + path: Path = None, + privileged: bool = False, + options: cli_runner.GlobalOptions = None, + env: dict = None, + ): + """Initialize the PodmanCommand. + + Args: + path (Path, optional): Path to the Podman executable. Defaults to None. + privileged (bool, optional): Whether to run commands with elevated privileges. Defaults to False. + options (cli_runner.GlobalOptions, optional): Global options for Podman commands. Defaults to a new instance of GlobalOptions. + env (dict, optional): Environment variables for the subprocess. Defaults to None. + """ + if options is None: + options = cli_runner.GlobalOptions() + self.runner = cli_runner.Runner( + path=path, + privileged=privileged, + options=options, + env=env, + ) + self.machine = machine_manager.MachineManager(self.runner) + + def run( + self, + cmd: list[str], + *, + check: bool = True, + capture_output=True, + wait=True, + **skwargs, + ) -> Union[str, subprocess.Popen]: + """Run the specified Podman command. + + Args: + cmd (list[str]): The command to run, as a list of strings. + check (bool, optional): Whether to check for errors. Defaults to True. + capture_output (bool, optional): Whether to capture output. Defaults to True. + wait (bool, optional): Whether to wait for the command to complete. Defaults to True. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + Optional[str]: The output of the command if captured, otherwise the + subprocess.Popen instance. + + Raises: + errors.CommandError: If the command fails. + """ + return self.runner.run( + cmd=cmd, check=check, capture_output=capture_output, wait=wait, **skwargs + ) + + @property + def options(self) -> cli_runner.GlobalOptions: + """Returns the global options for this Podman command instance.""" + return self.runner.options + + def start_service( + self, + uri: Optional[str] = None, + time: Optional[int] = None, + cors: Optional[str] = None, + **skwargs, + ) -> subprocess.Popen: + """Start the Podman system service. + + This method starts a REST API using Podman's `system service` command. + This method is available only on Linux systems. + + Args: + uri (str, optional): The URI for the service. Uses the default URI if not specified. + time (str, optional): How long should the service be up. Default is 5 seconds. + cors (str, optional): CORS settings for the service. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + subprocess.Popen: The process handle of the `podman system service` + command. + """ + if platform.system() != "Linux": + raise errors.PodmanError( + "The `podman system service` command is available only on Linux systems" + ) + + cmd = self.runner.construct("system", "service", uri, time=time, cors=cors) + return self.runner.run_raw(cmd, wait=False, **skwargs) + + def stop_service( + self, + proc: subprocess.Popen, + timeout: Optional[int] = None, + ) -> int: + """Stop the Podman system service. + + This method stops the Podman REST API service. + + Args: + proc (subprocess.Popen): The process handle for Podman's system service. + timeout (int, optional): How long to wait until the service stops. + + Returns: + int: The exit code of the service process. + """ + proc.terminate() + try: + ret = proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + ret = proc.wait() + return ret + + def wait_for_service( + self, + uri: str, + proc: subprocess.Popen, + timeout: int = None, + check_interval: float = 0.1, + ): + """Wait for the Podman system service to be operational. + + This method checks two things; if the system service is still running, + and if we can ping it successfully. + + Args: + uri (str): The URI for the service. + proc (subprocess.Popen): The process handle for Podman's system service. + timeout (int, optional): How long to wait until the service is operational + check_interval (float): The interval between health checks + + Returns: + int: The exit code of the service process. + """ + start = time.monotonic() + with client.PodmanClient(base_url=uri) as c: + while True: + if timeout and time.monotonic() - start > timeout: + raise errors.ServiceTimeout(timeout) + + ret = proc.poll() + if ret is not None: + raise errors.ServiceTerminated(ret) + + try: + if c.ping(): + break + except errors.APIError: + pass + time.sleep(check_interval) + + @contextlib.contextmanager + def service( + self, + uri: str, + cors: str = None, + ping_timeout: int = None, + stop_timeout: int = None, + **skwargs, + ) -> subprocess.Popen: + """Manage the Podman system service. + + This method starts a REST API using Podman's `system service` command + and yields the process back to the user. Once the user does not need + the REST API any more, it stops the Podman service. + + Args: + uri (str): The URI for the service. + cors (str, optional): CORS settings for the service. + ping_timeout (int, optional): How long to wait until the service is up. + stop_timeout (int, optional): How long to wait until the service stops. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + subprocess.Popen: The process handle of the `podman system service` command. + """ + proc = self.start_service(uri=uri, time=0, cors=cors, **skwargs) + try: + self.wait_for_service(uri, proc, timeout=ping_timeout) + except (errors.ServiceTimeout, errors.ServiceTerminated): + self.stop_service(proc, timeout=stop_timeout) + raise + + yield proc + self.stop_service(proc, timeout=stop_timeout) diff --git a/podman/command/machine_manager.py b/podman/command/machine_manager.py new file mode 100644 index 00000000..bed1002f --- /dev/null +++ b/podman/command/machine_manager.py @@ -0,0 +1,150 @@ +import builtins +import json +import subprocess +from pathlib import Path +from typing import Union + +from . import cli_runner + + +class MachineManager: + """Manager for handling Podman machine operations. + + Attributes: + runner (cli_runner.Runner): The runner instance to execute commands. + """ + + def __init__(self, runner: cli_runner.Runner): + """Initialize the MachineManager. + + Args: + runner (cli_runner.Runner): The runner instance to execute commands. + """ + self.runner = runner + + def list(self, all_providers: bool = False, **skwargs) -> dict: + """List all machines. + + Args: + all_providers (bool, optional): Whether to include all providers. Defaults to False. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + dict: A dictionary containing the list of machines in JSON format. + """ + cmd = self.runner.construct( + "machine", "list", format="json", all_providers=all_providers, **skwargs + ) + return json.loads(self.runner.run_raw(cmd)) + + def init( + self, + name: str = None, + cpus: int = None, + disk_size: int = None, + ignition_path: Path = None, + image: Union[str, Path] = None, + memory: int = None, + now: bool = False, + playbook: str = None, + rootful: bool = False, + timezone: str = None, + usb: str = None, + user_mode_networking: bool = False, + username: str = None, + volume: Union[str, builtins.list[str]] = None, + **skwargs, + ) -> Union[str, subprocess.Popen]: + """Initialize a new machine. + + Args: + name (str, optional): Name of the machine. + cpus (int, optional): Number of CPUs to allocate. + disk_size (int, optional): Size of the disk in bytes. + ignition_path (Path, optional): Path to the ignition file. + image (Union[str, Path], optional): Image to use for the machine. + memory (int, optional): Amount of memory in bytes. + now (bool, optional): Whether to start the machine immediately. Defaults to False. + playbook (str, optional): Path to an Ansible playbook file. + rootful (bool, optional): Whether to create a rootful machine. Defaults to False. + timezone (str, optional): Timezone for the machine. + usb (str, optional): USB device to attach. + user_mode_networking (bool, optional): Whether to use user mode networking. Defaults to False. + username (str, optional): Username for the machine. + volume (str | list[str], optional): Volume to attach to the machine. + **skwargs: Additional keyword arguments for subprocess. + + Returns: + Optional[str]: The output of the command if captured, otherwise the + subprocess.Popen instance. + """ + cmd = self.runner.construct( + "machine", + "init", + name, + cpus=cpus, + disk_size=disk_size, + ignition_path=ignition_path, + image=image, + memory=memory, + now=now, + playbook=playbook, + rootful=rootful, + timezone=timezone, + usb=usb, + vendor=vendor, + user_mode_networking=user_mode_networking, + username=username, + volume=volume, + ) + return self.runner.run_raw(cmd, **skwargs) + + def start(self, name: str = None, **skwargs) -> None: + """Start a machine. + + Args: + name (str, optional): Name of the machine to start. + """ + cmd = self.runner.construct("machine", "start", name) + self.runner.run_raw(cmd, **skwargs) + + def stop(self, name: str = None, **skwargs) -> None: + """Stop a machine. + + Args: + name (str, optional): Name of the machine to stop. + """ + cmd = self.runner.construct("machine", "stop", name) + self.runner.run_raw(cmd, **skwargs) + + def remove( + self, + name: str = None, + force: bool = False, + save_image: bool = False, + save_ignition: bool = False, + **skwargs, + ) -> None: + """Remove a machine. + + Args: + name (str, optional): Name of the machine to remove. + force (bool, optional): Whether to stop the machine. Defaults to False. + save_image (bool, optional): Whether to save the machine's image. Defaults to False. + save_ignition (bool, optional): Whether to save the ignition file. Defaults to False. + """ + cmd = self.runner.construct( + "machine", + "rm", + name, + force=force, + save_image=save_image, + save_ignition=save_ignition, + **skwargs, + ) + self.runner.run_raw(cmd, **skwargs) + + def reset(self, **skwargs) -> None: + """Reset Podman machines and environment.""" + cmd = self.runner.construct("machine", "reset", force=True) + self.runner.run_raw(cmd, **skwargs) diff --git a/podman/errors/__init__.py b/podman/errors/__init__.py index 7fdeb471..42295c5d 100644 --- a/podman/errors/__init__.py +++ b/podman/errors/__init__.py @@ -13,6 +13,7 @@ # isort: unique-list __all__ = [ 'APIError', + 'CommandError', 'BuildError', 'ContainerError', 'DockerException', @@ -21,18 +22,24 @@ 'NotFound', 'NotFoundError', 'PodmanError', + 'PodmanNotInstalled', + 'ServiceTerminatedServiceTimeout', 'StreamParseError', ] try: from .exceptions import ( APIError, + CommandError, BuildError, ContainerError, DockerException, InvalidArgument, NotFound, PodmanError, + PodmanNotInstalled, + ServiceTerminated, + ServiceTimeout, StreamParseError, ) except ImportError: diff --git a/podman/errors/exceptions.py b/podman/errors/exceptions.py index f92d886c..2efe31c7 100644 --- a/podman/errors/exceptions.py +++ b/podman/errors/exceptions.py @@ -1,5 +1,7 @@ """Podman API Errors.""" +import subprocess + from typing import Optional, Union, TYPE_CHECKING from collections.abc import Iterable @@ -148,3 +150,32 @@ class InvalidArgument(PodmanError): class StreamParseError(RuntimeError): def __init__(self, reason): self.msg = reason + + +class PodmanNotInstalled(PodmanError): + def __init__(self) -> None: + msg = f"The Podman command is not installed in the system" + super().__init__(msg) + + +class CommandError(PodmanError): + def __init__(self, error: subprocess.CalledProcessError) -> None: + self.error = error + msg = f"The Podman process failed with the following error: {error}" + if self.error.stdout: + msg += f"\nStdout: {self.error.stdout}" + if self.error.stderr: + msg += f"\nStderr: {self.error.stderr}" + super().__init__(msg) + + +class ServiceTimeout(PodmanError): + def __init__(self, timeout: int) -> None: + msg = f"The Podman service failed to reply to a ping within {timeout} seconds" + super().__init__(msg) + + +class ServiceTerminated(PodmanError): + def __init__(self, code: int) -> None: + msg = f"The Podman service has terminated with error code {code}" + super().__init__(msg) diff --git a/podman/tests/errors.py b/podman/tests/errors.py deleted file mode 100644 index 9ca89999..00000000 --- a/podman/tests/errors.py +++ /dev/null @@ -1,19 +0,0 @@ -# Copyright 2020 Red Hat, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# -"""test error exceptions""" - - -class PodmanNotInstalled(Exception): - """Exception when podman is not available""" diff --git a/podman/tests/integration/utils.py b/podman/tests/integration/utils.py index 7151a7f9..12931172 100644 --- a/podman/tests/integration/utils.py +++ b/podman/tests/integration/utils.py @@ -24,7 +24,7 @@ import time -from podman.tests import errors +from podman import PodmanCommand logger = logging.getLogger("podman.service") @@ -41,54 +41,27 @@ def __init__( log_level: str = "WARNING", ) -> None: """create a launcher and build podman command""" - podman_exe: str = podman_path - if not podman_exe: - podman_exe = shutil.which('podman') - if podman_exe is None: - raise errors.PodmanNotInstalled() + self.podman = PodmanCommand(path=podman_path, privileged=privileged) + self.timeout = timeout + self.socket_uri: str = socket_uri self.socket_file: str = socket_uri.replace('unix://', '') self.log_level = log_level self.proc: Optional[subprocess.Popen[bytes]] = None self.reference_id = hash(time.monotonic()) - self.cmd: list[str] = [] - if privileged: - self.cmd.append('sudo') - - self.cmd.append(podman_exe) - logger.setLevel(logging.getLevelName(log_level)) # Map from python to go logging levels, FYI trace level breaks cirrus logging - self.cmd.append(f"--log-level={log_level.lower()}") - + self.podman.options.log_level = log_level.lower() if os.environ.get("container") == "oci": - self.cmd.append("--storage-driver=vfs") - - self.cmd.extend( - [ - "system", - "service", - f"--time={timeout}", - socket_uri, - ] - ) + self.podman.options.storage_driver = "vfs" - process = subprocess.run( - [podman_exe, "--version"], check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT - ) - self.version = str(process.stdout.decode("utf-8")).strip().split()[2] + self.version = self.podman.run(["--version"]).split()[2] def start(self, check_socket=True) -> None: """start podman service""" - logger.info( - "Launching(%s) %s refid=%s", - self.version, - ' '.join(self.cmd), - self.reference_id, - ) def consume_lines(pipe, consume_fn): with pipe: @@ -98,32 +71,34 @@ def consume_lines(pipe, consume_fn): def consume(line: str): logger.debug(line.strip("\n") + f" refid={self.reference_id}") - self.proc = subprocess.Popen(self.cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) # pylint: disable=consider-using-with + self.proc = self.podman.start_service( + self.socket_uri, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + logger.info( + "Launched(%s) %s pid=%s refid=%s", + self.version, + ' '.join(self.proc.args), + self.proc.pid, + self.reference_id, + ) + threading.Thread(target=consume_lines, args=[self.proc.stdout, consume]).start() if not check_socket: return # wait for socket to be created - timeout = time.monotonic() + 30 - while not os.path.exists(self.socket_file): - if time.monotonic() > timeout: - raise subprocess.TimeoutExpired("podman service ", timeout) - time.sleep(0.2) + self.podman.wait_for_service(self.socket_uri, self.proc, timeout=30) def stop(self) -> None: """stop podman service""" if not self.proc: return - self.proc.terminate() - try: - return_code = self.proc.wait(timeout=15) - except subprocess.TimeoutExpired: - self.proc.kill() - return_code = self.proc.wait() - self.proc = None - + return_code = self.podman.stop_service(self.proc, timeout=15) with suppress(FileNotFoundError): os.remove(self.socket_file)