robojerk
c51819c836
- Add complete pytest testing framework with conftest.py and test files - Add performance monitoring and benchmarking capabilities - Add plugin system with ccache plugin example - Add comprehensive documentation (API, deployment, testing, etc.) - Add Docker API wrapper for service deployment - Add advanced configuration examples - Remove old wget package file - Update core modules with enhanced functionality
436 lines
15 KiB
Python
436 lines
15 KiB
Python
"""
|
|
sbuild wrapper for deb-mock
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import grp
|
|
import pwd
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from .exceptions import SbuildError
|
|
|
|
|
|
class SbuildWrapper:
|
|
"""Wrapper around sbuild for standardized package building"""
|
|
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self._check_sbuild_requirements()
|
|
|
|
def _check_sbuild_requirements(self):
|
|
"""Check if sbuild requirements are met"""
|
|
# Check if sbuild is available
|
|
if not self._is_sbuild_available():
|
|
raise SbuildError("sbuild not found. Please install sbuild package.")
|
|
|
|
# Check if user is in sbuild group
|
|
if not self._is_user_in_sbuild_group():
|
|
raise SbuildError(
|
|
"User not in sbuild group. Please run 'sudo sbuild-adduser $USER' "
|
|
"and start a new shell session."
|
|
)
|
|
|
|
# Check if sbuild configuration exists
|
|
if not self._is_sbuild_configured():
|
|
self._setup_sbuild_config()
|
|
|
|
def _is_sbuild_available(self) -> bool:
|
|
"""Check if sbuild is available in PATH"""
|
|
try:
|
|
subprocess.run(["sbuild", "--version"], capture_output=True, check=True)
|
|
return True
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
return False
|
|
|
|
def _is_user_in_sbuild_group(self) -> bool:
|
|
"""Check if current user is in sbuild group"""
|
|
try:
|
|
current_user = pwd.getpwuid(os.getuid()).pw_name
|
|
sbuild_group = grp.getgrnam("sbuild")
|
|
return current_user in sbuild_group.gr_mem
|
|
except (KeyError, OSError):
|
|
return False
|
|
|
|
def _is_sbuild_configured(self) -> bool:
|
|
"""Check if sbuild configuration exists"""
|
|
config_paths = [
|
|
os.path.expanduser("~/.config/sbuild/config.pl"),
|
|
os.path.expanduser("~/.sbuildrc"),
|
|
"/etc/sbuild/sbuild.conf"
|
|
]
|
|
return any(os.path.exists(path) for path in config_paths)
|
|
|
|
def _setup_sbuild_config(self):
|
|
"""Setup basic sbuild configuration"""
|
|
config_dir = os.path.expanduser("~/.config/sbuild")
|
|
config_file = os.path.join(config_dir, "config.pl")
|
|
|
|
try:
|
|
os.makedirs(config_dir, exist_ok=True)
|
|
|
|
# Create minimal config
|
|
config_content = """#!/usr/bin/perl
|
|
# deb-mock sbuild configuration
|
|
$chroot_mode = "schroot";
|
|
$schroot = "schroot";
|
|
"""
|
|
|
|
with open(config_file, "w") as f:
|
|
f.write(config_content)
|
|
|
|
os.chmod(config_file, 0o644)
|
|
|
|
except Exception as e:
|
|
raise SbuildError(f"Failed to create sbuild configuration: {e}")
|
|
|
|
def build_package(
|
|
self,
|
|
source_package: str,
|
|
chroot_name: str = None,
|
|
output_dir: str = None,
|
|
**kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""Build a Debian source package using sbuild"""
|
|
|
|
if chroot_name is None:
|
|
chroot_name = self.config.chroot_name
|
|
if output_dir is None:
|
|
output_dir = self.config.get_output_path()
|
|
|
|
# Ensure output directory exists
|
|
try:
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
except Exception as e:
|
|
# If we can't create the directory, use a fallback
|
|
output_dir = os.path.join(tempfile.gettempdir(), "deb-mock-output")
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
# Validate source package
|
|
if not self._is_valid_source_package(source_package):
|
|
raise SbuildError(f"Invalid source package: {source_package}")
|
|
|
|
# Prepare sbuild command
|
|
cmd = self._prepare_sbuild_command(source_package, chroot_name, output_dir, **kwargs)
|
|
|
|
# Prepare environment variables
|
|
env = os.environ.copy()
|
|
if kwargs.get("build_env"):
|
|
env.update(kwargs["build_env"])
|
|
env.update(self.config.build_env)
|
|
|
|
# Create temporary log file
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as log_path:
|
|
log_file = log_path.name
|
|
|
|
try:
|
|
# Execute sbuild
|
|
result = self._execute_sbuild(cmd, log_file, env)
|
|
|
|
# Parse build results
|
|
build_info = self._parse_build_results(output_dir, log_file, result)
|
|
|
|
return build_info
|
|
|
|
finally:
|
|
# Clean up temporary log file
|
|
if os.path.exists(log_file):
|
|
os.unlink(log_file)
|
|
|
|
def _is_valid_source_package(self, source_package: str) -> bool:
|
|
"""Check if source package is valid"""
|
|
# Check if it's a directory with debian/control
|
|
if os.path.isdir(source_package):
|
|
control_file = os.path.join(source_package, "debian", "control")
|
|
return os.path.exists(control_file)
|
|
|
|
# Check if it's a .dsc file
|
|
if source_package.endswith(".dsc"):
|
|
return os.path.exists(source_package)
|
|
|
|
return False
|
|
|
|
def _prepare_sbuild_command(self, source_package: str, chroot_name: str, output_dir: str, **kwargs) -> List[str]:
|
|
"""Prepare the sbuild command with all necessary options"""
|
|
|
|
cmd = ["sbuild"]
|
|
|
|
# Basic options
|
|
cmd.extend(["--chroot", chroot_name])
|
|
cmd.extend(["--dist", self.config.suite])
|
|
cmd.extend(["--arch", self.config.architecture])
|
|
|
|
# Output options
|
|
cmd.extend(["--build-dir", output_dir])
|
|
|
|
# Build options
|
|
if kwargs.get("verbose", self.config.verbose):
|
|
cmd.append("--verbose")
|
|
|
|
if kwargs.get("debug", self.config.debug):
|
|
cmd.append("--debug")
|
|
|
|
# Additional build options from config
|
|
for option in self.config.build_options:
|
|
cmd.extend(option.split())
|
|
|
|
# Custom build options
|
|
if kwargs.get("build_options"):
|
|
for option in kwargs["build_options"]:
|
|
cmd.extend(option.split())
|
|
|
|
# Source package
|
|
cmd.append(source_package)
|
|
|
|
return cmd
|
|
|
|
def _execute_sbuild(self, cmd: List[str], log_path: str, env: Dict[str, str] = None) -> subprocess.CompletedProcess:
|
|
"""Execute sbuild command"""
|
|
try:
|
|
# Redirect output to log file
|
|
with open(log_path, "w") as log_file:
|
|
result = subprocess.run(
|
|
cmd,
|
|
stdout=log_file,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
check=True,
|
|
env=env,
|
|
)
|
|
return result
|
|
except subprocess.CalledProcessError as e:
|
|
# Read log file for error details
|
|
with open(log_path, "r") as log_file:
|
|
log_content = log_file.read()
|
|
raise SbuildError(f"sbuild failed: {e}\nLog output:\n{log_content}")
|
|
except FileNotFoundError:
|
|
raise SbuildError("sbuild not found. Please install sbuild package.")
|
|
|
|
def _parse_build_results(
|
|
self, output_dir: str, log_path: str, result: subprocess.CompletedProcess
|
|
) -> Dict[str, Any]:
|
|
"""Parse build results and collect artifacts"""
|
|
|
|
build_info = {
|
|
"success": True,
|
|
"output_dir": output_dir,
|
|
"log_file": log_path,
|
|
"artifacts": [],
|
|
"metadata": {},
|
|
}
|
|
|
|
# Collect build artifacts
|
|
artifacts = self._collect_artifacts(output_dir)
|
|
build_info["artifacts"] = artifacts
|
|
|
|
# Parse build metadata
|
|
metadata = self._parse_build_metadata(log_path, output_dir)
|
|
build_info["metadata"] = metadata
|
|
|
|
return build_info
|
|
|
|
def _collect_artifacts(self, output_dir: str) -> List[str]:
|
|
"""Collect build artifacts from output directory"""
|
|
artifacts = []
|
|
|
|
if not os.path.exists(output_dir):
|
|
return artifacts
|
|
|
|
# Look for .deb files
|
|
for deb_file in Path(output_dir).glob("*.deb"):
|
|
artifacts.append(str(deb_file))
|
|
|
|
# Look for .changes files
|
|
for changes_file in Path(output_dir).glob("*.changes"):
|
|
artifacts.append(str(changes_file))
|
|
|
|
# Look for .buildinfo files
|
|
for buildinfo_file in Path(output_dir).glob("*.buildinfo"):
|
|
artifacts.append(str(buildinfo_file))
|
|
|
|
return artifacts
|
|
|
|
def _parse_build_metadata(self, log_path: str, output_dir: str) -> Dict[str, Any]:
|
|
"""Parse build metadata from log and artifacts"""
|
|
metadata = {
|
|
"build_time": None,
|
|
"package_name": None,
|
|
"package_version": None,
|
|
"architecture": self.config.architecture,
|
|
"suite": self.config.suite,
|
|
"chroot": self.config.chroot_name,
|
|
"dependencies": [],
|
|
"build_dependencies": [],
|
|
}
|
|
|
|
# Parse log file for metadata
|
|
if os.path.exists(log_path):
|
|
with open(log_path, "r") as log_file:
|
|
log_content = log_file.read()
|
|
metadata.update(self._extract_metadata_from_log(log_content))
|
|
|
|
# Parse .changes file for additional metadata
|
|
changes_files = list(Path(output_dir).glob("*.changes"))
|
|
if changes_files:
|
|
metadata.update(self._parse_changes_file(changes_files[0]))
|
|
|
|
return metadata
|
|
|
|
def _extract_metadata_from_log(self, log_content: str) -> Dict[str, Any]:
|
|
"""Extract metadata from sbuild log content"""
|
|
metadata = {}
|
|
|
|
# Extract build time
|
|
import re
|
|
|
|
time_match = re.search(r"Build started at (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", log_content)
|
|
if time_match:
|
|
metadata["build_time"] = time_match.group(1)
|
|
|
|
# Extract package name and version
|
|
package_match = re.search(r"Building (\S+) \((\S+)\)", log_content)
|
|
if package_match:
|
|
metadata["package_name"] = package_match.group(1)
|
|
metadata["package_version"] = package_match.group(2)
|
|
|
|
return metadata
|
|
|
|
def _parse_changes_file(self, changes_file: Path) -> Dict[str, Any]:
|
|
"""Parse .changes file for metadata"""
|
|
metadata = {}
|
|
|
|
try:
|
|
with open(changes_file, "r") as f:
|
|
content = f.read()
|
|
|
|
lines = content.split("\n")
|
|
for line in lines:
|
|
if line.startswith("Source:"):
|
|
metadata["source_package"] = line.split(":", 1)[1].strip()
|
|
elif line.startswith("Version:"):
|
|
metadata["source_version"] = line.split(":", 1)[1].strip()
|
|
elif line.startswith("Architecture:"):
|
|
metadata["architectures"] = line.split(":", 1)[1].strip().split()
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
return metadata
|
|
|
|
def check_dependencies(self, source_package: str, chroot_name: str = None) -> Dict[str, Any]:
|
|
"""Check build dependencies for a source package"""
|
|
if chroot_name is None:
|
|
chroot_name = self.config.chroot_name
|
|
|
|
# Use dpkg-checkbuilddeps to check dependencies
|
|
cmd = ["schroot", "-c", chroot_name, "--", "dpkg-checkbuilddeps"]
|
|
|
|
try:
|
|
subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return {"satisfied": True, "missing": [], "conflicts": []}
|
|
except subprocess.CalledProcessError as e:
|
|
# Parse missing dependencies from error output
|
|
missing = self._parse_missing_dependencies(e.stderr)
|
|
return {"satisfied": False, "missing": missing, "conflicts": []}
|
|
|
|
def _parse_missing_dependencies(self, stderr: str) -> List[str]:
|
|
"""Parse missing dependencies from dpkg-checkbuilddeps output"""
|
|
missing = []
|
|
|
|
for line in stderr.split("\n"):
|
|
if "Unmet build dependencies:" in line:
|
|
# Extract package names from the line
|
|
import re
|
|
|
|
packages = re.findall(r"\b[a-zA-Z0-9][a-zA-Z0-9+\-\.]*\b", line)
|
|
missing.extend(packages)
|
|
|
|
return missing
|
|
|
|
def install_build_dependencies(self, dependencies: List[str], chroot_name: str = None) -> None:
|
|
"""Install build dependencies in the chroot"""
|
|
if chroot_name is None:
|
|
chroot_name = self.config.chroot_name
|
|
|
|
if not dependencies:
|
|
return
|
|
|
|
cmd = [
|
|
"schroot",
|
|
"-c",
|
|
chroot_name,
|
|
"--",
|
|
"apt-get",
|
|
"install",
|
|
"-y",
|
|
] + dependencies
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True)
|
|
except subprocess.CalledProcessError as e:
|
|
raise SbuildError(f"Failed to install build dependencies: {e}")
|
|
|
|
def update_chroot(self, chroot_name: str = None) -> None:
|
|
"""Update the chroot to ensure it's current"""
|
|
if chroot_name is None:
|
|
chroot_name = self.config.chroot_name
|
|
|
|
try:
|
|
# Update package lists
|
|
cmd = ["schroot", "-c", chroot_name, "--", "apt-get", "update"]
|
|
subprocess.run(cmd, check=True)
|
|
|
|
# Upgrade packages
|
|
cmd = ["schroot", "-c", chroot_name, "--", "apt-get", "upgrade", "-y"]
|
|
subprocess.run(cmd, check=True)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
raise SbuildError(f"Failed to update chroot: {e}")
|
|
|
|
def get_chroot_info(self, chroot_name: str = None) -> Dict[str, Any]:
|
|
"""Get information about a chroot"""
|
|
if chroot_name is None:
|
|
chroot_name = self.config.chroot_name
|
|
|
|
info = {
|
|
"name": chroot_name,
|
|
"status": "unknown",
|
|
"architecture": None,
|
|
"distribution": None,
|
|
"packages": [],
|
|
}
|
|
|
|
try:
|
|
# Get chroot status
|
|
cmd = ["schroot", "-i", "-c", chroot_name]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
|
|
# Parse schroot info output
|
|
for line in result.stdout.split("\n"):
|
|
if ":" in line:
|
|
key, value = line.split(":", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
|
|
if key == "Status":
|
|
info["status"] = value
|
|
elif key == "Architecture":
|
|
info["architecture"] = value
|
|
elif key == "Distribution":
|
|
info["distribution"] = value
|
|
|
|
# Get package count
|
|
cmd = ["schroot", "-c", chroot_name, "--", "dpkg", "-l", "|", "wc", "-l"]
|
|
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
try:
|
|
info["package_count"] = int(result.stdout.strip())
|
|
except ValueError:
|
|
pass
|
|
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
|
|
return info
|