robojerk
8c585e2e33
- Add MockAPIClient and MockEnvironment for external integration - Implement EnvironmentManager with full lifecycle support - Enhance plugin system with registry and BasePlugin class - Add comprehensive test suite and documentation - Include practical usage examples and plugin development guide
427 lines
14 KiB
Python
427 lines
14 KiB
Python
"""
|
|
Stable Python API for deb-mock integration
|
|
|
|
This module provides a stable, well-documented API for external tools
|
|
to integrate with deb-mock for build environment management.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import tempfile
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Dict, List, Any, Optional, Union
|
|
from contextlib import contextmanager
|
|
|
|
from .core import DebMock
|
|
from .config import Config
|
|
from .exceptions import ConfigurationError, ChrootError, SbuildError
|
|
|
|
|
|
class MockEnvironment:
|
|
"""Represents a mock environment for building packages"""
|
|
|
|
def __init__(self, name: str, deb_mock: DebMock):
|
|
self.name = name
|
|
self.deb_mock = deb_mock
|
|
self._active = False
|
|
|
|
def __enter__(self):
|
|
"""Context manager entry"""
|
|
self.activate()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Context manager exit"""
|
|
self.deactivate()
|
|
|
|
def activate(self):
|
|
"""Activate the environment"""
|
|
if not self.deb_mock.chroot_manager.chroot_exists(self.name):
|
|
raise ChrootError(f"Environment '{self.name}' does not exist")
|
|
self._active = True
|
|
|
|
def deactivate(self):
|
|
"""Deactivate the environment"""
|
|
self._active = False
|
|
|
|
def is_active(self) -> bool:
|
|
"""Check if environment is active"""
|
|
return self._active
|
|
|
|
def execute(self, command: Union[str, List[str]],
|
|
capture_output: bool = True,
|
|
check: bool = True) -> subprocess.CompletedProcess:
|
|
"""Execute a command in the environment"""
|
|
if not self._active:
|
|
raise RuntimeError("Environment is not active")
|
|
|
|
if isinstance(command, str):
|
|
command = command.split()
|
|
|
|
return self.deb_mock.chroot_manager.execute_in_chroot(
|
|
self.name, command, capture_output=capture_output
|
|
)
|
|
|
|
def install_packages(self, packages: List[str]) -> Dict[str, Any]:
|
|
"""Install packages in the environment"""
|
|
if not self._active:
|
|
raise RuntimeError("Environment is not active")
|
|
|
|
return self.deb_mock.install_packages(packages)
|
|
|
|
def copy_in(self, source: str, destination: str) -> None:
|
|
"""Copy files into the environment"""
|
|
if not self._active:
|
|
raise RuntimeError("Environment is not active")
|
|
|
|
self.deb_mock.chroot_manager.copy_to_chroot(source, destination, self.name)
|
|
|
|
def copy_out(self, source: str, destination: str) -> None:
|
|
"""Copy files out of the environment"""
|
|
if not self._active:
|
|
raise RuntimeError("Environment is not active")
|
|
|
|
self.deb_mock.chroot_manager.copy_from_chroot(source, destination, self.name)
|
|
|
|
def get_info(self) -> Dict[str, Any]:
|
|
"""Get information about the environment"""
|
|
return self.deb_mock.chroot_manager.get_chroot_info(self.name)
|
|
|
|
|
|
class MockAPIClient:
|
|
"""
|
|
Stable API client for deb-mock integration
|
|
|
|
This class provides a stable interface for external tools to interact
|
|
with deb-mock for build environment management.
|
|
"""
|
|
|
|
def __init__(self, config: Optional[Config] = None):
|
|
"""
|
|
Initialize the API client
|
|
|
|
Args:
|
|
config: Optional configuration object. If None, uses default config.
|
|
"""
|
|
if config is None:
|
|
config = Config.default()
|
|
|
|
self.config = config
|
|
self.deb_mock = DebMock(config)
|
|
self._environments = {}
|
|
|
|
def create_environment(self, name: str,
|
|
arch: str = None,
|
|
suite: str = None,
|
|
packages: List[str] = None) -> MockEnvironment:
|
|
"""
|
|
Create a new mock environment
|
|
|
|
Args:
|
|
name: Name for the environment
|
|
arch: Target architecture (defaults to config.architecture)
|
|
suite: Debian suite (defaults to config.suite)
|
|
packages: List of packages to install initially
|
|
|
|
Returns:
|
|
MockEnvironment instance
|
|
"""
|
|
try:
|
|
# Create the chroot environment
|
|
self.deb_mock.init_chroot(name, arch, suite)
|
|
|
|
# Install initial packages if specified
|
|
if packages:
|
|
self.deb_mock.install_packages(packages)
|
|
|
|
# Create environment wrapper
|
|
env = MockEnvironment(name, self.deb_mock)
|
|
self._environments[name] = env
|
|
|
|
return env
|
|
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to create environment '{name}': {e}")
|
|
|
|
def get_environment(self, name: str) -> MockEnvironment:
|
|
"""
|
|
Get an existing environment
|
|
|
|
Args:
|
|
name: Name of the environment
|
|
|
|
Returns:
|
|
MockEnvironment instance
|
|
|
|
Raises:
|
|
ValueError: If environment doesn't exist
|
|
"""
|
|
if name not in self._environments:
|
|
if not self.deb_mock.chroot_manager.chroot_exists(name):
|
|
raise ValueError(f"Environment '{name}' does not exist")
|
|
|
|
# Create wrapper for existing environment
|
|
env = MockEnvironment(name, self.deb_mock)
|
|
self._environments[name] = env
|
|
|
|
return self._environments[name]
|
|
|
|
def list_environments(self) -> List[str]:
|
|
"""List all available environments"""
|
|
return self.deb_mock.list_chroots()
|
|
|
|
def remove_environment(self, name: str) -> None:
|
|
"""Remove an environment"""
|
|
if name in self._environments:
|
|
del self._environments[name]
|
|
|
|
self.deb_mock.clean_chroot(name)
|
|
|
|
def build_package(self, source_package: str,
|
|
environment: str = None,
|
|
output_dir: str = None,
|
|
**kwargs) -> Dict[str, Any]:
|
|
"""
|
|
Build a package in a mock environment
|
|
|
|
Args:
|
|
source_package: Path to source package (.dsc file or directory)
|
|
environment: Environment name (uses default if None)
|
|
output_dir: Output directory for artifacts
|
|
**kwargs: Additional build options
|
|
|
|
Returns:
|
|
Build result dictionary
|
|
"""
|
|
if environment:
|
|
kwargs['chroot_name'] = environment
|
|
|
|
if output_dir:
|
|
kwargs['output_dir'] = output_dir
|
|
|
|
return self.deb_mock.build(source_package, **kwargs)
|
|
|
|
def build_parallel(self, source_packages: List[str],
|
|
max_workers: int = None,
|
|
**kwargs) -> List[Dict[str, Any]]:
|
|
"""
|
|
Build multiple packages in parallel
|
|
|
|
Args:
|
|
source_packages: List of source package paths
|
|
max_workers: Maximum number of parallel workers
|
|
**kwargs: Additional build options
|
|
|
|
Returns:
|
|
List of build results
|
|
"""
|
|
return self.deb_mock.build_parallel(source_packages, max_workers, **kwargs)
|
|
|
|
def build_chain(self, source_packages: List[str], **kwargs) -> List[Dict[str, Any]]:
|
|
"""
|
|
Build a chain of packages that depend on each other
|
|
|
|
Args:
|
|
source_packages: List of source package paths in dependency order
|
|
**kwargs: Additional build options
|
|
|
|
Returns:
|
|
List of build results
|
|
"""
|
|
return self.deb_mock.build_chain(source_packages, **kwargs)
|
|
|
|
@contextmanager
|
|
def environment(self, name: str,
|
|
arch: str = None,
|
|
suite: str = None,
|
|
packages: List[str] = None):
|
|
"""
|
|
Context manager for environment operations
|
|
|
|
Args:
|
|
name: Environment name
|
|
arch: Target architecture
|
|
suite: Debian suite
|
|
packages: Initial packages to install
|
|
|
|
Yields:
|
|
MockEnvironment instance
|
|
"""
|
|
env = None
|
|
try:
|
|
# Try to get existing environment first
|
|
try:
|
|
env = self.get_environment(name)
|
|
except ValueError:
|
|
# Create new environment if it doesn't exist
|
|
env = self.create_environment(name, arch, suite, packages)
|
|
|
|
env.activate()
|
|
yield env
|
|
|
|
finally:
|
|
if env:
|
|
env.deactivate()
|
|
|
|
def get_cache_stats(self) -> Dict[str, Any]:
|
|
"""Get cache statistics"""
|
|
return self.deb_mock.get_cache_stats()
|
|
|
|
def cleanup_caches(self) -> Dict[str, int]:
|
|
"""Clean up old cache files"""
|
|
return self.deb_mock.cleanup_caches()
|
|
|
|
def get_performance_summary(self) -> Dict[str, Any]:
|
|
"""Get performance monitoring summary"""
|
|
if hasattr(self.deb_mock, 'performance_monitor'):
|
|
return self.deb_mock.performance_monitor.get_performance_summary()
|
|
return {}
|
|
|
|
def export_metrics(self, output_file: str = None) -> str:
|
|
"""Export performance metrics to file"""
|
|
if hasattr(self.deb_mock, 'performance_monitor'):
|
|
return self.deb_mock.performance_monitor.export_metrics(output_file)
|
|
raise RuntimeError("Performance monitoring not available")
|
|
|
|
|
|
class MockConfigBuilder:
|
|
"""Builder class for creating mock configurations"""
|
|
|
|
def __init__(self):
|
|
self._config = {}
|
|
|
|
def environment(self, name: str) -> 'MockConfigBuilder':
|
|
"""Set environment name"""
|
|
self._config['chroot_name'] = name
|
|
return self
|
|
|
|
def architecture(self, arch: str) -> 'MockConfigBuilder':
|
|
"""Set target architecture"""
|
|
self._config['architecture'] = arch
|
|
return self
|
|
|
|
def suite(self, suite: str) -> 'MockConfigBuilder':
|
|
"""Set Debian suite"""
|
|
self._config['suite'] = suite
|
|
return self
|
|
|
|
def mirror(self, url: str) -> 'MockConfigBuilder':
|
|
"""Set package mirror URL"""
|
|
self._config['mirror'] = url
|
|
return self
|
|
|
|
def packages(self, packages: List[str]) -> 'MockConfigBuilder':
|
|
"""Set initial packages to install"""
|
|
self._config['chroot_additional_packages'] = packages
|
|
return self
|
|
|
|
def output_dir(self, path: str) -> 'MockConfigBuilder':
|
|
"""Set output directory"""
|
|
self._config['output_dir'] = path
|
|
return self
|
|
|
|
def cache_enabled(self, enabled: bool = True) -> 'MockConfigBuilder':
|
|
"""Enable/disable caching"""
|
|
self._config['use_root_cache'] = enabled
|
|
return self
|
|
|
|
def parallel_jobs(self, jobs: int) -> 'MockConfigBuilder':
|
|
"""Set number of parallel jobs"""
|
|
self._config['parallel_jobs'] = jobs
|
|
return self
|
|
|
|
def verbose(self, enabled: bool = True) -> 'MockConfigBuilder':
|
|
"""Enable verbose output"""
|
|
self._config['verbose'] = enabled
|
|
return self
|
|
|
|
def debug(self, enabled: bool = True) -> 'MockConfigBuilder':
|
|
"""Enable debug output"""
|
|
self._config['debug'] = enabled
|
|
return self
|
|
|
|
def build(self) -> Config:
|
|
"""Build the configuration object"""
|
|
return Config(**self._config)
|
|
|
|
|
|
# Convenience functions for common operations
|
|
def create_client(config: Optional[Config] = None) -> MockAPIClient:
|
|
"""Create a new API client"""
|
|
return MockAPIClient(config)
|
|
|
|
|
|
def create_config() -> MockConfigBuilder:
|
|
"""Create a new configuration builder"""
|
|
return MockConfigBuilder()
|
|
|
|
|
|
def quick_build(source_package: str,
|
|
environment: str = "debian-trixie-amd64",
|
|
arch: str = "amd64",
|
|
suite: str = "trixie") -> Dict[str, Any]:
|
|
"""
|
|
Quick build function for simple use cases
|
|
|
|
Args:
|
|
source_package: Path to source package
|
|
environment: Environment name
|
|
arch: Target architecture
|
|
suite: Debian suite
|
|
|
|
Returns:
|
|
Build result dictionary
|
|
"""
|
|
config = MockConfigBuilder().environment(environment).architecture(arch).suite(suite).build()
|
|
client = MockAPIClient(config)
|
|
|
|
return client.build_package(source_package)
|
|
|
|
|
|
# Example usage and integration patterns
|
|
def example_integration():
|
|
"""Example of how to use the API for integration"""
|
|
|
|
# Create a configuration
|
|
config = (MockConfigBuilder()
|
|
.environment("my-build-env")
|
|
.architecture("amd64")
|
|
.suite("trixie")
|
|
.mirror("http://deb.debian.org/debian/")
|
|
.packages(["build-essential", "devscripts"])
|
|
.cache_enabled(True)
|
|
.parallel_jobs(4)
|
|
.verbose(True)
|
|
.build())
|
|
|
|
# Create API client
|
|
client = MockAPIClient(config)
|
|
|
|
# Create environment
|
|
env = client.create_environment("my-build-env")
|
|
|
|
# Use environment context manager
|
|
with client.environment("my-build-env") as env:
|
|
# Install additional packages
|
|
env.install_packages(["cmake", "ninja-build"])
|
|
|
|
# Execute commands
|
|
result = env.execute(["ls", "-la", "/usr/bin"])
|
|
print(f"Command output: {result.stdout}")
|
|
|
|
# Copy files
|
|
env.copy_in("/local/source", "/build/source")
|
|
|
|
# Build package
|
|
build_result = client.build_package("/build/source", "my-build-env")
|
|
print(f"Build successful: {build_result['success']}")
|
|
|
|
# Cleanup
|
|
client.remove_environment("my-build-env")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Example usage
|
|
example_integration()
|