Administration/pmbootstrap-meow
2
0
Fork 0
forked from Mirror/pmbootstrap
Code Pull requests Activity Actions

Revert "pmb: Make RunOutputTypeDefault and RunOutputTypePopen enums"

Browse source 
Revert the patch, as it breaks "pmbootstrap chroot".

This reverts commit 7d2f055bcb.
This commit is contained in:
Oliver Smith 2025-07-10 23:53:45 +02:00
parent ae0d162d31
commit 3ea5a3433b
No known key found for this signature in database
GPG key ID: 5AE7F5513E0885CB
20 changed files with 84 additions and 167 deletions
Download patch file Download diff file Expand all files Collapse all files

6
pmb/build/backend.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import Apkbuild, CrossCompile, Env
class BootstrapStage(enum.IntEnum):
@ -316,6 +316,4 @@ def run_abuild(
finally:
handle_csum_failure(apkbuild, buildchroot)
pmb.helpers.run.root(
["umount", buildchroot / "/mnt/sysroot"], output=RunOutputTypeDefault.NULL, check=False
)
pmb.helpers.run.root(["umount", buildchroot / "/mnt/sysroot"], output="null", check=False)

8
pmb/build/kconfig.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import Apkbuild, Env
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=RunOutputTypeDefault.INTERACTIVE,
output="interactive",
env={"CARCH": str(arch)},
)
@ -148,9 +148,7 @@ def _make(
logging.info("(native) make " + make_command)
pmb.chroot.user(
["make", str(make_command)], chroot, outputdir, output=RunOutputTypeDefault.TUI, env=env
)
pmb.chroot.user(["make", str(make_command)], chroot, outputdir, output="tui", env=env)
# Find the updated config
source = Chroot.native() / outputdir / ".config"

4
pmb/chroot/initfs.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import PmbArgs
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), RunOutputTypeDefault.STDOUT)
pmb.chroot.root(["ls", "-lahR", "."], suffix, Path(tmp), "stdout")
pmb.chroot.root(["rm", "-r", tmp], suffix)

8
pmb/chroot/run.py
View file

@ -26,7 +26,7 @@ def rootm(
cmds: Sequence[Sequence[PathString]],
chroot: Chroot = Chroot.native(),
working_dir: PurePath = PurePath("/"),
output: RunOutputType = RunOutputTypeDefault.LOG,
output: RunOutputType = "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 = RunOutputTypeDefault.LOG,
output: RunOutputType = "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 = RunOutputTypeDefault.LOG,
output: RunOutputType = "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 = RunOutputTypeDefault.LOG,
output: RunOutputType = "log",
output_return: bool = False,
check: bool | None = None,
env: Env = {},

10
pmb/ci/__init__.py
View file

@ -7,7 +7,7 @@ import os
from pathlib import Path
from typing import TypedDict
import pmb.chroot
from pmb.types import Env, RunOutputTypeDefault
from pmb.types import Env
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=RunOutputTypeDefault.TUI)
rc = pmb.helpers.run.user([script_path], topdir, output="tui")
continue
else:
# Run inside pmbootstrap chroot
@ -191,11 +191,7 @@ 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=RunOutputTypeDefault.TUI,
[script_path], check=False, env=env, working_dir=Path("/home/pmos/ci"), output="tui"
)
if rc:
logging.error(f"ERROR: CI script failed: {script_name}")

4
pmb/commands/log.py
View file

@ -3,7 +3,7 @@
from __future__ import annotations
from pmb import commands
from pmb.types import PathString, RunOutputTypeDefault
from pmb.types import PathString
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=RunOutputTypeDefault.TUI)
run.user(cmd, output="tui")

5
pmb/config/pmaports.py
View file

@ -17,7 +17,6 @@ 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:
@ -228,9 +227,7 @@ 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, RunOutputTypeDefault.INTERACTIVE, check=False
):
if pmb.helpers.run.user(["git", "checkout", branch_new], aports, "interactive", check=False):
raise RuntimeError(
"Failed to switch branch. Go to your pmaports and"
" fix what git complained about, then try again: "

5
pmb/flasher/frontend.py
View file

@ -17,7 +17,6 @@ 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(
@ -193,9 +192,7 @@ def flash_lk2nd(
pmb.flasher.init(deviceinfo.codename, method)
logging.info("(native) checking current fastboot product")
output = pmb.chroot.root(
["fastboot", "getvar", "product"],
output=RunOutputTypeDefault.INTERACTIVE,
output_return=True,
["fastboot", "getvar", "product"], output="interactive", output_return=True
)
# Variable "product" is e.g. "LK2ND_MSM8974" or "lk2nd-msm8226" depending
# on the lk2nd version.

3
pmb/flasher/run.py
View file

@ -4,7 +4,6 @@ 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:
@ -114,4 +113,4 @@ def run(
# Remove empty strings
command = [x for x in command if x != ""]
# Run the action
pmb.chroot.root(command, output=RunOutputTypeDefault.INTERACTIVE)
pmb.chroot.root(command, output="interactive")

8
pmb/helpers/apk.py
View file

@ -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, RunOutputTypePopen
from pmb.types import PathString
import pmb.helpers.cli
import pmb.helpers.repo
import pmb.helpers.run
@ -141,10 +141,8 @@ 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=RunOutputTypePopen.PIPE) as p_cat:
with pmb.helpers.run.root(
command_with_progress, output=RunOutputTypePopen.BACKGROUND
) as p_apk:
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:
while p_apk.poll() is None:
p_cat_stdout = p_cat.stdout
if p_cat_stdout is None:

4
pmb/helpers/frontend.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import Env, PmbArgs
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=RunOutputTypeDefault.STDOUT)
pmb.chroot.user(["ccache", "-s"], chroot, output="stdout")
def work_migrate(args: PmbArgs) -> None:

25
pmb/helpers/git.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import PathString
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=RunOutputTypeDefault.STDOUT)
pmb.helpers.run.user(command, output="stdout")
# FETCH_HEAD does not exist after initial clone. Create it, so
# is_outdated() can use it.
@ -77,9 +77,7 @@ 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=RunOutputTypeDefault.NULL if silent else RunOutputTypeDefault.LOG
)
rev = pmb.helpers.run.user_output(command, path, output="null" if silent else "log")
return rev.rstrip()
@ -97,17 +95,12 @@ 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=RunOutputTypeDefault.NULL if silent else RunOutputTypeDefault.LOG
)
== ""
)
return pmb.helpers.run.user_output(command, path, output="null" if silent else "log") == ""
def list_remotes(aports: Path) -> list[str]:
command = ["git", "remote", "-v"]
output = pmb.helpers.run.user_output(command, aports, output=RunOutputTypeDefault.NULL)
output = pmb.helpers.run.user_output(command, aports, output="null")
return output.splitlines()
@ -166,7 +159,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=RunOutputTypeDefault.STDOUT)
pmb.helpers.run.user(command, output="stdout")
# Intentionally lower case for case-insensitive comparison
@ -222,9 +215,7 @@ 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=RunOutputTypeDefault.NULL, check=False
)
stdout = pmb.helpers.run.user_output(command, aports, output="null", check=False)
try:
cfg.read_string(stdout)
except configparser.MissingSectionHeaderError:
@ -338,7 +329,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, RunOutputTypeDefault.STDOUT)
pmb.helpers.run.user(command, repo, "stdout")
return 0

3
pmb/helpers/lint.py
View file

@ -11,7 +11,6 @@ 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
@ -105,7 +104,7 @@ def check(pkgnames: Sequence[str]) -> None:
if pmb.chroot.user(
["apkbuild-lint", *apkbuild_paths],
check=False,
output=RunOutputTypeDefault.STDOUT,
output="stdout",
working_dir=dest_paths[pkgrepo_name(repo)],
env={"CUSTOM_VALID_OPTIONS": " ".join(get_custom_valid_options())},
):

6
pmb/helpers/run.py
View file

@ -19,7 +19,7 @@ from pmb.types import (
def user(
cmd: Sequence[PathString],
working_dir: Path | None = None,
output: RunOutputType = RunOutputTypeDefault.LOG,
output: RunOutputType = "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 = RunOutputTypeDefault.LOG,
output: RunOutputType = "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 = RunOutputTypeDefault.LOG,
output: RunOutputType = "log",
output_return: bool = False,
check: bool | None = None,
env: Env = {},

71
pmb/helpers/run_core.py
View file

@ -2,14 +2,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
import fcntl
from pmb.core.context import get_context
from pmb.types import (
PathString,
Env,
RunOutputType,
RunOutputTypeDefault,
RunOutputTypePopen,
RunReturnType,
)
from pmb.types import PathString, Env, RunOutputType, RunReturnType
from pmb.helpers import logging
import os
from pathlib import Path
@ -58,23 +51,25 @@ def flat_cmd(
def sanity_checks(
output: RunOutputType = RunOutputTypeDefault.LOG,
output_return: bool = False,
check: bool | None = None,
output: RunOutputType = "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 == RunOutputTypePopen.BACKGROUND:
if check is not None and output == "background":
raise RuntimeError("Can't use check with output: background")
if output_return and output in [RunOutputTypeDefault.TUI, RunOutputTypePopen.BACKGROUND]:
raise RuntimeError(f"Can't use output_return with output: {output}")
if output_return and output in ["tui", "background"]:
raise RuntimeError("Can't use output_return with output: " + output)
def background(
@ -366,7 +361,7 @@ def core(
log_message: str,
cmd: Sequence[PathString],
working_dir: Path | None = None,
output: RunOutputType = RunOutputTypeDefault.LOG,
output: RunOutputType = "log",
output_return: bool = False,
check: bool | None = None,
sudo: bool = False,
@ -385,18 +380,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").
@ -404,17 +399,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
@ -443,34 +438,34 @@ def core(
# raise e
# Background
if output == RunOutputTypePopen.BACKGROUND:
if output == "background":
return background(cmd, working_dir)
# Pipe
if output == RunOutputTypePopen.PIPE:
if output == "pipe":
return pipe(cmd, working_dir)
# Foreground
output_after_run = ""
if output == RunOutputTypeDefault.TUI:
if output == "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.is_to_stdout():
if not context.details_to_stdout and output in ["stdout", "interactive"]:
output_to_stdout = True
output_timeout = output.has_timeout() and not disable_timeout
output_timeout = output in ["log", "stdout"] and not disable_timeout
stdin = None if output.has_pass_stdin() else subprocess.DEVNULL
stdin = subprocess.DEVNULL if output in ["log", "stdout"] else None
(code, output_after_run) = foreground_pipe(
cmd,
working_dir,
output_to_stdout,
output_return,
output != RunOutputTypeDefault.NULL,
output != "null",
output_timeout,
sudo,
stdin,

10
pmb/install/_install.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import Env, PartitionLayout, PmbArgs
import pmb.helpers.devices
from pmb.helpers.mount import mount_device_rootfs
import pmb.helpers.run
@ -286,9 +286,7 @@ def setup_login(args: PmbArgs, config: Config, chroot: Chroot) -> None:
else:
while True:
try:
pmb.chroot.root(
["passwd", config.user], chroot, output=RunOutputTypeDefault.INTERACTIVE
)
pmb.chroot.root(["passwd", config.user], chroot, output="interactive")
break
except RuntimeError:
logging.info("WARNING: Failed to set the password. Try it one more time.")
@ -352,9 +350,7 @@ 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=RunOutputTypeDefault.INTERACTIVE
)
pmb.chroot.root(["setup-keymap", layout, variant], chroot, output="interactive")
# Check xorg config
xconfig = None

6
pmb/install/format.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import PartitionLayout, PmbArgs, PathString
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=RunOutputTypeDefault.INTERACTIVE)
pmb.chroot.root([*open_cmd, device, "pm_crypt"], output=RunOutputTypeDefault.INTERACTIVE)
pmb.chroot.root(format_cmd, output="interactive")
pmb.chroot.root([*open_cmd, device, "pm_crypt"], output="interactive")
finally:
if path_outside:
os.unlink(path_outside)

4
pmb/qemu/run.py
View file

@ -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, RunOutputTypeDefault
from pmb.types import Env, PathString, PmbArgs
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=RunOutputTypeDefault.TUI, env=env)
process = pmb.helpers.run.user(qemu, output="tui", env=env)
except KeyboardInterrupt:
# In addition to not showing a trace when pressing ^C, let user know
# they can override this behavior:

10
pmb/sideload/__init__.py
View file

@ -6,7 +6,7 @@ from pmb.core.arch import Arch
from pmb.helpers import logging
import shlex
from pmb.types import PathString, PmbArgs, RunOutputTypeDefault
from pmb.types import PathString, PmbArgs
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=RunOutputTypeDefault.INTERACTIVE)
pmb.helpers.run.user(command, output="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=RunOutputTypeDefault.TUI)
pmb.helpers.run.user(command, output="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=RunOutputTypeDefault.INTERACTIVE)
pmb.helpers.run.user(command, output="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=RunOutputTypeDefault.TUI)
pmb.helpers.run.user(command, output="tui")
def sideload(

51
pmb/types.py
View file

@ -58,55 +58,8 @@ class CrossCompile(enum.Enum):
return Chroot.native()
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
RunOutputTypeDefault = Literal["log", "stdout", "interactive", "tui", "null"]
RunOutputTypePopen = Literal["background", "pipe"]
RunOutputType = RunOutputTypeDefault | RunOutputTypePopen
RunReturnType = str | int | subprocess.Popen
PathString = Path | str