builder_peppermint_void/builder/core/command_runner.py
2025-04-28 12:17:18 +00:00

125 lines
4.9 KiB
Python

# -*- coding: utf-8 -*-
"""
SPDX-FileCopyrightText: 2023-2025 PeppermintOS Team
(peppermintosteam@proton.me)
SPDX-License-Identifier: GPL-3.0-or-later
This module provides a function for running commands, with support for chroot environments.
It includes functionality to execute shell commands and log their output.
Credits:
- PeppermintOS Team (peppermintosteam@proton.me) - Development and maintenance of the project.
License:
This code is distributed under the GNU General Public License version 3 or later (GPL-3.0-or-later).
For more details, please refer to the LICENSE file included in the project or visit:
https://www.gnu.org/licenses/gpl-3.0.html
"""
import os
import sys
import subprocess
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE_DIR)
try:
from builder.configs import logger_config
except ImportError as e:
print(f"Error importing necessary modules: {e}. Ensure your environment is set up correctly.")
sys.exit(1)
logger = logger_config.setup_logger('command_runner')
def run_command(command, env=None):
"""Executes a command and logs the output, allowing to pass environment variables."""
try:
if isinstance(command, list):
cmd_str = ' '.join(command)
else:
cmd_str = command
logger.info(f"=> Executing command: {cmd_str}")
process = subprocess.run(command, capture_output=True, text=True, check=True, env=env)
if process.stdout:
logger.debug(f"STDOUT:\n{process.stdout}")
if process.stderr:
logger.debug(f"STDERR:\n{process.stderr}")
return process
except subprocess.CalledProcessError as e:
logger.error(f"Error executing command: {cmd_str}")
logger.error(f"Retcode: {e.returncode}")
logger.error(f"STDOUT:\n{e.stdout}")
logger.error(f"STDERR:\n{e.stderr}")
raise
except FileNotFoundError as e:
logger.error(f"Error: Command not found: {command[0] if isinstance(command, list) else command}")
logger.error(f"Details: {e}")
raise
def run_command_in_target(rootfs_path, command, architecture, check=True, shell=False, cwd=None, env=None):
"""
Runs a command within the target root filesystem using chroot.
Sets necessary environment variables for the target architecture.
Args:
rootfs_path (str): Path to the target root filesystem directory.
command (list): The command and its arguments as a list (command path should be relative to target root).
architecture (str): The target architecture (e.g., 'aarch64').
check (bool): If True, raise CalledProcessError if the command returns a non-zero exit code.
shell (bool): If True, execute the command through the shell within the chroot.
cwd (str, optional): The current working directory *within the chroot*. (Note: Requires shell=True or `cd` prefix).
env (dict, optional): A dictionary of environment variables to set *inside the chroot*.
XBPS_ARCH is set automatically.
Returns:
subprocess.CompletedProcess: The result of the command execution.
Raises:
subprocess.CalledProcessError: If check is True and the command fails.
FileNotFoundError: If the chroot binary or the command within chroot is not found.
ValueError: If rootfs_path is invalid.
"""
if not rootfs_path or not os.path.isdir(rootfs_path):
logger.error(f"Invalid target root filesystem path for chroot: {rootfs_path}")
raise ValueError(f"Invalid target root filesystem path for chroot: {rootfs_path}")
chroot_env_list = [f"XBPS_ARCH={architecture}"]
if env:
for key, value in env.items():
if key != 'XBPS_ARCH':
chroot_env_list.append(f"{key}={value}")
full_command = ["chroot", rootfs_path, "env"] + chroot_env_list + command
logger.info(f"Executing command IN TARGET ({rootfs_path}): {' '.join(full_command)}")
try:
result = subprocess.run(
full_command,
capture_output=True,
text=True,
check=check,
shell=shell,
)
logger.debug(f"STDOUT:\n{result.stdout.strip()}")
if result.stderr.strip():
logger.debug(f"STDERR:\n{result.stderr.strip()}")
return result
except FileNotFoundError:
logger.error(f"Command not found in target chroot or chroot binary failed: {command[0] if command else 'N/A'}. Ensure base system is correctly installed.")
raise
except subprocess.CalledProcessError as e:
logger.error(f"Command failed in target with exit code {e.returncode}")
logger.error(f"Command: {' '.join(e.cmd)}")
logger.error(f"Stderr:\n{e.stderr.strip()}")
logger.error(f"Stdout:\n{e.stdout.strip()}")
raise
except Exception as e:
logger.error(f"An unexpected error occurred during command execution in target: {e}")
raise