diff --git a/deb_mock/__init__.py b/deb_mock/__init__.py index cf916bc..b9c3857 100644 --- a/deb_mock/__init__.py +++ b/deb_mock/__init__.py @@ -13,10 +13,27 @@ from .chroot import ChrootManager from .config import Config from .core import DebMock from .sbuild import SbuildWrapper +from .api import MockAPIClient, MockEnvironment, MockConfigBuilder, create_client, create_config, quick_build +from .environment_manager import EnvironmentManager, EnvironmentInfo, BuildResult, create_environment_manager __all__ = [ + # Core classes "DebMock", "Config", "ChrootManager", "SbuildWrapper", + + # API classes + "MockAPIClient", + "MockEnvironment", + "MockConfigBuilder", + "EnvironmentManager", + "EnvironmentInfo", + "BuildResult", + + # Convenience functions + "create_client", + "create_config", + "create_environment_manager", + "quick_build", ] diff --git a/deb_mock/api.py b/deb_mock/api.py new file mode 100644 index 0000000..5ecf1f6 --- /dev/null +++ b/deb_mock/api.py @@ -0,0 +1,427 @@ +""" +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() diff --git a/deb_mock/environment_manager.py b/deb_mock/environment_manager.py new file mode 100644 index 0000000..30d6a0d --- /dev/null +++ b/deb_mock/environment_manager.py @@ -0,0 +1,475 @@ +""" +Environment Management API for deb-mock + +This module provides comprehensive environment management capabilities +for external tools integrating with deb-mock. +""" + +import os +import sys +import json +import tempfile +import subprocess +import shutil +from pathlib import Path +from typing import Dict, List, Any, Optional, Union, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime + +from .core import DebMock +from .config import Config +from .exceptions import ConfigurationError, ChrootError, SbuildError + + +@dataclass +class EnvironmentInfo: + """Information about a mock environment""" + name: str + architecture: str + suite: str + status: str + created: Optional[datetime] = None + modified: Optional[datetime] = None + size: int = 0 + packages_installed: List[str] = None + mounts: List[Dict[str, str]] = None + + +@dataclass +class BuildResult: + """Result of a build operation""" + success: bool + artifacts: List[str] + output_dir: str + log_file: str + metadata: Dict[str, Any] + error: Optional[str] = None + duration: float = 0.0 + + +class EnvironmentManager: + """ + Comprehensive environment management for deb-mock + + This class provides a high-level interface for managing mock environments, + executing commands, and collecting artifacts. + """ + + def __init__(self, config: Optional[Config] = None): + """Initialize the environment manager""" + if config is None: + config = Config.default() + + self.config = config + self.deb_mock = DebMock(config) + self._active_environments = {} + + def create_environment(self, + name: str, + arch: str = None, + suite: str = None, + packages: List[str] = None, + force: bool = False) -> EnvironmentInfo: + """ + Create a new mock environment + + Args: + name: Name for the environment + arch: Target architecture + suite: Debian suite + packages: Initial packages to install + force: Force creation even if environment exists + + Returns: + EnvironmentInfo object + """ + if not force and self.environment_exists(name): + raise ValueError(f"Environment '{name}' already exists") + + # Remove existing environment if force is True + if force and self.environment_exists(name): + self.remove_environment(name) + + 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) + + # Get environment info + info = self.get_environment_info(name) + self._active_environments[name] = info + + return info + + except Exception as e: + raise RuntimeError(f"Failed to create environment '{name}': {e}") + + def environment_exists(self, name: str) -> bool: + """Check if an environment exists""" + return self.deb_mock.chroot_manager.chroot_exists(name) + + def get_environment_info(self, name: str) -> EnvironmentInfo: + """Get detailed information about an environment""" + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + # Get basic chroot info + chroot_info = self.deb_mock.chroot_manager.get_chroot_info(name) + + # Get installed packages + packages = self._get_installed_packages(name) + + # Get mount information + mounts = self.deb_mock.chroot_manager.list_mounts(name) + + return EnvironmentInfo( + name=name, + architecture=self.config.architecture, + suite=self.config.suite, + status=chroot_info.get('status', 'unknown'), + created=chroot_info.get('created'), + modified=chroot_info.get('modified'), + size=chroot_info.get('size', 0), + packages_installed=packages, + mounts=mounts + ) + + def list_environments(self) -> List[EnvironmentInfo]: + """List all available environments""" + environments = [] + + for name in self.deb_mock.list_chroots(): + try: + info = self.get_environment_info(name) + environments.append(info) + except Exception as e: + print(f"Warning: Failed to get info for environment '{name}': {e}") + + return environments + + def remove_environment(self, name: str, force: bool = False) -> None: + """Remove an environment""" + if not self.environment_exists(name): + if not force: + raise ValueError(f"Environment '{name}' does not exist") + return + + # Clean up active environment tracking + if name in self._active_environments: + del self._active_environments[name] + + # Remove the chroot + self.deb_mock.clean_chroot(name) + + def update_environment(self, name: str) -> None: + """Update packages in an environment""" + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + self.deb_mock.update_chroot(name) + + def execute_command(self, + name: str, + command: Union[str, List[str]], + capture_output: bool = True, + check: bool = True, + timeout: Optional[int] = None) -> subprocess.CompletedProcess: + """ + Execute a command in an environment + + Args: + name: Environment name + command: Command to execute + capture_output: Whether to capture output + check: Whether to raise exception on non-zero exit + timeout: Command timeout in seconds + + Returns: + CompletedProcess object + """ + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + if isinstance(command, str): + command = command.split() + + # Prepare command with timeout if specified + if timeout: + command = ['timeout', str(timeout)] + command + + try: + result = self.deb_mock.chroot_manager.execute_in_chroot( + name, command, capture_output=capture_output + ) + + if check and result.returncode != 0: + raise subprocess.CalledProcessError( + result.returncode, command, result.stdout, result.stderr + ) + + return result + + except subprocess.CalledProcessError as e: + if check: + raise + return e + + def install_packages(self, name: str, packages: List[str]) -> Dict[str, Any]: + """Install packages in an environment""" + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + return self.deb_mock.install_packages(packages) + + def copy_files(self, + name: str, + source: str, + destination: str, + direction: str = "in") -> None: + """ + Copy files to/from an environment + + Args: + name: Environment name + source: Source path + destination: Destination path + direction: "in" to copy into environment, "out" to copy out + """ + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + if direction == "in": + self.deb_mock.chroot_manager.copy_to_chroot(source, destination, name) + elif direction == "out": + self.deb_mock.chroot_manager.copy_from_chroot(source, destination, name) + else: + raise ValueError("Direction must be 'in' or 'out'") + + def collect_artifacts(self, + name: str, + source_patterns: List[str] = None, + output_dir: str = None) -> List[str]: + """ + Collect build artifacts from an environment + + Args: + name: Environment name + source_patterns: File patterns to search for + output_dir: Output directory for artifacts + + Returns: + List of collected artifact paths + """ + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + if source_patterns is None: + source_patterns = [ + '*.deb', + '*.changes', + '*.buildinfo', + '*.dsc', + '*.tar.*', + '*.orig.tar.*', + '*.debian.tar.*' + ] + + if output_dir is None: + output_dir = tempfile.mkdtemp(prefix='deb-mock-artifacts-') + + os.makedirs(output_dir, exist_ok=True) + artifacts = [] + + for pattern in source_patterns: + # Find files matching pattern + result = self.execute_command( + name, ['find', '/build', '-name', pattern, '-type', 'f'], + capture_output=True, check=False + ) + + if result.returncode == 0: + for line in result.stdout.strip().split('\n'): + if line.strip(): + source_path = line.strip() + filename = os.path.basename(source_path) + dest_path = os.path.join(output_dir, filename) + + # Copy artifact + self.copy_files(name, source_path, dest_path, "out") + artifacts.append(dest_path) + + return artifacts + + def build_package(self, + name: str, + source_package: str, + output_dir: str = None, + **kwargs) -> BuildResult: + """ + Build a package in an environment + + Args: + name: Environment name + source_package: Path to source package + output_dir: Output directory for build artifacts + **kwargs: Additional build options + + Returns: + BuildResult object + """ + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + start_time = datetime.now() + + try: + # Set chroot name for build + kwargs['chroot_name'] = name + if output_dir: + kwargs['output_dir'] = output_dir + + # Build the package + result = self.deb_mock.build(source_package, **kwargs) + + # Calculate duration + duration = (datetime.now() - start_time).total_seconds() + + return BuildResult( + success=result.get('success', False), + artifacts=result.get('artifacts', []), + output_dir=result.get('output_dir', ''), + log_file=result.get('log_file', ''), + metadata=result.get('metadata', {}), + duration=duration + ) + + except Exception as e: + duration = (datetime.now() - start_time).total_seconds() + return BuildResult( + success=False, + artifacts=[], + output_dir=output_dir or '', + log_file='', + metadata={}, + error=str(e), + duration=duration + ) + + @contextmanager + def environment(self, + name: str, + arch: str = None, + suite: str = None, + packages: List[str] = None, + create_if_missing: bool = True) -> Iterator[EnvironmentInfo]: + """ + Context manager for environment operations + + Args: + name: Environment name + arch: Target architecture + suite: Debian suite + packages: Initial packages to install + create_if_missing: Create environment if it doesn't exist + + Yields: + EnvironmentInfo object + """ + env_info = None + created = False + + try: + # Get or create environment + if self.environment_exists(name): + env_info = self.get_environment_info(name) + elif create_if_missing: + env_info = self.create_environment(name, arch, suite, packages) + created = True + else: + raise ValueError(f"Environment '{name}' does not exist") + + yield env_info + + finally: + # Clean up if we created the environment + if created and env_info: + try: + self.remove_environment(name) + except Exception as e: + print(f"Warning: Failed to cleanup environment '{name}': {e}") + + def _get_installed_packages(self, name: str) -> List[str]: + """Get list of installed packages in environment""" + try: + result = self.execute_command( + name, ['dpkg', '-l'], capture_output=True, check=False + ) + + if result.returncode == 0: + packages = [] + for line in result.stdout.split('\n'): + if line.startswith('ii'): + parts = line.split() + if len(parts) >= 3: + packages.append(parts[1]) + return packages + + except Exception: + pass + + return [] + + def export_environment(self, name: str, output_path: str) -> None: + """Export environment to a tar archive""" + if not self.environment_exists(name): + raise ValueError(f"Environment '{name}' does not exist") + + chroot_path = self.deb_mock.config.get_chroot_path() + + # Create tar archive + subprocess.run([ + 'tar', '-czf', output_path, '-C', chroot_path, '.' + ], check=True) + + def import_environment(self, name: str, archive_path: str) -> None: + """Import environment from a tar archive""" + if self.environment_exists(name): + raise ValueError(f"Environment '{name}' already exists") + + # Create environment directory + chroot_path = os.path.join(self.config.chroot_dir, name) + os.makedirs(chroot_path, exist_ok=True) + + # Extract archive + subprocess.run([ + 'tar', '-xzf', archive_path, '-C', chroot_path + ], check=True) + + # Create schroot configuration + self.deb_mock.chroot_manager._create_schroot_config( + name, chroot_path, self.config.architecture, self.config.suite + ) + + +# Convenience functions +def create_environment_manager(config: Optional[Config] = None) -> EnvironmentManager: + """Create a new environment manager""" + return EnvironmentManager(config) + + +def quick_environment(name: str = "quick-build", + arch: str = "amd64", + suite: str = "trixie", + packages: List[str] = None) -> EnvironmentManager: + """Create a quick environment manager with default settings""" + config = Config( + chroot_name=name, + architecture=arch, + suite=suite, + chroot_additional_packages=packages or [] + ) + return EnvironmentManager(config) diff --git a/deb_mock/plugins/example_plugin.py b/deb_mock/plugins/example_plugin.py new file mode 100644 index 0000000..cc84e1e --- /dev/null +++ b/deb_mock/plugins/example_plugin.py @@ -0,0 +1,234 @@ +""" +Example plugin for deb-mock + +This plugin demonstrates how to create custom plugins for deb-mock +and provides examples of common plugin patterns. +""" + +import os +import logging +from typing import Dict, Any, List + +from ..plugin import BasePlugin, HookStages + + +class ExamplePlugin(BasePlugin): + """ + Example plugin demonstrating deb-mock plugin capabilities + + This plugin shows how to: + - Register hooks for different stages + - Access configuration and deb-mock context + - Perform custom operations during build lifecycle + - Log information and errors + """ + + # Plugin metadata + requires_api_version = "1.0" + plugin_name = "example" + plugin_version = "1.0.0" + plugin_description = "Example plugin for deb-mock" + + def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + # Plugin-specific configuration + self.enabled = self.get_config('enabled', True) + self.log_level = self.get_config('log_level', 'INFO') + self.custom_option = self.get_config('custom_option', 'default_value') + + # Setup logging + self.logger.setLevel(getattr(logging, self.log_level.upper())) + + self.log_info(f"ExamplePlugin initialized with config: {config}") + + def _register_hooks(self): + """Register hooks for different build stages""" + + # Chroot lifecycle hooks + self.plugin_manager.add_hook(HookStages.PRECHROOT_INIT, self.prechroot_init) + self.plugin_manager.add_hook(HookStages.POSTCHROOT_INIT, self.postchroot_init) + self.plugin_manager.add_hook(HookStages.PRECHROOT_CLEAN, self.prechroot_clean) + self.plugin_manager.add_hook(HookStages.POSTCHROOT_CLEAN, self.postchroot_clean) + + # Build lifecycle hooks + self.plugin_manager.add_hook(HookStages.PREBUILD, self.prebuild) + self.plugin_manager.add_hook(HookStages.POSTBUILD, self.postbuild) + self.plugin_manager.add_hook(HookStages.BUILD_START, self.build_start) + self.plugin_manager.add_hook(HookStages.BUILD_END, self.build_end) + + # Package management hooks + self.plugin_manager.add_hook(HookStages.PRE_INSTALL_DEPS, self.pre_install_deps) + self.plugin_manager.add_hook(HookStages.POST_INSTALL_DEPS, self.post_install_deps) + + # Mount management hooks + self.plugin_manager.add_hook(HookStages.PRE_MOUNT, self.pre_mount) + self.plugin_manager.add_hook(HookStages.POST_MOUNT, self.post_mount) + + # Cache management hooks + self.plugin_manager.add_hook(HookStages.PRE_CACHE_CREATE, self.pre_cache_create) + self.plugin_manager.add_hook(HookStages.POST_CACHE_CREATE, self.post_cache_create) + + # Error handling hooks + self.plugin_manager.add_hook(HookStages.ON_ERROR, self.on_error) + self.plugin_manager.add_hook(HookStages.ON_WARNING, self.on_warning) + + self.log_debug("Registered all hooks for ExamplePlugin") + + def prechroot_init(self, chroot_name: str, **kwargs): + """Called before chroot initialization""" + self.log_info(f"Pre-chroot init for {chroot_name}") + + # Example: Create custom directory structure + if self.get_config('create_custom_dirs', False): + custom_dirs = self.get_config('custom_dirs', ['/build/custom']) + for dir_path in custom_dirs: + self.log_debug(f"Would create directory: {dir_path}") + + def postchroot_init(self, chroot_name: str, **kwargs): + """Called after chroot initialization""" + self.log_info(f"Post-chroot init for {chroot_name}") + + # Example: Install additional packages + extra_packages = self.get_config('extra_packages', []) + if extra_packages: + self.log_info(f"Installing extra packages: {extra_packages}") + try: + result = self.deb_mock.install_packages(extra_packages) + if result.get('success', False): + self.log_info("Extra packages installed successfully") + else: + self.log_warning(f"Failed to install extra packages: {result}") + except Exception as e: + self.log_error(f"Error installing extra packages: {e}") + + def prechroot_clean(self, chroot_name: str, **kwargs): + """Called before chroot cleanup""" + self.log_info(f"Pre-chroot clean for {chroot_name}") + + # Example: Backup important files before cleanup + if self.get_config('backup_before_clean', False): + self.log_info("Backing up important files before cleanup") + # Implementation would backup files here + + def postchroot_clean(self, chroot_name: str, **kwargs): + """Called after chroot cleanup""" + self.log_info(f"Post-chroot clean for {chroot_name}") + + def prebuild(self, source_package: str, **kwargs): + """Called before package build""" + self.log_info(f"Pre-build for {source_package}") + + # Example: Validate source package + if not os.path.exists(source_package): + self.log_error(f"Source package not found: {source_package}") + raise FileNotFoundError(f"Source package not found: {source_package}") + + # Example: Check build dependencies + if self.get_config('check_deps', True): + self.log_info("Checking build dependencies") + # Implementation would check dependencies here + + def postbuild(self, build_result: Dict[str, Any], source_package: str, **kwargs): + """Called after package build""" + self.log_info(f"Post-build for {source_package}") + + success = build_result.get('success', False) + if success: + self.log_info("Build completed successfully") + artifacts = build_result.get('artifacts', []) + self.log_info(f"Generated {len(artifacts)} artifacts") + else: + self.log_error("Build failed") + + def build_start(self, source_package: str, chroot_name: str, **kwargs): + """Called when build starts""" + self.log_info(f"Build started: {source_package} in {chroot_name}") + + # Example: Set build environment variables + if self.get_config('set_build_env', False): + env_vars = self.get_config('build_env_vars', {}) + for key, value in env_vars.items(): + os.environ[key] = value + self.log_debug(f"Set environment variable: {key}={value}") + + def build_end(self, build_result: Dict[str, Any], source_package: str, chroot_name: str, **kwargs): + """Called when build ends""" + self.log_info(f"Build ended: {source_package} in {chroot_name}") + + # Example: Collect build statistics + if self.get_config('collect_stats', True): + stats = { + 'package': source_package, + 'chroot': chroot_name, + 'success': build_result.get('success', False), + 'artifacts_count': len(build_result.get('artifacts', [])), + 'duration': build_result.get('duration', 0) + } + self.log_info(f"Build statistics: {stats}") + + def pre_install_deps(self, dependencies: List[str], chroot_name: str, **kwargs): + """Called before installing dependencies""" + self.log_info(f"Pre-install deps: {dependencies} in {chroot_name}") + + # Example: Filter dependencies + if self.get_config('filter_deps', False): + filtered_deps = [dep for dep in dependencies if not dep.startswith('lib')] + self.log_info(f"Filtered dependencies: {filtered_deps}") + return filtered_deps + + def post_install_deps(self, dependencies: List[str], chroot_name: str, **kwargs): + """Called after installing dependencies""" + self.log_info(f"Post-install deps: {dependencies} in {chroot_name}") + + def pre_mount(self, mount_type: str, mount_path: str, chroot_name: str, **kwargs): + """Called before mounting""" + self.log_debug(f"Pre-mount: {mount_type} at {mount_path} in {chroot_name}") + + def post_mount(self, mount_type: str, mount_path: str, chroot_name: str, **kwargs): + """Called after mounting""" + self.log_debug(f"Post-mount: {mount_type} at {mount_path} in {chroot_name}") + + def pre_cache_create(self, cache_path: str, chroot_name: str, **kwargs): + """Called before creating cache""" + self.log_info(f"Pre-cache create: {cache_path} for {chroot_name}") + + def post_cache_create(self, cache_path: str, chroot_name: str, **kwargs): + """Called after creating cache""" + self.log_info(f"Post-cache create: {cache_path} for {chroot_name}") + + def on_error(self, error: Exception, stage: str, **kwargs): + """Called when an error occurs""" + self.log_error(f"Error in {stage}: {error}") + + # Example: Send error notification + if self.get_config('notify_on_error', False): + self.log_info("Would send error notification") + + def on_warning(self, warning: str, stage: str, **kwargs): + """Called when a warning occurs""" + self.log_warning(f"Warning in {stage}: {warning}") + + def get_plugin_info(self) -> Dict[str, Any]: + """Return plugin information""" + return { + 'name': self.plugin_name, + 'version': self.plugin_version, + 'description': self.plugin_description, + 'enabled': self.enabled, + 'config': { + 'log_level': self.log_level, + 'custom_option': self.custom_option + } + } + + +# Plugin initialization function (required by deb-mock) +def init(plugin_manager, config, deb_mock): + """ + Initialize the plugin + + This function is called by deb-mock when the plugin is loaded. + It should create and return an instance of the plugin class. + """ + return ExamplePlugin(plugin_manager, config, deb_mock) diff --git a/deb_mock/plugins/registry.py b/deb_mock/plugins/registry.py index a4047bb..3fde544 100644 --- a/deb_mock/plugins/registry.py +++ b/deb_mock/plugins/registry.py @@ -1,361 +1,413 @@ """ -Plugin Registry for Deb-Mock Plugin System +Plugin registry and management for deb-mock -This module provides the plugin registration and management functionality -for the Deb-Mock plugin system, inspired by Fedora's Mock plugin architecture. +This module provides a centralized registry for managing deb-mock plugins, +including discovery, loading, and lifecycle management. """ +import os +import sys import importlib +import importlib.util import logging -from typing import Any, Dict, Optional, Type +from pathlib import Path +from typing import Dict, List, Any, Optional, Type, Callable +from dataclasses import dataclass +from datetime import datetime from .base import BasePlugin +from ..exceptions import PluginError -logger = logging.getLogger(__name__) + +@dataclass +class PluginInfo: + """Information about a registered plugin""" + name: str + version: str + description: str + author: str + requires_api_version: str + plugin_class: Type[BasePlugin] + init_function: Callable + file_path: str + loaded_at: datetime + enabled: bool = True + config: Dict[str, Any] = None class PluginRegistry: """ - Manages plugin registration and instantiation. - - This class provides the functionality for registering plugin classes - and creating plugin instances, following Mock's plugin system pattern. + Central registry for deb-mock plugins + + This class manages plugin discovery, loading, and lifecycle. """ - - def __init__(self): - """Initialize the plugin registry.""" - self.plugins: Dict[str, Type[BasePlugin]] = {} - self.plugin_metadata: Dict[str, Dict[str, Any]] = {} - - # Auto-register built-in plugins - self._register_builtin_plugins() - - def register( - self, - plugin_name: str, - plugin_class: Type[BasePlugin], - metadata: Optional[Dict[str, Any]] = None, - ) -> None: + + def __init__(self, plugin_dirs: List[str] = None): """ - Register a plugin class. - + Initialize the plugin registry + Args: - plugin_name: Name of the plugin - plugin_class: Plugin class to register - metadata: Optional metadata about the plugin - - Raises: - ValueError: If plugin_name is already registered - TypeError: If plugin_class is not a subclass of BasePlugin + plugin_dirs: List of directories to search for plugins """ - if not issubclass(plugin_class, BasePlugin): - raise TypeError("Plugin class must inherit from BasePlugin") - - if plugin_name in self.plugins: - raise ValueError(f"Plugin '{plugin_name}' is already registered") - - self.plugins[plugin_name] = plugin_class - self.plugin_metadata[plugin_name] = metadata or {} - - logger.debug(f"Registered plugin '{plugin_name}' with class {plugin_class.__name__}") - - def unregister(self, plugin_name: str) -> bool: + self.logger = logging.getLogger(__name__) + + # Default plugin directories + self.plugin_dirs = plugin_dirs or [ + '/usr/share/deb-mock/plugins', + '/usr/local/share/deb-mock/plugins', + os.path.join(os.path.expanduser('~'), '.local', 'share', 'deb-mock', 'plugins'), + os.path.join(os.getcwd(), 'plugins') + ] + + # Plugin storage + self._plugins: Dict[str, PluginInfo] = {} + self._loaded_plugins: Dict[str, BasePlugin] = {} + + # API version compatibility + self.current_api_version = "1.0" + self.min_api_version = "1.0" + self.max_api_version = "1.0" + + def discover_plugins(self) -> List[PluginInfo]: """ - Unregister a plugin. - + Discover available plugins in plugin directories + + Returns: + List of discovered plugin information + """ + discovered = [] + + for plugin_dir in self.plugin_dirs: + if not os.path.exists(plugin_dir): + continue + + self.logger.debug(f"Scanning plugin directory: {plugin_dir}") + + for file_path in Path(plugin_dir).glob("*.py"): + if file_path.name.startswith('_'): + continue + + try: + plugin_info = self._load_plugin_info(file_path) + if plugin_info: + discovered.append(plugin_info) + self.logger.debug(f"Discovered plugin: {plugin_info.name}") + + except Exception as e: + self.logger.warning(f"Failed to load plugin from {file_path}: {e}") + + return discovered + + def _load_plugin_info(self, file_path: Path) -> Optional[PluginInfo]: + """ + Load plugin information from a file + + Args: + file_path: Path to the plugin file + + Returns: + PluginInfo object or None if not a valid plugin + """ + try: + # Load module + spec = importlib.util.spec_from_file_location(file_path.stem, file_path) + if not spec or not spec.loader: + return None + + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + # Check if it's a valid plugin + if not hasattr(module, 'init'): + return None + + # Get plugin metadata + plugin_name = getattr(module, 'plugin_name', file_path.stem) + plugin_version = getattr(module, 'plugin_version', '1.0.0') + plugin_description = getattr(module, 'plugin_description', 'No description') + plugin_author = getattr(module, 'plugin_author', 'Unknown') + requires_api_version = getattr(module, 'requires_api_version', '1.0') + + # Check API version compatibility + if not self._is_api_version_compatible(requires_api_version): + self.logger.warning( + f"Plugin {plugin_name} requires API version {requires_api_version}, " + f"but current version is {self.current_api_version}" + ) + return None + + # Get plugin class + plugin_class = getattr(module, 'Plugin', None) + if not plugin_class: + # Look for classes that inherit from BasePlugin + for attr_name in dir(module): + attr = getattr(module, attr_name) + if (isinstance(attr, type) and + issubclass(attr, BasePlugin) and + attr != BasePlugin): + plugin_class = attr + break + + if not plugin_class: + return None + + return PluginInfo( + name=plugin_name, + version=plugin_version, + description=plugin_description, + author=plugin_author, + requires_api_version=requires_api_version, + plugin_class=plugin_class, + init_function=module.init, + file_path=str(file_path), + loaded_at=datetime.now(), + enabled=True + ) + + except Exception as e: + self.logger.error(f"Error loading plugin info from {file_path}: {e}") + return None + + def _is_api_version_compatible(self, required_version: str) -> bool: + """ + Check if a plugin's required API version is compatible + + Args: + required_version: Required API version string + + Returns: + True if compatible, False otherwise + """ + try: + required_major, required_minor = map(int, required_version.split('.')) + current_major, current_minor = map(int, self.current_api_version.split('.')) + + # Same major version, minor version can be higher + return required_major == current_major and required_minor <= current_minor + + except ValueError: + return False + + def register_plugin(self, plugin_info: PluginInfo) -> None: + """ + Register a plugin in the registry + + Args: + plugin_info: Plugin information + """ + self._plugins[plugin_info.name] = plugin_info + self.logger.info(f"Registered plugin: {plugin_info.name} v{plugin_info.version}") + + def unregister_plugin(self, plugin_name: str) -> None: + """ + Unregister a plugin from the registry + Args: plugin_name: Name of the plugin to unregister - - Returns: - True if plugin was unregistered, False if not found """ - if plugin_name not in self.plugins: - return False - - del self.plugins[plugin_name] - del self.plugin_metadata[plugin_name] - - logger.debug(f"Unregistered plugin '{plugin_name}'") - return True - - def get_plugin_class(self, plugin_name: str) -> Optional[Type[BasePlugin]]: + if plugin_name in self._plugins: + del self._plugins[plugin_name] + self.logger.info(f"Unregistered plugin: {plugin_name}") + + def get_plugin(self, plugin_name: str) -> Optional[PluginInfo]: """ - Get a registered plugin class. - + Get plugin information by name + Args: plugin_name: Name of the plugin - + Returns: - Plugin class if found, None otherwise + PluginInfo object or None if not found """ - return self.plugins.get(plugin_name) - - def get_plugins(self) -> Dict[str, Type[BasePlugin]]: + return self._plugins.get(plugin_name) + + def list_plugins(self) -> List[PluginInfo]: """ - Get all registered plugins. - + List all registered plugins + Returns: - Dictionary of registered plugin names and classes + List of plugin information """ - return self.plugins.copy() - - def get_plugin_names(self) -> list: + return list(self._plugins.values()) + + def list_enabled_plugins(self) -> List[PluginInfo]: """ - Get list of registered plugin names. - + List enabled plugins + Returns: - List of registered plugin names + List of enabled plugin information """ - return list(self.plugins.keys()) - - def create(self, plugin_name: str, config: Any, hook_manager: Any) -> Optional[BasePlugin]: + return [plugin for plugin in self._plugins.values() if plugin.enabled] + + def enable_plugin(self, plugin_name: str) -> None: """ - Create a plugin instance. - + Enable a plugin + Args: - plugin_name: Name of the plugin to create - config: Configuration object - hook_manager: Hook manager instance - - Returns: - Plugin instance if successful, None if plugin not found + plugin_name: Name of the plugin to enable """ - plugin_class = self.get_plugin_class(plugin_name) - if not plugin_class: - logger.warning(f"Plugin '{plugin_name}' not found") - return None - + if plugin_name in self._plugins: + self._plugins[plugin_name].enabled = True + self.logger.info(f"Enabled plugin: {plugin_name}") + + def disable_plugin(self, plugin_name: str) -> None: + """ + Disable a plugin + + Args: + plugin_name: Name of the plugin to disable + """ + if plugin_name in self._plugins: + self._plugins[plugin_name].enabled = False + self.logger.info(f"Disabled plugin: {plugin_name}") + + def load_plugin(self, plugin_name: str, plugin_manager, config: Dict[str, Any], deb_mock) -> BasePlugin: + """ + Load a plugin instance + + Args: + plugin_name: Name of the plugin to load + plugin_manager: Plugin manager instance + config: Plugin configuration + deb_mock: DebMock instance + + Returns: + Loaded plugin instance + + Raises: + PluginError: If plugin cannot be loaded + """ + if plugin_name not in self._plugins: + raise PluginError(f"Plugin '{plugin_name}' not found in registry") + + plugin_info = self._plugins[plugin_name] + + if not plugin_info.enabled: + raise PluginError(f"Plugin '{plugin_name}' is disabled") + + if plugin_name in self._loaded_plugins: + return self._loaded_plugins[plugin_name] + try: - plugin_instance = plugin_class(config, hook_manager) - logger.debug(f"Created plugin instance '{plugin_name}'") + # Create plugin instance + plugin_instance = plugin_info.init_function(plugin_manager, config, deb_mock) + + if not isinstance(plugin_instance, BasePlugin): + raise PluginError(f"Plugin '{plugin_name}' did not return a BasePlugin instance") + + # Store loaded plugin + self._loaded_plugins[plugin_name] = plugin_instance + + self.logger.info(f"Loaded plugin: {plugin_name}") return plugin_instance + except Exception as e: - logger.error(f"Failed to create plugin '{plugin_name}': {e}") - return None - - def create_all_enabled(self, config: Any, hook_manager: Any) -> Dict[str, BasePlugin]: + raise PluginError(f"Failed to load plugin '{plugin_name}': {e}") + + def unload_plugin(self, plugin_name: str) -> None: """ - Create instances of all enabled plugins. - + Unload a plugin instance + Args: - config: Configuration object - hook_manager: Hook manager instance - - Returns: - Dictionary of plugin names and instances + plugin_name: Name of the plugin to unload """ - enabled_plugins = {} - - for plugin_name in self.get_plugin_names(): - plugin_instance = self.create(plugin_name, config, hook_manager) - if plugin_instance and plugin_instance.enabled: - enabled_plugins[plugin_name] = plugin_instance - - logger.debug(f"Created {len(enabled_plugins)} enabled plugin instances") - return enabled_plugins - - def get_plugin_info(self, plugin_name: str) -> Dict[str, Any]: + if plugin_name in self._loaded_plugins: + del self._loaded_plugins[plugin_name] + self.logger.info(f"Unloaded plugin: {plugin_name}") + + def reload_plugin(self, plugin_name: str, plugin_manager, config: Dict[str, Any], deb_mock) -> BasePlugin: """ - Get information about a registered plugin. - + Reload a plugin + Args: - plugin_name: Name of the plugin - + plugin_name: Name of the plugin to reload + plugin_manager: Plugin manager instance + config: Plugin configuration + deb_mock: DebMock instance + Returns: - Dictionary with plugin information + Reloaded plugin instance """ - if plugin_name not in self.plugins: - return {"error": f'Plugin "{plugin_name}" not found'} - - plugin_class = self.plugins[plugin_name] - metadata = self.plugin_metadata[plugin_name] - - info = { - "name": plugin_name, - "class": plugin_class.__name__, - "module": plugin_class.__module__, - "metadata": metadata, - "docstring": plugin_class.__doc__ or "No documentation available", - } - - return info - - def get_all_plugin_info(self) -> Dict[str, Dict[str, Any]]: - """ - Get information about all registered plugins. - - Returns: - Dictionary mapping plugin names to their information - """ - return {name: self.get_plugin_info(name) for name in self.get_plugin_names()} - - def load_plugin_from_module(self, module_name: str, plugin_name: str) -> bool: - """ - Load a plugin from a module. - - Args: - module_name: Name of the module to load - plugin_name: Name of the plugin class in the module - - Returns: - True if plugin was loaded successfully, False otherwise - """ - try: - module = importlib.import_module(module_name) - plugin_class = getattr(module, plugin_name) - - # Use module name as plugin name if not specified - self.register(plugin_name, plugin_class) - return True - - except ImportError as e: - logger.error(f"Failed to import module '{module_name}': {e}") - return False - except AttributeError as e: - logger.error(f"Plugin class '{plugin_name}' not found in module '{module_name}': {e}") - return False - except Exception as e: - logger.error(f"Failed to load plugin from '{module_name}.{plugin_name}': {e}") - return False - - def load_plugins_from_config(self, config: Any) -> Dict[str, BasePlugin]: - """ - Load plugins based on configuration. - - Args: - config: Configuration object with plugin settings - - Returns: - Dictionary of loaded plugin instances - """ - loaded_plugins = {} - - if not hasattr(config, "plugins") or not config.plugins: - return loaded_plugins - - for plugin_name, plugin_config in config.plugins.items(): - if not isinstance(plugin_config, dict): - continue - - if plugin_config.get("enabled", False): - # Try to load from built-in plugins first - plugin_instance = self.create(plugin_name, config, None) - if plugin_instance: - loaded_plugins[plugin_name] = plugin_instance - else: - # Try to load from external module - module_name = plugin_config.get("module") - if module_name: - if self.load_plugin_from_module(module_name, plugin_name): - plugin_instance = self.create(plugin_name, config, None) - if plugin_instance: - loaded_plugins[plugin_name] = plugin_instance - - return loaded_plugins - - def _register_builtin_plugins(self) -> None: - """Register built-in plugins.""" - try: - # Import and register built-in plugins - from .bind_mount import BindMountPlugin - from .compress_logs import CompressLogsPlugin - from .root_cache import RootCachePlugin - from .tmpfs import TmpfsPlugin - - self.register( - "bind_mount", - BindMountPlugin, - { - "description": "Mount host directories into chroot", - "hooks": ["mount_root", "postumount"], - "builtin": True, - }, - ) - - self.register( - "compress_logs", - CompressLogsPlugin, - { - "description": "Compress build logs to save space", - "hooks": ["process_logs"], - "builtin": True, - }, - ) - - self.register( - "root_cache", - RootCachePlugin, - { - "description": "Root cache management for faster builds", - "hooks": [ - "preinit", - "postinit", - "postchroot", - "postshell", - "clean", - ], - "builtin": True, - }, - ) - - self.register( - "tmpfs", - TmpfsPlugin, - { - "description": "Use tmpfs for faster I/O operations", - "hooks": ["mount_root", "postumount"], - "builtin": True, - }, - ) - - logger.debug("Registered built-in plugins") - - except ImportError as e: - logger.warning(f"Some built-in plugins could not be loaded: {e}") - except Exception as e: - logger.warning(f"Error registering built-in plugins: {e}") - + self.unload_plugin(plugin_name) + return self.load_plugin(plugin_name, plugin_manager, config, deb_mock) + def get_plugin_statistics(self) -> Dict[str, Any]: """ - Get statistics about registered plugins. - + Get plugin registry statistics + Returns: Dictionary with plugin statistics """ - stats = { - "total_plugins": len(self.plugins), - "builtin_plugins": len([p for p in self.plugin_metadata.values() if p.get("builtin", False)]), - "external_plugins": len([p for p in self.plugin_metadata.values() if not p.get("builtin", False)]), - "plugins_by_hook": {}, + total_plugins = len(self._plugins) + enabled_plugins = len(self.list_enabled_plugins()) + loaded_plugins = len(self._loaded_plugins) + + return { + 'total_plugins': total_plugins, + 'enabled_plugins': enabled_plugins, + 'loaded_plugins': loaded_plugins, + 'disabled_plugins': total_plugins - enabled_plugins, + 'api_version': self.current_api_version, + 'plugin_directories': self.plugin_dirs } - - # Count plugins by hook usage - for plugin_name, metadata in self.plugin_metadata.items(): - hooks = metadata.get("hooks", []) - for hook in hooks: - if hook not in stats["plugins_by_hook"]: - stats["plugins_by_hook"][hook] = [] - stats["plugins_by_hook"][hook].append(plugin_name) - - return stats - - def validate_plugin_config(self, plugin_name: str, config: Any) -> bool: + + def validate_plugin_dependencies(self, plugin_name: str) -> List[str]: """ - Validate plugin configuration. - + Validate plugin dependencies + Args: - plugin_name: Name of the plugin - config: Configuration to validate - + plugin_name: Name of the plugin to validate + Returns: - True if configuration is valid, False otherwise + List of missing dependencies """ - if plugin_name not in self.plugins: - return False + if plugin_name not in self._plugins: + return [f"Plugin '{plugin_name}' not found"] + + plugin_info = self._plugins[plugin_name] + missing_deps = [] + + # Check if plugin file exists + if not os.path.exists(plugin_info.file_path): + missing_deps.append(f"Plugin file not found: {plugin_info.file_path}") + + # Check Python dependencies + try: + spec = importlib.util.spec_from_file_location(plugin_name, plugin_info.file_path) + if spec and spec.loader: + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + except Exception as e: + missing_deps.append(f"Failed to load plugin module: {e}") + + return missing_deps - # Basic validation - plugins can override this method - plugin_class = self.plugins[plugin_name] - if hasattr(plugin_class, "validate_config"): - return plugin_class.validate_config(config) - return True +# Global plugin registry instance +_global_registry = None + + +def get_plugin_registry() -> PluginRegistry: + """Get the global plugin registry instance""" + global _global_registry + if _global_registry is None: + _global_registry = PluginRegistry() + return _global_registry + + +def discover_plugins() -> List[PluginInfo]: + """Discover all available plugins""" + registry = get_plugin_registry() + return registry.discover_plugins() + + +def register_plugin(plugin_info: PluginInfo) -> None: + """Register a plugin in the global registry""" + registry = get_plugin_registry() + registry.register_plugin(plugin_info) + + +def get_plugin(plugin_name: str) -> Optional[PluginInfo]: + """Get plugin information by name""" + registry = get_plugin_registry() + return registry.get_plugin(plugin_name) \ No newline at end of file diff --git a/docs/API.md b/docs/API.md index 30c830c..1e6f284 100644 --- a/docs/API.md +++ b/docs/API.md @@ -1,1117 +1,508 @@ -# Deb-Mock API Reference +# deb-mock API Documentation -## Overview - -This document provides a comprehensive reference for the `deb-mock` API, including all classes, methods, and their usage. The API is designed to be intuitive and follows Python best practices. +This document provides comprehensive documentation for the deb-mock Python API, designed for external tools and integrations. ## Table of Contents -1. [Core Classes](#core-classes) -2. [Configuration](#configuration) -3. [Chroot Management](#chroot-management) -4. [Cache Management](#cache-management) -5. [Sbuild Integration](#sbuild-integration) -6. [Plugin System](#plugin-system) -7. [Performance Monitoring](#performance-monitoring) -8. [Benchmarking](#benchmarking) -9. [Exceptions](#exceptions) -10. [CLI Interface](#cli-interface) -11. [Examples](#examples) +- [Quick Start](#quick-start) +- [API Overview](#api-overview) +- [Configuration](#configuration) +- [Environment Management](#environment-management) +- [Command Execution](#command-execution) +- [Package Building](#package-building) +- [Advanced Features](#advanced-features) +- [Error Handling](#error-handling) +- [Examples](#examples) -## Core Classes - -### DebMock - -The main class that orchestrates the build process. +## Quick Start ```python -from deb_mock.core import DebMock +from deb_mock import create_client, MockConfigBuilder -# Initialize with configuration -deb_mock = DebMock(config) +# Create a configuration +config = (MockConfigBuilder() + .environment("my-build-env") + .architecture("amd64") + .suite("trixie") + .packages(["build-essential", "cmake"]) + .build()) + +# Create API client +client = create_client(config) + +# Create environment +env = client.create_environment("my-build-env") + +# Use environment +with client.environment("my-build-env") as env: + env.install_packages(["ninja-build"]) + result = env.execute(["cmake", "--version"]) + print(f"CMake version: {result.stdout}") # Build a package -result = deb_mock.build("package-name", "source-path") - -# Get status -status = deb_mock.get_status() - -# Cleanup -deb_mock.cleanup() +result = client.build_package("/path/to/package.dsc", "my-build-env") +print(f"Build successful: {result['success']}") ``` -#### Methods +## API Overview -##### `__init__(config)` +The deb-mock API consists of several main components: -Initialize a new DebMock instance. +### Core Classes -**Parameters:** -- `config`: Configuration object or path to config file +- **`MockAPIClient`**: Main API client for high-level operations +- **`MockEnvironment`**: Represents a mock environment +- **`EnvironmentManager`**: Comprehensive environment management +- **`MockConfigBuilder`**: Fluent configuration builder -**Example:** -```python -from deb_mock.config import Config +### Data Classes -config = Config.from_file("deb-mock.yaml") -deb_mock = DebMock(config) -``` +- **`EnvironmentInfo`**: Information about a mock environment +- **`BuildResult`**: Result of a build operation -##### `build(package_name, source_path, **kwargs)` +### Convenience Functions -Build a package from source. - -**Parameters:** -- `package_name`: Name of the package to build -- `source_path`: Path to the source package -- `**kwargs`: Additional build options - -**Options:** -- `chroot_name`: Custom chroot name -- `keep_chroot`: Whether to keep chroot after build -- `parallel`: Enable parallel building -- `clean_cache`: Clean cache before building - -**Returns:** -- `BuildResult`: Object containing build results - -**Example:** -```python -result = deb_mock.build( - "my-package", - "/path/to/source", - chroot_name="custom-chroot", - keep_chroot=True, - parallel=True -) - -print(f"Build successful: {result.success}") -print(f"Build duration: {result.duration:.2f}s") -print(f"Output files: {result.output_files}") -``` - -##### `build_chain(packages, **kwargs)` - -Build multiple packages sequentially. - -**Parameters:** -- `packages`: List of package specifications -- `**kwargs`: Additional build options - -**Example:** -```python -packages = [ - {"name": "package1", "source": "/path/to/pkg1"}, - {"name": "package2", "source": "/path/to/pkg2"}, - {"name": "package3", "source": "/path/to/pkg3"} -] - -results = deb_mock.build_chain(packages, parallel=False) -``` - -##### `build_parallel(packages, **kwargs)` - -Build multiple packages in parallel. - -**Parameters:** -- `packages`: List of package specifications -- `**kwargs**: Additional build options - -**Example:** -```python -packages = [ - {"name": "package1", "source": "/path/to/pkg1"}, - {"name": "package2", "source": "/path/to/pkg2"}, - {"name": "package3", "source": "/path/to/pkg3"} -] - -results = deb_mock.build_parallel(packages, max_workers=4) -``` - -##### `get_status()` - -Get current status of the deb-mock instance. - -**Returns:** -- `dict`: Status information - -**Example:** -```python -status = deb_mock.get_status() -print(f"Active chroots: {status['chroot_manager']['chroots']}") -print(f"Cache status: {status['cache_manager']['caches']}") -print(f"Performance monitoring: {status['performance_monitor']['enabled']}") -``` - -##### `cleanup()` - -Clean up resources and temporary files. - -**Example:** -```python -deb_mock.cleanup() -``` +- **`create_client()`**: Create a new API client +- **`create_config()`**: Create a configuration builder +- **`quick_build()`**: Quick build function for simple use cases ## Configuration -### Config +### Using MockConfigBuilder -Configuration management class. +The `MockConfigBuilder` provides a fluent interface for creating configurations: ```python -from deb_mock.config import Config +from deb_mock import MockConfigBuilder -# Load from file -config = Config.from_file("deb-mock.yaml") - -# Load from dictionary -config = Config.from_dict({ - "chroot": {"suite": "trixie", "architecture": "amd64"}, - "cache": {"enabled": True} -}) - -# Access configuration values -suite = config.chroot.suite -cache_enabled = config.cache.enabled +config = (MockConfigBuilder() + .environment("debian-trixie-amd64") + .architecture("amd64") + .suite("trixie") + .mirror("http://deb.debian.org/debian/") + .packages(["build-essential", "devscripts"]) + .output_dir("/tmp/build-output") + .cache_enabled(True) + .parallel_jobs(4) + .verbose(True) + .build()) ``` -#### Configuration Schema +### Configuration Options -```yaml -chroot: - base_dir: /var/lib/deb-mock/chroots - suite: trixie - architecture: amd64 - mirror: http://deb.debian.org/debian/ - components: [main, contrib, non-free] +| Method | Description | Default | +|--------|-------------|---------| +| `.environment(name)` | Environment name | "bookworm-amd64" | +| `.architecture(arch)` | Target architecture | "amd64" | +| `.suite(suite)` | Debian suite | "bookworm" | +| `.mirror(url)` | Package mirror URL | Debian official | +| `.packages(list)` | Initial packages to install | [] | +| `.output_dir(path)` | Output directory | "./output" | +| `.cache_enabled(bool)` | Enable caching | True | +| `.parallel_jobs(int)` | Number of parallel jobs | 4 | +| `.verbose(bool)` | Verbose output | False | +| `.debug(bool)` | Debug output | False | -cache: - enabled: true - base_dir: /var/cache/deb-mock - ccache_size_mb: 2048 - root_cache_size_mb: 5120 - package_cache_size_mb: 1024 +### Direct Configuration -sbuild: - enabled: true - user: sbuild - group: sbuild - chroot_suffix: -sbuild - build_user: buildd - -performance: - enable_performance_monitoring: true - performance_metrics_dir: ./performance-metrics - performance_retention_days: 30 - performance_auto_optimization: true - performance_benchmark_iterations: 10 - performance_reporting: true - -plugins: - enabled: true - plugin_dir: ./plugins - auto_load: true - -parallel: - enabled: true - max_parallel_builds: 4 - max_parallel_chroots: 8 - -mounts: - proc: true - sys: true - dev: true - tmpfs: true - bind_mounts: - - source: /var/cache/apt/archives - target: /var/cache/apt/archives - options: [ro] - overlay_mounts: - - source: /var/cache/deb-mock/overlay - target: /var/cache/deb-mock/overlay - -uid_management: - enabled: true - create_users: true - copy_host_users: true - privilege_escalation: true -``` - -## Chroot Management - -### ChrootManager - -Manages chroot environments for builds. +You can also create configurations directly: ```python -from deb_mock.chroot import ChrootManager +from deb_mock import Config -chroot_manager = ChrootManager(config) - -# Create a new chroot -chroot = chroot_manager.create_chroot("trixie-amd64") - -# List existing chroots -chroots = chroot_manager.list_chroots() - -# Clean up chroot -chroot_manager.cleanup_chroot("trixie-amd64") -``` - -#### Methods - -##### `create_chroot(name, suite=None, architecture=None)` - -Create a new chroot environment. - -**Parameters:** -- `name`: Name for the chroot -- `suite`: Debian suite (defaults to config) -- `architecture`: Architecture (defaults to config) - -**Returns:** -- `Chroot`: Chroot instance - -**Example:** -```python -chroot = chroot_manager.create_chroot( - "trixie-amd64", +config = Config( + chroot_name="my-env", + architecture="amd64", suite="trixie", - architecture="amd64" + mirror="http://deb.debian.org/debian/", + chroot_additional_packages=["build-essential"], + use_root_cache=True, + parallel_jobs=4 ) ``` -##### `get_chroot(name)` +## Environment Management -Get an existing chroot by name. +### Creating Environments -**Parameters:** -- `name`: Chroot name - -**Returns:** -- `Chroot`: Chroot instance or None - -**Example:** ```python -chroot = chroot_manager.get_chroot("trixie-amd64") -if chroot: - print(f"Chroot exists: {chroot.path}") +from deb_mock import create_client + +client = create_client() + +# Create environment with specific settings +env = client.create_environment( + name="build-env", + arch="amd64", + suite="trixie", + packages=["build-essential", "cmake"] +) ``` -##### `list_chroots()` +### Environment Information -List all available chroots. - -**Returns:** -- `list`: List of chroot names - -**Example:** ```python -chroots = chroot_manager.list_chroots() -for chroot_name in chroots: - print(f"Available chroot: {chroot_name}") +# Get environment info +info = env.get_info() +print(f"Environment: {info.name}") +print(f"Architecture: {info.architecture}") +print(f"Suite: {info.suite}") +print(f"Status: {info.status}") ``` -##### `cleanup_chroot(name)` +### Listing Environments -Remove a chroot environment. - -**Parameters:** -- `name`: Chroot name - -**Example:** ```python -chroot_manager.cleanup_chroot("trixie-amd64") +# List all environments +environments = client.list_environments() +for env_name in environments: + print(f"Environment: {env_name}") ``` -## Cache Management - -### CacheManager - -Manages build caches for improved performance. +### Removing Environments ```python -from deb_mock.cache import CacheManager - -cache_manager = CacheManager(config) - -# Create cache -cache = cache_manager.create_cache("package-name") - -# Restore cache -success = cache_manager.restore_cache("package-name") - -# Clean cache -cache_manager.clean_cache("package-name") +# Remove environment +client.remove_environment("build-env") ``` -#### Methods +## Command Execution -##### `create_cache(package_name)` +### Basic Command Execution -Create a cache for a package. - -**Parameters:** -- `package_name`: Name of the package - -**Returns:** -- `Cache`: Cache instance - -**Example:** ```python -cache = cache_manager.create_cache("my-package") +# Execute a command +result = env.execute(["ls", "-la", "/usr/bin"]) +print(f"Return code: {result.returncode}") +print(f"Output: {result.stdout}") + +# Execute string command +result = env.execute("dpkg -l | grep build-essential") ``` -##### `restore_cache(package_name)` +### Command Options -Restore cache for a package. - -**Parameters:** -- `package_name`: Name of the package - -**Returns:** -- `bool`: True if restoration was successful - -**Example:** ```python -if cache_manager.restore_cache("my-package"): - print("Cache restored successfully") -else: - print("Cache restoration failed") +# Execute with options +result = env.execute( + ["apt", "list", "--installed"], + capture_output=True, + check=True, + timeout=30 +) ``` -##### `clean_cache(package_name)` +### Using Environment Manager -Clean cache for a package. - -**Parameters:** -- `package_name`: Name of the package - -**Example:** ```python -cache_manager.clean_cache("my-package") +from deb_mock import create_environment_manager + +manager = create_environment_manager() + +# Execute command in environment +result = manager.execute_command( + "build-env", + ["cmake", "--version"], + capture_output=True +) ``` -## Sbuild Integration +## Package Building -### SbuildWrapper - -Wrapper for sbuild integration. +### Building Single Packages ```python -from deb_mock.sbuild import SbuildWrapper - -sbuild = SbuildWrapper(config) - -# Check sbuild availability -if sbuild.is_available(): - print("Sbuild is available") -else: - print("Sbuild is not available") - # Build a package -result = sbuild.build_package("source.dsc", chroot_name="trixie-amd64") -``` - -#### Methods - -##### `is_available()` - -Check if sbuild is available and properly configured. - -**Returns:** -- `bool`: True if sbuild is available - -**Example:** -```python -if sbuild.is_available(): - print("Sbuild is ready to use") -else: - print("Sbuild needs configuration") -``` - -##### `build_package(source_path, chroot_name=None, **kwargs)` - -Build a package using sbuild. - -**Parameters:** -- `source_path`: Path to source package -- `chroot_name`: Chroot to use for building -- `**kwargs`: Additional build options - -**Returns:** -- `SbuildResult`: Build result - -**Example:** -```python -result = sbuild.build_package( - "/path/to/package.dsc", - chroot_name="trixie-amd64", - verbose=True +result = client.build_package( + source_package="/path/to/package.dsc", + environment="build-env", + output_dir="/tmp/build-output" ) -if result.success: - print(f"Build successful: {result.output_files}") -else: - print(f"Build failed: {result.error}") +print(f"Build successful: {result['success']}") +print(f"Artifacts: {result['artifacts']}") ``` -##### `update_chroot(chroot_name)` +### Parallel Building -Update a chroot with latest packages. - -**Parameters:** -- `chroot_name`: Name of the chroot to update - -**Returns:** -- `bool`: True if update was successful - -**Example:** ```python -if sbuild.update_chroot("trixie-amd64"): - print("Chroot updated successfully") +# Build multiple packages in parallel +packages = [ + "/path/to/package1.dsc", + "/path/to/package2.dsc", + "/path/to/package3.dsc" +] + +results = client.build_parallel( + packages, + max_workers=2 +) + +for i, result in enumerate(results): + print(f"Package {i+1}: {'Success' if result['success'] else 'Failed'}") ``` -## Plugin System - -### PluginManager - -Manages plugins and their lifecycle. +### Chain Building ```python -from deb_mock.plugin import PluginManager +# Build packages in dependency order +chain = [ + "/path/to/base-package.dsc", + "/path/to/depends-on-base.dsc", + "/path/to/final-package.dsc" +] -plugin_manager = PluginManager(config) - -# Load plugins -plugin_manager.load_plugins() - -# Get plugin information -plugins = plugin_manager.list_plugins() - -# Enable/disable plugins -plugin_manager.enable_plugin("ccache_plugin") -plugin_manager.disable_plugin("debug_plugin") +results = client.build_chain(chain) ``` -#### Methods +## Advanced Features -##### `load_plugins()` +### Context Managers -Load all available plugins. - -**Example:** ```python -plugin_manager.load_plugins() -print(f"Loaded {len(plugin_manager.plugins)} plugins") +# Use environment context manager +with client.environment("build-env") as env: + env.install_packages(["cmake", "ninja-build"]) + result = env.execute(["cmake", "--version"]) + # Environment is automatically cleaned up ``` -##### `list_plugins()` +### File Operations -List all available plugins. - -**Returns:** -- `list`: List of plugin information - -**Example:** ```python -plugins = plugin_manager.list_plugins() -for plugin in plugins: - print(f"Plugin: {plugin['name']} - Status: {plugin['status']}") +# Copy files into environment +env.copy_in("/local/source", "/build/source") + +# Copy files out of environment +env.copy_out("/build/artifacts", "/local/artifacts") ``` -##### `enable_plugin(plugin_name)` +### Artifact Collection -Enable a specific plugin. - -**Parameters:** -- `plugin_name`: Name of the plugin to enable - -**Returns:** -- `bool`: True if plugin was enabled - -**Example:** ```python -if plugin_manager.enable_plugin("ccache_plugin"): - print("CCache plugin enabled") +from deb_mock import create_environment_manager + +manager = create_environment_manager() + +# Collect build artifacts +artifacts = manager.collect_artifacts( + "build-env", + source_patterns=["*.deb", "*.changes"], + output_dir="/tmp/artifacts" +) + +print(f"Collected {len(artifacts)} artifacts") ``` -##### `disable_plugin(plugin_name)` - -Disable a specific plugin. - -**Parameters:** -- `plugin_name`: Name of the plugin to disable - -**Returns:** -- `bool`: True if plugin was disabled - -**Example:** -```python -if plugin_manager.disable_plugin("debug_plugin"): - print("Debug plugin disabled") -``` - -### BasePlugin - -Base class for creating custom plugins. +### Performance Monitoring ```python -from deb_mock.plugin import BasePlugin, HookStages - -class MyCustomPlugin(BasePlugin): - name = "my_custom_plugin" - version = "1.0.0" - description = "A custom plugin for deb-mock" - - def __init__(self): - super().__init__() - self.register_hook(HookStages.PRE_BUILD, self.pre_build_hook) - self.register_hook(HookStages.POST_BUILD, self.post_build_hook) - - def pre_build_hook(self, context): - print("Pre-build hook executed") - return context - - def post_build_hook(self, context): - print("Post-build hook executed") - return context -``` - -#### Hook Stages - -- `PRE_CHROOT_CREATE`: Before chroot creation -- `POST_CHROOT_CREATE`: After chroot creation -- `PRE_BUILD`: Before build starts -- `POST_BUILD`: After build completes -- `PRE_CACHE_RESTORE`: Before cache restoration -- `POST_CACHE_RESTORE`: After cache restoration -- `PRE_CLEANUP`: Before cleanup -- `POST_CLEANUP`: After cleanup - -## Performance Monitoring - -### PerformanceMonitor - -Monitors and tracks performance metrics. - -```python -from deb_mock.performance import PerformanceMonitor - -monitor = PerformanceMonitor(config) - -# Start monitoring -monitor.start_monitoring() - -# Monitor an operation -with monitor.monitor_operation("build_operation"): - # Your operation here - pass +# Get cache statistics +cache_stats = client.get_cache_stats() +print(f"Cache stats: {cache_stats}") # Get performance summary -summary = monitor.get_performance_summary() +perf_summary = client.get_performance_summary() +print(f"Performance: {perf_summary}") + +# Export metrics +metrics_file = client.export_metrics("/tmp/metrics.json") +print(f"Metrics exported to: {metrics_file}") ``` -#### Methods +## Error Handling -##### `start_monitoring()` - -Start system monitoring. - -**Example:** -```python -monitor.start_monitoring() -``` - -##### `stop_monitoring()` - -Stop system monitoring. - -**Example:** -```python -monitor.stop_monitoring() -``` - -##### `monitor_operation(operation_name)` - -Context manager for monitoring operations. - -**Parameters:** -- `operation_name`: Name of the operation to monitor - -**Example:** -```python -with monitor.monitor_operation("package_build"): - # Build operation - result = build_package() -``` - -##### `get_performance_summary()` - -Get a summary of performance metrics. - -**Returns:** -- `dict`: Performance summary - -**Example:** -```python -summary = monitor.get_performance_summary() -print(f"Total operations: {summary['total_operations']}") -print(f"Average duration: {summary['average_duration']:.2f}s") -``` - -##### `create_build_profile(build_id, package_name, architecture, suite)` - -Create a build performance profile. - -**Parameters:** -- `build_id`: Unique build identifier -- `package_name`: Name of the package -- `architecture`: Build architecture -- `suite`: Debian suite - -**Returns:** -- `BuildProfile`: Build profile instance - -**Example:** -```python -profile = monitor.create_build_profile( - "build-123", - "my-package", - "amd64", - "trixie" -) -``` - -### PerformanceOptimizer - -Provides optimization suggestions based on performance data. +### Common Exceptions ```python -from deb_mock.performance import PerformanceOptimizer - -optimizer = PerformanceOptimizer(config) - -# Analyze build performance -analysis = optimizer.analyze_build_performance(profile) - -# Get optimization suggestions -suggestions = optimizer.get_optimization_suggestions(profile) -``` - -#### Methods - -##### `analyze_build_performance(profile)` - -Analyze build performance for optimization opportunities. - -**Parameters:** -- `profile`: BuildProfile instance - -**Returns:** -- `dict`: Performance analysis - -**Example:** -```python -analysis = optimizer.analyze_build_performance(profile) -print(f"Optimization score: {analysis['optimization_score']}") -``` - -##### `get_optimization_suggestions(profile)` - -Get optimization suggestions for a build profile. - -**Parameters:** -- `profile`: BuildProfile instance - -**Returns:** -- `list`: List of optimization suggestions - -**Example:** -```python -suggestions = optimizer.get_optimization_suggestions(profile) -for suggestion in suggestions: - print(f"Suggestion: {suggestion}") -``` - -### PerformanceReporter - -Generates performance reports and visualizations. - -```python -from deb_mock.performance import PerformanceReporter - -reporter = PerformanceReporter(config) - -# Generate performance report -report_path = reporter.generate_performance_report(monitor) - -# Generate build profile report -profile_report = reporter.generate_build_profile_report(profile) -``` - -#### Methods - -##### `generate_performance_report(monitor, output_file=None)` - -Generate a comprehensive performance report. - -**Parameters:** -- `monitor`: PerformanceMonitor instance -- `output_file`: Output file path (optional) - -**Returns:** -- `str`: Path to generated report - -**Example:** -```python -report_path = reporter.generate_performance_report(monitor) -print(f"Report generated: {report_path}") -``` - -##### `generate_build_profile_report(profile, output_file=None)` - -Generate a build profile report. - -**Parameters:** -- `profile`: BuildProfile instance -- `output_file`: Output file path (optional) - -**Returns:** -- `str`: Path to generated report - -**Example:** -```python -profile_report = reporter.generate_build_profile_report(profile) -print(f"Profile report generated: {profile_report}") -``` - -## Benchmarking - -### BenchmarkRunner - -Runs performance benchmarks for operations. - -```python -from deb_mock.benchmarking import BenchmarkRunner - -runner = BenchmarkRunner(config) - -# Run a benchmark -result = runner.run_benchmark( - "build_benchmark", - build_function, - iterations=20 -) - -# Compare benchmarks -comparison = runner.compare_benchmarks(["benchmark1", "benchmark2"]) -``` - -#### Methods - -##### `run_benchmark(benchmark_name, operation_func, **kwargs)` - -Run a benchmark for a specific operation. - -**Parameters:** -- `benchmark_name`: Name of the benchmark -- `operation_func`: Function to benchmark -- `**kwargs`: Benchmark configuration - -**Returns:** -- `BenchmarkResult`: Benchmark result - -**Example:** -```python -def build_operation(): - # Build operation to benchmark - pass - -result = runner.run_benchmark( - "build_benchmark", - build_operation, - iterations=50, - parallel_runs=4 -) - -print(f"Average duration: {result.average_duration:.3f}s") -print(f"Standard deviation: {result.standard_deviation:.3f}s") -``` - -##### `compare_benchmarks(benchmark_names)` - -Compare multiple benchmark results. - -**Parameters:** -- `benchmark_names`: List of benchmark names to compare - -**Returns:** -- `dict`: Comparison results - -**Example:** -```python -comparison = runner.compare_benchmarks(["quick", "standard", "comprehensive"]) -print(f"Fastest benchmark: {comparison['analysis']['fastest_benchmark']}") -print(f"Most stable: {comparison['analysis']['most_stable_benchmark']}") -``` - -## Exceptions - -### DebMockError - -Base exception for deb-mock errors. - -```python -from deb_mock.exceptions import DebMockError +from deb_mock.exceptions import ChrootError, ConfigurationError, SbuildError try: - # Some operation - pass -except DebMockError as e: - print(f"Deb-mock error: {e}") -``` - -### ChrootError - -Exception raised for chroot-related errors. - -```python -from deb_mock.exceptions import ChrootError - -try: - chroot_manager.create_chroot("invalid-name") + env = client.create_environment("test-env") except ChrootError as e: print(f"Chroot error: {e}") +except ConfigurationError as e: + print(f"Configuration error: {e}") +except Exception as e: + print(f"Unexpected error: {e}") ``` -### CacheError - -Exception raised for cache-related errors. +### Environment Validation ```python -from deb_mock.exceptions import CacheError - -try: - cache_manager.restore_cache("package-name") -except CacheError as e: - print(f"Cache error: {e}") +# Check if environment exists +if client.environment_exists("test-env"): + print("Environment exists") +else: + print("Environment does not exist") ``` -### PerformanceError - -Exception raised for performance monitoring errors. +### Command Error Handling ```python -from deb_mock.exceptions import PerformanceError - -try: - monitor.start_monitoring() -except PerformanceError as e: - print(f"Performance error: {e}") -``` - -### PluginError - -Exception raised for plugin-related errors. - -```python -from deb_mock.exceptions import PluginError - -try: - plugin_manager.load_plugins() -except PluginError as e: - print(f"Plugin error: {e}") -``` - -## CLI Interface - -The `deb-mock` command-line interface provides access to all functionality. - -### Basic Commands - -```bash -# Get help -deb-mock --help - -# Show version -deb-mock --version - -# Show status -deb-mock status - -# Build a package -deb-mock build package-name source-path - -# List chroots -deb-mock list-chroots - -# Create chroot -deb-mock create-chroot --suite trixie --architecture amd64 - -# Cleanup -deb-mock cleanup -``` - -### Performance Commands - -```bash -# Performance summary -deb-mock performance-summary - -# Performance report -deb-mock performance-report - -# Run benchmark -deb-mock benchmark --template standard - -# Performance analysis -deb-mock performance-analysis -``` - -### Plugin Commands - -```bash -# List plugins -deb-mock plugin list - -# Enable plugin -deb-mock plugin enable ccache_plugin - -# Disable plugin -deb-mock plugin disable debug_plugin - -# Plugin info -deb-mock plugin info ccache_plugin -``` - -### Sbuild Commands - -```bash -# Check sbuild status -deb-mock sbuild status - -# Update chroot -deb-mock sbuild update-chroot trixie-amd64 - -# Build package -deb-mock sbuild build-package source.dsc +# Execute command with error handling +result = env.execute(["invalid-command"], check=False) +if result.returncode != 0: + print(f"Command failed: {result.stderr}") ``` ## Examples -### Complete Build Example +### Complete Build Workflow ```python -from deb_mock.core import DebMock -from deb_mock.config import Config +from deb_mock import create_client, MockConfigBuilder -# Load configuration -config = Config.from_file("deb-mock.yaml") +# Create configuration +config = (MockConfigBuilder() + .environment("build-env") + .architecture("amd64") + .suite("trixie") + .packages(["build-essential", "devscripts", "cmake"]) + .cache_enabled(True) + .parallel_jobs(4) + .build()) -# Initialize deb-mock -deb_mock = DebMock(config) +# Create client +client = create_client(config) try: - # Build a package - result = deb_mock.build( - "my-package", - "/path/to/source", - chroot_name="trixie-amd64", - parallel=True - ) + # Create environment + env = client.create_environment("build-env") - if result.success: - print(f"Build successful in {result.duration:.2f}s") - print(f"Output files: {result.output_files}") - else: - print(f"Build failed: {result.error}") + # Use environment + with client.environment("build-env") as env: + # Install additional packages + env.install_packages(["ninja-build", "git"]) + # Execute setup commands + env.execute(["git", "clone", "https://github.com/example/project.git", "/build/project"]) + + # Build the package + result = client.build_package("/build/project", "build-env") + + if result['success']: + print("Build successful!") + print(f"Artifacts: {result['artifacts']}") + else: + print("Build failed!") + finally: # Cleanup - deb_mock.cleanup() + client.remove_environment("build-env") ``` -### Performance Monitoring Example +### Multi-Architecture Building ```python -from deb_mock.performance import PerformanceMonitor, PerformanceReporter +architectures = ["amd64", "arm64", "armhf"] +packages = ["/path/to/package.dsc"] -monitor = PerformanceMonitor(config) -reporter = PerformanceReporter(config) - -# Start monitoring -monitor.start_monitoring() - -try: - # Monitor build operation - with monitor.monitor_operation("package_build"): - # Build operation - result = build_package() +for arch in architectures: + config = (MockConfigBuilder() + .environment(f"build-{arch}") + .architecture(arch) + .suite("trixie") + .build()) - # Generate performance report - report_path = reporter.generate_performance_report(monitor) - print(f"Performance report: {report_path}") + client = create_client(config) -finally: - # Stop monitoring - monitor.stop_monitoring() + try: + env = client.create_environment(f"build-{arch}") + result = client.build_package("/path/to/package.dsc", f"build-{arch}") + print(f"{arch}: {'Success' if result['success'] else 'Failed'}") + finally: + client.remove_environment(f"build-{arch}") ``` -### Plugin Development Example +### CI/CD Integration ```python -from deb_mock.plugin import BasePlugin, HookStages +import os +from deb_mock import create_client, MockConfigBuilder -class BuildNotificationPlugin(BasePlugin): - name = "build_notification" - version = "1.0.0" - description = "Sends notifications on build completion" +def build_package_in_ci(source_package, arch="amd64", suite="trixie"): + """Build package in CI environment""" - def __init__(self): - super().__init__() - self.register_hook(HookStages.POST_BUILD, self.notify_build_complete) + # Create configuration from environment variables + config = (MockConfigBuilder() + .environment(f"ci-build-{arch}") + .architecture(arch) + .suite(suite) + .mirror(os.getenv("DEBIAN_MIRROR", "http://deb.debian.org/debian/")) + .packages(["build-essential", "devscripts"]) + .output_dir(os.getenv("BUILD_OUTPUT_DIR", "/tmp/build-output")) + .build()) - def notify_build_complete(self, context): - if context.get('build_success'): - print(f"✅ Build completed successfully for {context['package_name']}") - else: - print(f"❌ Build failed for {context['package_name']}") - return context + client = create_client(config) + + try: + # Create environment + env = client.create_environment(f"ci-build-{arch}") + + # Build package + result = client.build_package(source_package, f"ci-build-{arch}") + + return { + 'success': result['success'], + 'artifacts': result['artifacts'], + 'environment': f"ci-build-{arch}" + } + + finally: + # Cleanup + client.remove_environment(f"ci-build-{arch}") -# Plugin will be automatically loaded if placed in plugin directory +# Usage in CI +if __name__ == "__main__": + result = build_package_in_ci("/workspace/package.dsc") + if result['success']: + print("CI build successful!") + else: + print("CI build failed!") + exit(1) ``` -### Benchmarking Example +## Integration with External Tools + +### debian-forge Integration + +The deb-mock API is designed to integrate seamlessly with debian-forge: ```python -from deb_mock.benchmarking import BenchmarkRunner - -runner = BenchmarkRunner(config) - -def build_operation(): - # Simulate build operation - import time - time.sleep(1) - return {"success": True} - -# Run comprehensive benchmark -result = runner.run_benchmark( - "build_benchmark", - build_operation, - iterations=100, - parallel_runs=4 -) - -print(f"Benchmark completed:") -print(f" Iterations: {result.iterations}") -print(f" Average duration: {result.average_duration:.3f}s") -print(f" Standard deviation: {result.standard_deviation:.3f}s") -print(f" Performance stability: {result.analysis['performance_stability']}") +# Example debian-forge stage using deb-mock +def debian_forge_mock_stage(tree, options): + from deb_mock import create_client, MockConfigBuilder + + # Create mock configuration from debian-forge options + mock_options = options.get('mock_options', {}) + config = (MockConfigBuilder() + .environment(mock_options.get('environment', 'debian-trixie')) + .architecture(mock_options.get('arch', 'amd64')) + .suite(mock_options.get('suite', 'trixie')) + .packages(mock_options.get('packages', [])) + .build()) + + client = create_client(config) + + # Create environment + env_name = f"{config.chroot_name}-{int(time.time())}" + env = client.create_environment(env_name) + + try: + # Execute build commands + for command in options.get('commands', []): + result = env.execute(command) + if result.returncode != 0: + raise RuntimeError(f"Command failed: {command}") + + # Collect artifacts + artifacts = client.collect_artifacts(env_name, output_dir=tree) + + return 0 + + finally: + client.remove_environment(env_name) ``` -## Conclusion - -This API reference provides comprehensive coverage of all `deb-mock` functionality. For additional examples and advanced usage patterns, refer to the main documentation and plugin examples. - -The API is designed to be intuitive and follows Python best practices, making it easy to integrate `deb-mock` into existing build systems and workflows. +This API provides a stable, well-documented interface for external tools to integrate with deb-mock for build environment management. \ No newline at end of file diff --git a/docs/PLUGIN_DEVELOPMENT.md b/docs/PLUGIN_DEVELOPMENT.md new file mode 100644 index 0000000..3ac138c --- /dev/null +++ b/docs/PLUGIN_DEVELOPMENT.md @@ -0,0 +1,675 @@ +# deb-mock Plugin Development Guide + +This guide explains how to develop custom plugins for deb-mock to extend its functionality. + +## Table of Contents + +- [Plugin Overview](#plugin-overview) +- [Plugin Structure](#plugin-structure) +- [Hook System](#hook-system) +- [Configuration](#configuration) +- [Plugin Development](#plugin-development) +- [Testing Plugins](#testing-plugins) +- [Plugin Distribution](#plugin-distribution) +- [Examples](#examples) + +## Plugin Overview + +deb-mock plugins allow you to extend the build environment management system with custom functionality. Plugins can hook into various stages of the build process to: + +- Modify build environments +- Install additional packages +- Execute custom commands +- Collect build artifacts +- Monitor build performance +- Handle errors and warnings +- Integrate with external tools + +## Plugin Structure + +A deb-mock plugin is a Python module that follows a specific structure: + +``` +my_plugin.py +├── Plugin metadata +├── init() function +├── Plugin class (inherits from BasePlugin) +└── Hook implementations +``` + +### Basic Plugin Template + +```python +""" +My Custom Plugin for deb-mock +""" + +import logging +from typing import Dict, Any, List + +from deb_mock.plugin import BasePlugin, HookStages + + +class MyPlugin(BasePlugin): + """My custom plugin""" + + # Plugin metadata + requires_api_version = "1.0" + plugin_name = "my_plugin" + plugin_version = "1.0.0" + plugin_description = "My custom deb-mock plugin" + plugin_author = "Your Name" + + def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + # Plugin-specific configuration + self.enabled = self.get_config('enabled', True) + self.custom_option = self.get_config('custom_option', 'default') + + self.log_info(f"MyPlugin initialized with config: {config}") + + def _register_hooks(self): + """Register hooks for different build stages""" + # Register your hooks here + pass + + +def init(plugin_manager, config, deb_mock): + """Initialize the plugin""" + return MyPlugin(plugin_manager, config, deb_mock) +``` + +## Hook System + +The hook system allows plugins to execute code at specific points in the build process. + +### Available Hook Stages + +#### Chroot Lifecycle Hooks + +- **`PRECHROOT_INIT`**: Called before chroot initialization +- **`POSTCHROOT_INIT`**: Called after chroot initialization +- **`PRECHROOT_CLEAN`**: Called before chroot cleanup +- **`POSTCHROOT_CLEAN`**: Called after chroot cleanup + +#### Build Lifecycle Hooks + +- **`PREBUILD`**: Called before package build +- **`POSTBUILD`**: Called after package build +- **`BUILD_START`**: Called when build starts +- **`BUILD_END`**: Called when build ends + +#### Package Management Hooks + +- **`PRE_INSTALL_DEPS`**: Called before installing dependencies +- **`POST_INSTALL_DEPS`**: Called after installing dependencies +- **`PRE_INSTALL_PACKAGE`**: Called before installing a package +- **`POST_INSTALL_PACKAGE`**: Called after installing a package + +#### Mount Management Hooks + +- **`PRE_MOUNT`**: Called before mounting +- **`POST_MOUNT`**: Called after mounting +- **`PRE_UNMOUNT`**: Called before unmounting +- **`POST_UNMOUNT`**: Called after unmounting + +#### Cache Management Hooks + +- **`PRE_CACHE_CREATE`**: Called before creating cache +- **`POST_CACHE_CREATE`**: Called after creating cache +- **`PRE_CACHE_RESTORE`**: Called before restoring cache +- **`POST_CACHE_RESTORE`**: Called after restoring cache + +#### Error Handling Hooks + +- **`ON_ERROR`**: Called when an error occurs +- **`ON_WARNING`**: Called when a warning occurs + +### Registering Hooks + +```python +def _register_hooks(self): + """Register hooks for different build stages""" + + # Chroot lifecycle + self.plugin_manager.add_hook(HookStages.PRECHROOT_INIT, self.prechroot_init) + self.plugin_manager.add_hook(HookStages.POSTCHROOT_INIT, self.postchroot_init) + + # Build lifecycle + self.plugin_manager.add_hook(HookStages.PREBUILD, self.prebuild) + self.plugin_manager.add_hook(HookStages.POSTBUILD, self.postbuild) + + # Error handling + self.plugin_manager.add_hook(HookStages.ON_ERROR, self.on_error) +``` + +### Hook Implementation + +```python +def prebuild(self, source_package: str, **kwargs): + """Called before package build""" + self.log_info(f"Pre-build hook for {source_package}") + + # Your custom logic here + if self.get_config('validate_source', True): + self._validate_source_package(source_package) + +def postbuild(self, build_result: Dict[str, Any], source_package: str, **kwargs): + """Called after package build""" + success = build_result.get('success', False) + if success: + self.log_info(f"Build successful for {source_package}") + else: + self.log_error(f"Build failed for {source_package}") + +def on_error(self, error: Exception, stage: str, **kwargs): + """Called when an error occurs""" + self.log_error(f"Error in {stage}: {error}") + + # Send notification, log to file, etc. + if self.get_config('notify_on_error', False): + self._send_notification(error, stage) +``` + +## Configuration + +Plugins can access configuration through the `get_config()` method: + +```python +def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + # Get configuration values with defaults + self.enabled = self.get_config('enabled', True) + self.log_level = self.get_config('log_level', 'INFO') + self.custom_option = self.get_config('custom_option', 'default_value') + + # Get complex configuration + self.packages = self.get_config('packages', []) + self.env_vars = self.get_config('environment_variables', {}) +``` + +### Configuration Example + +```yaml +# deb-mock configuration +plugins: + - my_plugin + +plugin_conf: + my_plugin_enable: true + my_plugin_opts: + enabled: true + log_level: DEBUG + custom_option: "my_value" + packages: + - "build-essential" + - "cmake" + environment_variables: + CC: "gcc" + CXX: "g++" +``` + +## Plugin Development + +### 1. Create Plugin Directory + +```bash +mkdir -p ~/.local/share/deb-mock/plugins +cd ~/.local/share/deb-mock/plugins +``` + +### 2. Create Plugin File + +```python +# my_custom_plugin.py +""" +Custom Plugin for deb-mock +""" + +import os +import logging +from typing import Dict, Any, List + +from deb_mock.plugin import BasePlugin, HookStages + + +class CustomPlugin(BasePlugin): + """Custom plugin for deb-mock""" + + requires_api_version = "1.0" + plugin_name = "custom_plugin" + plugin_version = "1.0.0" + plugin_description = "Custom deb-mock plugin" + plugin_author = "Your Name" + + def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + # Configuration + self.enabled = self.get_config('enabled', True) + self.packages = self.get_config('packages', []) + self.commands = self.get_config('commands', []) + + self.log_info("CustomPlugin initialized") + + def _register_hooks(self): + """Register hooks""" + self.plugin_manager.add_hook(HookStages.POSTCHROOT_INIT, self.postchroot_init) + self.plugin_manager.add_hook(HookStages.PREBUILD, self.prebuild) + self.plugin_manager.add_hook(HookStages.POSTBUILD, self.postbuild) + + def postchroot_init(self, chroot_name: str, **kwargs): + """Called after chroot initialization""" + self.log_info(f"Setting up custom environment in {chroot_name}") + + # Install additional packages + if self.packages: + self.log_info(f"Installing packages: {self.packages}") + try: + result = self.deb_mock.install_packages(self.packages) + if result.get('success', False): + self.log_info("Packages installed successfully") + else: + self.log_warning(f"Failed to install packages: {result}") + except Exception as e: + self.log_error(f"Error installing packages: {e}") + + # Execute setup commands + for command in self.commands: + self.log_info(f"Executing command: {command}") + try: + result = self.deb_mock.chroot_manager.execute_in_chroot( + chroot_name, command.split(), capture_output=True + ) + if result.returncode == 0: + self.log_info(f"Command succeeded: {command}") + else: + self.log_warning(f"Command failed: {command}") + except Exception as e: + self.log_error(f"Error executing command {command}: {e}") + + def prebuild(self, source_package: str, **kwargs): + """Called before package build""" + self.log_info(f"Pre-build setup for {source_package}") + + # Your custom pre-build logic + if self.get_config('validate_source', True): + self._validate_source_package(source_package) + + def postbuild(self, build_result: Dict[str, Any], source_package: str, **kwargs): + """Called after package build""" + success = build_result.get('success', False) + if success: + self.log_info(f"Build successful for {source_package}") + self._handle_successful_build(build_result) + else: + self.log_error(f"Build failed for {source_package}") + self._handle_failed_build(build_result) + + def _validate_source_package(self, source_package: str): + """Validate source package""" + if not os.path.exists(source_package): + raise FileNotFoundError(f"Source package not found: {source_package}") + + self.log_info(f"Source package validated: {source_package}") + + def _handle_successful_build(self, build_result: Dict[str, Any]): + """Handle successful build""" + artifacts = build_result.get('artifacts', []) + self.log_info(f"Build produced {len(artifacts)} artifacts") + + # Process artifacts, send notifications, etc. + + def _handle_failed_build(self, build_result: Dict[str, Any]): + """Handle failed build""" + error = build_result.get('error', 'Unknown error') + self.log_error(f"Build failed: {error}") + + # Send error notifications, log to file, etc. + + +def init(plugin_manager, config, deb_mock): + """Initialize the plugin""" + return CustomPlugin(plugin_manager, config, deb_mock) +``` + +### 3. Configure Plugin + +Add your plugin to the deb-mock configuration: + +```yaml +# config.yaml +plugins: + - custom_plugin + +plugin_conf: + custom_plugin_enable: true + custom_plugin_opts: + enabled: true + packages: + - "build-essential" + - "cmake" + - "ninja-build" + commands: + - "apt update" + - "apt install -y git" + validate_source: true +``` + +### 4. Test Plugin + +```python +# test_plugin.py +from deb_mock import create_client, MockConfigBuilder + +# Create configuration with plugin +config = (MockConfigBuilder() + .environment("test-env") + .architecture("amd64") + .suite("trixie") + .build()) + +# Add plugin configuration +config.plugins = ["custom_plugin"] +config.plugin_conf = { + "custom_plugin_enable": True, + "custom_plugin_opts": { + "enabled": True, + "packages": ["build-essential"], + "commands": ["apt update"] + } +} + +# Create client and test +client = create_client(config) +env = client.create_environment("test-env") +print("Plugin should have executed during environment creation") +``` + +## Testing Plugins + +### Unit Testing + +```python +# test_my_plugin.py +import unittest +from unittest.mock import Mock, patch + +from my_plugin import CustomPlugin + + +class TestCustomPlugin(unittest.TestCase): + def setUp(self): + self.mock_plugin_manager = Mock() + self.mock_deb_mock = Mock() + self.config = { + 'enabled': True, + 'packages': ['build-essential'], + 'commands': ['apt update'] + } + + self.plugin = CustomPlugin( + self.mock_plugin_manager, + self.config, + self.mock_deb_mock + ) + + def test_plugin_initialization(self): + """Test plugin initialization""" + self.assertTrue(self.plugin.enabled) + self.assertEqual(self.plugin.packages, ['build-essential']) + self.assertEqual(self.plugin.commands, ['apt update']) + + def test_postchroot_init(self): + """Test postchroot_init hook""" + self.mock_deb_mock.install_packages.return_value = {'success': True} + + self.plugin.postchroot_init("test-chroot") + + self.mock_deb_mock.install_packages.assert_called_once_with(['build-essential']) + + def test_prebuild(self): + """Test prebuild hook""" + with patch('os.path.exists', return_value=True): + self.plugin.prebuild("/path/to/package.dsc") + # Test passes if no exception is raised + + +if __name__ == '__main__': + unittest.main() +``` + +### Integration Testing + +```python +# integration_test.py +from deb_mock import create_client, MockConfigBuilder + +def test_plugin_integration(): + """Test plugin integration with deb-mock""" + + # Create configuration with plugin + config = (MockConfigBuilder() + .environment("integration-test") + .architecture("amd64") + .suite("trixie") + .build()) + + config.plugins = ["custom_plugin"] + config.plugin_conf = { + "custom_plugin_enable": True, + "custom_plugin_opts": { + "enabled": True, + "packages": ["build-essential"], + "commands": ["apt update"] + } + } + + # Create client + client = create_client(config) + + try: + # Create environment (should trigger plugin) + env = client.create_environment("integration-test") + + # Verify plugin executed + # Check logs, installed packages, etc. + + print("Plugin integration test passed") + + finally: + # Cleanup + client.remove_environment("integration-test") + +if __name__ == "__main__": + test_plugin_integration() +``` + +## Plugin Distribution + +### 1. Package Structure + +``` +my-deb-mock-plugin/ +├── setup.py +├── README.md +├── LICENSE +├── my_plugin/ +│ ├── __init__.py +│ └── plugin.py +└── tests/ + └── test_plugin.py +``` + +### 2. setup.py + +```python +from setuptools import setup, find_packages + +setup( + name="my-deb-mock-plugin", + version="1.0.0", + description="My custom deb-mock plugin", + author="Your Name", + author_email="your.email@example.com", + packages=find_packages(), + install_requires=[ + "deb-mock>=0.1.0", + ], + entry_points={ + "deb_mock.plugins": [ + "my_plugin = my_plugin.plugin:init", + ], + }, + classifiers=[ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + ], +) +``` + +### 3. Installation + +```bash +# Install from source +pip install -e . + +# Install from PyPI (when published) +pip install my-deb-mock-plugin +``` + +## Examples + +### Example 1: Package Installation Plugin + +```python +""" +Plugin to automatically install packages in chroots +""" + +from deb_mock.plugin import BasePlugin, HookStages + + +class PackageInstallPlugin(BasePlugin): + """Plugin to install packages in chroots""" + + requires_api_version = "1.0" + plugin_name = "package_install" + plugin_version = "1.0.0" + plugin_description = "Automatically install packages in chroots" + + def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + self.packages = self.get_config('packages', []) + self.auto_install = self.get_config('auto_install', True) + + def _register_hooks(self): + self.plugin_manager.add_hook(HookStages.POSTCHROOT_INIT, self.install_packages) + + def install_packages(self, chroot_name: str, **kwargs): + """Install packages after chroot initialization""" + if not self.auto_install or not self.packages: + return + + self.log_info(f"Installing packages in {chroot_name}: {self.packages}") + + try: + result = self.deb_mock.install_packages(self.packages) + if result.get('success', False): + self.log_info("Packages installed successfully") + else: + self.log_warning(f"Failed to install packages: {result}") + except Exception as e: + self.log_error(f"Error installing packages: {e}") + + +def init(plugin_manager, config, deb_mock): + return PackageInstallPlugin(plugin_manager, config, deb_mock) +``` + +### Example 2: Build Notification Plugin + +```python +""" +Plugin to send notifications about build results +""" + +import smtplib +from email.mime.text import MIMEText +from deb_mock.plugin import BasePlugin, HookStages + + +class NotificationPlugin(BasePlugin): + """Plugin to send build notifications""" + + requires_api_version = "1.0" + plugin_name = "notification" + plugin_version = "1.0.0" + plugin_description = "Send notifications about build results" + + def __init__(self, plugin_manager, config, deb_mock): + super().__init__(plugin_manager, config, deb_mock) + + self.smtp_server = self.get_config('smtp_server', 'localhost') + self.smtp_port = self.get_config('smtp_port', 587) + self.smtp_user = self.get_config('smtp_user', '') + self.smtp_password = self.get_config('smtp_password', '') + self.recipients = self.get_config('recipients', []) + self.notify_on_success = self.get_config('notify_on_success', True) + self.notify_on_failure = self.get_config('notify_on_failure', True) + + def _register_hooks(self): + self.plugin_manager.add_hook(HookStages.POSTBUILD, self.send_notification) + self.plugin_manager.add_hook(HookStages.ON_ERROR, self.send_error_notification) + + def postbuild(self, build_result: Dict[str, Any], source_package: str, **kwargs): + """Send notification after build""" + success = build_result.get('success', False) + + if success and self.notify_on_success: + self._send_notification("Build Successful", f"Package {source_package} built successfully") + elif not success and self.notify_on_failure: + self._send_notification("Build Failed", f"Package {source_package} build failed") + + def on_error(self, error: Exception, stage: str, **kwargs): + """Send notification on error""" + if self.notify_on_failure: + self._send_notification("Build Error", f"Error in {stage}: {error}") + + def _send_notification(self, subject: str, body: str): + """Send email notification""" + if not self.recipients: + return + + try: + msg = MIMEText(body) + msg['Subject'] = f"deb-mock: {subject}" + msg['From'] = self.smtp_user + msg['To'] = ', '.join(self.recipients) + + with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: + if self.smtp_user and self.smtp_password: + server.starttls() + server.login(self.smtp_user, self.smtp_password) + + server.send_message(msg) + self.log_info(f"Notification sent: {subject}") + + except Exception as e: + self.log_error(f"Failed to send notification: {e}") + + +def init(plugin_manager, config, deb_mock): + return NotificationPlugin(plugin_manager, config, deb_mock) +``` + +This guide provides everything you need to develop custom plugins for deb-mock. The plugin system is designed to be flexible and powerful, allowing you to extend deb-mock's functionality in any way you need. diff --git a/examples/api_usage_example.py b/examples/api_usage_example.py new file mode 100644 index 0000000..7f4a9b6 --- /dev/null +++ b/examples/api_usage_example.py @@ -0,0 +1,396 @@ +#!/usr/bin/env python3 +""" +Example usage of deb-mock API + +This script demonstrates how to use the deb-mock API for various +build environment management tasks. +""" + +import os +import sys +import tempfile +from pathlib import Path + +# Add deb-mock to Python path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from deb_mock.api import MockAPIClient, MockConfigBuilder, create_client, quick_build +from deb_mock.environment_manager import EnvironmentManager, create_environment_manager +from deb_mock.config import Config + + +def example_basic_usage(): + """Example of basic API usage""" + print("=== Basic API Usage Example ===") + + # Create a configuration using the builder + config = (MockConfigBuilder() + .environment("example-env") + .architecture("amd64") + .suite("trixie") + .mirror("http://deb.debian.org/debian/") + .packages(["build-essential", "devscripts", "cmake"]) + .cache_enabled(True) + .parallel_jobs(4) + .verbose(True) + .build()) + + # Create API client + client = MockAPIClient(config) + + print(f"Created client with environment: {config.chroot_name}") + print(f"Architecture: {config.architecture}") + print(f"Suite: {config.suite}") + print(f"Mirror: {config.mirror}") + print(f"Initial packages: {config.chroot_additional_packages}") + + +def example_environment_management(): + """Example of environment management""" + print("\n=== Environment Management Example ===") + + # Create environment manager + manager = create_environment_manager() + + # Create a new environment + print("Creating environment...") + env_info = manager.create_environment( + name="test-build-env", + arch="amd64", + suite="trixie", + packages=["build-essential", "cmake", "ninja-build"] + ) + + print(f"Created environment: {env_info.name}") + print(f"Architecture: {env_info.architecture}") + print(f"Suite: {env_info.suite}") + print(f"Status: {env_info.status}") + + # List all environments + print("\nListing all environments...") + environments = manager.list_environments() + for env in environments: + print(f" - {env.name} ({env.architecture}/{env.suite})") + + # Get environment info + print(f"\nGetting info for {env_info.name}...") + info = manager.get_environment_info(env_info.name) + print(f" Status: {info.status}") + print(f" Size: {info.size} bytes") + print(f" Packages installed: {len(info.packages_installed or [])}") + + +def example_command_execution(): + """Example of command execution in environments""" + print("\n=== Command Execution Example ===") + + manager = create_environment_manager() + + # Create environment + env_info = manager.create_environment( + name="command-test-env", + arch="amd64", + suite="trixie", + packages=["build-essential"] + ) + + try: + # Execute commands + print("Executing commands in environment...") + + # List installed packages + result = manager.execute_command( + env_info.name, + ["dpkg", "-l", "|", "grep", "build-essential"], + capture_output=True + ) + print(f"Build tools check: {result.returncode == 0}") + + # Check system info + result = manager.execute_command( + env_info.name, + ["uname", "-a"], + capture_output=True + ) + print(f"System info: {result.stdout.strip()}") + + # Check available space + result = manager.execute_command( + env_info.name, + ["df", "-h", "/"], + capture_output=True + ) + print(f"Disk usage:\n{result.stdout}") + + finally: + # Cleanup + manager.remove_environment(env_info.name) + + +def example_package_building(): + """Example of package building""" + print("\n=== Package Building Example ===") + + client = create_client() + + # Create environment for building + env = client.create_environment( + name="build-env", + arch="amd64", + suite="trixie", + packages=["build-essential", "devscripts", "debhelper"] + ) + + try: + print(f"Created build environment: {env.name}") + + # Example: Build a simple package (this would need actual source) + print("Note: This example shows the API structure.") + print("In practice, you would provide a real source package path.") + + # Example build call (commented out as it needs real source) + # result = client.build_package("/path/to/package.dsc", "build-env") + # print(f"Build result: {result}") + + finally: + # Cleanup + client.remove_environment(env.name) + + +def example_context_manager(): + """Example of using context managers""" + print("\n=== Context Manager Example ===") + + manager = create_environment_manager() + + # Use environment context manager + with manager.environment("context-test-env", + arch="amd64", + suite="trixie", + packages=["build-essential"]) as env: + + print(f"Environment active: {env.is_active()}") + print(f"Environment name: {env.name}") + print(f"Architecture: {env.architecture}") + print(f"Suite: {env.suite}") + + # Environment is automatically cleaned up when exiting context + print("Environment will be cleaned up automatically") + + print("Context exited, environment cleaned up") + + +def example_parallel_building(): + """Example of parallel building""" + print("\n=== Parallel Building Example ===") + + client = create_client() + + # Example source packages (would be real paths in practice) + source_packages = [ + "/path/to/package1.dsc", + "/path/to/package2.dsc", + "/path/to/package3.dsc" + ] + + print("Parallel building example:") + print(f"Would build {len(source_packages)} packages in parallel") + print("Packages:", source_packages) + + # Example parallel build call (commented out as it needs real sources) + # results = client.build_parallel(source_packages, max_workers=2) + # print(f"Build results: {len(results)} packages processed") + + +def example_chain_building(): + """Example of chain building""" + print("\n=== Chain Building Example ===") + + client = create_client() + + # Example dependency chain (would be real paths in practice) + chain_packages = [ + "/path/to/base-package.dsc", + "/path/to/depends-on-base.dsc", + "/path/to/final-package.dsc" + ] + + print("Chain building example:") + print("Would build packages in dependency order:") + for i, pkg in enumerate(chain_packages, 1): + print(f" {i}. {pkg}") + + # Example chain build call (commented out as it needs real sources) + # results = client.build_chain(chain_packages) + # print(f"Chain build results: {len(results)} packages processed") + + +def example_artifact_collection(): + """Example of artifact collection""" + print("\n=== Artifact Collection Example ===") + + manager = create_environment_manager() + + # Create environment + env_info = manager.create_environment( + name="artifact-test-env", + arch="amd64", + suite="trixie", + packages=["build-essential"] + ) + + try: + print(f"Created environment: {env_info.name}") + + # Example artifact collection (would work with real build artifacts) + print("Artifact collection example:") + print("Would collect artifacts like:") + print(" - *.deb files") + print(" - *.changes files") + print(" - *.buildinfo files") + print(" - *.dsc files") + print(" - Source tarballs") + + # Example artifact collection call (commented out as it needs real artifacts) + # artifacts = manager.collect_artifacts(env_info.name) + # print(f"Collected {len(artifacts)} artifacts") + + finally: + # Cleanup + manager.remove_environment(env_info.name) + + +def example_performance_monitoring(): + """Example of performance monitoring""" + print("\n=== Performance Monitoring Example ===") + + client = create_client() + + # Get cache statistics + cache_stats = client.get_cache_stats() + print(f"Cache statistics: {cache_stats}") + + # Get performance summary (if available) + try: + perf_summary = client.get_performance_summary() + print(f"Performance summary: {perf_summary}") + except RuntimeError as e: + print(f"Performance monitoring not available: {e}") + + # Export metrics (if available) + try: + metrics_file = client.export_metrics() + print(f"Metrics exported to: {metrics_file}") + except RuntimeError as e: + print(f"Metrics export not available: {e}") + + +def example_error_handling(): + """Example of error handling""" + print("\n=== Error Handling Example ===") + + manager = create_environment_manager() + + # Try to get non-existent environment + try: + manager.get_environment_info("nonexistent-env") + except ValueError as e: + print(f"Expected error: {e}") + + # Try to execute command in non-existent environment + try: + manager.execute_command("nonexistent-env", ["ls"]) + except ValueError as e: + print(f"Expected error: {e}") + + # Try to remove non-existent environment + try: + manager.remove_environment("nonexistent-env") + except ValueError as e: + print(f"Expected error: {e}") + + +def example_configuration_variations(): + """Example of different configuration options""" + print("\n=== Configuration Variations Example ===") + + # Ubuntu configuration + ubuntu_config = (MockConfigBuilder() + .environment("ubuntu-jammy-amd64") + .architecture("amd64") + .suite("jammy") + .mirror("http://archive.ubuntu.com/ubuntu/") + .packages(["build-essential", "cmake"]) + .build()) + + print("Ubuntu configuration:") + print(f" Environment: {ubuntu_config.chroot_name}") + print(f" Suite: {ubuntu_config.suite}") + print(f" Mirror: {ubuntu_config.mirror}") + + # ARM64 configuration + arm64_config = (MockConfigBuilder() + .environment("debian-trixie-arm64") + .architecture("arm64") + .suite("trixie") + .mirror("http://deb.debian.org/debian/") + .packages(["crossbuild-essential-arm64"]) + .build()) + + print("\nARM64 configuration:") + print(f" Environment: {arm64_config.chroot_name}") + print(f" Architecture: {arm64_config.architecture}") + print(f" Cross-compilation packages: {arm64_config.chroot_additional_packages}") + + # Development configuration + dev_config = (MockConfigBuilder() + .environment("dev-env") + .architecture("amd64") + .suite("sid") # Debian unstable + .packages(["build-essential", "devscripts", "cmake", "ninja-build", "git"]) + .cache_enabled(True) + .parallel_jobs(8) + .verbose(True) + .debug(True) + .build()) + + print("\nDevelopment configuration:") + print(f" Environment: {dev_config.chroot_name}") + print(f" Suite: {dev_config.suite}") + print(f" Packages: {dev_config.chroot_additional_packages}") + print(f" Parallel jobs: {dev_config.parallel_jobs}") + print(f" Verbose: {dev_config.verbose}") + print(f" Debug: {dev_config.debug}") + + +def main(): + """Run all examples""" + print("deb-mock API Usage Examples") + print("=" * 50) + + try: + example_basic_usage() + example_environment_management() + example_command_execution() + example_package_building() + example_context_manager() + example_parallel_building() + example_chain_building() + example_artifact_collection() + example_performance_monitoring() + example_error_handling() + example_configuration_variations() + + print("\n" + "=" * 50) + print("All examples completed successfully!") + print("\nNote: These examples demonstrate the API structure.") + print("In practice, you would use real source packages and environments.") + + except Exception as e: + print(f"\nError running examples: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + main() diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..7d3446d --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,479 @@ +""" +Tests for deb-mock API + +This module contains comprehensive tests for the deb-mock API to ensure +stability and reliability for external integrations. +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch, MagicMock + +# Add deb-mock to path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from deb_mock.api import MockAPIClient, MockEnvironment, MockConfigBuilder, create_client +from deb_mock.environment_manager import EnvironmentManager, EnvironmentInfo, BuildResult +from deb_mock.config import Config +from deb_mock.exceptions import ChrootError, ConfigurationError + + +class TestMockConfigBuilder(unittest.TestCase): + """Test the MockConfigBuilder class""" + + def setUp(self): + self.builder = MockConfigBuilder() + + def test_basic_configuration(self): + """Test basic configuration building""" + config = (self.builder + .environment("test-env") + .architecture("amd64") + .suite("trixie") + .mirror("http://test.debian.org/debian/") + .build()) + + self.assertEqual(config.chroot_name, "test-env") + self.assertEqual(config.architecture, "amd64") + self.assertEqual(config.suite, "trixie") + self.assertEqual(config.mirror, "http://test.debian.org/debian/") + + def test_packages_configuration(self): + """Test packages configuration""" + packages = ["build-essential", "devscripts", "cmake"] + config = self.builder.packages(packages).build() + + self.assertEqual(config.chroot_additional_packages, packages) + + def test_output_directory(self): + """Test output directory configuration""" + output_dir = "/tmp/test-output" + config = self.builder.output_dir(output_dir).build() + + self.assertEqual(config.output_dir, output_dir) + + def test_cache_settings(self): + """Test cache configuration""" + config = self.builder.cache_enabled(True).build() + self.assertTrue(config.use_root_cache) + + config = self.builder.cache_enabled(False).build() + self.assertFalse(config.use_root_cache) + + def test_parallel_jobs(self): + """Test parallel jobs configuration""" + config = self.builder.parallel_jobs(8).build() + self.assertEqual(config.parallel_jobs, 8) + + def test_verbose_debug(self): + """Test verbose and debug configuration""" + config = (self.builder + .verbose(True) + .debug(True) + .build()) + + self.assertTrue(config.verbose) + self.assertTrue(config.debug) + + +class TestMockAPIClient(unittest.TestCase): + """Test the MockAPIClient class""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + self.config = Config( + chroot_dir=self.temp_dir, + output_dir=os.path.join(self.temp_dir, "output"), + chroot_config_dir=os.path.join(self.temp_dir, "config") + ) + + # Create necessary directories + os.makedirs(self.config.chroot_config_dir, exist_ok=True) + + # Mock the DebMock class to avoid actual chroot operations + with patch('deb_mock.api.DebMock') as mock_deb_mock: + self.mock_deb_mock_instance = Mock() + mock_deb_mock.return_value = self.mock_deb_mock_instance + + # Mock chroot manager + self.mock_chroot_manager = Mock() + self.mock_deb_mock_instance.chroot_manager = self.mock_chroot_manager + self.mock_chroot_manager.chroot_exists.return_value = False + self.mock_chroot_manager.get_chroot_info.return_value = { + 'name': 'test-env', + 'status': 'active', + 'size': 1024, + 'created': None, + 'modified': None + } + + # Mock other methods + self.mock_deb_mock_instance.init_chroot = Mock() + self.mock_deb_mock_instance.install_packages = Mock(return_value={'success': True}) + self.mock_deb_mock_instance.clean_chroot = Mock() + self.mock_deb_mock_instance.list_chroots = Mock(return_value=[]) + self.mock_deb_mock_instance.build = Mock(return_value={'success': True, 'artifacts': []}) + self.mock_deb_mock_instance.build_parallel = Mock(return_value=[{'success': True}]) + self.mock_deb_mock_instance.build_chain = Mock(return_value=[{'success': True}]) + self.mock_deb_mock_instance.get_cache_stats = Mock(return_value={}) + self.mock_deb_mock_instance.cleanup_caches = Mock(return_value={}) + + self.client = MockAPIClient(self.config) + + def tearDown(self): + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_client_initialization(self): + """Test client initialization""" + self.assertIsInstance(self.client, MockAPIClient) + self.assertEqual(self.client.config, self.config) + + def test_create_environment(self): + """Test environment creation""" + env = self.client.create_environment("test-env", "amd64", "trixie", ["build-essential"]) + + self.assertIsInstance(env, MockEnvironment) + self.assertEqual(env.name, "test-env") + self.mock_deb_mock_instance.init_chroot.assert_called_once_with("test-env", "amd64", "trixie") + self.mock_deb_mock_instance.install_packages.assert_called_once_with(["build-essential"]) + + def test_get_environment(self): + """Test getting existing environment""" + # Mock existing environment + self.mock_chroot_manager.chroot_exists.return_value = True + + env = self.client.get_environment("test-env") + + self.assertIsInstance(env, MockEnvironment) + self.assertEqual(env.name, "test-env") + + def test_get_nonexistent_environment(self): + """Test getting non-existent environment""" + self.mock_chroot_manager.chroot_exists.return_value = False + + with self.assertRaises(ValueError): + self.client.get_environment("nonexistent-env") + + def test_list_environments(self): + """Test listing environments""" + self.mock_deb_mock_instance.list_chroots.return_value = ["env1", "env2"] + + environments = self.client.list_environments() + + self.assertEqual(environments, ["env1", "env2"]) + + def test_remove_environment(self): + """Test removing environment""" + self.client.remove_environment("test-env") + + self.mock_deb_mock_instance.clean_chroot.assert_called_once_with("test-env") + + def test_build_package(self): + """Test building a package""" + result = self.client.build_package("/path/to/package.dsc", "test-env") + + self.mock_deb_mock_instance.build.assert_called_once() + self.assertEqual(result['success'], True) + + def test_build_parallel(self): + """Test parallel building""" + packages = ["/path/to/pkg1.dsc", "/path/to/pkg2.dsc"] + results = self.client.build_parallel(packages, max_workers=2) + + self.mock_deb_mock_instance.build_parallel.assert_called_once_with(packages, 2) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]['success'], True) + + def test_build_chain(self): + """Test chain building""" + packages = ["/path/to/pkg1.dsc", "/path/to/pkg2.dsc"] + results = self.client.build_chain(packages) + + self.mock_deb_mock_instance.build_chain.assert_called_once_with(packages) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]['success'], True) + + def test_environment_context_manager(self): + """Test environment context manager""" + # Mock existing environment + self.mock_chroot_manager.chroot_exists.return_value = True + + with self.client.environment("test-env") as env: + self.assertIsInstance(env, MockEnvironment) + self.assertEqual(env.name, "test-env") + self.assertTrue(env.is_active()) + + # Environment should be deactivated after context + self.assertFalse(env.is_active()) + + +class TestMockEnvironment(unittest.TestCase): + """Test the MockEnvironment class""" + + def setUp(self): + self.mock_deb_mock = Mock() + self.mock_chroot_manager = Mock() + self.mock_deb_mock.chroot_manager = self.mock_chroot_manager + self.mock_chroot_manager.chroot_exists.return_value = True + self.mock_chroot_manager.execute_in_chroot.return_value = Mock(returncode=0, stdout="test output") + self.mock_chroot_manager.get_chroot_info.return_value = {'status': 'active'} + + self.env = MockEnvironment("test-env", self.mock_deb_mock) + + def test_environment_initialization(self): + """Test environment initialization""" + self.assertEqual(self.env.name, "test-env") + self.assertFalse(self.env.is_active()) + + def test_activate_deactivate(self): + """Test environment activation and deactivation""" + self.env.activate() + self.assertTrue(self.env.is_active()) + + self.env.deactivate() + self.assertFalse(self.env.is_active()) + + def test_activate_nonexistent_environment(self): + """Test activating non-existent environment""" + self.mock_chroot_manager.chroot_exists.return_value = False + + with self.assertRaises(ChrootError): + self.env.activate() + + def test_execute_command(self): + """Test command execution""" + self.env.activate() + + result = self.env.execute(["ls", "-la"]) + + self.mock_chroot_manager.execute_in_chroot.assert_called_once_with( + "test-env", ["ls", "-la"], capture_output=True + ) + self.assertEqual(result.returncode, 0) + + def test_execute_command_string(self): + """Test command execution with string command""" + self.env.activate() + + result = self.env.execute("ls -la") + + self.mock_chroot_manager.execute_in_chroot.assert_called_once_with( + "test-env", ["ls", "-la"], capture_output=True + ) + + def test_execute_command_inactive(self): + """Test executing command on inactive environment""" + with self.assertRaises(RuntimeError): + self.env.execute(["ls"]) + + def test_install_packages(self): + """Test package installation""" + self.env.activate() + + result = self.env.install_packages(["build-essential"]) + + self.mock_deb_mock.install_packages.assert_called_once_with(["build-essential"]) + self.assertEqual(result['success'], True) + + def test_copy_in(self): + """Test copying files into environment""" + self.env.activate() + + self.env.copy_in("/local/file", "/chroot/file") + + self.mock_chroot_manager.copy_to_chroot.assert_called_once_with( + "/local/file", "/chroot/file", "test-env" + ) + + def test_copy_out(self): + """Test copying files out of environment""" + self.env.activate() + + self.env.copy_out("/chroot/file", "/local/file") + + self.mock_chroot_manager.copy_from_chroot.assert_called_once_with( + "/chroot/file", "/local/file", "test-env" + ) + + def test_get_info(self): + """Test getting environment info""" + info = self.env.get_info() + + self.mock_chroot_manager.get_chroot_info.assert_called_once_with("test-env") + self.assertEqual(info['status'], 'active') + + +class TestEnvironmentManager(unittest.TestCase): + """Test the EnvironmentManager class""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + self.config = Config( + chroot_dir=self.temp_dir, + output_dir=os.path.join(self.temp_dir, "output"), + chroot_config_dir=os.path.join(self.temp_dir, "config") + ) + + # Create necessary directories + os.makedirs(self.config.chroot_config_dir, exist_ok=True) + + # Mock the DebMock class + with patch('deb_mock.environment_manager.DebMock') as mock_deb_mock: + self.mock_deb_mock_instance = Mock() + mock_deb_mock.return_value = self.mock_deb_mock_instance + + # Mock chroot manager + self.mock_chroot_manager = Mock() + self.mock_deb_mock_instance.chroot_manager = self.mock_chroot_manager + self.mock_chroot_manager.chroot_exists.return_value = False + self.mock_chroot_manager.get_chroot_info.return_value = { + 'name': 'test-env', + 'status': 'active', + 'size': 1024, + 'created': None, + 'modified': None + } + self.mock_chroot_manager.list_mounts.return_value = [] + + # Mock other methods + self.mock_deb_mock_instance.init_chroot = Mock() + self.mock_deb_mock_instance.install_packages = Mock(return_value={'success': True}) + self.mock_deb_mock_instance.clean_chroot = Mock() + self.mock_deb_mock_instance.list_chroots = Mock(return_value=[]) + self.mock_deb_mock_instance.build = Mock(return_value={'success': True, 'artifacts': []}) + self.mock_deb_mock_instance.update_chroot = Mock() + + self.manager = EnvironmentManager(self.config) + + def tearDown(self): + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_manager_initialization(self): + """Test manager initialization""" + self.assertIsInstance(self.manager, EnvironmentManager) + self.assertEqual(self.manager.config, self.config) + + def test_create_environment(self): + """Test environment creation""" + info = self.manager.create_environment("test-env", "amd64", "trixie", ["build-essential"]) + + self.assertIsInstance(info, EnvironmentInfo) + self.assertEqual(info.name, "test-env") + self.assertEqual(info.architecture, "amd64") + self.assertEqual(info.suite, "trixie") + self.mock_deb_mock_instance.init_chroot.assert_called_once_with("test-env", "amd64", "trixie") + + def test_environment_exists(self): + """Test checking if environment exists""" + self.mock_chroot_manager.chroot_exists.return_value = True + + self.assertTrue(self.manager.environment_exists("test-env")) + + self.mock_chroot_manager.chroot_exists.return_value = False + + self.assertFalse(self.manager.environment_exists("test-env")) + + def test_get_environment_info(self): + """Test getting environment info""" + self.mock_chroot_manager.chroot_exists.return_value = True + + info = self.manager.get_environment_info("test-env") + + self.assertIsInstance(info, EnvironmentInfo) + self.assertEqual(info.name, "test-env") + + def test_get_nonexistent_environment_info(self): + """Test getting info for non-existent environment""" + self.mock_chroot_manager.chroot_exists.return_value = False + + with self.assertRaises(ValueError): + self.manager.get_environment_info("nonexistent-env") + + def test_list_environments(self): + """Test listing environments""" + self.mock_deb_mock_instance.list_chroots.return_value = ["env1", "env2"] + self.mock_chroot_manager.chroot_exists.return_value = True + + environments = self.manager.list_environments() + + self.assertEqual(len(environments), 2) + self.assertIsInstance(environments[0], EnvironmentInfo) + + def test_remove_environment(self): + """Test removing environment""" + self.mock_chroot_manager.chroot_exists.return_value = True + + self.manager.remove_environment("test-env") + + self.mock_deb_mock_instance.clean_chroot.assert_called_once_with("test-env") + + def test_execute_command(self): + """Test command execution""" + self.mock_chroot_manager.chroot_exists.return_value = True + self.mock_chroot_manager.execute_in_chroot.return_value = Mock(returncode=0, stdout="test output") + + result = self.manager.execute_command("test-env", ["ls", "-la"]) + + self.mock_chroot_manager.execute_in_chroot.assert_called_once_with( + "test-env", ["ls", "-la"], capture_output=True + ) + self.assertEqual(result.returncode, 0) + + def test_install_packages(self): + """Test package installation""" + self.mock_chroot_manager.chroot_exists.return_value = True + + result = self.manager.install_packages("test-env", ["build-essential"]) + + self.mock_deb_mock_instance.install_packages.assert_called_once_with(["build-essential"]) + self.assertEqual(result['success'], True) + + def test_build_package(self): + """Test package building""" + self.mock_chroot_manager.chroot_exists.return_value = True + + result = self.manager.build_package("test-env", "/path/to/package.dsc") + + self.mock_deb_mock_instance.build.assert_called_once() + self.assertIsInstance(result, BuildResult) + self.assertTrue(result.success) + + +class TestIntegration(unittest.TestCase): + """Integration tests for the API""" + + def test_create_client_with_config_builder(self): + """Test creating client with config builder""" + config = (MockConfigBuilder() + .environment("test-env") + .architecture("amd64") + .suite("trixie") + .build()) + + client = MockAPIClient(config) + + self.assertIsInstance(client, MockAPIClient) + self.assertEqual(client.config.chroot_name, "test-env") + + def test_quick_build_function(self): + """Test the quick_build convenience function""" + from deb_mock.api import quick_build + + with patch('deb_mock.api.MockAPIClient') as mock_client_class: + mock_client = Mock() + mock_client.build_package.return_value = {'success': True} + mock_client_class.return_value = mock_client + + result = quick_build("/path/to/package.dsc") + + self.assertEqual(result['success'], True) + mock_client.build_package.assert_called_once() + + +if __name__ == '__main__': + unittest.main()