From 7d2f055bcb4067bfc621fef9083c4746f3ba5ef2 Mon Sep 17 00:00:00 2001 From: Newbyte Date: Mon, 7 Jul 2025 17:00:58 +0200 Subject: [PATCH] pmb: Make RunOutputTypeDefault and RunOutputTypePopen enums This allows us to get rid of some of the validation in sanity_checks() as mypy handles this validation at "build time", and any typos in the enum instantiation would be a runtime error rather than a silent failure. Additionally, it allows us to encode some of the behaviour of the different output types into the type definition itself by using methods. Part-of: https://gitlab.postmarketos.org/postmarketOS/pmbootstrap/-/merge_requests/2642 --- pmb/build/backend.py | 6 ++-- pmb/build/kconfig.py | 8 +++-- pmb/chroot/initfs.py | 4 +-- pmb/chroot/run.py | 8 ++--- pmb/ci/__init__.py | 10 ++++-- pmb/commands/log.py | 4 +-- pmb/config/pmaports.py | 5 ++- pmb/flasher/frontend.py | 5 ++- pmb/flasher/run.py | 3 +- pmb/helpers/apk.py | 8 +++-- pmb/helpers/frontend.py | 4 +-- pmb/helpers/git.py | 25 +++++++++----- pmb/helpers/lint.py | 3 +- pmb/helpers/run.py | 6 ++-- pmb/helpers/run_core.py | 71 +++++++++++++++++++++------------------- pmb/install/_install.py | 10 ++++-- pmb/install/format.py | 6 ++-- pmb/qemu/run.py | 4 +-- pmb/sideload/__init__.py | 10 +++--- pmb/types.py | 51 +++++++++++++++++++++++++++-- 20 files changed, 167 insertions(+), 84 deletions(-) diff --git a/pmb/build/backend.py b/pmb/build/backend.py index ad426039..52ebdbd5 100644 --- a/pmb/build/backend.py +++ b/pmb/build/backend.py @@ -11,7 +11,7 @@ from pmb.core import Context from pmb.core.arch import Arch from pmb.core.chroot import Chroot from pmb.helpers import logging -from pmb.types import Apkbuild, CrossCompile, Env +from pmb.types import Apkbuild, CrossCompile, Env, RunOutputTypeDefault class BootstrapStage(enum.IntEnum): @@ -316,4 +316,6 @@ def run_abuild( finally: handle_csum_failure(apkbuild, buildchroot) - pmb.helpers.run.root(["umount", buildchroot / "/mnt/sysroot"], output="null", check=False) + pmb.helpers.run.root( + ["umount", buildchroot / "/mnt/sysroot"], output=RunOutputTypeDefault.NULL, check=False + ) diff --git a/pmb/build/kconfig.py b/pmb/build/kconfig.py index d5da0bd5..3a5c0a6e 100644 --- a/pmb/build/kconfig.py +++ b/pmb/build/kconfig.py @@ -18,7 +18,7 @@ import pmb.helpers.pmaports import pmb.helpers.run import pmb.parse from pmb.core import Chroot -from pmb.types import Apkbuild, Env +from pmb.types import Apkbuild, Env, RunOutputTypeDefault class KConfigUI(enum.Enum): @@ -130,7 +130,7 @@ def extract_and_patch_sources(pkgname: str, arch: Arch) -> None: pmb.chroot.user( ["abuild", "prepare"], working_dir=Path("/home/pmos/build"), - output="interactive", + output=RunOutputTypeDefault.INTERACTIVE, env={"CARCH": str(arch)}, ) @@ -148,7 +148,9 @@ def _make( logging.info("(native) make " + make_command) - pmb.chroot.user(["make", str(make_command)], chroot, outputdir, output="tui", env=env) + pmb.chroot.user( + ["make", str(make_command)], chroot, outputdir, output=RunOutputTypeDefault.TUI, env=env + ) # Find the updated config source = Chroot.native() / outputdir / ".config" diff --git a/pmb/chroot/initfs.py b/pmb/chroot/initfs.py index 5899caa2..2ac726f3 100644 --- a/pmb/chroot/initfs.py +++ b/pmb/chroot/initfs.py @@ -6,7 +6,7 @@ import pmb.chroot.initfs_hooks import pmb.chroot.other import pmb.chroot.apk import pmb.config.pmaports -from pmb.types import PmbArgs +from pmb.types import PmbArgs, RunOutputTypeDefault import pmb.helpers.cli from pmb.core import Chroot from pmb.core.context import get_context @@ -82,7 +82,7 @@ def ls(flavor: str | None, suffix: Chroot, extra: bool = False) -> None: if extra: tmp = "/tmp/initfs-extra-extracted" extract(flavor, suffix, extra) - pmb.chroot.root(["ls", "-lahR", "."], suffix, Path(tmp), "stdout") + pmb.chroot.root(["ls", "-lahR", "."], suffix, Path(tmp), RunOutputTypeDefault.STDOUT) pmb.chroot.root(["rm", "-r", tmp], suffix) diff --git a/pmb/chroot/run.py b/pmb/chroot/run.py index 260a60e4..1e42d205 100644 --- a/pmb/chroot/run.py +++ b/pmb/chroot/run.py @@ -26,7 +26,7 @@ def rootm( cmds: Sequence[Sequence[PathString]], chroot: Chroot = Chroot.native(), working_dir: PurePath = PurePath("/"), - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, @@ -161,7 +161,7 @@ def root( cmds: Sequence[PathString], chroot: Chroot = Chroot.native(), working_dir: PurePath = PurePath("/"), - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, @@ -185,7 +185,7 @@ def userm( cmds: Sequence[Sequence[PathString]], chroot: Chroot = Chroot.native(), working_dir: Path = Path("/"), - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, @@ -254,7 +254,7 @@ def user( cmd: Sequence[PathString], chroot: Chroot = Chroot.native(), working_dir: Path = Path("/"), - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, diff --git a/pmb/ci/__init__.py b/pmb/ci/__init__.py index 0adc1a0b..aa4841e1 100644 --- a/pmb/ci/__init__.py +++ b/pmb/ci/__init__.py @@ -7,7 +7,7 @@ import os from pathlib import Path from typing import TypedDict import pmb.chroot -from pmb.types import Env +from pmb.types import Env, RunOutputTypeDefault import pmb.helpers.cli from pmb.core import Chroot @@ -181,7 +181,7 @@ def run_scripts(topdir: Path, scripts: dict[str, CiScriptDescriptor]) -> None: logging.info(f"*** ({step}/{steps}) RUNNING CI SCRIPT: {script_path} [{where}] ***") if "native" in script["options"]: - rc = pmb.helpers.run.user([script_path], topdir, output="tui") + rc = pmb.helpers.run.user([script_path], topdir, output=RunOutputTypeDefault.TUI) continue else: # Run inside pmbootstrap chroot @@ -191,7 +191,11 @@ def run_scripts(topdir: Path, scripts: dict[str, CiScriptDescriptor]) -> None: env: Env = {"TESTUSER": "pmos"} rc = pmb.chroot.root( - [script_path], check=False, env=env, working_dir=Path("/home/pmos/ci"), output="tui" + [script_path], + check=False, + env=env, + working_dir=Path("/home/pmos/ci"), + output=RunOutputTypeDefault.TUI, ) if rc: logging.error(f"ERROR: CI script failed: {script_name}") diff --git a/pmb/commands/log.py b/pmb/commands/log.py index 16ffe0b1..402fa68b 100644 --- a/pmb/commands/log.py +++ b/pmb/commands/log.py @@ -3,7 +3,7 @@ from __future__ import annotations from pmb import commands -from pmb.types import PathString +from pmb.types import PathString, RunOutputTypeDefault from pmb.helpers import run from pmb.core.context import get_context import pmb.config @@ -38,4 +38,4 @@ class Log(commands.Command): # looks for an error / what's currently going on). cmd += [context.log] - run.user(cmd, output="tui") + run.user(cmd, output=RunOutputTypeDefault.TUI) diff --git a/pmb/config/pmaports.py b/pmb/config/pmaports.py index b91c127e..57330e10 100644 --- a/pmb/config/pmaports.py +++ b/pmb/config/pmaports.py @@ -17,6 +17,7 @@ from pmb.meta import Cache import pmb.helpers.git import pmb.helpers.pmaports import pmb.parse.version +from pmb.types import RunOutputTypeDefault def clone() -> None: @@ -227,7 +228,9 @@ def switch_to_channel_branch(channel_new: str) -> bool: # Attempt to switch branch (git gives a nice error message, mentioning # which files need to be committed/stashed, so just pass it through) - if pmb.helpers.run.user(["git", "checkout", branch_new], aports, "interactive", check=False): + if pmb.helpers.run.user( + ["git", "checkout", branch_new], aports, RunOutputTypeDefault.INTERACTIVE, check=False + ): raise RuntimeError( "Failed to switch branch. Go to your pmaports and" " fix what git complained about, then try again: " diff --git a/pmb/flasher/frontend.py b/pmb/flasher/frontend.py index e1b5f1b9..89a45e36 100644 --- a/pmb/flasher/frontend.py +++ b/pmb/flasher/frontend.py @@ -17,6 +17,7 @@ import pmb.helpers.frontend import pmb.helpers.mount import pmb.parse.kconfig from pmb.core import Chroot, ChrootType +from pmb.types import RunOutputTypeDefault def kernel( @@ -192,7 +193,9 @@ def flash_lk2nd( pmb.flasher.init(deviceinfo.codename, method) logging.info("(native) checking current fastboot product") output = pmb.chroot.root( - ["fastboot", "getvar", "product"], output="interactive", output_return=True + ["fastboot", "getvar", "product"], + output=RunOutputTypeDefault.INTERACTIVE, + output_return=True, ) # Variable "product" is e.g. "LK2ND_MSM8974" or "lk2nd-msm8226" depending # on the lk2nd version. diff --git a/pmb/flasher/run.py b/pmb/flasher/run.py index 8e16fa6c..569b624f 100644 --- a/pmb/flasher/run.py +++ b/pmb/flasher/run.py @@ -4,6 +4,7 @@ from pmb.parse.deviceinfo import Deviceinfo import pmb.flasher import pmb.chroot.initfs import pmb.helpers.args +from pmb.types import RunOutputTypeDefault def check_partition_blacklist(deviceinfo: Deviceinfo, key: str, value: str) -> None: @@ -113,4 +114,4 @@ def run( # Remove empty strings command = [x for x in command if x != ""] # Run the action - pmb.chroot.root(command, output="interactive") + pmb.chroot.root(command, output=RunOutputTypeDefault.INTERACTIVE) diff --git a/pmb/helpers/apk.py b/pmb/helpers/apk.py index c932cffe..24f0376e 100644 --- a/pmb/helpers/apk.py +++ b/pmb/helpers/apk.py @@ -10,7 +10,7 @@ import pmb.chroot import pmb.config.pmaports from pmb.core.arch import Arch from pmb.core.chroot import Chroot -from pmb.types import PathString +from pmb.types import PathString, RunOutputTypePopen import pmb.helpers.cli import pmb.helpers.repo import pmb.helpers.run @@ -141,8 +141,10 @@ def _apk_with_progress(command: list[str]) -> None: fifo = _prepare_fifo() command_with_progress = _create_command_with_progress(command, fifo) log_msg = " ".join(command) - with pmb.helpers.run.root(["cat", fifo], output="pipe") as p_cat: - with pmb.helpers.run.root(command_with_progress, output="background") as p_apk: + with pmb.helpers.run.root(["cat", fifo], output=RunOutputTypePopen.PIPE) as p_cat: + with pmb.helpers.run.root( + command_with_progress, output=RunOutputTypePopen.BACKGROUND + ) as p_apk: while p_apk.poll() is None: p_cat_stdout = p_cat.stdout if p_cat_stdout is None: diff --git a/pmb/helpers/frontend.py b/pmb/helpers/frontend.py index d0e4eefb..5835db5d 100644 --- a/pmb/helpers/frontend.py +++ b/pmb/helpers/frontend.py @@ -19,7 +19,7 @@ import pmb.chroot.other import pmb.ci import pmb.config from pmb.core import Config -from pmb.types import Env, PmbArgs +from pmb.types import Env, PmbArgs, RunOutputTypeDefault import pmb.export import pmb.flasher import pmb.helpers.aportupgrade @@ -516,7 +516,7 @@ def stats(args: PmbArgs) -> None: # Install ccache and display stats pmb.chroot.apk.install(["ccache"], chroot) logging.info(f"({chroot}) % ccache -s") - pmb.chroot.user(["ccache", "-s"], chroot, output="stdout") + pmb.chroot.user(["ccache", "-s"], chroot, output=RunOutputTypeDefault.STDOUT) def work_migrate(args: PmbArgs) -> None: diff --git a/pmb/helpers/git.py b/pmb/helpers/git.py index 6c7d9bd2..4dd1fc8d 100644 --- a/pmb/helpers/git.py +++ b/pmb/helpers/git.py @@ -16,7 +16,7 @@ import pmb.config import pmb.helpers.pmaports import pmb.helpers.run from pmb.meta import Cache -from pmb.types import PathString +from pmb.types import PathString, RunOutputTypeDefault re_branch_aports = re.compile(r"^\d+\.\d\d+-stable$") re_branch_pmaports = re.compile(r"^v\d\d\.\d\d$") @@ -56,7 +56,7 @@ def clone(name_repo: str) -> None: # Create parent dir and clone logging.info(f"Clone git repository: {url}") (get_context().config.work / "cache_git").mkdir(exist_ok=True) - pmb.helpers.run.user(command, output="stdout") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.STDOUT) # FETCH_HEAD does not exist after initial clone. Create it, so # is_outdated() can use it. @@ -77,7 +77,9 @@ def rev_parse( or (with ``--abbrev-ref``): the branch name, e.g. "master" """ command = ["git", "rev-parse", *extra_args, revision] - rev = pmb.helpers.run.user_output(command, path, output="null" if silent else "log") + rev = pmb.helpers.run.user_output( + command, path, output=RunOutputTypeDefault.NULL if silent else RunOutputTypeDefault.LOG + ) return rev.rstrip() @@ -95,12 +97,17 @@ def can_fast_forward(path: Path, branch_upstream: str, branch: str = "HEAD") -> def clean_worktree(path: Path, silent: bool = False) -> bool: """Check if there are not any modified files in the git dir.""" command = ["git", "status", "--porcelain"] - return pmb.helpers.run.user_output(command, path, output="null" if silent else "log") == "" + return ( + pmb.helpers.run.user_output( + command, path, output=RunOutputTypeDefault.NULL if silent else RunOutputTypeDefault.LOG + ) + == "" + ) def list_remotes(aports: Path) -> list[str]: command = ["git", "remote", "-v"] - output = pmb.helpers.run.user_output(command, aports, output="null") + output = pmb.helpers.run.user_output(command, aports, output=RunOutputTypeDefault.NULL) return output.splitlines() @@ -159,7 +166,7 @@ def set_remote_url(repo: Path, remote_name: str, remote_url: str, remote_type: R "--push" if remote_type == RemoteType.PUSH else "--no-push", ] - pmb.helpers.run.user(command, output="stdout") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.STDOUT) # Intentionally lower case for case-insensitive comparison @@ -215,7 +222,9 @@ def parse_channels_cfg(aports: Path) -> dict: cfg = configparser.ConfigParser() remote = get_upstream_remote(aports) command = ["git", "show", f"{remote}/master:channels.cfg"] - stdout = pmb.helpers.run.user_output(command, aports, output="null", check=False) + stdout = pmb.helpers.run.user_output( + command, aports, output=RunOutputTypeDefault.NULL, check=False + ) try: cfg.read_string(stdout) except configparser.MissingSectionHeaderError: @@ -329,7 +338,7 @@ def pull(repo_name: str) -> int: # Fast-forward now (should not fail due to checks above, so it's fine to # throw an exception on error) command = ["git", "merge", "--ff-only", branch_upstream] - pmb.helpers.run.user(command, repo, "stdout") + pmb.helpers.run.user(command, repo, RunOutputTypeDefault.STDOUT) return 0 diff --git a/pmb/helpers/lint.py b/pmb/helpers/lint.py index cdbc4eb5..01b4466a 100644 --- a/pmb/helpers/lint.py +++ b/pmb/helpers/lint.py @@ -11,6 +11,7 @@ from pmb.core.pkgrepo import ( from pmb.helpers import logging from pmb.helpers.exceptions import NonBugError from pmb.helpers.toml import load_toml_file +from pmb.types import RunOutputTypeDefault import os import pmb.chroot @@ -104,7 +105,7 @@ def check(pkgnames: Sequence[str]) -> None: if pmb.chroot.user( ["apkbuild-lint", *apkbuild_paths], check=False, - output="stdout", + output=RunOutputTypeDefault.STDOUT, working_dir=dest_paths[pkgrepo_name(repo)], env={"CUSTOM_VALID_OPTIONS": " ".join(get_custom_valid_options())}, ): diff --git a/pmb/helpers/run.py b/pmb/helpers/run.py index 3775ce1e..27a30c95 100644 --- a/pmb/helpers/run.py +++ b/pmb/helpers/run.py @@ -19,7 +19,7 @@ from pmb.types import ( def user( cmd: Sequence[PathString], working_dir: Path | None = None, - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, @@ -58,7 +58,7 @@ def user( def user_output( cmd: Sequence[PathString], working_dir: Path | None = None, - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, check: bool | None = None, env: Env = {}, sudo: bool = False, @@ -106,7 +106,7 @@ def root( def root( cmd: Sequence[PathString], working_dir: Path | None = None, - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, env: Env = {}, diff --git a/pmb/helpers/run_core.py b/pmb/helpers/run_core.py index 2d5bafd7..99d1da51 100644 --- a/pmb/helpers/run_core.py +++ b/pmb/helpers/run_core.py @@ -2,7 +2,14 @@ # SPDX-License-Identifier: GPL-3.0-or-later import fcntl from pmb.core.context import get_context -from pmb.types import PathString, Env, RunOutputType, RunReturnType +from pmb.types import ( + PathString, + Env, + RunOutputType, + RunOutputTypeDefault, + RunOutputTypePopen, + RunReturnType, +) from pmb.helpers import logging import os from pathlib import Path @@ -51,25 +58,23 @@ def flat_cmd( def sanity_checks( - output: RunOutputType = "log", output_return: bool = False, check: bool | None = None + output: RunOutputType = RunOutputTypeDefault.LOG, + output_return: bool = False, + check: bool | None = None, ) -> None: """Raise an exception if the parameters passed to core() don't make sense. (all parameters are described in core() below). """ - vals = ["log", "stdout", "interactive", "tui", "background", "pipe", "null"] - if output not in vals: - raise RuntimeError("Invalid output value: " + str(output)) - # Prevent setting the check parameter with output="background". # The exit code won't be checked when running in background, so it would # always by check=False. But we prevent it from getting set to check=False # as well, so it does not look like you could change it to check=True. - if check is not None and output == "background": + if check is not None and output == RunOutputTypePopen.BACKGROUND: raise RuntimeError("Can't use check with output: background") - if output_return and output in ["tui", "background"]: - raise RuntimeError("Can't use output_return with output: " + output) + if output_return and output in [RunOutputTypeDefault.TUI, RunOutputTypePopen.BACKGROUND]: + raise RuntimeError(f"Can't use output_return with output: {output}") def background( @@ -361,7 +366,7 @@ def core( log_message: str, cmd: Sequence[PathString], working_dir: Path | None = None, - output: RunOutputType = "log", + output: RunOutputType = RunOutputTypeDefault.LOG, output_return: bool = False, check: bool | None = None, sudo: bool = False, @@ -380,18 +385,18 @@ def core( :param working_dir: path in host system where the command should run :param output: where to write the output (stdout and stderr) of the process. We almost always write to the log file, which can - be read with "pmbootstrap log" (output values: "log", - "stdout", "interactive", "background"), so it's easy to + be read with "pmbootstrap log" (output values: LOG, + STDOUT, INTERACTIVE, BACKGROUND), so it's easy to trace what pmbootstrap does. - The exceptions are "tui" (text-based user interface), where + The exceptions are TUI (text-based user interface), where it does not make sense to write to the log file (think of ncurses UIs, such as "menuconfig") and "pipe" where the output is written to a pipe for manual asynchronous consumption by the caller. - When the output is not set to "interactive", "tui", - "background" or "pipe", we kill the process if it does not + When the output is not set to INTERACTIVE, TUI, + BACKGROUND or PIPE, we kill the process if it does not output anything for 5 minutes (time can be set with "pmbootstrap --timeout"). @@ -399,17 +404,17 @@ def core( their properties. "wait" indicates that we wait for the process to complete. - ============= ======= ========== ============= ==== ========== - output value timeout out to log out to stdout wait pass stdin - ============= ======= ========== ============= ==== ========== - "log" x x x - "stdout" x x x x - "interactive" x x x x - "tui" x x x - "background" x - "pipe" - "null" - ============= ======= ========== ============= ==== ========== + ============ ======= ========== ============= ==== ========== + output value timeout out to log out to stdout wait pass stdin + ============ ======= ========== ============= ==== ========== + LOG x x x + STDOUT x x x x + INTERACTIVE x x x x + TUI x x x + BACKGROUND x + PIPE + NULL + ============ ======= ========== ============= ==== ========== :param output_return: in addition to writing the program's output to the destinations above in real time, write to a buffer and return it as string when the @@ -438,34 +443,34 @@ def core( # raise e # Background - if output == "background": + if output == RunOutputTypePopen.BACKGROUND: return background(cmd, working_dir) # Pipe - if output == "pipe": + if output == RunOutputTypePopen.PIPE: return pipe(cmd, working_dir) # Foreground output_after_run = "" - if output == "tui": + if output == RunOutputTypeDefault.TUI: # Foreground TUI code = foreground_tui(cmd, working_dir) else: # Foreground pipe (always redirects to the error log file) output_to_stdout = False - if not context.details_to_stdout and output in ["stdout", "interactive"]: + if not context.details_to_stdout and output.is_to_stdout(): output_to_stdout = True - output_timeout = output in ["log", "stdout"] and not disable_timeout + output_timeout = output.has_timeout() and not disable_timeout - stdin = subprocess.DEVNULL if output in ["log", "stdout"] else None + stdin = None if output.has_pass_stdin() else subprocess.DEVNULL (code, output_after_run) = foreground_pipe( cmd, working_dir, output_to_stdout, output_return, - output != "null", + output != RunOutputTypeDefault.NULL, output_timeout, sudo, stdin, diff --git a/pmb/install/_install.py b/pmb/install/_install.py index 1921c520..73fbe032 100644 --- a/pmb/install/_install.py +++ b/pmb/install/_install.py @@ -20,7 +20,7 @@ import pmb.config.pmaports from pmb.helpers.locale import get_xkb_layout from pmb.parse.deviceinfo import Deviceinfo from pmb.core import Config -from pmb.types import Env, PartitionLayout, PmbArgs +from pmb.types import Env, PartitionLayout, PmbArgs, RunOutputTypeDefault import pmb.helpers.devices from pmb.helpers.mount import mount_device_rootfs import pmb.helpers.run @@ -286,7 +286,9 @@ def setup_login(args: PmbArgs, config: Config, chroot: Chroot) -> None: else: while True: try: - pmb.chroot.root(["passwd", config.user], chroot, output="interactive") + pmb.chroot.root( + ["passwd", config.user], chroot, output=RunOutputTypeDefault.INTERACTIVE + ) break except RuntimeError: logging.info("WARNING: Failed to set the password. Try it one more time.") @@ -350,7 +352,9 @@ def setup_keymap(config: Config) -> None: options = deviceinfo.keymaps.split(" ") if config.keymap != "" and config.keymap is not None and config.keymap in options: layout, variant = config.keymap.split("/") - pmb.chroot.root(["setup-keymap", layout, variant], chroot, output="interactive") + pmb.chroot.root( + ["setup-keymap", layout, variant], chroot, output=RunOutputTypeDefault.INTERACTIVE + ) # Check xorg config xconfig = None diff --git a/pmb/install/format.py b/pmb/install/format.py index e598f7fc..776ed6f4 100644 --- a/pmb/install/format.py +++ b/pmb/install/format.py @@ -5,7 +5,7 @@ from pmb.helpers.devices import get_device_category_by_name import pmb.chroot from pmb.core import Chroot from pmb.core.context import get_context -from pmb.types import PartitionLayout, PmbArgs, PathString +from pmb.types import PartitionLayout, PmbArgs, PathString, RunOutputTypeDefault import os import tempfile @@ -81,8 +81,8 @@ def format_luks_root(args: PmbArgs, device: str) -> None: open_cmd += ["--key-file", str(path)] try: - pmb.chroot.root(format_cmd, output="interactive") - pmb.chroot.root([*open_cmd, device, "pm_crypt"], output="interactive") + pmb.chroot.root(format_cmd, output=RunOutputTypeDefault.INTERACTIVE) + pmb.chroot.root([*open_cmd, device, "pm_crypt"], output=RunOutputTypeDefault.INTERACTIVE) finally: if path_outside: os.unlink(path_outside) diff --git a/pmb/qemu/run.py b/pmb/qemu/run.py index f87ba866..9fb3f57a 100644 --- a/pmb/qemu/run.py +++ b/pmb/qemu/run.py @@ -24,7 +24,7 @@ import pmb.chroot.initfs import pmb.config import pmb.config.pmaports import pmb.install.losetup -from pmb.types import Env, PathString, PmbArgs +from pmb.types import Env, PathString, PmbArgs, RunOutputTypeDefault import pmb.helpers.run import pmb.parse.cpuinfo from pmb.core import Chroot, ChrootType @@ -470,7 +470,7 @@ def run(args: PmbArgs) -> None: process = None try: signal.signal(signal.SIGTERM, sigterm_handler) - process = pmb.helpers.run.user(qemu, output="tui", env=env) + process = pmb.helpers.run.user(qemu, output=RunOutputTypeDefault.TUI, env=env) except KeyboardInterrupt: # In addition to not showing a trace when pressing ^C, let user know # they can override this behavior: diff --git a/pmb/sideload/__init__.py b/pmb/sideload/__init__.py index b28c7ebc..88a44b82 100644 --- a/pmb/sideload/__init__.py +++ b/pmb/sideload/__init__.py @@ -6,7 +6,7 @@ from pmb.core.arch import Arch from pmb.helpers import logging import shlex -from pmb.types import PathString, PmbArgs +from pmb.types import PathString, PmbArgs, RunOutputTypeDefault import pmb.helpers.run import pmb.helpers.run_core import pmb.parse.apkindex @@ -30,7 +30,7 @@ def scp_abuild_key(args: PmbArgs, user: str, host: str, port: str) -> None: logging.info(f"Copying signing key ({key_name}) to {user}@{host}") command: list[PathString] = ["scp", "-P", port, key, f"{user}@{host}:/tmp"] - pmb.helpers.run.user(command, output="interactive") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.INTERACTIVE) logging.info(f"Installing signing key at {user}@{host}") keyname = os.path.join("/tmp", os.path.basename(key)) @@ -43,7 +43,7 @@ def scp_abuild_key(args: PmbArgs, user: str, host: str, port: str) -> None: remote_cmd = pmb.helpers.run_core.flat_cmd([remote_cmd_l]) full_cmd = shlex.quote(f"{su_cmd} {remote_cmd}") command = ["ssh", "-t", "-p", port, f"{user}@{host}", f"sh -c {full_cmd}"] - pmb.helpers.run.user(command, output="tui") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.TUI) def ssh_find_arch(args: PmbArgs, user: str, host: str, port: str) -> Arch: @@ -77,7 +77,7 @@ def ssh_install_apks(args: PmbArgs, user: str, host: str, port: str, paths: list logging.info(f"Copying packages to {user}@{host}") command: list[PathString] = ["scp", "-P", port, *paths, f"{user}@{host}:/tmp"] - pmb.helpers.run.user(command, output="interactive") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.INTERACTIVE) logging.info(f"Installing packages at {user}@{host}") add_cmd_list = ["apk", "--wait", "30", "add", *remote_paths] @@ -86,7 +86,7 @@ def ssh_install_apks(args: PmbArgs, user: str, host: str, port: str, paths: list add_cmd_complete = shlex.quote(f"{su_cmd} {add_cmd} rc=$?; {clean_cmd} exit $rc") # Run apk command in a subshell in case the foreign device has a non-POSIX shell. command = ["ssh", "-t", "-p", port, f"{user}@{host}", f"sh -c {add_cmd_complete}"] - pmb.helpers.run.user(command, output="tui") + pmb.helpers.run.user(command, output=RunOutputTypeDefault.TUI) def sideload( diff --git a/pmb/types.py b/pmb/types.py index 3bb533fe..db994d14 100644 --- a/pmb/types.py +++ b/pmb/types.py @@ -58,8 +58,55 @@ class CrossCompile(enum.Enum): return Chroot.native() -RunOutputTypeDefault = Literal["log", "stdout", "interactive", "tui", "null"] -RunOutputTypePopen = Literal["background", "pipe"] +class RunOutputTypeDefault(enum.Enum): + LOG = enum.auto() + STDOUT = enum.auto() + INTERACTIVE = enum.auto() + TUI = enum.auto() + NULL = enum.auto() + + def is_to_stdout(self) -> bool: + match self: + case self.STDOUT | self.INTERACTIVE: + return True + case self.LOG | self.TUI | self.NULL: + return False + case _: + raise AssertionError + + def has_timeout(self) -> bool: + match self: + case self.LOG | self.STDOUT: + return True + case self.INTERACTIVE | self.TUI | self.NULL: + return False + case _: + raise AssertionError + + def has_pass_stdin(self) -> bool: + match self: + case self.INTERACTIVE | self.TUI: + return True + case self.LOG | self.STDOUT | self.NULL: + return False + case _: + raise AssertionError + + +class RunOutputTypePopen(enum.Enum): + BACKGROUND = enum.auto() + PIPE = enum.auto() + + def is_to_stdout(self) -> bool: + return False + + def has_timeout(self) -> bool: + return False + + def has_pass_stdin(self) -> bool: + return False + + RunOutputType = RunOutputTypeDefault | RunOutputTypePopen RunReturnType = str | int | subprocess.Popen PathString = Path | str