deb-mock/docs/PLUGIN_DEVELOPMENT.md
robojerk 8c585e2e33
Some checks failed
Build Deb-Mock Package / build (push) Failing after 59s
Lint Code / Lint All Code (push) Failing after 2s
Test Deb-Mock Build / test (push) Failing after 41s
Add stable Python API and comprehensive environment management
- 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
2025-09-04 10:04:16 -07:00

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

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 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

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.