[elbe-devel] [PATCH 08/11] elbepack: migrate command_out to subprocess package
Thomas Weißschuh
thomas.weissschuh at linutronix.de
Wed Mar 13 16:54:57 CET 2024
The subprocess APIs are more powerful, better documented and
standardized.
Signed-off-by: Thomas Weißschuh <thomas.weissschuh at linutronix.de>
---
elbepack/commands/check-build.py | 15 ++++++-------
elbepack/commands/test.py | 8 +++----
elbepack/daemons/soap/esoap.py | 20 ++++++++++-------
elbepack/shellhelper.py | 48 ----------------------------------------
4 files changed, 23 insertions(+), 68 deletions(-)
diff --git a/elbepack/commands/check-build.py b/elbepack/commands/check-build.py
index 39417f5bb91e..2f417f004ffd 100644
--- a/elbepack/commands/check-build.py
+++ b/elbepack/commands/check-build.py
@@ -16,15 +16,12 @@ from elbepack import qemu_firmware
from elbepack.directories import elbe_exe
from elbepack.filesystem import TmpdirFilesystem
from elbepack.log import elbe_logging
-from elbepack.shellhelper import command_out, do, get_command_out
+from elbepack.shellhelper import do, get_command_out
from elbepack.treeutils import etree
import pexpect
-DEVNULL = open(os.devnull, 'w')
-
-
def run_command(argv):
oparser = optparse.OptionParser(usage='usage: %prog check-build <cmd> <build-dir>')
@@ -131,8 +128,9 @@ class CheckCdroms(CheckBase):
@staticmethod
def cmp_version(v1, v2):
- return command_out(
- f'dpkg --compare-versions {v1} eq {v2}', output=DEVNULL)[0]
+ return subprocess.run([
+ 'dpkg', '--compare-versions', v1, 'eq', v2,
+ ]).returncode
def do_src(self, sources, src_total):
"""Check for sources in src-cdrom*"""
@@ -390,8 +388,9 @@ class CheckImage(CheckBase):
@staticmethod
def open_tgz(path):
tmp = tempfile.NamedTemporaryFile(prefix='elbe')
- command_out(
- f'tar --to-stdout --extract --gunzip --file {path}', output=tmp)
+ subprocess.run([
+ 'tar', '--to-stdout', '--extract', '--gunzip', '--file', path,
+ ], check=True, stdout=tmp)
return tmp
def open_img(self, path):
diff --git a/elbepack/commands/test.py b/elbepack/commands/test.py
index 108ff7727bb5..a2a025a6101f 100644
--- a/elbepack/commands/test.py
+++ b/elbepack/commands/test.py
@@ -4,7 +4,7 @@
# elbepack/commands/test.py - Elbe unit test wrapper
-from elbepack.shellhelper import command_out
+import subprocess
class ElbeTestException(Exception):
@@ -20,6 +20,6 @@ class ElbeTestException(Exception):
def system(cmd, allow_fail=False):
- ret, out = command_out(cmd)
- if ret != 0 and not allow_fail:
- raise ElbeTestException(cmd, ret, out)
+ ps = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ if ps.returncode != 0 and not allow_fail:
+ raise ElbeTestException(ps.args, ps.returncode, ps.stdout)
diff --git a/elbepack/daemons/soap/esoap.py b/elbepack/daemons/soap/esoap.py
index cae16827ae6c..da3f51af06a6 100644
--- a/elbepack/daemons/soap/esoap.py
+++ b/elbepack/daemons/soap/esoap.py
@@ -6,12 +6,13 @@
import binascii
import fnmatch
import os
+import subprocess
import tarfile
from tempfile import NamedTemporaryFile
from elbepack.elbexml import ValidationMode
from elbepack.filesystem import hostfs
-from elbepack.shellhelper import command_out, system
+from elbepack.shellhelper import system
from elbepack.version import elbe_version, is_devel
from spyne.decorator import rpc
@@ -69,20 +70,23 @@ class ESoap (ServiceBase):
hostfs.write_file('usr/sbin/policy-rc.d',
0o755, '#!/bin/sh\nexit 101\n')
try:
- env = {'LANG': 'C',
- 'LANGUAGE': 'C',
- 'LC_ALL': 'C',
- 'DEBIAN_FRONTEND': 'noninteractive',
- 'DEBCONF_NONINTERACTIVE_SEEN': 'true'}
+ env = os.environ()
+ env.update({
+ 'LANG': 'C',
+ 'LANGUAGE': 'C',
+ 'LC_ALL': 'C',
+ 'DEBIAN_FRONTEND': 'noninteractive',
+ 'DEBCONF_NONINTERACTIVE_SEEN': 'true',
+ })
cmd = ('apt-get update; '
f"apt-get install -y --allow-downgrades {' '.join(pkgs)}")
- ret, out = command_out(cmd, env_add=env)
+ ps = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
finally:
hostfs.remove('usr/sbin/policy-rc.d')
- return SoapCmdReply(ret, out)
+ return SoapCmdReply(ps.returncode, ps.stdout)
@rpc(String, String, String, String, Boolean)
@soap_faults
diff --git a/elbepack/shellhelper.py b/elbepack/shellhelper.py
index 5449b8bea243..13afc19bf488 100644
--- a/elbepack/shellhelper.py
+++ b/elbepack/shellhelper.py
@@ -6,7 +6,6 @@
import logging
import os
import subprocess
-from io import BytesIO, TextIOWrapper
from subprocess import PIPE, Popen, STDOUT
from elbepack.log import async_logging_ctx
@@ -39,53 +38,6 @@ def system(cmd, allow_fail=False, env_add=None):
subprocess.run(cmd, shell=True, env=new_env, check=not allow_fail)
-def command_out(cmd, stdin=None, output=PIPE, env_add=None):
- """command_out() - Execute cmd in a shell.
-
- Returns a tuple with the exitcode and the output of cmd.
-
- --
-
- >>> command_out("true")
- (0, '')
-
- >>> command_out("$TRUE && true", env_add={"TRUE":"true"})
- (0, '')
-
- >>> command_out("cat -", stdin=b"ELBE")
- (0, 'ELBE')
-
- >>> command_out("2>&1 cat -", stdin=b"ELBE")
- (0, 'ELBE')
-
- >>> command_out("2>&1 cat -", stdin="ELBE")
- (0, 'ELBE')
-
- >>> command_out("false")
- (1, '')
-
- """
- new_env = os.environ.copy()
- if env_add:
- new_env.update(env_add)
-
- if isinstance(stdin, str):
- stdin = stdin.encode()
-
- if stdin is None:
- p = Popen(cmd, shell=True,
- stdout=output, stderr=STDOUT, env=new_env)
- out, _ = p.communicate()
- else:
- p = Popen(cmd, shell=True,
- stdout=output, stderr=STDOUT, stdin=PIPE, env=new_env)
- out, _ = p.communicate(input=stdin)
-
- out = TextIOWrapper(BytesIO(out), encoding='utf-8', errors='replace').read()
-
- return p.returncode, out
-
-
def do(cmd, allow_fail=False, stdin=None, env_add=None, log_cmd=None):
"""do() - Execute cmd in a shell and redirect outputs to logging.
--
2.44.0
More information about the elbe-devel
mailing list