Replace CompletedProcess with wrapper class CompletedExec

main
Felix Stupp 2 years ago
parent 90306b97b5
commit 0af7d128ce
Signed by: zocker
GPG Key ID: 93E1BD26F6B02FB7

@ -26,12 +26,14 @@ from __future__ import annotations
import argparse import argparse
from functools import cache, cached_property, wraps from functools import cache, cached_property, wraps
import json
import os import os
from pathlib import Path from pathlib import Path
import shlex import shlex
from subprocess import CompletedProcess, PIPE, run from subprocess import CompletedProcess, PIPE, run
import sys import sys
from typing import ( from typing import (
Any,
Callable, Callable,
Dict, Dict,
Iterable, Iterable,
@ -125,6 +127,21 @@ PODMAN_COMPOSE_EXEC = CommandArgs(
# === helpers # === helpers
@define()
class CompletedExec:
completed_process: CompletedProcess
@property
def returncode(self) -> int:
return self.completed_process.returncode
def check_returncode(self) -> None:
return self.completed_process.check_returncode()
def to_json(self) -> Mapping[str, Any]:
return json.loads(self.completed_process.stdout)
def filter_cmds(command: Iterable[Optional[str]]) -> CommandArgs: def filter_cmds(command: Iterable[Optional[str]]) -> CommandArgs:
return CommandArgs([arg for arg in command if arg is not None]) return CommandArgs([arg for arg in command if arg is not None])
@ -135,18 +152,19 @@ def combine_cmds(*commands: CommandArgs | List[Optional[str]]) -> CommandArgs:
def exec_cmd( def exec_cmd(
command: CommandArgs, check: bool = True, capture_stdout: bool = True command: CommandArgs, check: bool = True, capture_stdout: bool = True
) -> CompletedProcess: ) -> CompletedExec:
res = run( return CompletedExec(
args=command, run(
check=check, args=command,
shell=False, check=check,
stdout=PIPE if capture_stdout else None, shell=False,
stdout=PIPE if capture_stdout else None,
)
) )
return res
def process_tester( def process_tester(
exec: Callable[[CommandArgs], CompletedProcess] exec: Callable[[CommandArgs], CompletedExec]
) -> Callable[[CommandArgs], bool]: ) -> Callable[[CommandArgs], bool]:
return lambda command: exec(command).returncode == 0 return lambda command: exec(command).returncode == 0
@ -178,7 +196,7 @@ def search_shell() -> str:
def exec_shell( def exec_shell(
shell_cmd: ShellCommand, check: bool = True, capture_stdout: bool = True shell_cmd: ShellCommand, check: bool = True, capture_stdout: bool = True
) -> CompletedProcess: ) -> CompletedExec:
return exec_cmd( return exec_cmd(
CommandArgs([search_shell(), "-c", shell_cmd]), CommandArgs([search_shell(), "-c", shell_cmd]),
check=check, check=check,
@ -242,7 +260,7 @@ class PodmanClient:
self, self,
command: CommandArgs, command: CommandArgs,
check: bool = True, check: bool = True,
) -> CompletedProcess: ) -> CompletedExec:
return exec_cmd( return exec_cmd(
command=CommandArgs(self.podman_cmd + command), command=CommandArgs(self.podman_cmd + command),
check=check, check=check,
@ -253,7 +271,7 @@ class PodmanClient:
self, self,
command: CommandArgs, command: CommandArgs,
check: bool = True, check: bool = True,
) -> CompletedProcess: ) -> CompletedExec:
return exec_cmd( return exec_cmd(
command=CommandArgs(self.podman_compose_cmd + command), command=CommandArgs(self.podman_compose_cmd + command),
check=check, check=check,
@ -330,7 +348,7 @@ class ComposeFile:
self, self,
command: CommandArgs, command: CommandArgs,
check: bool = True, check: bool = True,
) -> CompletedProcess: ) -> CompletedExec:
return self.podman.exec_compose( return self.podman.exec_compose(
command=combine_cmds( command=combine_cmds(
[f"--project-name=self.project_name"], [f"--project-name=self.project_name"],
@ -364,7 +382,7 @@ class ComposeContainer:
command: CommandArgs, command: CommandArgs,
check: bool = True, check: bool = True,
workdir: Optional[str] = None, workdir: Optional[str] = None,
) -> CompletedProcess: ) -> CompletedExec:
return self.compose.podman.exec_podman( return self.compose.podman.exec_podman(
command=combine_cmds( command=combine_cmds(
[ [

Loading…
Cancel
Save