[elbe-devel] [PATCH 5/5] elbepack: migrate shellhelper.system to subprocess package
Thomas Weißschuh
thomas.weissschuh at linutronix.de
Fri Apr 5 11:53:59 CEST 2024
The subprocess APIs are more powerful, better documented and
standardized.
Signed-off-by: Thomas Weißschuh <thomas.weissschuh at linutronix.de>
---
elbepack/archivedir.py | 13 +++---
elbepack/commands/check_updates.py | 6 +--
elbepack/commands/chroot.py | 13 +++---
elbepack/daemons/soap/esoap.py | 3 +-
elbepack/debinstaller.py | 32 ++++++++------
elbepack/efilesystem.py | 16 ++++---
elbepack/egpg.py | 15 ++++---
elbepack/hashes.py | 4 +-
elbepack/initvmaction.py | 86 +++++++++++++++++++-------------------
elbepack/pbuilderaction.py | 59 ++++++++++++--------------
elbepack/pkgutils.py | 6 +--
elbepack/shellhelper.py | 30 +++----------
elbepack/updated.py | 9 ++--
13 files changed, 138 insertions(+), 154 deletions(-)
diff --git a/elbepack/archivedir.py b/elbepack/archivedir.py
index f5013e379547..718cf9ac8f46 100644
--- a/elbepack/archivedir.py
+++ b/elbepack/archivedir.py
@@ -5,6 +5,7 @@
import bz2
import os
import re
+import subprocess
import sys
from base64 import encodebytes, standard_b64decode
from subprocess import CalledProcessError
@@ -12,7 +13,6 @@ from tempfile import NamedTemporaryFile
from urllib.parse import urljoin, urlparse
from elbepack.filesystem import TmpdirFilesystem
-from elbepack.shellhelper import system
from elbepack.treeutils import etree
@@ -29,12 +29,11 @@ def enbase(fname, compress=True):
def collect(tararchive, path, keep):
- if keep:
- cmd = 'tar rf ' + tararchive + ' -C '
- else:
- cmd = 'tar rf ' + tararchive + ' --owner=root --group=root -C '
- cmd += path + ' .'
- system(cmd)
+ subprocess.run([
+ 'tar', 'rf', tararchive,
+ *([] if keep else ['--owner=root', '--group=root']),
+ '-C', path, '.',
+ ])
def chg_archive(xml, path, keep):
diff --git a/elbepack/commands/check_updates.py b/elbepack/commands/check_updates.py
index 45e6ef9691b8..8f391d40fe56 100644
--- a/elbepack/commands/check_updates.py
+++ b/elbepack/commands/check_updates.py
@@ -2,6 +2,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
# SPDX-FileCopyrightText: 2013-2018 Linutronix GmbH
+import subprocess
import sys
from optparse import OptionParser
@@ -10,7 +11,6 @@ from elbepack.aptpkgutils import XMLPackage
from elbepack.changelogxml import changelogs_xml
from elbepack.elbexml import ElbeXML
from elbepack.pkgutils import ChangelogNeedsDependency, extract_pkg_changelog
-from elbepack.shellhelper import system
from elbepack.validate import validate_xml
@@ -121,7 +121,7 @@ def run_command(argv):
if errors > 0:
print(f'{errors} Errors occured, xml files needs fixing')
if opt.script:
- system(f'{opt.script} ERRORS {args[0]}', allow_fail=True)
+ subprocess.run([opt.script, 'ERRORS', args[0]])
elif required_updates > 0:
print(f'{required_updates} updates required')
@@ -129,6 +129,6 @@ def run_command(argv):
build_changelog_xml(v, opt, update_packages)
if opt.script:
- system(f'{opt.script} UPDATE {args[0]}', allow_fail=True)
+ subprocess.run([opt.script, 'UPDATE', args[0]])
else:
print('No Updates available')
diff --git a/elbepack/commands/chroot.py b/elbepack/commands/chroot.py
index b5a441cf1cee..c7b7002ab211 100644
--- a/elbepack/commands/chroot.py
+++ b/elbepack/commands/chroot.py
@@ -11,7 +11,6 @@ from optparse import OptionParser
from elbepack.elbeproject import ElbeProject
from elbepack.elbexml import ValidationError, ValidationMode
from elbepack.log import elbe_logging
-from elbepack.shellhelper import system
def run_command(argv):
@@ -49,19 +48,19 @@ def run_command(argv):
# TODO: howto set env in chroot?
os.environ['PS1'] = project.xml.text('project/name') + r': \w\$'
- cmd = '/bin/bash'
+ chroot_args = ['/bin/bash']
if len(args) > 1:
- cmd = ''
- cmd2 = args[1:]
- for c in cmd2:
- cmd += (c + ' ')
+ chroot_args = args[1:]
chroot, path = (project.targetfs, project.targetpath) if opt.target else \
(project.buildenv, project.chrootpath)
try:
with chroot:
- system(f'/usr/sbin/chroot {path} {cmd}')
+ subprocess.run([
+ '/usr/sbin/chroot', path,
+ *chroot_args,
+ ], check=True)
except subprocess.CalledProcessError as e:
print(repr(e))
diff --git a/elbepack/daemons/soap/esoap.py b/elbepack/daemons/soap/esoap.py
index da3f51af06a6..1b0bde8e1cf2 100644
--- a/elbepack/daemons/soap/esoap.py
+++ b/elbepack/daemons/soap/esoap.py
@@ -12,7 +12,6 @@ from tempfile import NamedTemporaryFile
from elbepack.elbexml import ValidationMode
from elbepack.filesystem import hostfs
-from elbepack.shellhelper import system
from elbepack.version import elbe_version, is_devel
from spyne.decorator import rpc
@@ -372,7 +371,7 @@ class ESoap (ServiceBase):
@authenticated_admin
@soap_faults
def shutdown_initvm(self):
- system('systemctl --no-block poweroff')
+ subprocess.run(['systemctl', '--no-block', 'poweroff'], check=True)
@rpc(String)
@authenticated_uid
diff --git a/elbepack/debinstaller.py b/elbepack/debinstaller.py
index 4e7493421e3e..4e03c450cd2f 100644
--- a/elbepack/debinstaller.py
+++ b/elbepack/debinstaller.py
@@ -4,6 +4,7 @@
import os
import re
+import shutil
import subprocess
import sys
from shutil import copyfile
@@ -12,7 +13,6 @@ from urllib.request import urlopen
from elbepack.egpg import OverallStatus, check_signature
from elbepack.filesystem import TmpdirFilesystem
from elbepack.hashes import HashValidationFailed, HashValidator
-from elbepack.shellhelper import system
from gpg import core
from gpg.constants import PROTOCOL_OpenPGP
@@ -82,27 +82,31 @@ def setup_apt_keyring(gpg_home, keyring_fname):
sys.exit(115)
if os.path.exists('/etc/apt/trusted.gpg'):
- system(f'cp /etc/apt/trusted.gpg "{ring_path}"')
+ shutil.copyfile('/etc/apt/trusted.gpg', ring_path)
- gpg_options = f'--keyring "{ring_path}" --no-auto-check-trustdb ' \
- '--trust-model always --no-default-keyring ' \
- '--batch ' \
- f'--homedir "{gpg_home}"'
+ gpg_options = [
+ '--keyring', ring_path,
+ '--no-auto-check-trustdb',
+ '--trust-model', 'always', '--no-default-keyring',
+ '--batch',
+ '--homedir', gpg_home,
+ ]
trustkeys = os.listdir('/etc/apt/trusted.gpg.d')
for key in trustkeys:
print(f'Import {key}: ')
try:
- system(
- f'gpg {gpg_options} '
- f'--import "{os.path.join("/etc/apt/trusted.gpg.d", key)}"')
+ subprocess.run([
+ 'gpg', *gpg_options,
+ '--import', os.path.join('/etc/apt/trusted.gpg.d', key),
+ ], check=True)
except subprocess.CalledProcessError:
print(f'adding keyring "{key}" to keyring "{ring_path}" failed')
def download(url, local_fname):
try:
- system(f'wget -O "{local_fname}" "{url}"')
+ subprocess.run(['wget', '-O', local_fname, url], check=True)
except subprocess.CalledProcessError:
raise NoKinitrdException(f'Failed to download {url}')
@@ -207,9 +211,11 @@ def copy_kinitrd(prj, target_dir):
try:
tmp = TmpdirFilesystem()
if prj.has('mirror/cdrom'):
- system(
- f'7z x -o{tmp.fname("/")} "{prj.text("mirror/cdrom")}" '
- 'initrd-cdrom.gz vmlinuz')
+ subprocess.run([
+ '7z', 'x', f'-o{tmp.fname("/")}',
+ prj.text('mirror/cdrom'),
+ 'initrd-cdrom.gz', 'vmlinuz',
+ ], check=True)
# initrd.gz needs to be cdrom version !
copyfile(tmp.fname('initrd-cdrom.gz'),
diff --git a/elbepack/efilesystem.py b/elbepack/efilesystem.py
index 2f22f1d80d42..a71c368e475f 100644
--- a/elbepack/efilesystem.py
+++ b/elbepack/efilesystem.py
@@ -15,7 +15,7 @@ from elbepack.fstab import fstabentry
from elbepack.hdimg import do_hdimg
from elbepack.licencexml import copyright_xml
from elbepack.packers import default_packer
-from elbepack.shellhelper import chroot, do, get_command_out, system
+from elbepack.shellhelper import chroot, do, get_command_out
from elbepack.version import elbe_version
@@ -332,10 +332,14 @@ class ChRootFilesystem(ElbeFilesystem):
if self.path == '/':
return
try:
- system(f'mount -t proc none {self.path}/proc')
- system(f'mount -t sysfs none {self.path}/sys')
- system(f'mount -o bind /dev {self.path}/dev')
- system(f'mount -o bind /dev/pts {self.path}/dev/pts')
+ subprocess.run(['mount', '-t', 'proc', 'none', os.path.join(self.path, 'proc')],
+ check=True)
+ subprocess.run(['mount', '-t', 'sysfs', 'none', os.path.join(self.path, 'sys')],
+ check=True)
+ subprocess.run(['mount', '-o', 'bind', '/dev', os.path.join(self.path, 'dev')],
+ check=True)
+ subprocess.run(['mount', '-o', 'bind', '/dev/pts', os.path.join(self.path, 'dev/pts')],
+ check=True)
except BaseException:
self.umount()
raise
@@ -358,7 +362,7 @@ class ChRootFilesystem(ElbeFilesystem):
def _umount(self, path):
path = os.path.join(self.path, path)
if os.path.ismount(path):
- system(f'umount {path}')
+ subprocess.run(['umount', path], check=True)
def umount(self):
if self.path == '/':
diff --git a/elbepack/egpg.py b/elbepack/egpg.py
index 1a8a01f06c34..e0bcaf827f72 100644
--- a/elbepack/egpg.py
+++ b/elbepack/egpg.py
@@ -7,7 +7,7 @@ import os
import subprocess
from elbepack.filesystem import hostfs
-from elbepack.shellhelper import get_command_out, system
+from elbepack.shellhelper import env_add, get_command_out
from gpg import core
from gpg.constants import PROTOCOL_OpenPGP, sig, sigsum
@@ -186,9 +186,10 @@ def unlock_key(fingerprint):
'/var/cache/elbe/gnupg')
key = ctx.get_key(fingerprint, secret=True)
keygrip = key.subkeys[0].keygrip
- system('/usr/lib/gnupg/gpg-preset-passphrase '
- f'--preset -P requiredToAvoidUserInput {keygrip}',
- env_add={'GNUPGHOME': '/var/cache/elbe/gnupg'})
+ subprocess.run([
+ '/usr/lib/gnupg/gpg-preset-passphrase',
+ '--preset', '-P', 'requiredToAvoidUserInput', keygrip,
+ ], check=True, env=env_add({'GNUPGHOME': '/var/cache/elbe/gnupg'}))
def sign(infile, outfile, fingerprint):
@@ -278,8 +279,10 @@ def generate_elbe_internal_key():
def export_key(fingerprint, outfile):
- system(f'/usr/bin/gpg -a -o {outfile} --export {fingerprint}',
- env_add={'GNUPGHOME': '/var/cache/elbe/gnupg'})
+ subprocess.run([
+ '/usr/bin/gpg', '-a', '-o', outfile,
+ '--export', fingerprint,
+ ], check=True, env=env_add({'GNUPGHOME': '/var/cache/elbe/gnupg'}))
def unarmor_openpgp_keyring(armored):
diff --git a/elbepack/hashes.py b/elbepack/hashes.py
index a1b5c1b9dc58..eaf8e548bd69 100644
--- a/elbepack/hashes.py
+++ b/elbepack/hashes.py
@@ -5,8 +5,6 @@
import hashlib
import subprocess
-from elbepack.shellhelper import system
-
class HashValidationFailed(Exception):
pass
@@ -45,7 +43,7 @@ class HashValidator:
def download_and_validate_file(self, upstream_fname, local_fname):
url = self.base_url + upstream_fname
try:
- system(f'wget -O "{local_fname}" "{url}"')
+ subprocess.run(['wget', '-O', local_fname, url], check=True)
except subprocess.CalledProcessError as e:
raise HashValidationFailed(f'Failed to download {url}') from e
diff --git a/elbepack/initvmaction.py b/elbepack/initvmaction.py
index 059cf5c41054..a0c491de88c4 100644
--- a/elbepack/initvmaction.py
+++ b/elbepack/initvmaction.py
@@ -18,7 +18,6 @@ from elbepack.directories import elbe_exe, run_elbe
from elbepack.elbexml import ElbeXML, ValidationError, ValidationMode
from elbepack.filesystem import TmpdirFilesystem
from elbepack.repodir import Repodir, RepodirError
-from elbepack.shellhelper import system
from elbepack.treeutils import etree
from elbepack.xmlpreprocess import PreprocessWrapper
@@ -310,7 +309,7 @@ class EnsureAction(InitVMAction):
import libvirt
if self.initvm_state() == libvirt.VIR_DOMAIN_SHUTOFF:
- system(f'{sys.executable} {elbe_exe} initvm start')
+ run_elbe(['initvm', 'start'], check=True)
elif self.initvm_state() == libvirt.VIR_DOMAIN_RUNNING:
test_soap_communication()
else:
@@ -426,7 +425,8 @@ class AttachAction(InitVMAction):
sys.exit(126)
print('Attaching to initvm console.')
- system(f'virsh --connect qemu:///system console {cfg["initvm_domain"]}')
+ subprocess.run(['virsh', '--connect', 'qemu:///system', 'console', cfg['initvm_domain']],
+ check=True)
def execute(self, initvmdir, opt, _args):
if opt.qemu_mode:
@@ -485,7 +485,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
if cdrom is not None:
print('Uploading CDROM. This might take a while')
try:
- system(f'{sys.executable} {elbe_exe} control set_cdrom "{prjdir}" "{cdrom}"')
+ run_elbe(['control', 'set_cdrom', prjdir, cdrom], check=True)
except subprocess.CalledProcessError:
print('elbe control set_cdrom Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -493,16 +493,16 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
print('Upload finished')
- build_opts = ''
+ build_opts = []
if opt.build_bin:
- build_opts += '--build-bin '
+ build_opts.append('--build-bin')
if opt.build_sources:
- build_opts += '--build-sources '
+ build_opts.append('--build-sources')
if cdrom:
- build_opts += '--skip-pbuilder '
+ build_opts.append('--skip-pbuilder')
try:
- system(f'{sys.executable} {elbe_exe} control build "{prjdir}" {build_opts}')
+ run_elbe(['control', 'build', prjdir, *build_opts], check=True)
except subprocess.CalledProcessError:
print('elbe control build Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -511,7 +511,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
print('Build started, waiting till it finishes')
try:
- system(f'{sys.executable} {elbe_exe} control wait_busy "{prjdir}"')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control wait_busy Failed', file=sys.stderr)
print('', file=sys.stderr)
@@ -536,7 +536,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
if opt.build_sdk:
try:
- system(f'{sys.executable} {elbe_exe} control build_sdk "{prjdir}" {build_opts}')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control build_sdk Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -545,7 +545,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
print('SDK Build started, waiting till it finishes')
try:
- system(f'{sys.executable} {elbe_exe} control wait_busy "{prjdir}"')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control wait_busy Failed, while waiting for the SDK',
file=sys.stderr)
@@ -571,14 +571,14 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
print('')
try:
- system(f'{sys.executable} {elbe_exe} control dump_file "{prjdir}" validation.txt')
+ run_elbe(['control', 'dump_file', prjdir, 'validation.txt'], check=True)
except subprocess.CalledProcessError:
print(
'Project failed to generate validation.txt',
file=sys.stderr)
print('Getting log.txt', file=sys.stderr)
try:
- system(f'{sys.executable} {elbe_exe} control dump_file "{prjdir}" log.txt')
+ run_elbe(['control', 'dump_file', prjdir, 'log.txt'], check=True)
except subprocess.CalledProcessError:
print('Failed to dump log.txt', file=sys.stderr)
@@ -590,7 +590,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
print('Listing available files:')
print('')
try:
- system(f'{sys.executable} {elbe_exe} control get_files "{prjdir}"')
+ run_elbe(['control', 'get_files', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control get_files Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -606,9 +606,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
ensure_outdir(opt)
try:
- system(
- f'{sys.executable} {elbe_exe} control get_files --output "{opt.outdir}" '
- f'"{prjdir}"')
+ run_elbe(['control', 'get_files', '--output', opt.outdir, prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control get_files Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -616,7 +614,7 @@ def submit_and_dl_result(xmlfile, cdrom, opt):
if not opt.keep_files:
try:
- system(f'{sys.executable} {elbe_exe} control del_project "{prjdir}"')
+ run_elbe(['control', 'del_project', prjdir], check=True)
except subprocess.CalledProcessError:
print('remove project from initvm failed',
file=sys.stderr)
@@ -639,7 +637,7 @@ def extract_cdrom(cdrom):
iso.get_file_from_iso(extracted, iso_path=f'/{in_iso_name.upper()};1')
iso.close()
except ImportError:
- system(f'7z x -o{tmp.path} "{cdrom}" {in_iso_name}')
+ subprocess.run(['7z', 'x', f'-o{tmp.path}', cdrom, in_iso_name], check=True)
print('', file=sys.stderr)
@@ -701,12 +699,13 @@ class CreateAction(InitVMAction):
# Upgrade from older versions which used tmux
try:
- system('tmux has-session -t ElbeInitVMSession 2>/dev/null')
+ subprocess.run(['tmux', 'has-session', '-t', 'ElbeInitVMSession'],
+ stderr=subprocess.DEVNULL, check=True)
print('ElbeInitVMSession exists in tmux. '
'It may belong to an old elbe version. '
'Please stop it to prevent interfering with this version.', file=sys.stderr)
sys.exit(143)
- except subprocess.CalledProcessError:
+ except (subprocess.CalledProcessError, FileNotFoundError):
pass
# Init cdrom to None, if we detect it, we set it
@@ -743,29 +742,27 @@ class CreateAction(InitVMAction):
'init/default-init.xml')
try:
- init_opts = ''
+ init_opts = []
if opt.devel:
- init_opts += ' --devel'
+ init_opts.append('--devel')
if opt.nesting:
- init_opts += ' --nesting'
+ init_opts.append('--nesting')
if not opt.build_bin:
- init_opts += ' --skip-build-bin'
+ init_opts.append('--skip-build-bin')
if not opt.build_sources:
- init_opts += ' --skip-build-source'
+ init_opts.append(' --skip-build-source')
+
+ if cdrom:
+ cdrom_opts = ['--cdrom', cdrom]
+ else:
+ cdrom_opts = []
with PreprocessWrapper(xmlfile, opt) as ppw:
- if cdrom:
- system(
- f'{sys.executable} {elbe_exe} init {init_opts} '
- f'--directory "{initvmdir}" --cdrom "{cdrom}" '
- f'"{ppw.preproc}"')
- else:
- system(
- f'{sys.executable} {elbe_exe} init {init_opts} '
- f'--directory "{initvmdir}" "{ppw.preproc}"')
+ run_elbe(['init', *init_opts, '--directory', initvmdir, *cdrom_opts, ppw.preproc],
+ check=True)
except subprocess.CalledProcessError:
print("'elbe init' Failed", file=sys.stderr)
@@ -790,7 +787,7 @@ class CreateAction(InitVMAction):
# Build initvm
try:
- system(f'cd "{initvmdir}"; make')
+ subprocess.run(['make'], cwd=initvmdir, check=True)
except subprocess.CalledProcessError:
print('Building the initvm Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -879,12 +876,15 @@ class SyncAction(InitVMAction):
def execute(self, _initvmdir, _opt, _args):
top_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+ excludes = ['.git*', '*.pyc', 'elbe-build*', 'initvm', '__pycache__', 'docs', 'examples']
try:
- system('rsync --info=name1,stats1 --archive --times '
- "--exclude='.git*' --exclude='*.pyc' --exclude='elbe-build*' "
- "--exclude='initvm' --exclude='__pycache__' --exclude='docs' "
- "--exclude='examples' "
- f"--rsh='ssh -p {cfg['sshport']}' --chown=root:root "
- f'{top_dir}/ root at localhost:/var/cache/elbe/devel')
+ subprocess.run([
+ 'rsync', '--info=name1,stats1', '--archive', '--times',
+ *[arg for e in excludes for arg in ('--exclude', e)],
+ f'--rsh=ssh -p {cfg["sshport"]}',
+ '--chown=root:root',
+ f'{top_dir}/',
+ 'root at localhost:/var/cache/elbe/devel'
+ ], check=True)
except subprocess.CalledProcessError as E:
print(E)
diff --git a/elbepack/pbuilderaction.py b/elbepack/pbuilderaction.py
index 35f0220095c4..56f5c5c5ba28 100644
--- a/elbepack/pbuilderaction.py
+++ b/elbepack/pbuilderaction.py
@@ -6,9 +6,8 @@ import os
import subprocess
import sys
-from elbepack.directories import elbe_exe, run_elbe
+from elbepack.directories import run_elbe
from elbepack.filesystem import TmpdirFilesystem
-from elbepack.shellhelper import system
from elbepack.xmlpreprocess import PreprocessWrapper
@@ -64,15 +63,13 @@ class CreateAction(PBuilderAction):
PBuilderAction.__init__(self, node)
def execute(self, opt, _args):
- crossopt = ''
+ crossopt = []
if opt.cross:
- crossopt = '--cross'
+ crossopt = ['--cross']
if opt.noccache:
- ccacheopt = '--no-ccache'
- ccachesize = ''
+ ccacheopt = ['--no-ccache']
else:
- ccacheopt = '--ccache-size'
- ccachesize = opt.ccachesize
+ ccacheopt = ['--ccache-size', opt.ccachesize]
if opt.xmlfile:
try:
@@ -116,15 +113,15 @@ class CreateAction(PBuilderAction):
print('Creating pbuilder')
try:
- system(f'{sys.executable} {elbe_exe} control '
- f'build_pbuilder "{prjdir}" {crossopt} {ccacheopt} {ccachesize}')
+ run_elbe(['control', 'build_pbuilder', prjdir, *crossopt, *ccacheopt],
+ check=True)
except subprocess.CalledProcessError:
print('elbe control build_pbuilder Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
sys.exit(156)
try:
- system(f'{sys.executable} {elbe_exe} control wait_busy "{prjdir}"')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control wait_busy Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -156,7 +153,7 @@ class UpdateAction(PBuilderAction):
print('Updating pbuilder')
try:
- system(f'{sys.executable} {elbe_exe} control update_pbuilder "{prjdir}"')
+ run_elbe(['control', 'update_pbuilder', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control update_pbuilder Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -179,9 +176,9 @@ class BuildAction(PBuilderAction):
def execute(self, opt, _args):
- crossopt = ''
+ crossopt = []
if opt.cross:
- crossopt = '--cross'
+ crossopt = ['--cross']
tmp = TmpdirFilesystem()
if opt.xmlfile:
@@ -196,14 +193,14 @@ class BuildAction(PBuilderAction):
prjdir = ps.stdout.strip()
try:
- system(f'{sys.executable} {elbe_exe} control build_pbuilder "{prjdir}"')
+ run_elbe(['control', 'build_pbuilder', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control build_pbuilder Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
sys.exit(161)
try:
- system(f'{sys.executable} {elbe_exe} control wait_busy "{prjdir}"')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control wait_busy Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -214,7 +211,7 @@ class BuildAction(PBuilderAction):
print('')
elif opt.project:
prjdir = opt.project
- system(f'{sys.executable} {elbe_exe} control rm_log {prjdir}')
+ run_elbe(['control', 'rm_log', prjdir], check=True)
else:
print(
'you need to specify --project or --xmlfile option',
@@ -225,7 +222,7 @@ class BuildAction(PBuilderAction):
print('Packing Source into tmp archive')
print('')
try:
- system(f'tar cfz "{tmp.fname("pdebuild.tar.gz")}" .')
+ subprocess.run(['tar', 'cfz', tmp.fname('pdebuild.tar.gz'), '.'], check=True)
except subprocess.CalledProcessError:
print('tar Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -236,8 +233,7 @@ class BuildAction(PBuilderAction):
print(f"Pushing orig file '{of}' into pbuilder")
print('')
try:
- system(
- f'{sys.executable} {elbe_exe} control set_orig "{prjdir}" "{of}"')
+ run_elbe(['control', 'set_orig', prjdir, of], check=True)
except subprocess.CalledProcessError:
print('elbe control set_orig Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -248,16 +244,17 @@ class BuildAction(PBuilderAction):
print('')
try:
- system(
- f'{sys.executable} {elbe_exe} control set_pdebuild --cpuset "{opt.cpuset}" '
- f'--profile "{opt.profile}" {crossopt} '
- f'"{prjdir}" "{tmp.fname("pdebuild.tar.gz")}"')
+ run_elbe([
+ 'control', 'set_pdebuild', '--cpuset', str(opt.cpuset),
+ '--profile', opt.profile, *crossopt,
+ prjdir, tmp.fname('pdebuild.tar.gz'),
+ ], check=True)
except subprocess.CalledProcessError:
print('elbe control set_pdebuild Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
sys.exit(166)
try:
- system(f'{sys.executable} {elbe_exe} control wait_busy "{prjdir}"')
+ run_elbe(['control', 'wait_busy', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control wait_busy Failed', file=sys.stderr)
print('Giving up', file=sys.stderr)
@@ -271,15 +268,14 @@ class BuildAction(PBuilderAction):
print('Listing available files:')
print('')
try:
- system(
- f'{sys.executable} {elbe_exe} control --pbuilder-only get_files "{prjdir}"')
+ run_elbe(['control', '--pbuilder-only', 'get_files', prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control get_files Failed', file=sys.stderr)
print('', file=sys.stderr)
print('dumping logfile', file=sys.stderr)
try:
- system(f'{sys.executable} {elbe_exe} control dump_file "{prjdir}" log.txt')
+ run_elbe(['control', 'dump_file', prjdir, 'log.txt'], check=True)
except subprocess.CalledProcessError:
print('elbe control dump_file Failed', file=sys.stderr)
print('', file=sys.stderr)
@@ -297,16 +293,15 @@ class BuildAction(PBuilderAction):
ensure_outdir(opt)
try:
- system(
- f'{sys.executable} {elbe_exe} control --pbuilder-only get_files '
- f'--output "{opt.outdir}" "{prjdir}"')
+ run_elbe(['control', '--pbuilder-only', 'get_files',
+ '--output', opt.outdir, prjdir], check=True)
except subprocess.CalledProcessError:
print('elbe control get_files Failed', file=sys.stderr)
print('', file=sys.stderr)
print('dumping logfile', file=sys.stderr)
try:
- system(f'{sys.executable} {elbe_exe} control dump_file "{prjdir}" log.txt')
+ run_elbe(['control', 'dump_file', prjdir, 'log.txt'], check=True)
except subprocess.CalledProcessError:
print('elbe control dump_file Failed', file=sys.stderr)
print('', file=sys.stderr)
diff --git a/elbepack/pkgutils.py b/elbepack/pkgutils.py
index 23b76544c007..465305d5031f 100644
--- a/elbepack/pkgutils.py
+++ b/elbepack/pkgutils.py
@@ -4,11 +4,11 @@
import os
import re
+import subprocess
from apt_pkg import TagFile
from elbepack.filesystem import TmpdirFilesystem
-from elbepack.shellhelper import system
class NoPackageException(Exception):
@@ -89,9 +89,9 @@ def extract_pkg_changelog(fname, extra_pkg=None):
if extra_pkg:
print('with extra ' + extra_pkg)
- system(f'dpkg -x "{extra_pkg}" "{fs.fname("/")}"')
+ subprocess.run(['dpkg', '-x', extra_pkg, fs.fname('/')], check=True)
- system(f'dpkg -x "{fname}" "{fs.fname("/")}"')
+ subprocess.run(['dpkg', '-x', fname, fs.fname('/')], check=True)
dch_dir = f'/usr/share/doc/{pkgname}'
diff --git a/elbepack/shellhelper.py b/elbepack/shellhelper.py
index 13afc19bf488..1b32fd784467 100644
--- a/elbepack/shellhelper.py
+++ b/elbepack/shellhelper.py
@@ -14,30 +14,6 @@ log = logging.getLogger('log')
soap = logging.getLogger('soap')
-def system(cmd, allow_fail=False, env_add=None):
- """system() - Execute cmd in a shell.
-
- Throws a subprocess.CalledProcessError if cmd returns none-zero and allow_fail=False
-
- --
-
- >>> system("true")
-
- >>> system("false", allow_fail=True)
-
- >>> system("$FALSE", env_add={"FALSE":"false"}) # doctest: +ELLIPSIS
- Traceback (most recent call last):
- ...
- subprocess.CalledProcessError: ...
-
- """
- new_env = os.environ.copy()
- if env_add:
- new_env.update(env_add)
-
- subprocess.run(cmd, shell=True, env=new_env, check=not allow_fail)
-
-
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.
@@ -171,3 +147,9 @@ def get_command_out(cmd, stdin=None, allow_fail=False, env_add=None):
raise subprocess.CalledProcessError(p.returncode, cmd)
return stdout
+
+
+def env_add(d):
+ env = os.environ.copy()
+ env.update(d)
+ return env
diff --git a/elbepack/updated.py b/elbepack/updated.py
index 203b5831ca83..cee21f466e25 100644
--- a/elbepack/updated.py
+++ b/elbepack/updated.py
@@ -24,7 +24,6 @@ from elbepack.aptprogress import (
)
from elbepack.config import cfg
from elbepack.egpg import unsign_file
-from elbepack.shellhelper import system
from elbepack.treeutils import etree
from packaging import version
@@ -172,7 +171,7 @@ class rw_access:
if self.mount_orig == 'ro':
self.status.log(f'remount {self.mount} read/writeable')
try:
- system(f'mount -o remount,rw {self.mount}')
+ subprocess.run(['mount', '-o', 'remount,rw', self.mount], check=True)
except subprocess.CalledProcessError as e:
self.status.log(repr(e))
@@ -180,11 +179,11 @@ class rw_access:
if self.mount_orig == 'ro':
self.status.log(f'remount {self.mount} readonly')
try:
- system('sync')
+ subprocess.run(['sync'], check=True)
except subprocess.CalledProcessError as e:
self.status.log(repr(e))
try:
- system(f'mount -o remount,ro {self.mount}')
+ subprocess.run(['mount', '-o', 'remount,ro', self.mount], check=True)
except subprocess.CalledProcessError as e:
self.status.log(repr(e))
@@ -428,7 +427,7 @@ def apply_update(fname, status):
# don't use execute() here, it results in an error that the apt-cache
# is locked. We currently don't understand this behaviour :(
try:
- system('apt-get clean')
+ subprocess.run(['apt-get', 'clean'], check=True)
except subprocess.CalledProcessError as e:
status.log(repr(e))
if p.exitcode != 0:
--
2.44.0
More information about the elbe-devel
mailing list