#!/usr/bin/env python3 """ Debian Module Adapter This module provides a compatibility layer between the new Debian modules and existing Debian tools, ensuring seamless integration. """ import json import os import subprocess import tempfile from typing import Dict, List, Optional, Any, Union from dataclasses import dataclass, asdict from pathlib import Path import logging @dataclass class ModuleConfig: """Represents a module configuration""" type: str config: Dict[str, Any] module_path: str working_directory: str @dataclass class ModuleResult: """Represents the result of module execution""" success: bool output: str error: Optional[str] artifacts: List[str] metadata: Dict[str, Any] class DebianModuleAdapter: """Adapter for Debian modules to integrate with existing tools""" def __init__(self, config_dir: str = "./config/modules"): self.config_dir = Path(config_dir) self.config_dir.mkdir(parents=True, exist_ok=True) self.logger = self._setup_logging() # Module type mappings self.module_mappings = { "apt": "apt", "apt-ostree": "apt-ostree", "deb-mock": "deb-mock", "dnf": "apt", # Map dnf to apt for compatibility "rpm-ostree": "apt-ostree", # Map rpm-ostree to apt-ostree "mock": "deb-mock" # Map mock to deb-mock } def _setup_logging(self) -> logging.Logger: """Setup logging for the adapter""" logger = logging.getLogger('debian-module-adapter') logger.setLevel(logging.INFO) if not logger.handlers: handler = logging.FileHandler(self.config_dir / "adapter.log") formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) return logger def adapt_module_config(self, original_config: Dict[str, Any]) -> ModuleConfig: """Adapt a module configuration for Debian compatibility""" module_type = original_config.get("type", "") # Map module types if needed if module_type in self.module_mappings: adapted_type = self.module_mappings[module_type] self.logger.info(f"Adapting module type from {module_type} to {adapted_type}") else: adapted_type = module_type # Create adapted configuration adapted_config = self._create_adapted_config(original_config, adapted_type) return ModuleConfig( type=adapted_type, config=adapted_config, module_path=f"modules/{adapted_type}", working_directory=os.getcwd() ) def _create_adapted_config(self, original_config: Dict[str, Any], adapted_type: str) -> Dict[str, Any]: """Create an adapted configuration for the target module type""" adapted_config = original_config.copy() adapted_config["type"] = adapted_type # Handle specific adaptations based on module type if adapted_type == "apt": adapted_config = self._adapt_to_apt(original_config) elif adapted_type == "apt-ostree": adapted_config = self._adapt_to_apt_ostree(original_config) elif adapted_type == "deb-mock": adapted_config = self._adapt_to_deb_mock(original_config) return adapted_config def _adapt_to_apt(self, config: Dict[str, Any]) -> Dict[str, Any]: """Adapt configuration for apt module""" adapted = config.copy() # Convert RPM-specific fields to Debian equivalents if "group-install" in adapted: adapted["task-install"] = { "packages": adapted.pop("group-install")["packages"], "with-optional": adapted.pop("group-install").get("with-optional", False) } if "group-remove" in adapted: adapted["task-remove"] = { "packages": adapted.pop("group-remove")["packages"] } # Convert COPR repos to PPA if "repos" in adapted and "copr" in adapted["repos"]: if "ppa" not in adapted["repos"]: adapted["repos"]["ppa"] = [] copr_repos = adapted["repos"]["copr"] if isinstance(copr_repos, list): for copr in copr_repos: # Convert COPR format to PPA format ppa_name = f"ppa:{copr.replace('/', '/')}" adapted["repos"]["ppa"].append(ppa_name) # Remove COPR section del adapted["repos"]["copr"] # Convert RPM Fusion to Debian backports if "repos" in adapted and "nonfree" in adapted["repos"]: if adapted["repos"]["nonfree"] == "rpmfusion": adapted["repos"]["backports"] = True del adapted["repos"]["nonfree"] return adapted def _adapt_to_apt_ostree(self, config: Dict[str, Any]) -> Dict[str, Any]: """Adapt configuration for apt-ostree module""" adapted = config.copy() # Convert RPM-specific fields to Debian equivalents if "group-install" in adapted: adapted["task-install"] = { "packages": adapted.pop("group-install")["packages"], "with-optional": adapted.pop("group-install").get("with-optional", False) } if "group-remove" in adapted: adapted["task-remove"] = { "packages": adapted.pop("group-remove")["packages"] } return adapted def _adapt_to_deb_mock(self, config: Dict[str, Any]) -> Dict[str, Any]: """Adapt configuration for deb-mock module""" adapted = config.copy() # Convert RPM-specific fields to Debian equivalents if "chroot" in adapted: adapted["environment"] = adapted.pop("chroot") if "builddeps" in adapted: adapted["packages"] = adapted.pop("builddeps") if "resultdir" in adapted: adapted["artifacts"] = adapted.pop("resultdir") return adapted def execute_module(self, module_config: ModuleConfig) -> ModuleResult: """Execute a Debian module""" try: self.logger.info(f"Executing module: {module_config.type}") # Set up environment variables env = os.environ.copy() env["BLUEBUILD_MODULE_CONFIG"] = json.dumps(module_config.config) # Execute the module result = subprocess.run( ["docker", "run", "--rm", "-v", f"{module_config.working_directory}:/workspace", "-w", "/workspace", "-e", f"BLUEBUILD_MODULE_CONFIG={json.dumps(module_config.config)}", f"ghcr.io/blue-build/modules/{module_config.type}:latest"], capture_output=True, text=True, env=env, cwd=module_config.working_directory ) if result.returncode == 0: return ModuleResult( success=True, output=result.stdout, error=None, artifacts=self._extract_artifacts(result.stdout), metadata={"return_code": result.returncode} ) else: return ModuleResult( success=False, output=result.stdout, error=result.stderr, artifacts=[], metadata={"return_code": result.returncode} ) except Exception as e: self.logger.error(f"Error executing module {module_config.type}: {e}") return ModuleResult( success=False, output="", error=str(e), artifacts=[], metadata={"error_type": type(e).__name__} ) def _extract_artifacts(self, output: str) -> List[str]: """Extract artifact information from module output""" artifacts = [] # Look for artifact patterns in output lines = output.split('\n') for line in lines: if "artifact:" in line.lower() or "created:" in line.lower(): # Extract file paths from the line parts = line.split() for part in parts: if part.startswith('/') or part.endswith(('.deb', '.dsc', '.tar.gz')): artifacts.append(part) return artifacts def validate_module_config(self, module_config: ModuleConfig) -> bool: """Validate a module configuration""" try: # Check if module type is supported if module_config.type not in ["apt", "apt-ostree", "deb-mock"]: self.logger.error(f"Unsupported module type: {module_config.type}") return False # Check if required fields are present required_fields = self._get_required_fields(module_config.type) for field in required_fields: if field not in module_config.config: self.logger.error(f"Missing required field: {field}") return False return True except Exception as e: self.logger.error(f"Error validating module config: {e}") return False def _get_required_fields(self, module_type: str) -> List[str]: """Get required fields for a module type""" if module_type == "apt": return ["type"] elif module_type == "apt-ostree": return ["type"] elif module_type == "deb-mock": return ["type"] else: return [] def get_module_info(self, module_type: str) -> Dict[str, Any]: """Get information about a module type""" module_info = { "apt": { "description": "Debian package management with apt", "capabilities": ["package_installation", "repository_management", "task_management"], "supported_formats": [".deb", "package_names"], "repository_types": ["ppa", "backports", "custom"] }, "apt-ostree": { "description": "Debian package layering with ostree", "capabilities": ["package_layering", "ostree_integration", "atomic_updates"], "supported_formats": [".deb", "package_names"], "repository_types": ["ppa", "backports", "custom"] }, "deb-mock": { "description": "Debian build environment management", "capabilities": ["build_environment", "package_building", "artifact_collection"], "supported_formats": ["source_packages", ".deb"], "build_tools": ["pbuilder", "sbuild", "schroot"] } } return module_info.get(module_type, {})