mirror of
https://github.com/peridotbuild/pv2.git
synced 2024-11-25 06:31:26 +00:00
185d144567
Adds the importutil module that allows targetting a specific source RPM
file and importing and tagging. It aims to keep the same structure of
git.centos.org.
Other changes:
* constants.py: New constants added for git and rpm
* error.py: New git and rpm error classes added
* fileutil.py:
* Add filter_files_inverse (matches everything but the filter)
* Add get_magic_file (returns magic data from a file)
* Add get_magic_content (returns magic data from data/content)
* generic.py: Add safe_encoding to return a urlquote string
* processor.py:
* Add run_proc_foreground_shell to support shell calls
* Add run_proc_no_output_shell to support shell calls
* rpmutil.py:
* get_rpm_header now supports verify_signature parameter (default
false). If set to true and key is not available, raises exception.
* Add verify_rpm_signature, which allows local rpm verification
without ingesting the whole header into a usable object.
* Add add_rpm_key, which enables a user to add a key to the rpm
keyring.
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
# -*-:python; coding:utf-8; -*-
|
|
# author: Louis Abel <label@rockylinux.org>
|
|
"""
|
|
Provides subprocess utilities
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
from pv2.util import error as err
|
|
|
|
# todo: remove python 3.6 checks. nodes won't be on el8.
|
|
|
|
def run_proc_foreground(command: list):
|
|
"""
|
|
Takes in the command in the form of a list and runs it via subprocess.
|
|
Everything should be in the foreground. The return is just for the exit
|
|
code.
|
|
"""
|
|
try:
|
|
processor = subprocess.run(args=command, check=False)
|
|
except Exception as exc:
|
|
raise err.GenericError(f'There was an error with your command: {exc}')
|
|
|
|
return processor
|
|
|
|
def run_proc_foreground_shell(command: str):
|
|
"""
|
|
Takes in the command in the form of a list and runs it via subprocess.
|
|
Everything should be in the foreground. The return is just for the exit
|
|
code.
|
|
"""
|
|
try:
|
|
processor = subprocess.run(args=command, shell=True, check=False)
|
|
except Exception as exc:
|
|
raise err.GenericError(f'There was an error with your command: {exc}')
|
|
|
|
return processor
|
|
|
|
def run_proc_no_output(command: list):
|
|
"""
|
|
Output will be stored in stdout and stderr as needed.
|
|
"""
|
|
try:
|
|
if sys.version_info <= (3, 6):
|
|
processor = subprocess.run(args=command, check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=True)
|
|
else:
|
|
processor = subprocess.run(args=command, check=False, capture_output=True,
|
|
text=True)
|
|
except Exception as exc:
|
|
raise err.GenericError(f'There was an error with your command: {exc}')
|
|
|
|
return processor
|
|
|
|
def run_proc_no_output_shell(command: str):
|
|
"""
|
|
Output will be stored in stdout and stderr as needed.
|
|
"""
|
|
try:
|
|
if sys.version_info <= (3, 6):
|
|
processor = subprocess.run(args=command, check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
shell=True)
|
|
else:
|
|
processor = subprocess.run(args=command, check=False, capture_output=True,
|
|
text=True, shell=True)
|
|
except Exception as exc:
|
|
raise err.GenericError(f'There was an error with your command: {exc}')
|
|
|
|
return processor
|
|
|
|
|
|
def popen_proc_no_output(command: list):
|
|
"""
|
|
This opens a process, but is non-blocking.
|
|
"""
|
|
try:
|
|
if sys.version_info <= (3, 6):
|
|
processor = subprocess.Popen(args=command, stdout=subprocess.PIPE,
|
|
universal_newlines=True)
|
|
else:
|
|
# pylint: disable=consider-using-with
|
|
processor = subprocess.Popen(args=command, stdout=subprocess.PIPE,
|
|
text=True)
|
|
except Exception as exc:
|
|
raise err.GenericError(f'There was an error with your command: {exc}')
|
|
|
|
return processor
|
|
|
|
def run_check_call(command: list) -> int:
|
|
"""
|
|
Runs subprocess check_call and returns an integer.
|
|
"""
|
|
env = os.environ
|
|
try:
|
|
subprocess.check_call(command, env=env)
|
|
except subprocess.CalledProcessError as exc:
|
|
sys.stderr.write(f'Run failed: {exc}\n')
|
|
return 1
|
|
return 0
|