From b2e1e411f109861b5a70a023625b3faa81fb2278 Mon Sep 17 00:00:00 2001 From: Jan Lindemann Date: Mon, 23 Mar 2026 13:10:07 +0100 Subject: [PATCH] lib.pm.*.query_packages(): Make it non-interactive lib.pm.query_packages() uses a TTY for doing its thing and outputs half-digested stuff to the terminal, fix that. Signed-off-by: Jan Lindemann --- src/python/jw/pkg/lib/pm/dpkg.py | 4 ++-- src/python/jw/pkg/lib/pm/rpm.py | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/python/jw/pkg/lib/pm/dpkg.py b/src/python/jw/pkg/lib/pm/dpkg.py index c67010d7..c65b269b 100644 --- a/src/python/jw/pkg/lib/pm/dpkg.py +++ b/src/python/jw/pkg/lib/pm/dpkg.py @@ -7,8 +7,8 @@ from typing import Iterable, TYPE_CHECKING if TYPE_CHECKING: from ..ExecContext import ExecContext +from ..ExecContext import InputMode from ..util import run_cmd, run_sudo - from ..Package import Package, meta_tags _meta_map: dict[str, str]|None = None @@ -37,7 +37,7 @@ async def run_dpkg_query(args: list[str], sudo: bool=False, ec: ExecContext=None cmd.extend(args) if sudo: return await run_sudo(cmd) - return (await run_cmd(cmd, ec=ec)).decode() + return (await run_cmd(cmd, ec=ec, cmd_input=InputMode.NonInteractive)).decode() async def query_packages(names: Iterable[str] = [], ec: ExecContext=None) -> Iterable[Package]: # export fmt_str = '|'.join([(f'${{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n' diff --git a/src/python/jw/pkg/lib/pm/rpm.py b/src/python/jw/pkg/lib/pm/rpm.py index d253851d..276511ab 100644 --- a/src/python/jw/pkg/lib/pm/rpm.py +++ b/src/python/jw/pkg/lib/pm/rpm.py @@ -8,7 +8,7 @@ if TYPE_CHECKING: from ..ExecContext import ExecContext from ..util import run_cmd, run_sudo - +from ..ExecContext import InputMode from ..Package import Package, meta_tags _meta_map: dict[str, str]|None = None @@ -25,21 +25,21 @@ def meta_map(): }) return _meta_map -async def run_rpm(args: list[str], sudo: bool=False, ec: ExecContext=None, **kwargs): # export +async def run_rpm(args: list[str], sudo: bool=False, ec: ExecContext=None, mode: InputMode=InputMode.OptInteractive, **kwargs): # export cmd = ['/usr/bin/rpm'] cmd.extend(args) if sudo: - return await run_sudo(cmd, ec=ec, **kwargs) - return await run_cmd(cmd, ec=ec, **kwargs) + return await run_sudo(cmd, ec=ec, cmd_input=mode, **kwargs) + return await run_cmd(cmd, ec=ec, cmd_input=mode, **kwargs) async def query_packages(names: Iterable[str] = [], ec: ExecContext=None) -> Iterable[Package]: # export fmt_str = '|'.join([(f'%{{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n' opts = ['-q', '--queryformat', fmt_str] if not names: opts.append('-a') - specs, stderr, status = await run_rpm([*opts, *names], throw=True, sudo=False, ec=ec) + specs, stderr, status = await run_rpm([*opts, *names], throw=True, sudo=False, mode=InputMode.NonInteractive, ec=ec) return Package.parse_specs_str(specs.decode()) async def list_files(pkg: str, ec: ExecContext=None) -> list[str]: # export - stdout, stderr, status = await run_rpm(['-ql', pkg], throw=True, sudo=False, ec=ec) + stdout, stderr, status = await run_rpm(['-ql', pkg], throw=True, sudo=False, mode=InputMode.NonInteractive, ec=ec) return stdout.decode().splitlines()