- Add MockAPIClient and MockEnvironment for external integration - Implement EnvironmentManager with full lifecycle support - Enhance plugin system with registry and BasePlugin class - Add comprehensive test suite and documentation - Include practical usage examples and plugin development guide
20 KiB
20 KiB
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 Structure
- Hook System
- Configuration
- Plugin Development
- Testing Plugins
- Plugin Distribution
- 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
"""
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 initializationPOSTCHROOT_INIT: Called after chroot initializationPRECHROOT_CLEAN: Called before chroot cleanupPOSTCHROOT_CLEAN: Called after chroot cleanup
Build Lifecycle Hooks
PREBUILD: Called before package buildPOSTBUILD: Called after package buildBUILD_START: Called when build startsBUILD_END: Called when build ends
Package Management Hooks
PRE_INSTALL_DEPS: Called before installing dependenciesPOST_INSTALL_DEPS: Called after installing dependenciesPRE_INSTALL_PACKAGE: Called before installing a packagePOST_INSTALL_PACKAGE: Called after installing a package
Mount Management Hooks
PRE_MOUNT: Called before mountingPOST_MOUNT: Called after mountingPRE_UNMOUNT: Called before unmountingPOST_UNMOUNT: Called after unmounting
Cache Management Hooks
PRE_CACHE_CREATE: Called before creating cachePOST_CACHE_CREATE: Called after creating cachePRE_CACHE_RESTORE: Called before restoring cachePOST_CACHE_RESTORE: Called after restoring cache
Error Handling Hooks
ON_ERROR: Called when an error occursON_WARNING: Called when a warning occurs
Registering Hooks
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
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:
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
# 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
mkdir -p ~/.local/share/deb-mock/plugins
cd ~/.local/share/deb-mock/plugins
2. Create Plugin File
# 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:
# 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
# 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
# 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
# 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
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
# Install from source
pip install -e .
# Install from PyPI (when published)
pip install my-deb-mock-plugin
Examples
Example 1: Package Installation Plugin
"""
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
"""
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.