414 lines
No EOL
11 KiB
Python
414 lines
No EOL
11 KiB
Python
"""
|
|
Base Plugin Class for Deb-Mock Plugin System
|
|
|
|
This module provides the base plugin class that all Deb-Mock plugins should inherit from,
|
|
inspired by Fedora's Mock plugin architecture but adapted for Debian-based systems.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BasePlugin:
|
|
"""
|
|
Base class for all Deb-Mock plugins.
|
|
|
|
This class provides the foundation for all plugins in the Deb-Mock system,
|
|
following the same patterns as Fedora's Mock plugins but adapted for Debian workflows.
|
|
|
|
Plugins should inherit from this class and override the hook methods they need.
|
|
"""
|
|
|
|
def __init__(self, config, hook_manager):
|
|
"""
|
|
Initialize the plugin.
|
|
|
|
Args:
|
|
config: Configuration object
|
|
hook_manager: Hook manager instance
|
|
"""
|
|
self.config = config
|
|
self.hook_manager = hook_manager
|
|
self.enabled = self._is_enabled()
|
|
self.plugin_name = self.__class__.__name__.lower()
|
|
|
|
# Register hooks if plugin is enabled
|
|
if self.enabled:
|
|
self._register_hooks()
|
|
logger.debug(f"Plugin {self.plugin_name} initialized and enabled")
|
|
else:
|
|
logger.debug(f"Plugin {self.plugin_name} initialized but disabled")
|
|
|
|
def _is_enabled(self) -> bool:
|
|
"""
|
|
Check if plugin is enabled in configuration.
|
|
|
|
Returns:
|
|
True if plugin is enabled, False otherwise
|
|
"""
|
|
plugin_config = getattr(self.config, 'plugins', {})
|
|
plugin_name = self.plugin_name
|
|
|
|
# Check if plugin is explicitly enabled
|
|
if plugin_name in plugin_config:
|
|
return plugin_config[plugin_name].get('enabled', False)
|
|
|
|
# Check if plugin is enabled via global plugin settings
|
|
return getattr(self.config, 'enable_plugins', {}).get(plugin_name, False)
|
|
|
|
def _register_hooks(self):
|
|
"""
|
|
Register plugin hooks with the hook manager.
|
|
|
|
Override this method in subclasses to register specific hooks.
|
|
"""
|
|
# Override in subclasses to register hooks
|
|
pass
|
|
|
|
def _get_plugin_config(self) -> Dict[str, Any]:
|
|
"""
|
|
Get plugin-specific configuration.
|
|
|
|
Returns:
|
|
Plugin configuration dictionary
|
|
"""
|
|
plugin_config = getattr(self.config, 'plugins', {})
|
|
return plugin_config.get(self.plugin_name, {})
|
|
|
|
def _log_info(self, message: str):
|
|
"""Log an info message with plugin context."""
|
|
logger.info(f"[{self.plugin_name}] {message}")
|
|
|
|
def _log_debug(self, message: str):
|
|
"""Log a debug message with plugin context."""
|
|
logger.debug(f"[{self.plugin_name}] {message}")
|
|
|
|
def _log_warning(self, message: str):
|
|
"""Log a warning message with plugin context."""
|
|
logger.warning(f"[{self.plugin_name}] {message}")
|
|
|
|
def _log_error(self, message: str):
|
|
"""Log an error message with plugin context."""
|
|
logger.error(f"[{self.plugin_name}] {message}")
|
|
|
|
# ============================================================================
|
|
# Hook Method Stubs - Override in subclasses as needed
|
|
# ============================================================================
|
|
|
|
def clean(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Clean up plugin resources.
|
|
|
|
Called after chroot cleanup.
|
|
|
|
Args:
|
|
context: Context dictionary with cleanup information
|
|
"""
|
|
pass
|
|
|
|
def earlyprebuild(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Very early build stage.
|
|
|
|
Called before SRPM rebuild, before dependencies.
|
|
|
|
Args:
|
|
context: Context dictionary with early build information
|
|
"""
|
|
pass
|
|
|
|
def initfailed(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Chroot initialization failed.
|
|
|
|
Called when chroot creation fails.
|
|
|
|
Args:
|
|
context: Context dictionary with error information
|
|
"""
|
|
pass
|
|
|
|
def list_snapshots(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
List available snapshots.
|
|
|
|
Called when --list-snapshots is used.
|
|
|
|
Args:
|
|
context: Context dictionary with snapshot information
|
|
"""
|
|
pass
|
|
|
|
def make_snapshot(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Create a snapshot.
|
|
|
|
Called when snapshot creation is requested.
|
|
|
|
Args:
|
|
context: Context dictionary with snapshot creation parameters
|
|
"""
|
|
pass
|
|
|
|
def mount_root(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Mount chroot directory.
|
|
|
|
Called before preinit, chroot exists.
|
|
|
|
Args:
|
|
context: Context dictionary with mount information
|
|
"""
|
|
pass
|
|
|
|
def postbuild(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After build completion.
|
|
|
|
Called after RPM/SRPM build (success/failure).
|
|
|
|
Args:
|
|
context: Context dictionary with build results
|
|
"""
|
|
pass
|
|
|
|
def postchroot(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After chroot command.
|
|
|
|
Called after mock chroot command.
|
|
|
|
Args:
|
|
context: Context dictionary with chroot command results
|
|
"""
|
|
pass
|
|
|
|
def postclean(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After chroot cleanup.
|
|
|
|
Called after chroot content deletion.
|
|
|
|
Args:
|
|
context: Context dictionary with cleanup information
|
|
"""
|
|
pass
|
|
|
|
def postdeps(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After dependency installation.
|
|
|
|
Called when dependencies installed, before build.
|
|
|
|
Args:
|
|
context: Context dictionary with dependency information
|
|
"""
|
|
pass
|
|
|
|
def postinit(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After chroot initialization.
|
|
|
|
Called when chroot ready for dependencies.
|
|
|
|
Args:
|
|
context: Context dictionary with initialization results
|
|
"""
|
|
pass
|
|
|
|
def postshell(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After shell exit.
|
|
|
|
Called after mock shell command.
|
|
|
|
Args:
|
|
context: Context dictionary with shell session information
|
|
"""
|
|
pass
|
|
|
|
def postupdate(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After package updates.
|
|
|
|
Called after successful package updates.
|
|
|
|
Args:
|
|
context: Context dictionary with update information
|
|
"""
|
|
pass
|
|
|
|
def postumount(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After unmounting.
|
|
|
|
Called when all inner mounts unmounted.
|
|
|
|
Args:
|
|
context: Context dictionary with unmount information
|
|
"""
|
|
pass
|
|
|
|
def postapt(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
After APT operations.
|
|
|
|
Called after any package manager action.
|
|
|
|
Args:
|
|
context: Context dictionary with APT operation results
|
|
"""
|
|
pass
|
|
|
|
def prebuild(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Before build starts.
|
|
|
|
Called after BuildRequires, before RPM build.
|
|
|
|
Args:
|
|
context: Context dictionary with build preparation information
|
|
"""
|
|
pass
|
|
|
|
def prechroot(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Before chroot command.
|
|
|
|
Called before mock chroot command.
|
|
|
|
Args:
|
|
context: Context dictionary with chroot command parameters
|
|
"""
|
|
pass
|
|
|
|
def preinit(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Before chroot initialization.
|
|
|
|
Called when only chroot/result dirs exist.
|
|
|
|
Args:
|
|
context: Context dictionary with initialization parameters
|
|
"""
|
|
pass
|
|
|
|
def preshell(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Before shell prompt.
|
|
|
|
Called before mock shell prompt.
|
|
|
|
Args:
|
|
context: Context dictionary with shell session parameters
|
|
"""
|
|
pass
|
|
|
|
def preapt(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Before APT operations.
|
|
|
|
Called before any package manager action.
|
|
|
|
Args:
|
|
context: Context dictionary with APT operation parameters
|
|
"""
|
|
pass
|
|
|
|
def process_logs(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Process build logs.
|
|
|
|
Called after build log completion.
|
|
|
|
Args:
|
|
context: Context dictionary with log information
|
|
"""
|
|
pass
|
|
|
|
def remove_snapshot(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Remove snapshot.
|
|
|
|
Called when snapshot removal requested.
|
|
|
|
Args:
|
|
context: Context dictionary with snapshot removal parameters
|
|
"""
|
|
pass
|
|
|
|
def rollback_to(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Rollback to snapshot.
|
|
|
|
Called when rollback requested.
|
|
|
|
Args:
|
|
context: Context dictionary with rollback parameters
|
|
"""
|
|
pass
|
|
|
|
def scrub(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Scrub chroot.
|
|
|
|
Called when chroot scrubbing requested.
|
|
|
|
Args:
|
|
context: Context dictionary with scrub parameters
|
|
"""
|
|
pass
|
|
|
|
# ============================================================================
|
|
# Plugin Lifecycle Methods
|
|
# ============================================================================
|
|
|
|
def setup(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Setup plugin before build.
|
|
|
|
Called once during plugin initialization.
|
|
|
|
Args:
|
|
context: Context dictionary with setup information
|
|
"""
|
|
pass
|
|
|
|
def teardown(self, context: Dict[str, Any]) -> None:
|
|
"""
|
|
Cleanup plugin after build.
|
|
|
|
Called once during plugin cleanup.
|
|
|
|
Args:
|
|
context: Context dictionary with teardown information
|
|
"""
|
|
pass
|
|
|
|
def validate_config(self, config: Any) -> bool:
|
|
"""
|
|
Validate plugin configuration.
|
|
|
|
Args:
|
|
config: Configuration to validate
|
|
|
|
Returns:
|
|
True if configuration is valid, False otherwise
|
|
"""
|
|
return True
|
|
|
|
def get_plugin_info(self) -> Dict[str, Any]:
|
|
"""
|
|
Get plugin information.
|
|
|
|
Returns:
|
|
Dictionary with plugin information
|
|
"""
|
|
return {
|
|
'name': self.plugin_name,
|
|
'class': self.__class__.__name__,
|
|
'enabled': self.enabled,
|
|
'docstring': self.__class__.__doc__ or 'No documentation available'
|
|
} |