deb-mock/deb_mock/plugins/hook_manager.py

"""
Hook Manager for Deb-Mock Plugin System

This module provides the hook management functionality for the Deb-Mock plugin system,
inspired by Fedora's Mock plugin hooks but adapted for Debian-based workflows.
"""

import logging
from typing import Any, Callable, Dict, List, Optional

logger = logging.getLogger(__name__)


class HookManager:
    """
    Manages plugin hooks and their execution.

    This class provides the core functionality for registering and executing
    plugin hooks at specific points in the build lifecycle, following the
    same pattern as Mock's plugin hook system.
    """

    def __init__(self):
        """Initialize the hook manager."""
        self.hooks: Dict[str, List[Callable]] = {}
        self.hook_contexts: Dict[str, Dict[str, Any]] = {}

        # Define available hook points (based on Mock's hook system)
        self.available_hooks = {
            "clean": "Clean up plugin resources",
            "earlyprebuild": "Very early build stage",
            "initfailed": "Chroot initialization failed",
            "list_snapshots": "List available snapshots",
            "make_snapshot": "Create a snapshot",
            "mount_root": "Mount chroot directory",
            "postbuild": "After build completion",
            "postchroot": "After chroot command",
            "postclean": "After chroot cleanup",
            "postdeps": "After dependency installation",
            "postinit": "After chroot initialization",
            "postshell": "After shell exit",
            "postupdate": "After package updates",
            "postumount": "After unmounting",
            "postapt": "After APT operations",
            "prebuild": "Before build starts",
            "prechroot": "Before chroot command",
            "preinit": "Before chroot initialization",
            "preshell": "Before shell prompt",
            "preapt": "Before APT operations",
            "process_logs": "Process build logs",
            "remove_snapshot": "Remove snapshot",
            "rollback_to": "Rollback to snapshot",
            "scrub": "Scrub chroot",
        }

    def add_hook(self, hook_name: str, callback: Callable) -> None:
        """
        Register a hook callback.

        Args:
            hook_name: Name of the hook to register for
            callback: Function to call when hook is triggered

        Raises:
            ValueError: If hook_name is not a valid hook point
        """
        if hook_name not in self.available_hooks:
            raise ValueError(f"Invalid hook name: {hook_name}. Available hooks: {list(self.available_hooks.keys())}")

        if hook_name not in self.hooks:
            self.hooks[hook_name] = []

        self.hooks[hook_name].append(callback)
        logger.debug(f"Registered hook '{hook_name}' with callback {callback.__name__}")

    def call_hook(self, hook_name: str, context: Optional[Dict[str, Any]] = None) -> None:
        """
        Execute all registered hooks for a given hook name.

        Args:
            hook_name: Name of the hook to trigger
            context: Context dictionary to pass to hook callbacks

        Note:
            Hook execution errors are logged but don't fail the build,
            following Mock's behavior.
        """
        if hook_name not in self.hooks:
            logger.debug(f"No hooks registered for '{hook_name}'")
            return

        context = context or {}
        logger.debug(f"Calling {len(self.hooks[hook_name])} hooks for '{hook_name}'")

        for i, callback in enumerate(self.hooks[hook_name]):
            try:
                logger.debug(f"Executing hook {i + 1}/{len(self.hooks[hook_name])}: {callback.__name__}")
                callback(context)
                logger.debug(f"Successfully executed hook: {callback.__name__}")
            except Exception as e:
                logger.warning(f"Hook '{hook_name}' failed in {callback.__name__}: {e}")
                # Continue with other hooks - don't fail the build

    def call_hook_with_result(self, hook_name: str, context: Optional[Dict[str, Any]] = None) -> List[Any]:
        """
        Execute all registered hooks and collect their results.

        Args:
            hook_name: Name of the hook to trigger
            context: Context dictionary to pass to hook callbacks

        Returns:
            List of results from hook callbacks (None for failed hooks)
        """
        if hook_name not in self.hooks:
            return []

        context = context or {}
        results = []

        for callback in self.hooks[hook_name]:
            try:
                result = callback(context)
                results.append(result)
            except Exception as e:
                logger.warning(f"Hook '{hook_name}' failed in {callback.__name__}: {e}")
                results.append(None)

        return results

    def get_hook_names(self) -> List[str]:
        """
        Get list of available hook names.

        Returns:
            List of hook names that have been registered
        """
        return list(self.hooks.keys())

    def get_available_hooks(self) -> Dict[str, str]:
        """
        Get all available hook points with descriptions.

        Returns:
            Dictionary mapping hook names to descriptions
        """
        return self.available_hooks.copy()

    def get_hook_info(self, hook_name: str) -> Dict[str, Any]:
        """
        Get information about a specific hook.

        Args:
            hook_name: Name of the hook

        Returns:
            Dictionary with hook information
        """
        if hook_name not in self.available_hooks:
            return {"error": f'Hook "{hook_name}" not found'}

        info = {
            "name": hook_name,
            "description": self.available_hooks[hook_name],
            "registered_callbacks": len(self.hooks.get(hook_name, [])),
            "callbacks": [],
        }

        if hook_name in self.hooks:
            for callback in self.hooks[hook_name]:
                info["callbacks"].append({"name": callback.__name__, "module": callback.__module__})

        return info

    def remove_hook(self, hook_name: str, callback: Callable) -> bool:
        """
        Remove a specific hook callback.

        Args:
            hook_name: Name of the hook
            callback: Callback function to remove

        Returns:
            True if callback was removed, False if not found
        """
        if hook_name not in self.hooks:
            return False

        try:
            self.hooks[hook_name].remove(callback)
            logger.debug(f"Removed hook '{hook_name}' callback {callback.__name__}")
            return True
        except ValueError:
            return False

    def clear_hooks(self, hook_name: Optional[str] = None) -> None:
        """
        Clear all hooks or hooks for a specific hook name.

        Args:
            hook_name: Specific hook name to clear, or None to clear all
        """
        if hook_name is None:
            self.hooks.clear()
            logger.debug("Cleared all hooks")
        elif hook_name in self.hooks:
            self.hooks[hook_name].clear()
            logger.debug(f"Cleared hooks for '{hook_name}'")

    def get_hook_statistics(self) -> Dict[str, Any]:
        """
        Get statistics about hook usage.

        Returns:
            Dictionary with hook statistics
        """
        stats = {
            "total_hooks": len(self.hooks),
            "total_callbacks": sum(len(callbacks) for callbacks in self.hooks.values()),
            "hooks_with_callbacks": len([h for h in self.hooks.values() if h]),
            "available_hooks": len(self.available_hooks),
            "hook_details": {},
        }

        for hook_name in self.available_hooks:
            stats["hook_details"][hook_name] = {
                "description": self.available_hooks[hook_name],
                "registered": hook_name in self.hooks,
                "callback_count": len(self.hooks.get(hook_name, [])),
            }

        return stats

    def validate_hook_name(self, hook_name: str) -> bool:
        """
        Validate if a hook name is valid.

        Args:
            hook_name: Name of the hook to validate

        Returns:
            True if hook name is valid, False otherwise
        """
        return hook_name in self.available_hooks

    def get_hook_suggestions(self, partial_name: str) -> List[str]:
        """
        Get hook name suggestions based on partial input.

        Args:
            partial_name: Partial hook name

        Returns:
            List of matching hook names
        """
        return [name for name in self.available_hooks.keys() if name.startswith(partial_name)]