""" Plugin registry and management for deb-mock 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 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 @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: """ Central registry for deb-mock plugins This class manages plugin discovery, loading, and lifecycle. """ def __init__(self, plugin_dirs: List[str] = None): """ Initialize the plugin registry Args: plugin_dirs: List of directories to search for plugins """ 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]: """ 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 """ 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 plugin information by name Args: plugin_name: Name of the plugin Returns: PluginInfo object or None if not found """ return self._plugins.get(plugin_name) def list_plugins(self) -> List[PluginInfo]: """ List all registered plugins Returns: List of plugin information """ return list(self._plugins.values()) def list_enabled_plugins(self) -> List[PluginInfo]: """ List enabled plugins Returns: List of enabled plugin information """ return [plugin for plugin in self._plugins.values() if plugin.enabled] def enable_plugin(self, plugin_name: str) -> None: """ Enable a plugin Args: plugin_name: Name of the plugin to enable """ 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: # 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: raise PluginError(f"Failed to load plugin '{plugin_name}': {e}") def unload_plugin(self, plugin_name: str) -> None: """ Unload a plugin instance Args: plugin_name: Name of the plugin to unload """ 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: """ Reload a plugin Args: plugin_name: Name of the plugin to reload plugin_manager: Plugin manager instance config: Plugin configuration deb_mock: DebMock instance Returns: Reloaded plugin instance """ 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 plugin registry statistics Returns: Dictionary with plugin statistics """ 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 } def validate_plugin_dependencies(self, plugin_name: str) -> List[str]: """ Validate plugin dependencies Args: plugin_name: Name of the plugin to validate Returns: List of missing dependencies """ 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 # 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)